License/backend/app/main.py

import asyncio
import logging
import math
from contextlib import asynccontextmanager
from datetime import datetime, timezone, timedelta

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

from app.config import settings
from app.routers import domains, monitoring, license as license_router, auth as auth_router
from app.routers.license import public_router as public_license_router
from app.routers import sa_balance as sa_balance_router
from app.database import async_session
from app.services.license import _notify_on_expired, _notify_on_warning
from app.services.domain_fetch import fetch_domain_transactions
from app.services.sms import send_sa_balance_warning, send_sa_balance_depleted
from app.services.compensation_service import process_pending
from app.models.license import SuperAdminLicense
from app.models.monitoring import FetchScheduleConfig, FetchLog, SuperAdmin
from app.models.domain import MonitoredDomain
from app.models.visitor import VisitorInfo
from app.redis import close_redis
from sqlalchemy import select, text
from decimal import Decimal

logger = logging.getLogger(__name__)


CST = timezone(timedelta(hours=8))


async def _check_licenses():
    """定期检查 License：过期状态更新 + 7 天预警"""
    while True:
        try:
            async with async_session() as session:
                # 1. 检查已过期
                expired_result = await session.execute(
                    select(SuperAdminLicense).where(
                        SuperAdminLicense.status == "active",
                        SuperAdminLicense.expires_at <= text("NOW()"),
                    )
                )
                for lic in expired_result.scalars().all():
                    logger.info("检测到 License #%d 已过期，更新状态并发送预警短信", lic.id)
                    lic.status = "expired"
                    await session.commit()
                    await _notify_on_expired(session, lic)

                # 2. 检查剩余 7 天预警（未发送过的）
                warning_result = await session.execute(
                    select(SuperAdminLicense).where(
                        SuperAdminLicense.status == "active",
                        SuperAdminLicense.warning_sent == False,
                        SuperAdminLicense.expires_at > text("NOW()"),
                        SuperAdminLicense.expires_at <= text("NOW() + INTERVAL '7 days'"),
                    )
                )
                for lic in warning_result.scalars().all():
                    days_left = math.ceil((lic.expires_at - datetime.now(timezone.utc)).total_seconds() / 86400)
                    logger.info("检测到 License #%d 剩余 %d 天，发送预警短信", lic.id, days_left)
                    await _notify_on_warning(session, lic, days_left)
                    lic.warning_sent = True
                    await session.commit()
        except Exception:
            logger.exception("定时检查 License 异常")

        await asyncio.sleep(24 * 3600)


async def _daily_fetch():
    """定时爬取：每分钟检查 DB 配置，到了目标时间就爬取当天流水"""
    while True:
        try:
            async with async_session() as session:
                result = await session.execute(select(FetchScheduleConfig).limit(1))
                config = result.scalar_one_or_none()

                if config and config.enabled:
                    h, m = map(int, config.schedule_time.split(":"))
                    now = datetime.now(CST)
                    today = now.strftime("%Y-%m-%d")

                    # 已过目标时间且今天还没爬过
                    if now.hour * 60 + now.minute >= h * 60 + m and config.last_fetch_date != today:
                        logger.info("开始定时爬取当天流水: %s", today)

                        # 查询今日已失败的域名（跳过）
                        failed_result = await session.execute(
                            select(FetchLog.domain).where(
                                FetchLog.status == "failed",
                                FetchLog.created_at >= text(f"'{today} 00:00:00+08'"),
                            ).distinct()
                        )
                        failed_domains = {r[0] for r in failed_result.all()}

                        domain_result = await session.execute(
                            select(MonitoredDomain).where(MonitoredDomain.is_active == True)
                        )
                        for d in domain_result.scalars().all():
                            if d.domain in failed_domains:
                                logger.info("域名 %s 今日已失败，跳过", d.domain)
                                continue
                            try:
                                await fetch_domain_transactions(d.domain, session, fetch_date=today)
                                logger.info("域名 %s 当天流水爬取完成", d.domain)
                                session.add(FetchLog(
                                    domain=d.domain, status="success", message="当天流水爬取完成"
                                ))
                            except Exception as e:
                                error_msg = str(e)[:500]
                                logger.exception("域名 %s 当天流水爬取失败", d.domain)
                                session.add(FetchLog(
                                    domain=d.domain, status="failed", message=error_msg
                                ))
                        config.last_fetch_date = today
                        await session.commit()
                        last_fetch_date = today
                        logger.info("当天定时爬取全部完成")
        except Exception:
            logger.exception("定时爬取异常")

        await asyncio.sleep(60)


async def _check_sa_balances():
    """定期检查超管余额：余额预警 + 余额耗尽通知（每 10 分钟）"""
    while True:
        try:
            async with async_session() as session:
                result = await session.execute(select(SuperAdmin).order_by(SuperAdmin.id))
                threshold = Decimal(str(settings.sa_balance_warning_threshold))

                for sa in result.scalars().all():
                    balance = Decimal(str(sa.balance or 0))
                    company = sa.remark or sa.username or str(sa.id)

                    # 手机号：优先用超管自身 phone，否则从域名关联的 VisitorInfo 获取
                    phone = sa.phone
                    if not phone:
                        domain_row = (await session.execute(
                            select(MonitoredDomain).where(
                                MonitoredDomain.super_admin_id == sa.id,
                                MonitoredDomain.is_active == True,
                            ).limit(1)
                        )).scalar_one_or_none()
                        if domain_row:
                            visitor = (await session.execute(
                                select(VisitorInfo).where(VisitorInfo.domain_id == domain_row.id)
                            )).scalar_one_or_none()
                            if visitor:
                                phone = visitor.phone

                    if balance <= 0:
                        # 余额耗尽
                        if not sa.balance_depleted_sent and phone:
                            ok, reason = await send_sa_balance_depleted(phone, company)
                            if ok:
                                sa.balance_depleted_sent = True
                                await session.commit()
                                logger.info("超管 %d(%s) 余额耗尽，短信已发送", sa.id, company)
                            else:
                                logger.warning("超管 %d 耗尽短信发送失败: %s", sa.id, reason)
                    elif balance <= threshold:
                        # 余额不足但未耗尽
                        if not sa.balance_warning_sent and phone:
                            ok, reason = await send_sa_balance_warning(phone, company, f"{balance:.2f}")
                            if ok:
                                sa.balance_warning_sent = True
                                await session.commit()
                                logger.info("超管 %d(%s) 余额预警(%.2f)，短信已发送", sa.id, company, balance)
                            else:
                                logger.warning("超管 %d 预警短信发送失败: %s", sa.id, reason)
                    else:
                        # 余额恢复，重置标记
                        if sa.balance_warning_sent or sa.balance_depleted_sent:
                            sa.balance_warning_sent = False
                            sa.balance_depleted_sent = False
                            await session.commit()
                            logger.info("超管 %d(%s) 余额恢复至 %.2f，重置预警标记", sa.id, company, balance)
        except Exception:
            logger.exception("定时检查超管余额异常")

        await asyncio.sleep(600)  # 10 分钟


async def _process_compensation():
    """定期处理待补偿扣减记录（每 30 秒）"""
    while True:
        try:
            async with async_session() as session:
                count = await process_pending(session)
                if count > 0:
                    logger.info("补偿任务处理了 %d 条记录", count)
        except Exception:
            logger.exception("补偿任务异常")

        await asyncio.sleep(30)


@asynccontextmanager
async def lifespan(app: FastAPI):
    # 启动时立即检查一次
    try:
        async with async_session() as session:
            # 1. 检查已过期
            expired_result = await session.execute(
                select(SuperAdminLicense).where(
                    SuperAdminLicense.status == "active",
                    SuperAdminLicense.expires_at <= text("NOW()"),
                )
            )
            for lic in expired_result.scalars().all():
                logger.info("启动时检测到 License #%d 已过期，更新状态并发送预警短信", lic.id)
                lic.status = "expired"
                await session.commit()
                await _notify_on_expired(session, lic)

            # 2. 检查剩余 7 天预警（未发送过的）
            from datetime import timedelta
            warning_result = await session.execute(
                select(SuperAdminLicense).where(
                    SuperAdminLicense.status == "active",
                    SuperAdminLicense.warning_sent == False,
                    SuperAdminLicense.expires_at > text("NOW()"),
                    SuperAdminLicense.expires_at <= text("NOW() + INTERVAL '7 days'"),
                )
            )
            for lic in warning_result.scalars().all():
                days_left = math.ceil((lic.expires_at - datetime.now(timezone.utc)).total_seconds() / 86400)
                logger.info("启动时检测到 License #%d 剩余 %d 天，发送预警短信", lic.id, days_left)
                await _notify_on_warning(session, lic, days_left)
                lic.warning_sent = True
                await session.commit()

            logger.info("启动检查完成")
    except Exception:
        logger.exception("启动时检查 License 异常")

    # 启动后台定时任务（错开启动，避免同时抢连接）
    license_task = asyncio.create_task(_check_licenses())
    await asyncio.sleep(2)

    fetch_task = asyncio.create_task(_daily_fetch())
    await asyncio.sleep(2)

    balance_task = asyncio.create_task(_check_sa_balances())
    await asyncio.sleep(2)

    compensation_task = asyncio.create_task(_process_compensation())

    logger.info("后台任务已启动：License 检查 + 定时爬取 + 超管余额预警 + 扣减补偿")
    yield
    license_task.cancel()
    fetch_task.cancel()
    balance_task.cancel()
    compensation_task.cancel()
    for task in (license_task, fetch_task, balance_task, compensation_task):
        try:
            await task
        except asyncio.CancelledError:
            pass
    await close_redis()


app = FastAPI(
    title="域名流水监控",
    version="0.1.0",
    debug=settings.debug,
    lifespan=lifespan,
)

# CORS 配置，允许前端 Vite 开发服务器访问
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 注册路由
app.include_router(domains.router)
app.include_router(monitoring.router)
app.include_router(license_router.router)
app.include_router(public_license_router)
app.include_router(auth_router.router)
app.include_router(sa_balance_router.router)
app.include_router(sa_balance_router.public_router)


@app.get("/health")
async def health():
    """健康检查接口"""
    return {"status": "ok"}