group_fqcd_jr/app/worker/risk_scan_scheduler.py

"""风控规则扫描定时 Worker。

该进程不依赖 Web 进程生命周期，通过 MySQL 咨询锁保证同一时刻只有一个
调度者执行扫描。默认关闭，配置开启后才执行。
"""

from __future__ import annotations

import argparse
import asyncio
import logging
from collections.abc import AsyncIterator, Awaitable, Callable
from contextlib import AbstractAsyncContextManager, asynccontextmanager
from datetime import UTC, datetime, timedelta
from typing import Any
from uuid import uuid4

from sqlalchemy import text

from app.core.config import get_settings
from app.core.contracts import RequestContext
from app.infrastructure.db import SessionFactory, engine
from app.model.audit import InteractionAudit
from app.service.risk_scan_schedule_config import (
    RiskScanScheduleConfig,
    load_risk_scan_schedule_config,
)
from app.service.risk_scan_service import RiskScanService

SCAN_LOCK_NAME = "jr_risk_scan_schedule"
logger = logging.getLogger(__name__)

ConfigLoader = Callable[[], Awaitable[RiskScanScheduleConfig]]
ScanExecutor = Callable[[], Awaitable[dict[str, int | str]]]
AuditWriter = Callable[[str, dict[str, Any]], Awaitable[None]]
LockFactory = Callable[[], AbstractAsyncContextManager[bool]]


@asynccontextmanager
async def mysql_scan_lock() -> AsyncIterator[bool]:
    """使用 MySQL 连接级咨询锁约束跨进程并发。"""
    async with SessionFactory() as session:
        acquired = bool(
            await session.scalar(
                text("SELECT GET_LOCK(:name, 0)"),
                {"name": SCAN_LOCK_NAME},
            )
        )
        try:
            yield acquired
        finally:
            if acquired:
                await session.scalar(
                    text("SELECT RELEASE_LOCK(:name)"),
                    {"name": SCAN_LOCK_NAME},
                )


async def default_scan_executor() -> dict[str, int | str]:
    context = RequestContext(
        user_id="0",
        trace_id=f"risk-scan-schedule-{uuid4()}",
        roles=("system",),
        permissions=("risk:alert:scan",),
        data_scope="all",
        portal="worker",
    )
    async with SessionFactory() as session:
        return await RiskScanService(session).scan(context)


async def default_audit_writer(status: str, detail: dict[str, Any]) -> None:
    async with SessionFactory() as session, session.begin():
        session.add(
            InteractionAudit(
                actor_type="system",
                actor_id=None,
                target_customer_id=None,
                portal="worker",
                action_type=f"risk_scan_scheduled_{status}",
                detail=detail,
                created_at=datetime.now(UTC).replace(tzinfo=None),
            )
        )


class RiskScanSchedulerWorker:
    def __init__(
        self,
        *,
        config_loader: ConfigLoader = load_risk_scan_schedule_config,
        scan_executor: ScanExecutor = default_scan_executor,
        audit_writer: AuditWriter = default_audit_writer,
        lock_factory: LockFactory = mysql_scan_lock,
        now: Callable[[], datetime] | None = None,
    ) -> None:
        self.config_loader = config_loader
        self.scan_executor = scan_executor
        self.audit_writer = audit_writer
        self.lock_factory = lock_factory
        self.now = now or (lambda: datetime.now(UTC))
        self.last_run_at: datetime | None = None

    async def run_once(self, *, force: bool = False) -> bool:
        config = await self.config_loader()
        if not config.enabled:
            return False
        current = self.now()
        if not force and not self._is_due(config, current):
            return False

        async with self.lock_factory() as acquired:
            if not acquired:
                logger.info("风控定时扫描由其他 Worker 执行，本轮跳过")
                return False
            return await self._execute_with_retry(config)

    def _is_due(
        self,
        config: RiskScanScheduleConfig,
        current: datetime,
    ) -> bool:
        if self.last_run_at is None:
            # 进程刚起来，不知道自己上次是什么时候跑的 —— `last_run_at` 只存在内存里。
            # **保守地视为 due**：多跑一次的最坏后果是重复扫描，而扫描本身是幂等的
            # （每条规则先 `_exists` 查重）并且有 MySQL 级锁；反过来"不跑"的后果可能是
            # **永远不跑**：原先这里返回 `config.run_immediately`（默认 False），
            # 重启之后 `_is_due` 恒为假，调度器形同虚设 —— 而且没有任何告警，
            # 现场只会表现为"风控好像没在扫描"。
            #
            # `config.run_immediately` 因此不再承担"首次是否执行"的语义（它原本想表达的
            # 是"启动后别马上跑"，但那与"永远不跑"在实现上无法区分）。字段保留，
            # 以免破坏既有配置。
            return True
        return current - self.last_run_at >= timedelta(minutes=config.interval_minutes)

    async def _execute_with_retry(
        self,
        config: RiskScanScheduleConfig,
    ) -> bool:
        attempts = config.retry_limit + 1
        last_error: Exception | None = None
        for attempt in range(1, attempts + 1):
            try:
                result = await self.scan_executor()
                self.last_run_at = self.now()
                await self.audit_writer(
                    "succeeded",
                    {
                        **result,
                        "attempt": attempt,
                        "trace_id": f"risk-scan-schedule-{uuid4()}",
                    },
                )
                return True
            except Exception as error:
                last_error = error
                logger.warning(
                    "风控定时扫描失败 attempt=%s/%s",
                    attempt,
                    attempts,
                    exc_info=True,
                )
        await self.audit_writer(
            "failed",
            {
                "attempts": attempts,
                "error_type": type(last_error).__name__ if last_error else "unknown",
                "trace_id": f"risk-scan-schedule-{uuid4()}",
            },
        )
        return False


async def serve(*, once: bool = False, force: bool = False) -> None:
    worker = RiskScanSchedulerWorker()
    try:
        while True:
            try:
                await worker.run_once(force=force)
            except Exception:
                logger.warning("风控定时扫描 Worker 轮次失败", exc_info=True)
                if once:
                    raise
            if once:
                return
            await asyncio.sleep(get_settings().risk_scan_poll_seconds)
    finally:
        await engine.dispose()


def main() -> None:
    parser = argparse.ArgumentParser(description="奶龙风控规则定时扫描 Worker")
    parser.add_argument("--once", action="store_true", help="执行一轮后退出")
    parser.add_argument("--force", action="store_true", help="忽略间隔，立即执行一次")
    args = parser.parse_args()
    logging.basicConfig(level=logging.INFO)
    try:
        asyncio.run(serve(once=args.once, force=args.force))
    except KeyboardInterrupt:
        pass


if __name__ == "__main__":
    main()