guanlan-src/app_ontology/app_ontology_guanlan/sop/scenarios.py

"""场景级注册表 纯函数 (① 体检 workflow 升格; 批量装配 SOP §5)。

消费 configs/scenario_registry.yaml。答: 给定商业场合(scenario) →
  出哪些报告章 (chapter_subset) · 必产哪些交付域 (required_domains) ·
  什么审级 (review_tier, 走 SOP §6.1 RV-1) · 什么裁决基调 (flavor)。
编号 namespace (SOP §4.-1): 域=交付九域 D1-D9; 章=报告7章; 审级 RV-1。
纯函数 (coverage 扫描只读文件, 无写副作用)。机器位 CLI = scripts/run_scenario.py。
"""

try:
    from app_common.app_common_guanlan.api import install_root as _install_root
except ImportError:                     # 理论不可达；包结构异常时回退到按位置上跳
    from pathlib import Path as _P
    def _install_root(_f): return _P(_f).resolve().parents[4]
from src import paths as P
import os
from pathlib import Path
import yaml

REPO = _install_root(__file__)
REGISTRY = P.config("scenario_registry.yaml")        # 配置唯一取用口 (src/paths.py)

# 交付九域 → 证据关键词 (扫 outputs/<farm>/sop/ 文件名判该域是否已产)。SOP §4.-1 crosswalk。
DOMAIN_EVIDENCE = {
    "D1": ["风资源", "风况", "wind_resource"],
    "D2": ["功率", "power", "mast", "偏航", "yaw", "变桨", "pitch", "回桨", "气动", "aero", "达成"],
    "D3": ["温度", "temp", "振动", "vib", "轴承", "低速轴", "主轴", "1P", "探头"],
    "D4": ["故障", "停机", "fault", "pareto", "帕累托", "事件"],
    "D5": ["可利用率", "可靠", "mtbf", "avail", "复发"],
    "D6": ["安全", "前兆", "safety", "安全链"],
    "D7": ["运维", "备件", "检修", "承包商"],
    "D8": ["寿命", "rul", "延寿", "役龄"],
    "D9": ["方案", "隐患清单", "隐患排查", "治理", "体检报告", "报告"],
    # 前置/横切 (非交付域, 但 workflow 前置门必查)
    "限电": ["限电", "curtail", "弃风"],
    "损失": ["损失", "电量", "waterfall", "瀑布"],
}

VALID_DOMAINS = {f"D{i}" for i in range(1, 10)}
TIER_ORDER = {"lite": 0, "双模型": 1, "AdvCI": 2}


def load_registry(path=REGISTRY):
    with open(path, encoding="utf-8") as f:
        return yaml.safe_load(f)


def resolve(scenario_id, reg=None):
    """按 id 或 alias 找场景 dict; 未知 → KeyError。"""
    reg = reg or load_registry()
    for s in reg["scenarios"]:
        if s["id"] == scenario_id or scenario_id in (s.get("aliases") or []):
            return s
    raise KeyError(f"未知场景 {scenario_id!r} (合法: {[s['id'] for s in reg['scenarios']]})")


def required_domains(scenario_id, reg=None):
    s = resolve(scenario_id, reg)
    return list(s.get("required_domains") or s.get("domain_weight", {}).get("主攻", []))


def chapter_subset(scenario_id, reg=None):
    return list(resolve(scenario_id, reg).get("chapter_subset") or [])


def review_tier(*, has_economic, has_confirmed, systemic, destination):
    """SOP §6.1 RV-1 审级门 (唯一正本逻辑; 场景 default 仅 doc 提示, 实际 tier 由此算)。
    destination ∈ {内部样板, 业主实名, 上云公开}。返回 (tier, 理由)。"""
    if destination == "上云公开" or has_economic or has_confirmed:
        why = []
        if destination == "上云公开": why.append("上云/公开")
        if has_economic: why.append("含经济数字¥")
        if has_confirmed: why.append("确诊异常")
        return "AdvCI", "RV-1: " + "∨".join(why) + " → AdvCI-PR (独立 C/D/R subagent)"
    if systemic:
        return "双模型", "RV-1: 跨机系统性(≥3台同机制) → 双模型审 (Gemini+DeepSeek)"
    return "lite", "RV-1: 健康清场/无确诊无¥ → lite (1 独立 subagent, 禁主 Claude 自审)"


def check_coverage(farm, scenario_id, reg=None, sop_dir=None):
    """扫 outputs/<farm>/sop/ 文件名, 判每个 required_domain 是否已产。
    返回 {domain: {"covered": bool, "evidence": [文件名...]}} + 前置门(限电/损失)。
    coverage = §0.8 required_outputs / 九域矩阵'缺一中止'的机器化 (缺 = 显式声明, 非静默跳过)。"""
    sop_dir = Path(sop_dir) if sop_dir else P.sop(farm)
    files = [p.name for p in sop_dir.iterdir()] if sop_dir.is_dir() else []

    def hits(keys):
        return [fn for fn in files if any(k.lower() in fn.lower() for k in keys)]

    req = required_domains(scenario_id, reg)
    out = {}
    for d in req + ["限电", "损失"]:
        ev = hits(DOMAIN_EVIDENCE.get(d, []))
        out[d] = {"covered": bool(ev), "evidence": ev}
    return out