feat: 增加风控定时扫描 Worker 与环境配置

app/core/config.py

@@ -58,6 +58,11 @@ class Settings(BaseSettings):
     worker_lease_seconds: int = Field(default=60, gt=0)
     worker_retry_limit: int = Field(default=3, ge=0)
     worker_poll_seconds: float = Field(default=1, gt=0)
+    risk_scan_schedule_enabled: bool = False
+    risk_scan_interval_minutes: int = Field(default=5, ge=1, le=1440)
+    risk_scan_run_immediately: bool = False
+    risk_scan_retry_limit: int = Field(default=2, ge=0, le=5)
+    risk_scan_poll_seconds: float = Field(default=30, gt=0)
 
     model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
 

app/service/risk_scan_schedule_config.py

@@ -0,0 +1,25 @@
+"""风控定时扫描环境配置读取。"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from app.core.config import get_settings
+
+
+@dataclass(frozen=True, slots=True)
+class RiskScanScheduleConfig:
+    enabled: bool = False
+    interval_minutes: int = 5
+    run_immediately: bool = False
+    retry_limit: int = 2
+
+async def load_risk_scan_schedule_config() -> RiskScanScheduleConfig:
+    """读取环境变量配置；缺失时使用默认关闭值。"""
+    settings = get_settings()
+    return RiskScanScheduleConfig(
+        enabled=settings.risk_scan_schedule_enabled,
+        interval_minutes=settings.risk_scan_interval_minutes,
+        run_immediately=settings.risk_scan_run_immediately,
+        retry_limit=settings.risk_scan_retry_limit,
+    )

app/worker/risk_scan_scheduler.py

@@ -0,0 +1,195 @@
+"""风控规则扫描定时 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:
+            return config.run_immediately
+        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()