webapp/app/services/webhook_service.py

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.queue_manager import queue_manager
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.status.in_(["pending", "grabbed", "running", "completed"]),
            )
            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)
            await db.flush()
            await db.refresh(task)
            queue_manager.put(
                queue_name=routing.routing_key,
                task_id=task.id,
                priority=task.priority
            )
            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,
        )