selfrelease
74e35fed96
- 新增场景基类、配置加载、注册表与原语模块 - 添加 r10_refund_split 规则及场景 JSON Schema - 扩展 scan 引擎与 scenarios API - 新增场景注册表/配置/集成测试 - 更新前端 App、api、labels 支持新场景
119 lines
4.4 KiB
Python
119 lines
4.4 KiB
Python
"""YAML 配置驱动场景（B 快捷通道）：用「配置 + 原语库」声明同构场景，免写 Python。
|
||
|
||
加载 scenarios/rules/*.yaml，经 pydantic 校验后为每条配置动态生成一个 BaseScenario
|
||
子类并注册。新增此类场景 = 新增一个 YAML 文件，无需改动任何 Python 代码。
|
||
运行时与代码场景一致：实例化后注入数据行 rows，由 run_scenarios 统一调度落库。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import uuid
|
||
from pathlib import Path
|
||
|
||
import yaml
|
||
from pydantic import BaseModel, Field, ValidationError
|
||
|
||
from app.scenarios.base import BaseScenario, ClueDraft, ScanOutcome
|
||
from app.scenarios.primitives import get_primitive
|
||
from app.scenarios.registry import register_scenario
|
||
|
||
_DEFAULT_RULES_DIR = Path(__file__).parent / "rules"
|
||
|
||
|
||
class ScenarioConfig(BaseModel):
|
||
"""YAML 场景配置的 schema（pydantic 校验，字段缺失/类型错误即报错）。"""
|
||
|
||
code: str = Field(min_length=1)
|
||
title: str = Field(min_length=1)
|
||
risk_domain: str = Field(min_length=1)
|
||
label: str = Field(min_length=1)
|
||
detector: str = Field(min_length=1)
|
||
params: dict = Field(default_factory=dict)
|
||
threshold: float = 0.5
|
||
rationale_template: str = "{summary}"
|
||
rule_version: str | None = None
|
||
|
||
|
||
class ConfigScenario(BaseScenario):
|
||
"""由 YAML 配置驱动的场景：运行时按 detector 调用原语，模板生成人话理由。
|
||
|
||
config 由动态子类（build_scenario_class）设置；数据行 rows 在实例化时注入。
|
||
"""
|
||
|
||
config: ScenarioConfig
|
||
|
||
def __init__(self, rows: list[dict] | None = None) -> None:
|
||
self._rows = rows or []
|
||
|
||
def with_rows(self, rows: list[dict]) -> ConfigScenario:
|
||
"""注入待扫描的数据行（链式返回自身）。"""
|
||
self._rows = rows
|
||
return self
|
||
|
||
def scan(self, session, *, data_version_id: uuid.UUID | None = None) -> ScanOutcome:
|
||
fn = get_primitive(self.config.detector)
|
||
if fn is None:
|
||
raise ValueError(f"未知检测原语: {self.config.detector}")
|
||
result = fn(self._rows, self.config.params)
|
||
drafts: list[ClueDraft] = []
|
||
if result.hit and result.score > 0:
|
||
try:
|
||
rationale = self.config.rationale_template.format(
|
||
summary=result.summary, **result.metrics
|
||
)
|
||
except (KeyError, IndexError, ValueError):
|
||
rationale = result.summary
|
||
drafts.append(
|
||
ClueDraft(
|
||
score=result.score,
|
||
rationale=rationale,
|
||
evidence=result.evidence,
|
||
subjects=result.subjects,
|
||
)
|
||
)
|
||
return ScanOutcome(scanned_count=len(self._rows), drafts=drafts)
|
||
|
||
|
||
def build_scenario_class(config: ScenarioConfig) -> type[ConfigScenario]:
|
||
"""按配置动态生成并注册一个 ConfigScenario 子类（类属性承载元数据）。"""
|
||
attrs = {
|
||
"code": config.code,
|
||
"title": config.title,
|
||
"risk_domain": config.risk_domain,
|
||
"label": config.label,
|
||
"score_threshold": config.threshold,
|
||
"rule_version": config.rule_version,
|
||
"config": config,
|
||
}
|
||
cls = type(f"ConfigScenario_{config.code}", (ConfigScenario,), attrs)
|
||
return register_scenario(cls)
|
||
|
||
|
||
def load_scenario_config(path: Path) -> ScenarioConfig:
|
||
"""读取并校验单个 YAML 场景配置文件。"""
|
||
raw = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
|
||
try:
|
||
config = ScenarioConfig(**raw)
|
||
except ValidationError as exc:
|
||
raise ValueError(f"场景配置 {path.name} 校验失败: {exc}") from exc
|
||
if get_primitive(config.detector) is None:
|
||
raise ValueError(
|
||
f"场景配置 {path.name} 引用了未注册的检测原语: {config.detector}"
|
||
)
|
||
return config
|
||
|
||
|
||
def load_yaml_scenarios(rules_dir: Path | str | None = None) -> list[type[ConfigScenario]]:
|
||
"""加载 rules 目录下全部 *.yaml 场景配置，动态生成并注册场景类。
|
||
|
||
新增 YAML 文件即被发现，无需改动代码。返回新注册的场景类列表（便于测试）。
|
||
"""
|
||
rules_dir = Path(rules_dir) if rules_dir else _DEFAULT_RULES_DIR
|
||
classes: list[type[ConfigScenario]] = []
|
||
if not rules_dir.exists():
|
||
return classes
|
||
for path in sorted(rules_dir.glob("*.yaml")):
|
||
config = load_scenario_config(path)
|
||
classes.append(build_scenario_class(config))
|
||
return classes
|