327 lines
10 KiB
Python
327 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, Iterable, List, Mapping, Optional
|
|
|
|
from . import cryptopay, freekassa, heleket, platega, severpay, stars, wata, yookassa
|
|
from .base import (
|
|
PaymentProviderPresentation,
|
|
PaymentProviderSpec,
|
|
ProviderConfigBundle,
|
|
ProviderManifestField,
|
|
ServiceFactoryContext,
|
|
)
|
|
|
|
PAYMENT_PROVIDER_SPECS: tuple[PaymentProviderSpec, ...] = (
|
|
freekassa.SPEC,
|
|
platega.SBP_SPEC,
|
|
platega.CRYPTO_SPEC,
|
|
severpay.SPEC,
|
|
wata.SPEC,
|
|
yookassa.SPEC,
|
|
stars.SPEC,
|
|
cryptopay.SPEC,
|
|
heleket.SPEC,
|
|
)
|
|
|
|
|
|
# Provider configs (env-loaded BaseSettings models) live here as a process-wide
|
|
# singleton populated by build_provider_configs() on startup. Modules that need
|
|
# to read provider configs without changing call signatures (e.g. presentation
|
|
# resolution from arbitrary callers) can look them up via current_provider_configs().
|
|
#
|
|
# Keyed by ``service_key`` because multiple SPECs (Platega SBP / Platega Crypto)
|
|
# can share the same backing service. Per-SPEC presentation overrides live in
|
|
# ``_provider_presentations`` instead, indexed by ``spec.id``.
|
|
_provider_configs: Dict[str, ProviderConfigBundle] = {}
|
|
_provider_presentations: Dict[str, Any] = {}
|
|
|
|
|
|
def iter_provider_specs() -> Iterable[PaymentProviderSpec]:
|
|
return PAYMENT_PROVIDER_SPECS
|
|
|
|
|
|
def get_provider_spec(method: str) -> Optional[PaymentProviderSpec]:
|
|
normalized = str(method or "").strip().lower()
|
|
for spec in PAYMENT_PROVIDER_SPECS:
|
|
if normalized in spec.method_ids:
|
|
return spec
|
|
return None
|
|
|
|
|
|
def build_provider_configs() -> Dict[str, ProviderConfigBundle]:
|
|
"""Instantiate per-provider BaseSettings models declared on each SPEC.
|
|
|
|
Returns a mapping ``service_key`` → ``ProviderConfigBundle(config, presentation)``.
|
|
For SPECs that share a service (Platega SBP + Platega Crypto), only the
|
|
first one's presentation lands in the shared bundle — per-SPEC presentation
|
|
overrides live separately in ``_provider_presentations`` keyed by ``spec.id``.
|
|
"""
|
|
from .base import provider_env_file
|
|
|
|
env_file = provider_env_file()
|
|
init_kwargs = {"_env_file": env_file}
|
|
|
|
bundles: Dict[str, ProviderConfigBundle] = {}
|
|
presentations: Dict[str, Any] = {}
|
|
seen_services: set[str] = set()
|
|
for spec in PAYMENT_PROVIDER_SPECS:
|
|
if spec.presentation_class is not None:
|
|
presentations[spec.id] = spec.presentation_class(**init_kwargs)
|
|
|
|
if not spec.service_key or spec.service_key in seen_services:
|
|
continue
|
|
if spec.config_class is None and spec.presentation_class is None:
|
|
continue
|
|
seen_services.add(spec.service_key)
|
|
bundles[spec.service_key] = ProviderConfigBundle(
|
|
config=spec.config_class(**init_kwargs) if spec.config_class else None,
|
|
presentation=presentations.get(spec.id),
|
|
)
|
|
_provider_configs.clear()
|
|
_provider_configs.update(bundles)
|
|
_provider_presentations.clear()
|
|
_provider_presentations.update(presentations)
|
|
return bundles
|
|
|
|
|
|
def get_spec_presentation(spec_id: str) -> Optional[Any]:
|
|
return _provider_presentations.get(spec_id)
|
|
|
|
|
|
def current_provider_configs() -> Mapping[str, ProviderConfigBundle]:
|
|
return _provider_configs
|
|
|
|
|
|
def get_provider_bundle(service_key: Optional[str]) -> Optional[ProviderConfigBundle]:
|
|
if not service_key:
|
|
return None
|
|
return _provider_configs.get(service_key)
|
|
|
|
|
|
def _setting_value(source: Any, key: str) -> Optional[str]:
|
|
if source is None:
|
|
return None
|
|
value = getattr(source, key, None)
|
|
if value is None:
|
|
return None
|
|
value = str(value).strip()
|
|
return value or None
|
|
|
|
|
|
def _presentation_setting(spec: PaymentProviderSpec, suffix: str) -> str:
|
|
return f"PAYMENT_{spec.settings_key}_{suffix}"
|
|
|
|
|
|
def _normalize_language(language: Optional[str], settings: Any = None) -> str:
|
|
value = language or getattr(settings, "DEFAULT_LANGUAGE", None) or "ru"
|
|
normalized = str(value).strip().lower().split("-", 1)[0].split("_", 1)[0]
|
|
return normalized or "ru"
|
|
|
|
|
|
def _presentation_attr(suffix: str, language: Optional[str] = None) -> str:
|
|
"""Attribute name on a provider's presentation BaseSettings model.
|
|
|
|
Mirrors the legacy ``PAYMENT_<ID>_<suffix>`` env name, but without the
|
|
``PAYMENT_<ID>_`` prefix (which is supplied by env_prefix on the model).
|
|
"""
|
|
if language:
|
|
return f"{suffix}_{language.upper()}"
|
|
return suffix
|
|
|
|
|
|
def _provider_presentation_value(
|
|
spec: PaymentProviderSpec,
|
|
suffix: str,
|
|
*,
|
|
language: Optional[str] = None,
|
|
) -> Optional[str]:
|
|
presentation = _provider_presentations.get(spec.id)
|
|
if presentation is None:
|
|
bundle = _provider_configs.get(spec.service_key) if spec.service_key else None
|
|
presentation = bundle.presentation if bundle else None
|
|
if presentation is None:
|
|
return None
|
|
attr = _presentation_attr(suffix, language=language)
|
|
return _setting_value(presentation, attr)
|
|
|
|
|
|
def _localized_setting_value(
|
|
settings: Any,
|
|
spec: PaymentProviderSpec,
|
|
suffix: str,
|
|
language: str,
|
|
) -> Optional[str]:
|
|
# 1) Per-provider presentation model (new pattern) takes priority.
|
|
provider_value = _provider_presentation_value(spec, suffix, language=language)
|
|
if provider_value is not None:
|
|
return provider_value
|
|
# 2) Legacy: PAYMENT_<ID>_<suffix>_<LANG> on the global Settings.
|
|
return _setting_value(
|
|
settings,
|
|
_presentation_setting(spec, f"{suffix}_{language.upper()}"),
|
|
)
|
|
|
|
|
|
def _bare_setting_value(
|
|
settings: Any,
|
|
spec: PaymentProviderSpec,
|
|
suffix: str,
|
|
) -> Optional[str]:
|
|
provider_value = _provider_presentation_value(spec, suffix)
|
|
if provider_value is not None:
|
|
return provider_value
|
|
return _setting_value(settings, _presentation_setting(spec, suffix))
|
|
|
|
|
|
def _localized_default(
|
|
values: Optional[Mapping[str, str]],
|
|
language: str,
|
|
fallback: Optional[str],
|
|
) -> Optional[str]:
|
|
if not values:
|
|
return fallback
|
|
return (
|
|
values.get(language)
|
|
or values.get("en")
|
|
or values.get("ru")
|
|
or next(iter(values.values()), None)
|
|
or fallback
|
|
)
|
|
|
|
|
|
def resolve_provider_presentation(
|
|
spec: PaymentProviderSpec,
|
|
settings: Any = None,
|
|
*,
|
|
language: Optional[str] = None,
|
|
) -> PaymentProviderPresentation:
|
|
lang = _normalize_language(language, settings)
|
|
webapp_label = (
|
|
_localized_setting_value(settings, spec, "WEBAPP_LABEL", lang)
|
|
or _localized_default(spec.webapp_labels, lang, spec.webapp_label)
|
|
or spec.label
|
|
)
|
|
webapp_icon = (
|
|
_bare_setting_value(settings, spec, "WEBAPP_ICON")
|
|
or spec.webapp_icon
|
|
)
|
|
telegram_label_override = _localized_setting_value(
|
|
settings,
|
|
spec,
|
|
"TELEGRAM_LABEL",
|
|
lang,
|
|
)
|
|
telegram_emoji_override = _bare_setting_value(settings, spec, "TELEGRAM_EMOJI")
|
|
telegram_label = (
|
|
telegram_label_override
|
|
or _localized_default(spec.telegram_labels, lang, None)
|
|
or spec.label
|
|
)
|
|
telegram_emoji = telegram_emoji_override or spec.default_telegram_emoji
|
|
|
|
return PaymentProviderPresentation(
|
|
webapp_label=webapp_label,
|
|
webapp_icon=webapp_icon,
|
|
telegram_label=telegram_label,
|
|
telegram_emoji=telegram_emoji,
|
|
telegram_customized=bool(telegram_label_override or telegram_emoji_override),
|
|
)
|
|
|
|
|
|
def provider_telegram_button_text(
|
|
spec: PaymentProviderSpec,
|
|
settings: Any,
|
|
*,
|
|
language: Optional[str] = None,
|
|
) -> str:
|
|
presentation = resolve_provider_presentation(spec, settings, language=language)
|
|
if presentation.telegram_emoji:
|
|
return f"{presentation.telegram_emoji} {presentation.telegram_label}".strip()
|
|
return presentation.telegram_label
|
|
|
|
|
|
def iter_unique_provider_routers():
|
|
seen: set[int] = set()
|
|
for spec in PAYMENT_PROVIDER_SPECS:
|
|
router = spec.load_router()
|
|
if not router:
|
|
continue
|
|
marker = id(router)
|
|
if marker in seen:
|
|
continue
|
|
seen.add(marker)
|
|
yield router
|
|
|
|
|
|
def iter_service_keys() -> Iterable[str]:
|
|
seen: set[str] = set()
|
|
for spec in PAYMENT_PROVIDER_SPECS:
|
|
if not spec.service_key or spec.service_key in seen:
|
|
continue
|
|
seen.add(spec.service_key)
|
|
yield spec.service_key
|
|
|
|
|
|
def iter_service_specs() -> Iterable[PaymentProviderSpec]:
|
|
seen: set[str] = set()
|
|
for spec in PAYMENT_PROVIDER_SPECS:
|
|
if not spec.service_key or not spec.create_service or spec.service_key in seen:
|
|
continue
|
|
seen.add(spec.service_key)
|
|
yield spec
|
|
|
|
|
|
def build_provider_services(ctx: ServiceFactoryContext) -> Dict[str, Any]:
|
|
services: Dict[str, Any] = {}
|
|
for spec in iter_service_specs():
|
|
services[spec.service_key] = spec.create_service(ctx)
|
|
return services
|
|
|
|
|
|
def provider_label_map(settings: Any = None, language: Optional[str] = None) -> Dict[str, str]:
|
|
labels: Dict[str, str] = {}
|
|
for spec in PAYMENT_PROVIDER_SPECS:
|
|
presentation = resolve_provider_presentation(
|
|
spec,
|
|
settings,
|
|
language=language,
|
|
)
|
|
label = presentation.telegram_label if presentation.telegram_customized else spec.label
|
|
labels.setdefault(spec.provider_key, label)
|
|
for method in spec.method_ids:
|
|
labels.setdefault(method, label)
|
|
return labels
|
|
|
|
|
|
def provider_emoji_map(settings: Any = None) -> Dict[str, str]:
|
|
emojis: Dict[str, str] = {}
|
|
for spec in PAYMENT_PROVIDER_SPECS:
|
|
emoji = resolve_provider_presentation(spec, settings).telegram_emoji
|
|
emojis.setdefault(spec.provider_key, emoji)
|
|
for method in spec.method_ids:
|
|
emojis.setdefault(method, emoji)
|
|
return emojis
|
|
|
|
|
|
def pending_statuses() -> List[str]:
|
|
statuses = ["pending"]
|
|
for spec in PAYMENT_PROVIDER_SPECS:
|
|
if spec.pending_status not in statuses:
|
|
statuses.append(spec.pending_status)
|
|
return statuses
|
|
|
|
|
|
def iter_provider_manifest_fields() -> Iterable[tuple[PaymentProviderSpec, ProviderManifestField]]:
|
|
"""Yield (spec, manifest_field) for every fragment declared on a provider SPEC."""
|
|
for spec in PAYMENT_PROVIDER_SPECS:
|
|
for field in spec.manifest_fields:
|
|
yield spec, field
|
|
|
|
|
|
def find_manifest_owner(key: str) -> Optional[tuple[PaymentProviderSpec, ProviderManifestField]]:
|
|
"""Find which provider owns a manifest key (if any)."""
|
|
for spec, field in iter_provider_manifest_fields():
|
|
if field.key == key:
|
|
return spec, field
|
|
return None
|