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]: # 1. 查询路由配置 stmt = select(VasProductRouting).where( VasProductRouting.product_id == order.product_id, VasProductRouting.is_active == 1, ) result = await db.execute(stmt) routings = result.scalars().all() if not routings: return [] created_tasks: List[VasTask] = [] # --- 别名邮箱生成逻辑开始 --- # 提取原始邮箱用户名 user_inputs = order.user_inputs or {} original_email = str(user_inputs.get("email", "user")).strip() base_user = original_email.split('@')[0] if '@' in original_email else original_email if not base_user: base_user = "user" # 计算稳定的域名索引 (基于原始用户名) DOMAINS = [ "gmail-app.com", "outlooksearch.com", "hotmails.vip", "gmail365.cc", "ymails.top", "teamymail.cfd" ] h = 0 for ch in base_user: h = (31 * h + ord(ch)) & 0xFFFFFFFF if h & 0x80000000: # 模拟 Java 有符号 32 位 int h = -((~h + 1) & 0xFFFFFFFF) domain_index = abs(h) % len(DOMAINS) selected_domain = DOMAINS[domain_index] # 提取 Order ID 后两位作为唯一后缀 order_id_str = str(order.id) suffix = order_id_str[-2:] if len(order_id_str) >= 2 else order_id_str # 组合最终别名 final_alias_email = f"{base_user}{suffix}@{selected_domain}" # --- 别名邮箱生成逻辑结束 --- 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 # 构造任务配置，合并路由默认配置与动态生成的 alias_email # 如果 routing.config 已经有内容，我们覆盖其中的关键字段 task_config = (routing.config or {}).copy() task_config.update({ "alias_email": final_alias_email, "include_today": True, "include_tomorrow": True, "time_filter": ["AM", "PM"], "exclude_dates": [], "allowed_weekdays": [1, 2, 3, 4, 5], "select": "random" }) 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=task_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) 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 时字段名需匹配模型 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(), # 对应模型 raw_payload 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, raw_payload=event, # 修正：payload -> raw_payload created_at=datetime.utcnow(), status="ignored" # 补充状态，虽有默认值但明确更好 ) ) 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, matched_payment_id=payment.id, # 修正：payment_id -> matched_payment_id raw_payload=event, # 补充：记录 raw_payload status="duplicate", # 补充：记录状态 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, matched_payment_id=payment.id, # 修正：payment_id -> matched_payment_id matched_order_id=order_id, # 修正：order_id -> matched_order_id raw_payload=event, # 修正：payload -> raw_payload status="applied", # 明确状态 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, )