274 lines
11 KiB
Python
274 lines
11 KiB
Python
"""Клиент Digiseller: проверка оплаты по уникальному коду."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import hashlib
|
||
import logging
|
||
import time
|
||
from dataclasses import dataclass
|
||
from typing import Any
|
||
from urllib.parse import parse_qs, urlencode, urlparse, urlunparse
|
||
|
||
import httpx
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
API_BASE = "https://api.digiseller.com"
|
||
API_BASE_FALLBACK = "https://oplata.info"
|
||
|
||
|
||
class DigisellerError(RuntimeError):
|
||
pass
|
||
|
||
|
||
@dataclass(frozen=True, slots=True)
|
||
class DigisellerPurchase:
|
||
unique_code: str
|
||
inv: int
|
||
product_id: int
|
||
amount: float
|
||
currency: str
|
||
cnt_goods: float
|
||
email: str | None
|
||
state: int
|
||
raw: dict[str, Any]
|
||
|
||
@property
|
||
def delivered(self) -> bool:
|
||
# 2/3/4 — уже отмечено как выдано
|
||
return self.state in {2, 3, 4}
|
||
|
||
|
||
class Digiseller:
|
||
def __init__(
|
||
self,
|
||
seller_id: str | int,
|
||
api_key: str,
|
||
*,
|
||
product_ids: tuple[int, ...] = (),
|
||
pay_url: str = "",
|
||
credit_from: str = "cnt_goods",
|
||
) -> None:
|
||
self.seller_id = str(seller_id or "").strip()
|
||
self.api_key = (api_key or "").strip()
|
||
self.product_ids = product_ids
|
||
self.pay_url = (pay_url or "").strip()
|
||
self.credit_from = (credit_from or "cnt_goods").strip().lower()
|
||
self._token: str | None = None
|
||
self._token_expires: float = 0.0
|
||
|
||
@property
|
||
def enabled(self) -> bool:
|
||
return bool(self.seller_id and self.api_key)
|
||
|
||
def _sign(self, timestamp: int) -> str:
|
||
raw = f"{self.api_key}{timestamp}".encode("utf-8")
|
||
return hashlib.sha256(raw).hexdigest()
|
||
|
||
async def _login(self) -> str:
|
||
now = time.time()
|
||
if self._token and now < self._token_expires - 60:
|
||
return self._token
|
||
|
||
timestamp = int(now)
|
||
payload = {
|
||
"seller_id": int(self.seller_id),
|
||
"timestamp": timestamp,
|
||
"sign": self._sign(timestamp),
|
||
}
|
||
data = await self._request("POST", "/api/apilogin", json_body=payload, auth=False)
|
||
if int(data.get("retval") or -1) != 0 or not data.get("token"):
|
||
raise DigisellerError(data.get("desc") or data.get("retdesc") or "login failed")
|
||
self._token = str(data["token"])
|
||
# токен живёт ~2 часа
|
||
self._token_expires = now + 110 * 60
|
||
return self._token
|
||
|
||
async def _request(
|
||
self,
|
||
method: str,
|
||
path: str,
|
||
*,
|
||
json_body: dict[str, Any] | None = None,
|
||
params: dict[str, Any] | None = None,
|
||
auth: bool = True,
|
||
) -> dict[str, Any]:
|
||
if not self.enabled:
|
||
raise DigisellerError("Digiseller не настроен")
|
||
|
||
headers = {"Accept": "application/json", "Content-Type": "application/json"}
|
||
query = dict(params or {})
|
||
if auth:
|
||
query["token"] = await self._login()
|
||
|
||
last_exc: Exception | None = None
|
||
for base in (API_BASE, API_BASE_FALLBACK):
|
||
try:
|
||
async with httpx.AsyncClient(timeout=30.0) as client:
|
||
resp = await client.request(
|
||
method,
|
||
f"{base}{path}",
|
||
headers=headers,
|
||
params=query or None,
|
||
json=json_body,
|
||
)
|
||
if resp.status_code == 401 and auth:
|
||
self._token = None
|
||
query["token"] = await self._login()
|
||
resp = await client.request(
|
||
method,
|
||
f"{base}{path}",
|
||
headers=headers,
|
||
params=query,
|
||
json=json_body,
|
||
)
|
||
data = resp.json()
|
||
if not isinstance(data, dict):
|
||
raise DigisellerError(f"bad response: {resp.text[:200]}")
|
||
return data
|
||
except Exception as exc: # noqa: BLE001
|
||
last_exc = exc
|
||
logger.warning("Digiseller %s %s via %s failed: %s", method, path, base, exc)
|
||
raise DigisellerError(str(last_exc or "request failed"))
|
||
|
||
async def get_by_unique_code(self, unique_code: str) -> DigisellerPurchase:
|
||
code = "".join(ch for ch in unique_code.strip() if ch.isalnum())
|
||
if len(code) < 10:
|
||
raise DigisellerError("Некорректный уникальный код")
|
||
|
||
data = await self._request(
|
||
"GET",
|
||
f"/api/purchases/unique-code/{code}",
|
||
)
|
||
retval = int(data.get("retval") if data.get("retval") is not None else -1)
|
||
if retval != 0:
|
||
raise DigisellerError(
|
||
str(data.get("retdesc") or data.get("desc") or f"код ошибки {retval}")
|
||
)
|
||
|
||
state_obj = data.get("unique_code_state") or {}
|
||
state = int(state_obj.get("state") or 0)
|
||
inv = int(data.get("inv") or 0)
|
||
product_id = int(data.get("id_goods") or 0)
|
||
if not inv:
|
||
raise DigisellerError("В ответе Digiseller нет номера счёта (inv)")
|
||
|
||
return DigisellerPurchase(
|
||
unique_code=code,
|
||
inv=inv,
|
||
product_id=product_id,
|
||
amount=float(data.get("amount") or 0),
|
||
currency=str(data.get("type_curr") or ""),
|
||
cnt_goods=float(data.get("cnt_goods") or 0),
|
||
email=(str(data.get("email")).strip() if data.get("email") else None),
|
||
state=state,
|
||
raw=data,
|
||
)
|
||
|
||
def assert_payable(self, purchase: DigisellerPurchase) -> None:
|
||
"""Проверка, что код можно принять к зачислению."""
|
||
# 1 — ещё не верифицирован (первый запрос), 5 — верифицирован, товар не выдан
|
||
# 2/3 — уже отмечен выданным (допускаем для дозачисления после сбоя)
|
||
# 4 — оспорен
|
||
if purchase.state == 4:
|
||
raise DigisellerError("Оплата по этому коду оспорена")
|
||
if purchase.state not in {1, 2, 3, 5}:
|
||
raise DigisellerError(
|
||
f"Код ещё не готов к выдаче (state={purchase.state}). "
|
||
"Оплати товар и попробуй снова."
|
||
)
|
||
date_pay = str((purchase.raw or {}).get("date_pay") or "").strip()
|
||
if not date_pay and purchase.state == 1:
|
||
raise DigisellerError("Код ещё не оплачен")
|
||
|
||
async def mark_delivered(self, unique_code: str) -> None:
|
||
code = "".join(ch for ch in unique_code.strip() if ch.isalnum())
|
||
data = await self._request(
|
||
"PUT",
|
||
f"/api/purchases/unique-code/{code}/deliver",
|
||
)
|
||
retval = int(data.get("retval") if data.get("retval") is not None else -1)
|
||
# 4 — уже выдано ранее (идемпотентно ок)
|
||
if retval not in (0, 4):
|
||
raise DigisellerError(
|
||
str(data.get("retdesc") or data.get("desc") or f"deliver error {retval}")
|
||
)
|
||
|
||
def credit_amount(self, purchase: DigisellerPurchase) -> float:
|
||
"""Сколько рублей зачислить на баланс бота."""
|
||
if self.product_ids and purchase.product_id not in self.product_ids:
|
||
raise DigisellerError(
|
||
f"Товар id={purchase.product_id} не разрешён для пополнения"
|
||
)
|
||
|
||
if self.credit_from == "amount":
|
||
value = purchase.amount
|
||
else:
|
||
# Типичная схема: товар без фикс. цены, cnt_goods = сумма в ₽
|
||
value = purchase.cnt_goods if purchase.cnt_goods > 0 else purchase.amount
|
||
|
||
if value <= 0:
|
||
raise DigisellerError("Сумма пополнения равна 0")
|
||
return float(value)
|
||
|
||
def default_pay_url(self) -> str:
|
||
if self.pay_url:
|
||
return self.pay_url
|
||
if self.product_ids:
|
||
return f"https://oplata.info/asp2/pay.asp?id_d={self.product_ids[0]}"
|
||
return "https://my.digiseller.com/"
|
||
|
||
def pay_url_for(self, telegram_id: int) -> str:
|
||
"""Ссылка на оплату с привязкой к Telegram ID (уходит в query_string)."""
|
||
base = self.default_pay_url()
|
||
parsed = urlparse(base)
|
||
q = parse_qs(parsed.query, keep_blank_values=True)
|
||
q["telegram_id"] = [str(int(telegram_id))]
|
||
# совместимость с разными интеграциями Digiseller
|
||
q["tg_id"] = [str(int(telegram_id))]
|
||
new_query = urlencode({k: v[-1] for k, v in q.items()})
|
||
return urlunparse(parsed._replace(query=new_query))
|
||
|
||
@staticmethod
|
||
def extract_telegram_id(purchase: DigisellerPurchase) -> int | None:
|
||
"""Достать telegram_id из query_string / options платежа."""
|
||
raw = purchase.raw or {}
|
||
|
||
qs_b64 = str(raw.get("query_string") or "").strip()
|
||
if qs_b64:
|
||
try:
|
||
pad = "=" * (-len(qs_b64) % 4)
|
||
decoded = base64.b64decode(qs_b64 + pad).decode("utf-8", "ignore")
|
||
params = parse_qs(decoded, keep_blank_values=True)
|
||
for key in ("telegram_id", "tg_id", "tg", "user_id", "uid"):
|
||
vals = params.get(key) or []
|
||
if vals:
|
||
tid = int(str(vals[0]).strip())
|
||
if tid > 0:
|
||
return tid
|
||
except Exception: # noqa: BLE001
|
||
logger.debug("Digiseller query_string parse failed", exc_info=True)
|
||
|
||
for opt in raw.get("options") or []:
|
||
if not isinstance(opt, dict):
|
||
continue
|
||
name = str(opt.get("name") or "").strip().lower()
|
||
value = str(opt.get("value") or "").strip()
|
||
if not value:
|
||
continue
|
||
if any(
|
||
x in name
|
||
for x in ("telegram", "телеграм", "tg id", "tg_id", "userid", "user id")
|
||
):
|
||
digits = "".join(ch for ch in value if ch.isdigit())
|
||
if digits:
|
||
tid = int(digits)
|
||
if tid > 0:
|
||
return tid
|
||
# иногда покупатель просто вписывает числовой ID
|
||
if value.isdigit() and len(value) >= 5:
|
||
return int(value)
|
||
|
||
return None
|