131 lines
4.3 KiB
Python
131 lines
4.3 KiB
Python
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 _get_json(url: str, headers: dict, timeout: int = 15) -> dict:
|
|
req = urllib.request.Request(url, method="GET", 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 check_yookassa_payment(external_id: str) -> str:
|
|
auth = base64.b64encode(f"{YOOKASSA_SHOP_ID}:{YOOKASSA_SECRET_KEY}".encode()).decode()
|
|
data = _get_json(
|
|
f"https://api.yookassa.ru/v3/payments/{external_id}",
|
|
{"Authorization": f"Basic {auth}"},
|
|
)
|
|
return data.get("status", "")
|
|
|
|
|
|
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 check_platega_payment(external_id: str) -> str:
|
|
data = _get_json(
|
|
f"https://app.platega.io/transaction/{external_id}",
|
|
{"X-MerchantId": PLATEGA_MERCHANT_ID, "X-Secret": PLATEGA_SECRET},
|
|
)
|
|
return data.get("status", "")
|
|
|
|
|
|
PAID_STATUSES = {"succeeded", "CONFIRMED"}
|
|
FAILED_STATUSES = {"canceled", "CANCELED", "CHARGEBACKED"}
|
|
|
|
|
|
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}")
|
|
|
|
|
|
def check_payment_status(provider: str, external_id: str) -> str:
|
|
if provider == "yookassa":
|
|
return check_yookassa_payment(external_id)
|
|
if provider == "platega":
|
|
return check_platega_payment(external_id)
|
|
raise ValueError(f"unknown provider: {provider}")
|