webapp/app/services/webhook_service.py

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
        )