import re import json from datetime import datetime, timedelta from sqlalchemy.orm import Session from typing import List, Optional from decimal import Decimal, ROUND_HALF_UP 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: @staticmethod def _create_task_if_not_exists( db: Session, order: VasOrder ): routings = ( db.query(VasProductRouting) .filter( VasProductRouting.product_id == order.product_id, VasProductRouting.is_active == 1 ) .all() ) if not routings: return [] created_tasks = [] for routing in routings: # ---------- 2. 幂等判断 ---------- exists = ( db.query(VasTask) .filter( VasTask.order_id == order.id, VasTask.routing_key == routing.routing_key, VasTask.script_version == routing.script_version, ) .first() ) if exists: continue # ---------- 3. 创建 task ---------- task = VasTask( order_id=order.id, routing_key=routing.routing_key, script_version=routing.script_version, priority=10, 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 @staticmethod def smshelper_payment_webhook(db: Session, payload: SMSHelperWebhookPayload): """ webhook payload 示例： title=【微信】微信支付 content=【SM-E5260】个人收款码到账¥0.01 """ title = payload.title content = payload.content 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") 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() return None now = datetime.utcnow() if payment.expire_at and now > payment.expire_at: payment.status = "late_paid" else: payment.status = "succeeded" # ---------- 写入原始 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" WebhookService._create_task_if_not_exists(db, order) event.status = "applied" event.matched_payment_id = payment.id event.matched_order_id = payment.order_id db.commit() db.refresh(payment) return PaymentWebhookOut( status=True, order_id=order.id, user_id=order.user_id, payment_id=payment.id, notify=True ) @staticmethod def stripe_payment_webhook(db: Session, event): """ Stripe webhook handler """ event_id = event["id"] event_type = event["type"] data = event["data"]["object"] # ---------- 1. 幂等（事件级） ---------- existed_event = ( db.query(VasPaymentEvent) .filter(VasPaymentEvent.provider == "stripe") .filter(VasPaymentEvent.event_id == event_id) .first() ) if existed_event: return None # ---------- 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() return None # ---------- 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() return None # ---------- 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" 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 WebhookService._create_task_if_not_exists(db, order) # ---------- 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() db.refresh(payment) return PaymentWebhookOut( status=True, order_id=order.id, user_id=order.user_id, payment_id=payment.id, notify=True )