packages/core/subscription_matcher.py

import asyncio
import os
import httpx
from datetime import datetime, timezone
from typing import List
from sqlalchemy import select
from packages.core.logger import app_logger
from packages.core.models import StandardTradeRecord, Subscription
from packages.core.database import AsyncSessionLocal

# 订阅通知配置
SENDGRID_API_KEY = os.getenv("SENDGRID_API_KEY")
NOTIFICATION_FROM_EMAIL = os.getenv("NOTIFICATION_FROM_EMAIL", "noreply@customs-data.com")

async def match_and_notify_subscriptions(records: List[StandardTradeRecord]):
    """
    匹配新入库的标准记录与用户订阅，并触发通知。
    """
    if not records:
        return

    # 提取本次批次中涉及的 HS Code 和 企业名
    hs_codes = set(r.hs_code for r in records if r.hs_code)
    
    # 真实场景下，企业订阅可能基于 standard_entity_id，这里 MVP 阶段先用名称模糊匹配或精确匹配
    # 为了简化，我们假设 target_entity_id 存的是企业名称
    company_names = set()
    for r in records:
        if r.importer_name:
            company_names.add(r.importer_name)
        if r.exporter_name:
            company_names.add(r.exporter_name)

    if not hs_codes and not company_names:
        return

    async with AsyncSessionLocal() as session:
        # 查找可能命中的订阅
        stmt = select(Subscription).where(
            (Subscription.target_hs_code.in_(hs_codes)) |
            (Subscription.target_entity_id.in_(company_names))
        )
        result = await session.execute(stmt)
        subscriptions = result.scalars().all()

        if not subscriptions:
            return

        # 组织通知内容
        notifications = []
        for sub in subscriptions:
            matched_records = []
            for r in records:
                if sub.target_hs_code and r.hs_code == sub.target_hs_code:
                    matched_records.append(r)
                elif sub.target_entity_id and (r.importer_name == sub.target_entity_id or r.exporter_name == sub.target_entity_id):
                    matched_records.append(r)
            
            if matched_records:
                notifications.append({
                    "user_email": sub.user_email,
                    "subscription_id": sub.id,
                    "matched_count": len(matched_records),
                    "matched_records": matched_records[:5],  # 最多显示 5 条样本
                    "subscription": sub,
                })

        # 触发通知
        notified_at = datetime.now(timezone.utc)
        for notif in notifications:
            # 发送邮件通知
            await _send_email_notification(notif)
            
            # 发送 Webhook 通知（如果配置）
            if notif["subscription"].webhook_url:
                await _send_webhook_notification(notif)
            
            # 记录日志
            app_logger.info(
                f"[SUBSCRIPTION NOTIFY] Sent alert to {notif['user_email']} "
                f"for sub {notif['subscription_id']}. Matched {notif['matched_count']} new records."
            )
            
            # 更新通知时间
            notif["subscription"].last_notified_at = notified_at

        if notifications:
            await session.commit()


async def _send_email_notification(notif: dict):
    """
    发送邮件通知
    """
    if not SENDGRID_API_KEY:
        app_logger.warning("SENDGRID_API_KEY not configured, skipping email notification")
        return
    
    try:
        subscription = notif["subscription"]
        matched_records = notif["matched_records"]
        
        # 构建邮件内容
        records_html = "<ul>"
        for record in matched_records:
            records_html += f"""
            <li>
                <strong>日期:</strong> {record.trade_date.strftime('%Y-%m-%d') if record.trade_date else 'N/A'}<br>
                <strong>国家:</strong> {record.source_country} ({record.trade_direction})<br>
                <strong>HS编码:</strong> {record.hs_code or 'N/A'}<br>
                <strong>商品:</strong> {record.product_name or 'N/A'}<br>
                <strong>进口商:</strong> {record.importer_name or 'N/A'}<br>
                <strong>出口商:</strong> {record.exporter_name or 'N/A'}<br>
                <strong>金额:</strong> {record.amount} {record.currency or ''}<br>
            </li>
            """
        records_html += "</ul>"
        
        if notif["matched_count"] > len(matched_records):
            records_html += f"<p><em>...还有 {notif['matched_count'] - len(matched_records)} 条匹配记录</em></p>"
        
        # 构建订阅条件描述
        subscription_desc = []
        if subscription.target_hs_code:
            subscription_desc.append(f"HS编码: {subscription.target_hs_code}")
        if subscription.target_entity_id:
            subscription_desc.append(f"企业: {subscription.target_entity_id}")
        
        payload = {
            "personalizations": [{
                "to": [{"email": notif["user_email"]}],
                "subject": f"海关数据订阅提醒：发现 {notif['matched_count']} 条新匹配记录"
            }],
            "from": {
                "email": NOTIFICATION_FROM_EMAIL,
                "name": "海关数据系统"
            },
            "content": [{
                "type": "text/html",
                "value": f"""
                <h2>您的订阅有新数据</h2>
                <p><strong>订阅条件:</strong> {' | '.join(subscription_desc)}</p>
                <p><strong>匹配记录数:</strong> {notif['matched_count']}</p>
                
                <h3>样本记录:</h3>
                {records_html}
                
                <hr>
                <p><small>如需停止此订阅，请登录系统管理您的订阅设置。</small></p>
                """
            }]
        }
        
        async with httpx.AsyncClient() as client:
            response = await client.post(
                "https://api.sendgrid.com/v3/mail/send",
                headers={
                    "Authorization": f"Bearer {SENDGRID_API_KEY}",
                    "Content-Type": "application/json"
                },
                json=payload,
                timeout=10.0
            )
            
            if response.status_code == 202:
                app_logger.info(f"Subscription email sent to {notif['user_email']}")
            else:
                app_logger.error(f"Failed to send subscription email: {response.text}")
                
    except Exception as e:
        app_logger.error(f"Error sending subscription email: {e}")


async def _send_webhook_notification(notif: dict):
    """
    发送 Webhook 通知
    """
    try:
        subscription = notif["subscription"]
        webhook_url = subscription.webhook_url
        
        if not webhook_url:
            return
        
        # 构建 Webhook payload
        payload = {
            "subscription_id": notif["subscription_id"],
            "user_email": notif["user_email"],
            "matched_count": notif["matched_count"],
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "subscription": {
                "target_hs_code": subscription.target_hs_code,
                "target_entity_id": subscription.target_entity_id,
            },
            "sample_records": [
                {
                    "record_id": r.record_id,
                    "source_country": r.source_country,
                    "trade_direction": r.trade_direction,
                    "trade_date": r.trade_date.isoformat() if r.trade_date else None,
                    "hs_code": r.hs_code,
                    "product_name": r.product_name,
                    "importer_name": r.importer_name,
                    "exporter_name": r.exporter_name,
                    "amount": r.amount,
                    "currency": r.currency
                }
                for r in notif["matched_records"]
            ]
        }
        
        async with httpx.AsyncClient() as client:
            response = await client.post(
                webhook_url,
                json=payload,
                timeout=10.0,
                headers={"Content-Type": "application/json"}
            )
            
            if response.status_code in (200, 201, 202, 204):
                app_logger.info(f"Webhook notification sent to {webhook_url}")
            else:
                app_logger.error(f"Webhook notification failed: {response.status_code} - {response.text}")
                
    except Exception as e:
        app_logger.error(f"Error sending webhook notification: {e}")