payments: Platega + YooKassa integration, opt-in via PAYMENTS_ENABLED
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
parent
c9955b2774
commit
5b30d1f148
6 changed files with 326 additions and 8 deletions
96
payments.py
Normal file
96
payments.py
Normal file
|
|
@ -0,0 +1,96 @@
|
|||
import base64
|
||||
import hashlib
|
||||
import hmac
|
||||
import json
|
||||
import secrets
|
||||
import urllib.request
|
||||
|
||||
from config import (
|
||||
PANEL_DOMAIN,
|
||||
YOOKASSA_ENABLED, YOOKASSA_SHOP_ID, YOOKASSA_SECRET_KEY,
|
||||
PLATEGA_ENABLED, PLATEGA_MERCHANT_ID, PLATEGA_SECRET,
|
||||
)
|
||||
|
||||
PROVIDER_NAMES = {"yookassa": "ЮKassa", "platega": "Platega"}
|
||||
|
||||
|
||||
def available_providers() -> list[str]:
|
||||
providers = []
|
||||
if YOOKASSA_ENABLED:
|
||||
providers.append("yookassa")
|
||||
if PLATEGA_ENABLED:
|
||||
providers.append("platega")
|
||||
return providers
|
||||
|
||||
|
||||
def new_payment_id() -> str:
|
||||
return secrets.token_hex(16)
|
||||
|
||||
|
||||
def _post_json(url: str, body: dict, headers: dict, timeout: int = 15) -> dict:
|
||||
data = json.dumps(body).encode()
|
||||
req = urllib.request.Request(url, data=data, method="POST", headers=headers)
|
||||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||||
return json.loads(resp.read().decode())
|
||||
|
||||
|
||||
def create_yookassa_payment(payment_id: str, amount_rub: int, description: str) -> str:
|
||||
auth = base64.b64encode(f"{YOOKASSA_SHOP_ID}:{YOOKASSA_SECRET_KEY}".encode()).decode()
|
||||
data = _post_json(
|
||||
"https://api.yookassa.ru/v3/payments",
|
||||
{
|
||||
"amount": {"value": f"{amount_rub}.00", "currency": "RUB"},
|
||||
"confirmation": {"type": "redirect", "return_url": f"https://{PANEL_DOMAIN}/pay/done"},
|
||||
"capture": True,
|
||||
"description": description,
|
||||
"metadata": {"payment_id": payment_id},
|
||||
},
|
||||
{
|
||||
"Content-Type": "application/json",
|
||||
"Authorization": f"Basic {auth}",
|
||||
"Idempotence-Key": payment_id,
|
||||
},
|
||||
)
|
||||
external_id = data["id"]
|
||||
pay_url = data["confirmation"]["confirmation_url"]
|
||||
return external_id, pay_url
|
||||
|
||||
|
||||
def verify_yookassa_notification(body: dict) -> bool:
|
||||
return body.get("event") == "payment.succeeded" and "object" in body
|
||||
|
||||
|
||||
def create_platega_payment(payment_id: str, amount_rub: int, description: str) -> str:
|
||||
data = _post_json(
|
||||
"https://app.platega.io/transaction/process",
|
||||
{
|
||||
"paymentMethod": 2,
|
||||
"id": payment_id,
|
||||
"paymentDetails": {"amount": amount_rub, "currency": "RUB"},
|
||||
"description": description,
|
||||
"return": f"https://{PANEL_DOMAIN}/pay/done",
|
||||
},
|
||||
{
|
||||
"Content-Type": "application/json",
|
||||
"X-MerchantId": PLATEGA_MERCHANT_ID,
|
||||
"X-Secret": PLATEGA_SECRET,
|
||||
},
|
||||
)
|
||||
external_id = data.get("id") or data.get("transactionId")
|
||||
pay_url = data.get("redirectUrl") or data.get("url")
|
||||
return external_id, pay_url
|
||||
|
||||
|
||||
def verify_platega_signature(raw_body: bytes, signature: str) -> bool:
|
||||
if not signature:
|
||||
return False
|
||||
expected = hmac.new(PLATEGA_SECRET.encode(), raw_body, hashlib.sha256).hexdigest()
|
||||
return hmac.compare_digest(expected, signature)
|
||||
|
||||
|
||||
def create_payment_link(provider: str, payment_id: str, amount_rub: int, description: str):
|
||||
if provider == "yookassa":
|
||||
return create_yookassa_payment(payment_id, amount_rub, description)
|
||||
if provider == "platega":
|
||||
return create_platega_payment(payment_id, amount_rub, description)
|
||||
raise ValueError(f"unknown provider: {provider}")
|
||||
Loading…
Add table
Add a link
Reference in a new issue