import json from sqlalchemy.orm import Session from typing import List, Optional from app.core.biz_exception import NotFoundError, PermissionDeniedError, BizLogicError from app.models.order import VasOrder from app.models.vas_task import VasTask from app.models.product import VasProduct 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: def create_task_if_not_exists( db: Session, order: VasOrder, routing_key: str, script_version: str, config: dict, ): existing = ( db.query(VasTask) .filter( VasTask.order_id == order.id, VasTask.routing_key == routing_key, VasTask.script_version == script_version, ) .first() ) if existing: return existing # 幂等命中，直接返回 task = VasTask( order_id=order.id, routing_key=routing_key, script_version=script_version, config=config, user_inputs=order.user_inputs, status="pending", priority=10, expire_at=None, ) db.add(task) return task @staticmethod def smshelper_payment_webhook(db: Session, payload: SMSHelperWebhookPayload): """ webhook payload 示例： title=【微信】微信支付 content=【SM-E5260】个人收款码到账¥0.01 """ title = payload.title content = payload.content webhook_out = PaymentWebhookOut() if "微信" in title: provider = "wechat" elif "支付宝" in title: provider = "alipay" 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) db.commit() db.refresh(event) payment_qr = ( db.query(VasPaymentQR) .filter( VasPaymentQR.provider == provider, VasPaymentQR.device == device_id, VasPaymentQR.is_active == 1 ) .first() ) if not payment_qr: event.status = "failed" event.error_message = "QR not found" db.commit() raise BizLogicError("QR not found") payment = ( db.query(VasPayment) .filter( VasPayment.provider == provider, VasPayment.amount == amount_cent, VasPayment.qr_id == payment_qr.id, VasPayment.status == "pending" ) .order_by(VasPayment.created_at.desc()) .first() ) if not payment: event.status = "failed" event.error_message = "No matching pending payment" db.commit() raise BizLogicError("Payment not found") webhook_out.payment_id = payment.id if payment.status in ("succeeded", "late_paid"): event.status = "duplicate" event.matched_payment_id = payment.id event.matched_order_id = payment.order_id db.commit() webhook_out.status = True webhook_out.notify = False return webhook_out now = datetime.utcnow() if payment.expire_at and now > payment.expire_at: payment.status = "late_paid" else: payment.status = "succeeded" _create_task_if_not_exists(db, order) # ---------- 写入原始 payload ---------- payment.provider_payload = { "title": title, "content": content, "device_id": device_id, "received_at": now.isoformat(), } order = db.query(VasOrder).filter(VasOrder.id == payment.order_id).first() if order and order.status != "paid": order.status = "paid" event.status = "applied" event.matched_payment_id = payment.id event.matched_order_id = payment.order_id db.commit() db.refresh(payment) webhook_out.status = True webhook_out.order_id = order.id webhook_out.user_id = order.user_id webhook_out.notify = True return webhook_out @staticmethod def stripe_payment_webhook(db: Session, event): """ Stripe webhook handler """ event_id = event["id"] event_type = event["type"] data = event["data"]["object"] webhook_out = PaymentWebhookOut() # ---------- 1. 幂等（事件级） ---------- existed_event = ( db.query(VasPaymentEvent) .filter(VasPaymentEvent.provider == "stripe") .filter(VasPaymentEvent.event_id == event_id) .first() ) if existed_event: webhook_out.status = True webhook_out.notify = False return webhook_out # ---------- 2. 只处理关心的事件 ---------- 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(), ) ) db.commit() webhook_out.status = True webhook_out.notify = False return webhook_out # ---------- 3. 解析 metadata ---------- 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") # ---------- 4. 查找 payment（业务级幂等） ---------- payment = ( db.query(VasPayment) .filter(VasPayment.id == int(payment_id)) .first() ) 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, payload=event, payment_id=payment.id, created_at=datetime.utcnow(), ) ) db.commit() webhook_out.status = True webhook_out.notify = False return webhook_out # ---------- 5. 金额校验 ---------- paid_amount = data["amount_total"] # 单位：cent 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}, got {paid_amount} {paid_currency}") # ---------- 6. 判断是否超时 ---------- now = datetime.utcnow() if payment.expire_at and now > payment.expire_at: payment.status = "late_paid" else: payment.status = "succeeded" _create_task_if_not_exists(db, order) payment.provider_payload = event payment.updated_at = now # ---------- 7. 更新 order ---------- order = db.query(VasOrder).filter(VasOrder.id == order_id).first() if order and order.status != "paid": order.status = "paid" order.updated_at = now # ---------- 8. 写 payment_event ---------- 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, ) ) db.commit() webhook_out.status = True webhook_out.order_id = order.id webhook_out.user_id = order.user_id webhook_out.notify = True return webhook_out