290 lines
8.7 KiB
Python
290 lines
8.7 KiB
Python
"""Клиент 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),
|
|
)
|