import re import json from datetime import datetime, timedelta from typing import List, Optional from decimal import Decimal from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from app.core.biz_exception import NotFoundError, BizLogicError from app.models.order import VasOrder from app.models.vas_task import VasTask from app.models.product_routing import VasProductRouting from app.models.payment_event import VasPaymentEvent from app.models.payment import VasPayment from app.models.payment_qr import VasPaymentQR from app.schemas.webhook import SMSHelperWebhookPayload, PaymentWebhookOut class WebhookService: # ========================================================= # 内部方法：创建 Task（幂等） # ========================================================= @staticmethod async def _create_task_if_not_exists( db: AsyncSession, order: VasOrder, ) -> List[VasTask]: stmt = select(VasProductRouting).where( VasProductRouting.product_id == order.product_id, VasProductRouting.is_active == 1, ) result = await db.execute(stmt) routings = result.scalars().all() print(f'routings = {routings}') if not routings: return [] created_tasks: List[VasTask] = [] for routing in routings: exists_stmt = select(VasTask).where( VasTask.order_id == order.id, VasTask.routing_key == routing.routing_key, VasTask.script_version == routing.script_version, ) exists_result = await db.execute(exists_stmt) exists = exists_result.scalar_one_or_none() if exists: continue task = VasTask( order_id=order.id, routing_key=routing.routing_key, script_version=routing.script_version, priority=routing.priority, status="pending", user_inputs=order.user_inputs, config=routing.config, attempt_count=0, notify_count=0, expire_at=datetime.utcnow() + timedelta(days=60), created_at=datetime.utcnow(), ) db.add(task) created_tasks.append(task) return created_tasks # ========================================================= # SMSHelper 微信 / 支付宝 收款 webhook # ========================================================= @staticmethod async def smshelper_payment_webhook( db: AsyncSession, payload: SMSHelperWebhookPayload, ) -> Optional[PaymentWebhookOut]: title = payload.title content = payload.content if "微信" in title: provider = "wechat" elif "支付宝" in title: provider = "alipay" else: raise BizLogicError("Unknown payment provider") device_match = re.search(r"【(.+?)】", content) device_id = device_match.group(1) if device_match else None amount_match = re.search(r"¥([\d.]+)", content) if not amount_match: raise BizLogicError("Amount not found in webhook content") amount_yuan = Decimal(amount_match.group(1)) amount_cent = int(amount_yuan * 100) event = VasPaymentEvent( provider=provider, event_type="payment_received", title=title, content=content, parsed_amount=amount_cent, parsed_currency="CNY", parsed_device=device_id, raw_payload=payload.dict(), status="received", ) db.add(event) await db.commit() await db.refresh(event) # ---------- 查找 QR ---------- qr_stmt = select(VasPaymentQR).where( VasPaymentQR.provider == provider, VasPaymentQR.device == device_id, VasPaymentQR.is_active == 1, ) qr_result = await db.execute(qr_stmt) payment_qr = qr_result.scalar_one_or_none() if not payment_qr: event.status = "failed" event.error_message = "QR not found" await db.commit() raise BizLogicError("QR not found") # ---------- 查找 payment ---------- pay_stmt = ( select(VasPayment) .where( VasPayment.provider == provider, VasPayment.amount == amount_cent, VasPayment.qr_id == payment_qr.id, VasPayment.status == "pending", ) .order_by(VasPayment.created_at.desc()) ) pay_result = await db.execute(pay_stmt) payment = pay_result.scalar_one_or_none() if not payment: event.status = "failed" event.error_message = "No matching pending payment" await db.commit() raise BizLogicError("Payment not found") if payment.status in ("succeeded", "late_paid"): event.status = "duplicate" event.matched_payment_id = payment.id event.matched_order_id = payment.order_id await db.commit() return None now = datetime.utcnow() payment.status = "late_paid" if payment.expire_at and now > payment.expire_at else "succeeded" payment.provider_payload = { "title": title, "content": content, "device_id": device_id, "received_at": now.isoformat(), } order_stmt = select(VasOrder).where(VasOrder.id == payment.order_id) order_result = await db.execute(order_stmt) order = order_result.scalar_one_or_none() if order and order.status != "paid": order.status = "paid" await WebhookService._create_task_if_not_exists(db, order) event.status = "applied" event.matched_payment_id = payment.id event.matched_order_id = payment.order_id await db.commit() await db.refresh(payment) return PaymentWebhookOut( status=True, order_id=order.id, user_id=order.user_id, payment_id=payment.id, notify=True, ) # ========================================================= # Stripe webhook # ========================================================= @staticmethod async def stripe_payment_webhook( db: AsyncSession, event: dict, ) -> Optional[PaymentWebhookOut]: event_id = event["id"] event_type = event["type"] data = event["data"]["object"] existed_stmt = select(VasPaymentEvent).where( VasPaymentEvent.provider == "stripe", VasPaymentEvent.event_id == event_id, ) existed_result = await db.execute(existed_stmt) if existed_result.scalar_one_or_none(): return None if event_type != "checkout.session.completed": db.add( VasPaymentEvent( provider="stripe", event_id=event_id, event_type=event_type, payload=event, created_at=datetime.utcnow(), ) ) await db.commit() return None metadata = data.get("metadata", {}) payment_id = metadata.get("payment_id") order_id = metadata.get("order_id") if not payment_id or not order_id: raise BizLogicError("Missing payment_id or order_id in metadata") pay_stmt = select(VasPayment).where(VasPayment.id == int(payment_id)) pay_result = await db.execute(pay_stmt) payment = pay_result.scalar_one_or_none() if not payment: raise NotFoundError("Payment not found") if payment.status == "succeeded": db.add( VasPaymentEvent( provider="stripe", event_id=event_id, event_type=event_type, payment_id=payment.id, created_at=datetime.utcnow(), ) ) await db.commit() return None paid_amount = data["amount_total"] paid_currency = data["currency"].upper() if paid_amount != payment.amount or paid_currency != payment.currency: raise BizLogicError( f"Amount mismatch, expected {payment.amount} {payment.currency}, " f"got {paid_amount} {paid_currency}" ) now = datetime.utcnow() payment.status = "late_paid" if payment.expire_at and now > payment.expire_at else "succeeded" payment.provider_payload = event payment.updated_at = now order_stmt = select(VasOrder).where(VasOrder.id == order_id) order_result = await db.execute(order_stmt) order = order_result.scalar_one_or_none() if order and order.status != "paid": order.status = "paid" order.updated_at = now await WebhookService._create_task_if_not_exists(db, order) db.add( VasPaymentEvent( provider="stripe", event_id=event_id, event_type=event_type, payment_id=payment.id, order_id=order_id, payload=event, created_at=now, ) ) await db.commit() await db.refresh(payment) return PaymentWebhookOut( status=True, order_id=order.id, user_id=order.user_id, payment_id=payment.id, notify=True, )