"""Клиент AnyPay (https://anypay.io) — SCI + API create-payment.""" from __future__ import annotations import hashlib import hmac import logging from dataclasses import dataclass from typing import Any from urllib.parse import urlencode import httpx logger = logging.getLogger(__name__) API_BASE = "https://anypay.io/api" MERCHANT_URL = "https://anypay.io/merchant" PAID_STATUSES = frozenset({"paid", "success"}) ANYPAY_IPS = frozenset({"185.162.128.38", "185.162.128.39", "185.162.128.88"}) class AnypayError(RuntimeError): pass @dataclass(frozen=True, slots=True) class AnypayInvoice: pay_id: str pay_url: str transaction_id: str | None = None status: str = "waiting" amount: float = 0.0 def _fmt_amount(amount: float) -> str: return f"{float(amount):.2f}" def sign_create_payment( *, api_id: str, project_id: str, pay_id: str, amount: str, currency: str, desc: str, method: str, api_key: str, ) -> str: raw = ( f"create-payment{api_id}{project_id}{pay_id}{amount}" f"{currency}{desc}{method}{api_key}" ) return hashlib.sha256(raw.encode("utf-8")).hexdigest() def sign_sci( *, project_id: str, pay_id: str, amount: str, currency: str, desc: str, success_url: str, fail_url: str, api_key: str, ) -> str: raw = ":".join( [ str(project_id), str(pay_id), amount, currency, desc, success_url or "", fail_url or "", api_key, ] ) return hashlib.sha256(raw.encode("utf-8")).hexdigest() def verify_notification_sign( data: dict[str, Any], *, project_id: str, api_key: str, ) -> bool: """Проверка подписи IPN (SHA256 или MD5).""" received = str(data.get("sign") or "").strip() if not received or not api_key: return False currency = str(data.get("currency") or "") amount = str(data.get("amount") or "") pay_id = str(data.get("pay_id") or "") merchant_id = str(data.get("merchant_id") or project_id) status = str(data.get("status") or "paid") # SHA256: currency:amount:pay_id:merchant_id:status:secret sha_raw = ":".join([currency, amount, pay_id, str(merchant_id), status, api_key]) sha_expected = hashlib.sha256(sha_raw.encode("utf-8")).hexdigest() if hmac.compare_digest(received.lower(), sha_expected.lower()): return True if status.lower() != "paid": sha_paid = ":".join( [currency, amount, pay_id, str(merchant_id), "paid", api_key] ) sha_paid_expected = hashlib.sha256(sha_paid.encode("utf-8")).hexdigest() if hmac.compare_digest(received.lower(), sha_paid_expected.lower()): return True # MD5: merchant_id:amount:pay_id:secret md5_raw = f"{merchant_id}:{amount}:{pay_id}:{api_key}" md5_expected = hashlib.md5(md5_raw.encode("utf-8")).hexdigest() return hmac.compare_digest(received.lower(), md5_expected.lower()) def client_ip_allowed(remote: str | None, *, forwarded: str | None = None) -> bool: candidates: list[str] = [] if remote: candidates.append(remote.strip()) if forwarded: first = forwarded.split(",")[0].strip() if first: candidates.append(first) return any(ip in ANYPAY_IPS for ip in candidates) class Anypay: def __init__( self, api_id: str, api_key: str, project_id: str, *, currency: str = "RUB", method: str = "", success_url: str = "", fail_url: str = "", default_email: str = "pay@anypay.local", check_ip: bool = False, ) -> None: self.api_id = str(api_id or "").strip() self.api_key = str(api_key or "").strip() self.project_id = str(project_id or "").strip() self.currency = (currency or "RUB").strip().upper() or "RUB" self.method = (method or "").strip().lower() self.success_url = (success_url or "").strip() self.fail_url = (fail_url or "").strip() self.default_email = (default_email or "pay@anypay.local").strip() self.check_ip = bool(check_ip) @property def enabled(self) -> bool: return bool(self.api_id and self.api_key and self.project_id) def build_sci_pay_url( self, *, pay_id: str, amount_rub: float, desc: str, email: str = "", ) -> str: if not self.enabled: raise AnypayError("AnyPay не настроен") amount = _fmt_amount(amount_rub) currency = self.currency description = (desc or "Topup")[:150] sign = sign_sci( project_id=self.project_id, pay_id=str(pay_id), amount=amount, currency=currency, desc=description, success_url=self.success_url, fail_url=self.fail_url, api_key=self.api_key, ) params: dict[str, str] = { "merchant_id": self.project_id, "pay_id": str(pay_id), "amount": amount, "currency": currency, "desc": description, "email": (email or self.default_email)[:120], "sign": sign, "lang": "ru", } if self.success_url: params["success_url"] = self.success_url if self.fail_url: params["fail_url"] = self.fail_url if self.method: params["method"] = self.method return f"{MERCHANT_URL}?{urlencode(params)}" async def create_invoice( self, *, pay_id: str, amount_rub: float, desc: str, email: str = "", ) -> AnypayInvoice: """Создать счёт: API (если задан method) или SCI-ссылка.""" if not self.enabled: raise AnypayError("AnyPay не настроен (ANYPAY_API_ID / KEY / PROJECT_ID)") pay_id_s = "".join(ch for ch in str(pay_id) if ch.isdigit()) if not pay_id_s or len(pay_id_s) > 15: raise AnypayError("pay_id: только цифры, до 15 символов") amount = _fmt_amount(amount_rub) description = (desc or "Topup")[:150] mail = (email or self.default_email)[:120] # Без method — SCI (покупатель выбирает способ на странице AnyPay) if not self.method: url = self.build_sci_pay_url( pay_id=pay_id_s, amount_rub=amount_rub, desc=description, email=mail, ) return AnypayInvoice( pay_id=pay_id_s, pay_url=url, amount=float(amount), status="waiting", ) sign = sign_create_payment( api_id=self.api_id, project_id=self.project_id, pay_id=pay_id_s, amount=amount, currency=self.currency, desc=description, method=self.method, api_key=self.api_key, ) params: dict[str, Any] = { "project_id": self.project_id, "pay_id": pay_id_s, "amount": amount, "currency": self.currency, "desc": description, "email": mail, "method": self.method, "sign": sign, "lang": "ru", } if self.success_url: params["success_url"] = self.success_url if self.fail_url: params["fail_url"] = self.fail_url url = f"{API_BASE}/create-payment/{self.api_id}" async with httpx.AsyncClient(timeout=30.0) as client: resp = await client.get( url, params=params, headers={"Accept": "application/json"} ) try: data = resp.json() except Exception as exc: # noqa: BLE001 raise AnypayError(f"AnyPay bad JSON: {resp.text[:200]}") from exc if not isinstance(data, dict): raise AnypayError(f"AnyPay unexpected response: {data!r}") if data.get("error"): raise AnypayError(str(data.get("error") or data)) result = data.get("result") if not isinstance(result, dict): raise AnypayError(f"AnyPay: нет result ({data})") pay_url = str(result.get("payment_url") or "").strip() if not pay_url: raise AnypayError("AnyPay: пустой payment_url") return AnypayInvoice( pay_id=str(result.get("pay_id") or pay_id_s), pay_url=pay_url, transaction_id=str(result.get("transaction_id") or "") or None, status=str(result.get("status") or "waiting"), amount=float(amount), )