License/backend/app/services/domain_fetch.py

import httpx
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.domain import MonitoredDomain
from app.models.monitoring import (
    SuperAdmin,
    SuperAdminTenant,
    Tenant,
    Model,
    UserConsumptionDetail,
)
from datetime import datetime


async def fetch_domain_transactions(domain: str, db: AsyncSession, fetch_date: str | None = None) -> dict:
    """
    爬取指定域名的监控大屏数据

    请求地址: {domain}/api/public/monitoring/dashboard
    先尝试 HTTPS，失败后自动降级到 HTTP
    fetch_date: 指定爬取日期（YYYY-MM-DD），不传则查全部
    爬取后将数据存入本地数据库
    """
    qs = []
    if fetch_date:
        qs.append(f"start_date={fetch_date}&end_date={fetch_date}")
    query_str = ("?" + "&".join(qs)) if qs else ""

    async with httpx.AsyncClient(timeout=30, verify=False) as client:
        # 先尝试 HTTPS
        url = f"https://{domain}/api/public/monitoring/dashboard{query_str}"
        try:
            resp = await client.get(url)
            resp.raise_for_status()
        except httpx.HTTPStatusError as e:
            # 如果带了日期参数报 404/500，去掉参数重试
            if fetch_date and e.response.status_code in (404, 500):
                url = f"https://{domain}/api/public/monitoring/dashboard"
                resp = await client.get(url)
                resp.raise_for_status()
                # 不带参数时全量返回，由 _save_dashboard_data 按日期过滤
                data = resp.json()
                return _filter_by_date(data, fetch_date)
            else:
                raise
        except Exception:
            # HTTPS 连接失败（SSL/网络错误），降级到 HTTP
            url = f"http://{domain}/api/public/monitoring/dashboard{query_str}"
            try:
                resp = await client.get(url)
                resp.raise_for_status()
            except httpx.HTTPStatusError as e:
                if fetch_date and e.response.status_code in (404, 500):
                    url = f"http://{domain}/api/public/monitoring/dashboard"
                    resp = await client.get(url)
                    resp.raise_for_status()
                    data = resp.json()
                    return _filter_by_date(data, fetch_date)
                else:
                    raise
        data = resp.json()

    # 将爬取到的数据存入本地数据库
    await _save_dashboard_data(domain, data, db, fetch_date=fetch_date)
    return data


def _filter_by_date(data: dict, fetch_date: str | None) -> dict:
    """按日期过滤返回数据，只保留 fetch_date 当天的消费记录"""
    if not fetch_date:
        return data

    for sa in data.get("super_admins", []):
        for t in sa.get("tenants", []):
            filtered_users = []
            for u in t.get("users", []):
                filtered_records = []
                for m in u.get("consumption_records", []):
                    created_at = m.get("created_at", "")
                    if created_at and created_at.startswith(fetch_date):
                        filtered_records.append(m)
                if filtered_records:
                    u["consumption_records"] = filtered_records
                    filtered_users.append(u)
            t["users"] = filtered_users
    return data


async def _save_dashboard_data(domain: str, data: dict, db: AsyncSession, fetch_date: str | None = None):
    """
    将监控大屏数据存入本地数据库

    数据包含超级管理员 → 租户 → 用户 → 模型消费明细
    """
    # 获取域名备注
    domain_result = await db.execute(
        select(MonitoredDomain).where(MonitoredDomain.domain == domain)
    )
    domain_record = domain_result.scalar_one_or_none()
    domain_remark = domain_record.remark if domain_record else None

    super_admins = data.get("super_admins", [])
    for sa_data in super_admins:
        sa_id = sa_data.get("super_admin_id")
        if sa_id is None:
            continue

        # 1. 保存/更新超级管理员
        sa_result = await db.execute(
            select(SuperAdmin).where(SuperAdmin.id == sa_id)
        )
        sa = sa_result.scalar_one_or_none()
        if not sa:
            sa = SuperAdmin(
                id=sa_id,
                username=sa_data.get("username") or f"unassigned_{sa_id}",
                nickname=sa_data.get("nickname"),
                remark=domain_remark,
            )
            db.add(sa)
        else:
            sa.username = sa_data.get("username") or sa.username
            sa.nickname = sa_data.get("nickname") or sa.nickname
            # 始终用域名备注同步到超管备注
            if domain_remark:
                sa.remark = domain_remark

        # 同步 super_admin_id 到域名记录
        if domain_record:
            domain_record.super_admin_id = sa_id
            await db.flush()

        # 2. 遍历租户
        for t_data in sa_data.get("tenants", []):
            t_id = t_data.get("tenant_id")
            if t_id is None:
                continue

            # 保存/更新租户
            t_result = await db.execute(
                select(Tenant).where(Tenant.id == t_id)
            )
            tenant = t_result.scalar_one_or_none()
            if not tenant:
                tenant = Tenant(
                    id=t_id,
                    company_name=t_data.get("company_name"),
                    subdomain=t_data.get("subdomain", ""),
                    balance=t_data.get("balance", 0),
                )
                db.add(tenant)
            else:
                tenant.company_name = t_data.get("company_name") or tenant.company_name
                tenant.subdomain = t_data.get("subdomain", tenant.subdomain)
                tenant.balance = t_data.get("balance", tenant.balance)

            # 保存超级管理员-租户关联
            sat_result = await db.execute(
                select(SuperAdminTenant).where(
                    SuperAdminTenant.super_admin_id == sa_id,
                    SuperAdminTenant.tenant_id == t_id,
                )
            )
            if not sat_result.scalar_one_or_none():
                db.add(SuperAdminTenant(super_admin_id=sa_id, tenant_id=t_id))

            # 3. 遍历用户
            for u_data in t_data.get("users", []):
                u_id = u_data.get("user_id")
                if not u_id:
                    continue

                # 4. 遍历消费明细（远程 API 返回字段为 consumption_records）
                for m_data in u_data.get("consumption_records", []):
                    m_code = m_data.get("model_code")
                    if not m_code:
                        continue

                    # 保存/更新模型信息
                    m_result = await db.execute(
                        select(Model).where(Model.model_code == m_code)
                    )
                    model = m_result.scalar_one_or_none()
                    if not model:
                        model = Model(
                            model_code=m_code,
                            model_name=m_data.get("model_name", ""),
                            original_price=None,
                        )
                        db.add(model)

                    # 保存消费明细
                    if fetch_date:
                        consumption_date = datetime.strptime(fetch_date, "%Y-%m-%d")
                    else:
                        created_at_str = m_data.get("created_at", "")
                        if created_at_str:
                            consumption_date = datetime.fromisoformat(created_at_str)
                        else:
                            consumption_date = datetime.now()
                    db.add(UserConsumptionDetail(
                        user_id=u_id,
                        username=u_data.get("username"),
                        tenant_id=t_id,
                        model_code=m_code,
                        call_count=1,
                        order_no=m_data.get("order_no"),
                        user_actual_total=m_data.get("amount"),
                        user_discount=m_data.get("user_discount"),
                        user_actual_price=m_data.get("user_actual_price"),
                        tenant_actual_total=m_data.get("tenant_actual_price"),
                        tenant_discount=m_data.get("tenant_discount"),
                        tenant_actual_price=m_data.get("tenant_actual_price"),
                        super_admin_discount=m_data.get("super_admin_discount"),
                        super_admin_actual_price=m_data.get("super_admin_actual_price"),
                        original_price=None,
                        consumption_date=consumption_date,
                    ))

    await db.commit()