178 lines
4.3 KiB
Python
178 lines
4.3 KiB
Python
import os
|
|
from uuid import uuid4
|
|
from typing import Dict, Any
|
|
|
|
import httpx
|
|
from fastapi import FastAPI, HTTPException, Request
|
|
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
|
|
from jinja2 import Template
|
|
|
|
app = FastAPI(title="YooKassa Mock")
|
|
|
|
payments_db: Dict[str, Dict[str, Any]] = {}
|
|
|
|
WEBHOOK_URL = os.getenv("WEBHOOK_URL", "").strip()
|
|
MOCK_HOST = os.getenv("MOCK_HOST", "").strip()
|
|
MOCK_PORT = int(os.getenv("MOCK_PORT", "8081"))
|
|
|
|
CHECKOUT_TEMPLATE = Template("""
|
|
<!doctype html>
|
|
<html lang="ru">
|
|
<head>
|
|
<meta charset="utf-8">
|
|
<meta name="viewport" content="width=device-width,initial-scale=1">
|
|
<title>Оплата заказа</title>
|
|
<style>
|
|
body {
|
|
font-family: Arial, sans-serif;
|
|
background: #f6f7fb;
|
|
margin: 0;
|
|
padding: 40px 20px;
|
|
color: #222;
|
|
}
|
|
.card {
|
|
max-width: 480px;
|
|
margin: 0 auto;
|
|
background: #fff;
|
|
border-radius: 12px;
|
|
padding: 32px;
|
|
box-shadow: 0 8px 24px rgba(0,0,0,0.08);
|
|
}
|
|
h1 {
|
|
margin-top: 0;
|
|
font-size: 24px;
|
|
}
|
|
.amount {
|
|
font-size: 28px;
|
|
font-weight: 700;
|
|
margin: 16px 0 24px;
|
|
}
|
|
button {
|
|
width: 100%;
|
|
border: 0;
|
|
border-radius: 10px;
|
|
padding: 14px 18px;
|
|
font-size: 16px;
|
|
font-weight: 700;
|
|
cursor: pointer;
|
|
background: #006efc;
|
|
color: white;
|
|
}
|
|
.meta {
|
|
margin-top: 20px;
|
|
font-size: 13px;
|
|
color: #666;
|
|
word-break: break-all;
|
|
}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<div class="card">
|
|
<h1>Тестовая оплата</h1>
|
|
<div>Сумма к оплате:</div>
|
|
<div class="amount">{{ amount }} {{ currency }}</div>
|
|
|
|
<form method="post" action="/process/{{ payment_id }}">
|
|
<button type="submit">Оплатить</button>
|
|
</form>
|
|
|
|
<div class="meta">
|
|
payment_id: {{ payment_id }}
|
|
</div>
|
|
</div>
|
|
</body>
|
|
</html>
|
|
""")
|
|
|
|
|
|
def build_base_url(request: Request) -> str:
|
|
if MOCK_HOST:
|
|
return f"http://{MOCK_HOST}:{MOCK_PORT}"
|
|
return str(request.base_url).rstrip("/")
|
|
|
|
|
|
@app.post("/v3/payments")
|
|
async def create_payment(request: Request):
|
|
payload = await request.json()
|
|
|
|
payment_id = str(uuid4())
|
|
|
|
amount_data = payload.get("amount") or {}
|
|
confirmation_data = payload.get("confirmation") or {}
|
|
|
|
value = amount_data.get("value", "0.00")
|
|
currency = amount_data.get("currency", "RUB")
|
|
return_url = confirmation_data.get("return_url")
|
|
|
|
if not return_url:
|
|
raise HTTPException(status_code=400, detail="confirmation.return_url is required")
|
|
|
|
payments_db[payment_id] = {
|
|
"return_url": return_url,
|
|
"amount": value,
|
|
"currency": currency,
|
|
}
|
|
|
|
base_url = build_base_url(request)
|
|
confirmation_url = f"{base_url}/checkout?payment_id={payment_id}"
|
|
|
|
response = {
|
|
"id": payment_id,
|
|
"status": "pending",
|
|
"paid": False,
|
|
"amount": {
|
|
"value": value,
|
|
"currency": currency,
|
|
},
|
|
"confirmation": {
|
|
"type": "redirect",
|
|
"confirmation_url": confirmation_url,
|
|
},
|
|
"test": True,
|
|
}
|
|
|
|
return JSONResponse(content=response)
|
|
|
|
|
|
@app.get("/checkout", response_class=HTMLResponse)
|
|
async def checkout(payment_id: str):
|
|
payment = payments_db.get(payment_id)
|
|
if not payment:
|
|
raise HTTPException(status_code=404, detail="payment not found")
|
|
|
|
html = CHECKOUT_TEMPLATE.render(
|
|
payment_id=payment_id,
|
|
amount=payment["amount"],
|
|
currency=payment["currency"],
|
|
)
|
|
return HTMLResponse(content=html)
|
|
|
|
|
|
@app.post("/process/{payment_id}")
|
|
async def process_payment(payment_id: str):
|
|
payment = payments_db.get(payment_id)
|
|
if not payment:
|
|
raise HTTPException(status_code=404, detail="payment not found")
|
|
|
|
if WEBHOOK_URL:
|
|
webhook_payload = {
|
|
"event": "payment.succeeded",
|
|
"type": "notification",
|
|
"object": {
|
|
"id": payment_id,
|
|
"status": "succeeded",
|
|
},
|
|
}
|
|
try:
|
|
async with httpx.AsyncClient(timeout=10.0) as client:
|
|
await client.post(WEBHOOK_URL, json=webhook_payload)
|
|
except Exception as e:
|
|
print(f"Webhook send failed: {e}")
|
|
|
|
return RedirectResponse(url=payment["return_url"], status_code=303)
|
|
|
|
|
|
@app.get("/health")
|
|
async def health():
|
|
return {"status": "ok"}
|