InternalAuditInterprise/backend/app/engines/scan.py

"""全量穿透扫描编排（P1.5）。

把场景检测器的结果转化为线索，记录扫描覆盖范围（证明全量性）与数据版本（可追溯）。
当前为同步执行；后续可包装为 Celery 异步任务（接口保持不变）。
"""

from __future__ import annotations

import uuid
from dataclasses import dataclass

from sqlalchemy.orm import Session

from app.clues import service as clue_svc
from app.clues.models import Clue
from app.scenarios import churn_fraud as cf
from app.scenarios import split_contract as sc
from app.scenarios.base import BaseScenario

MODEL_VERSION = "mock-llm@0.1"


@dataclass
class ScanResult:
    scenario_code: str
    scanned_count: int
    clue: Clue | None


def run_split_contract_scan(
    session: Session,
    contracts: list[sc.ContractRecord],
    approval_threshold: float,
    shared_controller: bool = False,
    data_version_id: uuid.UUID | None = None,
) -> ScanResult:
    """场景一拆单扫描：检测→评分→（命中则）生成线索。"""
    finding = sc.detect_threshold_edge(contracts, approval_threshold)
    score = sc.split_risk_score(finding, shared_controller)
    clue = None
    if score > 0:
        rationale = sc.build_rationale(finding, approval_threshold, shared_controller)
        clue = clue_svc.create_clue(
            session,
            title="疑似政企拆单规避审批",
            risk_domain="收入",
            scenario_code="R8",
            score=score,
            rationale=rationale,
            evidence={
                "near_threshold_contracts": [c.contract_id for c in finding.near_threshold],
                "edge_ratio": finding.ratio,
                "near_threshold_amount": finding.total_amount,
                "approval_threshold": approval_threshold,
                "shared_controller": shared_controller,
            },
            subjects={"customers": sorted({c.customer_key for c in finding.near_threshold})},
            amount_involved=finding.total_amount,
            model_version=MODEL_VERSION,
            data_version_id=data_version_id,
        )
    return ScanResult("R8", len(contracts), clue)


def run_churn_scan(
    session: Session,
    retention_curve: list[cf.CohortPoint],
    commission_paid: float,
    active_ratio: float,
    zero_usage_ratio: float,
    channel_key: str,
    data_version_id: uuid.UUID | None = None,
) -> ScanResult:
    """场景二养卡骗补扫描：时序断崖 + 佣金质量不匹配→线索。"""
    finding = cf.detect_pulse_decay(retention_curve)
    mismatch = cf.commission_quality_mismatch(commission_paid, active_ratio, zero_usage_ratio)
    score = cf.churn_risk_score(finding, mismatch)
    clue = None
    if score >= 0.5:
        rationale = cf.build_rationale(finding, mismatch)
        clue = clue_svc.create_clue(
            session,
            title="疑似养卡骗补（脉冲增长+规律退订）",
            risk_domain="成本",
            scenario_code="R9",
            score=score,
            rationale=rationale,
            evidence={
                "cliff_month": finding.cliff_month,
                "max_drop": finding.max_drop,
                "commission_paid": commission_paid,
                "active_ratio": active_ratio,
                "zero_usage_ratio": zero_usage_ratio,
                "mismatch": mismatch,
            },
            subjects={"channel": channel_key},
            amount_involved=commission_paid,
            model_version=MODEL_VERSION,
            data_version_id=data_version_id,
        )
    return ScanResult("R9", len(retention_curve), clue)


@dataclass
class ScenarioRunResult:
    """通用调度下单个场景的落库结果：扫描覆盖数 + 生成的线索列表。"""

    scenario_code: str
    scanned_count: int
    clues: list[Clue]


def run_scenarios(
    session: Session,
    scenarios: list[BaseScenario],
    *,
    data_version_id: uuid.UUID | None = None,
    actor: str = "system",
) -> list[ScenarioRunResult]:
    """通用场景调度器：遍历已实例化的场景，检测→按阈值过滤→统一落库。

    新增场景无需改动本函数：实现 BaseScenario 子类并经 @register_scenario 注册即可。
    每个场景自报扫描覆盖数（证明全量性），超过阈值的草稿由本函数统一调 create_clue 落库。
    """
    results: list[ScenarioRunResult] = []
    for scenario in scenarios:
        outcome = scenario.scan(session, data_version_id=data_version_id)
        created: list[Clue] = []
        for draft in outcome.drafts:
            if draft.score >= scenario.score_threshold:
                created.append(
                    clue_svc.create_clue(
                        session,
                        title=draft.title or scenario.title,
                        risk_domain=scenario.risk_domain,
                        scenario_code=scenario.code,
                        score=draft.score,
                        rationale=draft.rationale,
                        evidence=draft.evidence,
                        subjects=draft.subjects,
                        amount_involved=draft.amount_involved,
                        model_version=MODEL_VERSION,
                        rule_version=scenario.rule_version,
                        data_version_id=data_version_id,
                        actor=actor,
                    )
                )
        results.append(
            ScenarioRunResult(scenario.code, outcome.scanned_count, created)
        )
    return results