201 lines
6.8 KiB
Python
201 lines
6.8 KiB
Python
"""Клиент Lava.ru Business API (карта / СБП)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import logging
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
API_BASE = "https://api.lava.ru"
|
|
PAID_STATUSES = frozenset({"success", "paid"})
|
|
|
|
|
|
class LavaError(RuntimeError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class LavaInvoice:
|
|
invoice_id: str
|
|
order_id: str
|
|
amount: float
|
|
status: str
|
|
pay_url: str
|
|
expired: str | None = None
|
|
|
|
|
|
def _json_body(payload: dict[str, Any]) -> bytes:
|
|
# Порядок ключей и separators критичны для Signature
|
|
return json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
|
|
|
|
|
|
def sign_request(payload: dict[str, Any], secret_key: str) -> str:
|
|
return hmac.new(
|
|
secret_key.encode("utf-8"),
|
|
_json_body(payload),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
|
|
|
|
def verify_webhook(
|
|
*,
|
|
data: dict[str, Any],
|
|
raw_body: bytes,
|
|
authorization: str | None,
|
|
additional_key: str,
|
|
) -> bool:
|
|
"""Проверка подписи webhook (два формата Lava)."""
|
|
key = (additional_key or "").strip()
|
|
if not key:
|
|
# без доп. ключа принимаем только если нет подписи вообще (нежелательно)
|
|
return not data.get("sign") and not (authorization or "").strip()
|
|
|
|
# Новый формат SDK: HMAC SHA256(raw body) в Authorization
|
|
auth = (authorization or "").strip()
|
|
if auth:
|
|
expected = hmac.new(key.encode("utf-8"), raw_body, hashlib.sha256).hexdigest()
|
|
if hmac.compare_digest(auth.lower(), expected.lower()):
|
|
return True
|
|
|
|
# Классический формат: md5(invoice_id:amount:pay_time:secret_key_2)
|
|
received = str(data.get("sign") or "").strip()
|
|
if received:
|
|
invoice_id = str(data.get("invoice_id") or "")
|
|
amount = data.get("amount")
|
|
pay_time = data.get("pay_time")
|
|
# amount иногда float — приводим как в доке (как пришло / без лишних нулей)
|
|
amount_s = str(amount)
|
|
raw = f"{invoice_id}:{amount_s}:{pay_time}:{key}"
|
|
expected_md5 = hashlib.md5(raw.encode("utf-8")).hexdigest()
|
|
if hmac.compare_digest(received.lower(), expected_md5.lower()):
|
|
return True
|
|
|
|
return False
|
|
|
|
|
|
class Lava:
|
|
def __init__(
|
|
self,
|
|
shop_id: str,
|
|
secret_key: str,
|
|
*,
|
|
additional_key: str = "",
|
|
hook_url: str = "",
|
|
success_url: str = "",
|
|
fail_url: str = "",
|
|
expire_minutes: int = 60,
|
|
) -> None:
|
|
self.shop_id = (shop_id or "").strip()
|
|
self.secret_key = (secret_key or "").strip()
|
|
self.additional_key = (additional_key or "").strip()
|
|
self.hook_url = (hook_url or "").strip().rstrip("/")
|
|
self.success_url = (success_url or "").strip()
|
|
self.fail_url = (fail_url or "").strip()
|
|
self.expire_minutes = max(5, min(int(expire_minutes or 60), 5 * 24 * 60))
|
|
|
|
@property
|
|
def enabled(self) -> bool:
|
|
return bool(self.shop_id and self.secret_key)
|
|
|
|
async def _post(self, path: str, payload: dict[str, Any]) -> dict[str, Any]:
|
|
if not self.enabled:
|
|
raise LavaError("Lava не настроен (LAVA_SHOP_ID / LAVA_SECRET_KEY)")
|
|
body = _json_body(payload)
|
|
headers = {
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
"Signature": sign_request(payload, self.secret_key),
|
|
}
|
|
async with httpx.AsyncClient(timeout=30.0) as client:
|
|
resp = await client.post(
|
|
f"{API_BASE}{path}",
|
|
headers=headers,
|
|
content=body,
|
|
)
|
|
try:
|
|
data = resp.json()
|
|
except Exception as exc: # noqa: BLE001
|
|
raise LavaError(f"HTTP {resp.status_code}: {resp.text[:300]}") from exc
|
|
|
|
if not isinstance(data, dict):
|
|
raise LavaError(f"Неожиданный ответ: {data}")
|
|
|
|
if resp.status_code >= 400 or not data.get("status_check", True):
|
|
err = data.get("error") or data.get("message") or data
|
|
raise LavaError(str(err))
|
|
|
|
result = data.get("data")
|
|
if not isinstance(result, dict):
|
|
raise LavaError(f"Нет data в ответе: {data}")
|
|
return result
|
|
|
|
async def create_invoice(
|
|
self,
|
|
*,
|
|
amount_rub: float,
|
|
order_id: str,
|
|
comment: str = "",
|
|
custom_fields: str = "",
|
|
) -> LavaInvoice:
|
|
# Порядок ключей фиксируем — от него зависит Signature
|
|
payload: dict[str, Any] = {
|
|
"shopId": self.shop_id,
|
|
"sum": float(f"{amount_rub:.2f}"),
|
|
"orderId": order_id,
|
|
}
|
|
if comment:
|
|
payload["comment"] = comment[:255]
|
|
if custom_fields:
|
|
payload["customFields"] = custom_fields[:500]
|
|
if self.fail_url:
|
|
payload["failUrl"] = self.fail_url
|
|
if self.success_url:
|
|
payload["successUrl"] = self.success_url
|
|
if self.hook_url:
|
|
payload["hookUrl"] = self.hook_url
|
|
payload["expire"] = self.expire_minutes
|
|
|
|
result = await self._post("/business/invoice/create", payload)
|
|
return self._parse(result, order_id_fallback=order_id)
|
|
|
|
async def get_invoice(
|
|
self,
|
|
*,
|
|
order_id: str | None = None,
|
|
invoice_id: str | None = None,
|
|
) -> LavaInvoice | None:
|
|
payload: dict[str, Any] = {"shopId": self.shop_id}
|
|
if order_id:
|
|
payload["orderId"] = order_id
|
|
if invoice_id:
|
|
payload["invoiceId"] = invoice_id
|
|
if "orderId" not in payload and "invoiceId" not in payload:
|
|
raise LavaError("Нужен order_id или invoice_id")
|
|
try:
|
|
result = await self._post("/business/invoice/status", payload)
|
|
except LavaError:
|
|
return None
|
|
return self._parse(result, order_id_fallback=order_id or "")
|
|
|
|
@staticmethod
|
|
def is_paid(status: str) -> bool:
|
|
return str(status or "").lower() in PAID_STATUSES
|
|
|
|
@staticmethod
|
|
def _parse(raw: dict[str, Any], *, order_id_fallback: str = "") -> LavaInvoice:
|
|
return LavaInvoice(
|
|
invoice_id=str(raw.get("id") or raw.get("invoice_id") or ""),
|
|
order_id=str(raw.get("orderId") or raw.get("order_id") or order_id_fallback),
|
|
amount=float(raw.get("amount") or raw.get("sum") or 0),
|
|
status=str(raw.get("status") or ""),
|
|
pay_url=str(raw.get("url") or raw.get("payUrl") or ""),
|
|
expired=str(raw.get("expired") or raw.get("expire") or "") or None,
|
|
)
|