From 5684df3cafbb706c6f38703009bf1336983af064 Mon Sep 17 00:00:00 2001 From: wuyang6 Date: Tue, 19 May 2026 11:59:49 +0800 Subject: [PATCH] Add intervention data skill --- skills/intervention-data/SKILL.md | 136 ++++++++ .../knowledge/type_match_agent.json | 211 +++++++++++ .../scripts/analyze_intervention_sources.py | 31 ++ .../scripts/audit_intervention_sources.py | 247 +++++++++++++ .../generate_intervention_candidate.py | 141 ++++++++ .../scripts/intervention_common.py | 330 ++++++++++++++++++ .../scripts/validate_intervention_records.py | 220 ++++++++++++ 7 files changed, 1316 insertions(+) create mode 100644 skills/intervention-data/SKILL.md create mode 100644 skills/intervention-data/knowledge/type_match_agent.json create mode 100644 skills/intervention-data/scripts/analyze_intervention_sources.py create mode 100644 skills/intervention-data/scripts/audit_intervention_sources.py create mode 100644 skills/intervention-data/scripts/generate_intervention_candidate.py create mode 100644 skills/intervention-data/scripts/intervention_common.py create mode 100644 skills/intervention-data/scripts/validate_intervention_records.py diff --git a/skills/intervention-data/SKILL.md b/skills/intervention-data/SKILL.md new file mode 100644 index 0000000..33eedf5 --- /dev/null +++ b/skills/intervention-data/SKILL.md @@ -0,0 +1,136 @@ +--- +name: intervention-data +description: 生成和校验单句精确干预、正则干预候选 TSV，复用标签大师的 target 语法校验，并从历史干预表推断 code 到下发 domain 的固定映射。 +when_to_use: 当用户希望新增、检查或整理干预规则，例如给某个 query 或多轮正则生成干预条目、确认设备和标签是否合法、推断下发 domain 时使用。 +aliases: intervention, rule-intervention, 干预数据 +allowed_tools: read_file, write_file, edit_file, grep_search, glob_search, ask_user_question, python_exec +--- + +使用这个 skill 处理“query/多轮表达 -> 干预 TSV 候选 -> 合法性校验”的任务。 + +干预数据目前有两类： + +- **单句精确干预**：一行四列，格式为 `设备\tquery\tcode\t下发domain`。 +- **正则干预**：一行三列，格式为 `设备\t正则\t标签`。 + +其中单句精确干预的 `code` 和正则干预的 `标签` 使用同一套 target 语法，和 `label-master` 的输出一致，但不包含复杂度判断。校验时必须复用 `skills/label-master/scripts/validate_label_output.py` 的逻辑。 + +## 关键约定 + +- `unify` 表示全部设备。 +- 设备必须优先从历史干预文件中归纳，不要随手发明设备名。 +- 单句精确干预如果用户没有提供 `下发domain`，必须从历史精确干预表中按 `code -> 下发domain` 的多数映射推断。 +- 正则干预支持多轮。多轮正则使用 `beforeQuery#上一轮#query#当前轮` 这种片段表达，最终通常包在 `^...$` 中。 +- 默认只生成候选 TSV 和校验结果，不直接追加到历史干预文件。只有用户明确要求“追加到某个文件末尾”或“修改某个文件”时，才使用通用文件编辑能力。 +- 所有产物优先写到当前会话的 `output/` 或 `scratchpad/` 下，不要写到项目根目录的 `output/`。 + +## 推荐流程 + +1. 先确认用户要哪种干预方式： + - 单句精确干预：query 必须完全命中。 + - 正则干预：适合一类表达、多轮上下文或需要泛化的高频 case。 +2. 确认设备。用户没说设备时，必须问设备；可以提示 `unify` 表示全部设备。 +3. 确认目标标签/code。它必须是完整 target，例如 `Agent(tag="地图导航")`、`QA()`、`Chat()` 或合法 function 调用。 +4. 如果是单句精确干预： + - 确认 `query`。 + - 如果没有 `下发domain`，调用脚本根据历史数据推断。 +5. 如果是正则干预： + - 确认当前轮 query。 + - 如果有多轮，确认前序 query 顺序。 + - 用户给了原始正则就保留；没给时由脚本生成 `^query#...$` 或 `^beforeQuery#...#query#...$`。 +6. 调用 `scripts/generate_intervention_candidate.py` 生成候选 TSV 和校验报告。 +7. 展示时只展示必要信息：干预类型、设备、候选 TSV、校验是否通过、需要用户确认的问题。 + +## 脚本能力 + +```text +skills/intervention-data/ + SKILL.md + knowledge/ + type_match_agent.json + scripts/ + intervention_common.py + analyze_intervention_sources.py + audit_intervention_sources.py + generate_intervention_candidate.py + validate_intervention_records.py +``` + +### analyze_intervention_sources.py + +读取历史干预文件，输出设备枚举、`code -> 下发domain` 多数映射、常见标签，以及 `knowledge/type_match_agent.json` 中的下发表。 + +示例： + +```bash +python skills/intervention-data/scripts/analyze_intervention_sources.py +``` + +### generate_intervention_candidate.py + +从 JSON 输入生成候选 TSV，并自动校验。输入可以来自文件或 stdin。 + +### audit_intervention_sources.py + +审计现有两个历史干预文件，只输出摘要，不修改文件。适合验证当前 skill 的校验规则和存量数据是否贴合。 + +示例： + +```bash +python skills/intervention-data/scripts/audit_intervention_sources.py +``` + +单句精确干预示例： + +```json +{ + "mode": "exact", + "device": "unify", + "query": "播放周杰伦的歌", + "target": "Agent(tag=\"音乐\")" +} +``` + +正则多轮干预示例： + +```json +{ + "mode": "regex", + "device": "miCar", + "before_queries": ["第一个"], + "query": "红绿灯少的路线", + "target": "Agent(tag=\"地图导航\")" +} +``` + +### validate_intervention_records.py + +校验候选 TSV。支持 `--mode exact` 或 `--mode regex`。 + +规则包括： + +- 列数正确。 +- 设备合法。 +- `code` / `标签` 优先通过 label-master 校验；如果 label-master 未收录但历史干预表中已经稳定出现，允许作为候选通过，并给出 warning 让用户确认知识库是否需要补齐。 +- 单句精确干预的 `下发domain` 与历史多数映射一致，默认不一致为 warning，`--strict-domain` 时为 error。 +- 单句精确干预还会用 `type_match_agent.json` 做配置校对；配置值可以是更宽泛前缀，例如 `音乐 -> contentCopilot` 可以兼容历史里的 `contentCopilot|music`。 +- 正则可编译，且包含 `query#` 片段。 +- 与历史文件重复或冲突时给出 warning/error。 + +## 展示格式 + +生成候选后，优先用这个格式回复： + +```text +我生成了一条候选干预，先不写入历史文件。 +- 类型：单句精确干预 / 正则干预 +- 设备：xxx +- 标签：xxx +- 校验：通过 / 有问题 + +候选 TSV：`设备 query code 下发domain` + +需要确认：是否写入某个文件，或是否调整设备/标签/正则。 +``` + +如果校验失败，先解释失败原因，不要要求用户直接落盘。 diff --git a/skills/intervention-data/knowledge/type_match_agent.json b/skills/intervention-data/knowledge/type_match_agent.json new file mode 100644 index 0000000..6e30224 --- /dev/null +++ b/skills/intervention-data/knowledge/type_match_agent.json @@ -0,0 +1,211 @@ +{ + "音乐": "contentCopilot", + "音乐#tvV2": "contentCopilot|video", + "音乐#miCar": "contentCopilot|station", + "音乐#screenSoundbox": "contentCopilot|music", + "音乐#glass": "contentCopilot|ancientPoem", + "电台": "contentCopilot", + "电台#tvV2": "contentCopilot|video", + "视频": "contentCopilot", + "古诗": "contentCopilot", + "歌单": "contentCopilot|controlCopilot|soundboxControl|smartApp", + "应用播放": "contentCopilot|controlCopilot|soundboxControl|smartApp", + "新闻": "contentCopilot", + "笑话": "contentCopilot", + "内容控制": "contentCopilot|soundboxControl|controlCopilot", + "内容控制|系统控制": "contentCopilot|soundboxControl|controlCopilot", + "内容控制|应用控制": "contentCopilot|smartApp|controlCopilot", + "播放状态查询": "contentCopilot|soundboxControl|controlCopilot", + "内容问答": "contentCopilot|qabot|dialogCopilot", + "内容问答#tvV2": "contentCopilot|qabot|dialogCopilot|video", + "内容问答#soundbox": "contentCopilot|qabot|dialogCopilot|michat", + "赛事播放": "contentCopilot|qabot|dialogCopilot|sports", + "赛事预约": "contentCopilot|qabot|dialogCopilot|sports", + "声音": "contentCopilot", + "人物": "dialogCopilot", + "人物#tvV2": "dialogCopilot|video", + "问答": "dialogCopilot|open-class|shopping|productAgent", + "问答#tvV2": "dialogCopilot|open-class|shopping|productAgent|video", + "百科": "dialogCopilot", + "百科#tvV2": "dialogCopilot|video", + "医疗": "dialogCopilot", + "星座": "dialogCopilot", + "菜谱": "dialogCopilot", + "民俗": "dialogCopilot", + "运动": "contentCopilot|dialogCopilot|sports|qabot", + "体育赛事问答": "contentCopilot|dialogCopilot|sports|qabot", + "词典": "dialogCopilot", + "彩票": "dialogCopilot", + "投资": "dialogCopilot", + "股票": "dialogCopilot", + "闲聊": "dialogCopilot|feedback|scenes|fitnessHealth", + "闲聊#tvV2": "dialogCopilot|video|feedback|scenes|help|fitnessHealth|shopping|productAgent", + "用户反馈": "dialogCopilot|feedback", + "角色扮演": "dialogCopilot", + "重说": "dialogCopilot", + "图片问答": "dialogCopilot|imageLU|mapCopilot", + "屏幕问答": "dialogCopilot|imageLU|mapCopilot|smartApp|controlCopilot|productAgent|auto|lifeCopilot|lifeAgent", + "拍照问答": "dialogCopilot|imageLU|mapCopilot", + "搜索": "dialogCopilot|search", + "搜索#tvV2": "dialogCopilot|search|video|contentCopilot", + "搜索#soundbox": "dialogCopilot|search|music|contentCopilot", + "健康监控#glass": "dialogCopilot|toolsCopilot", + "地图导航": "mapCopilot|lifeCopilot|lifeAgent", + "地图导航#tvV2": "mapCopilot|lifeCopilot|lifeAgent|mapApp", + "地图问答": "mapCopilot|qabot|dialogCopilot", + "走哪问哪": "mapCopilot|qabot|dialogCopilot", + "地图设置": "mapCopilot|soundboxControl|controlCopilot|toolsCopilot|todolist", + "违章查询": "mapCopilot", + "限号": "mapCopilot", + "地址设置": "toolsCopilot|mapCopilot|mapApp", + "图片": "aicreativeCopilot|dialogCopilot|qabot", + "图像创作": "aicreativeCopilot|dialogCopilot|qabot", + "视频创作": "aicreativeCopilot|dialogCopilot|qabot", + "作文": "aicreativeCopilot", + "文本创作": "aicreativeCopilot|dialogCopilot|toolsCopilot", + "代码创作": "aicreativeCopilot", + "图像编辑": "aicreativeCopilot", + "时间": "toolsCopilot", + "天气": "toolsCopilot|dialogCopilot|qabot", + "数学": "toolsCopilot", + "数学应用题": "toolsCopilot", + "电话": "toolsCopilot", + "看图打电话": "toolsCopilot", + "翻译": "toolsCopilot", + "翻译#glass": "toolsCopilot|dialogCopilot|qabot", + "闹钟": "toolsCopilot", + "备忘录": "toolsCopilot|dialogCopilot|michat", + "记忆": "toolsCopilot", + "结构记忆": "toolsCopilot|dialogCopilot|michat", + "视频收藏": "toolsCopilot|controlCopilot|smartApp", + "电话问答": "toolsCopilot|dialogCopilot|qabot", + "课程表": "toolsCopilot", + "内容总结": "toolsCopilot|aicreativeCopilot|dialogCopilot|qabot", + "留言": "toolsCopilot|message", + "健康监控": "toolsCopilot|fitnessHealth|todolist|dialogCopilot|qabot", + "健康控制": "toolsCopilot|fitnessHealth|todolist|controlCopilot|soundboxControl|smartApp", + "声纹": "controlCopilot|voiceprint", + "家庭传声": "controlCopilot|iotCopilot|smartMiot|dialogCopilot|qabot|michat|contentCopilot|music|station", + "系统控制": "controlCopilot|iotCopilot|productAgent", + "萌宠控制": "controlCopilot|iotCopilot|productAgent", + "屏幕操作": "controlCopilot|iotCopilot|productAgent", + "音色切换": "controlCopilot|iotCopilot|productAgent", + "车载控制": "controlCopilot|iotCopilot|productAgent", + "车载设备状态查询": "controlCopilot|iotCopilot|productAgent", + "系统查询": "controlCopilot|iotCopilot|productAgent", + "设备控制": "controlCopilot|iotCopilot", + "家用设备状态查询": "controlCopilot|iotCopilot|productAgent", + "应用控制": "controlCopilot|iotCopilot|lifeCopilot|lifeAgent|fitnessHealth", + "生活缴费": "controlCopilot", + "应用窗口控制": "controlCopilot", + "应用控制#tvV2": "controlCopilot|iotCopilot|lifeCopilot|lifeAgent|fitnessHealth|video|soundboxControl|phonecall", + "应用控制#soundbox": "controlCopilot|contentCopilot|music|station|dialogCopilot|michat|qabot|video|fitnessHealth|lifeCopilot", + "应用控制搜索": "controlCopilot|smartApp", + "浏览器搜索": "controlCopilot|search|smartApp", + "相机": "controlCopilot", + "设备查找": "controlCopilot|smartMiot|michat", + "系统定时控制": "controlCopilot|iotCopilot|productAgent", + "设备定时控制": "controlCopilot|iotCopilot", + "应用定时控制": "controlCopilot|iotCopilot|lifeCopilot|lifeAgent|fitnessHealth", + "帮助": "productAgent|soundboxControl|controlCopilot", + "汽车手册": "productAgent|dialogCopilot|qabot", + "客服问答": "productAgent|soundboxControl|controlCopilot", + "购物": "productAgent|soundboxControl|controlCopilot", + "产品问答": "productAgent", + "手车互联": "productAgent|soundboxControl|controlCopilot", + "产品推荐": "productAgent", + "产品购买": "productAgent", + "外卖": "lifeCopilot", + "外卖|应用": "lifeCopilot|controlCopilot|smartApp", + "美食": "lifeCopilot", + "美食|应用": "lifeCopilot|controlCopilot|smartApp", + "景点": "lifeCopilot", + "景点|应用": "lifeCopilot|controlCopilot|smartApp", + "购票": "lifeCopilot", + "购票|应用": "lifeCopilot|controlCopilot|smartApp", + "电影": "lifeCopilot", + "电影|应用": "lifeCopilot|controlCopilot|smartApp", + "酒店": "lifeCopilot", + "酒店|应用": "lifeCopilot|controlCopilot|smartApp", + "生活休闲": "lifeCopilot", + "车服": "lifeCopilot", + "汽车服务": "lifeCopilot", + "团购": "lifeCopilot", + "打车": "lifeCopilot", + "航班查询": "lifeCopilot", + "通讯服务": "lifeCopilot", + "快递服务": "lifeCopilot|productAgent|shopping", + "产品推荐|参数对比": "lifeCopilot|productAgent|shopping", + "产品推荐|信息对比": "lifeCopilot|productAgent|shopping", + "平台比价": "productAgent|shopping", + "找同款": "lifeCopilot|productAgent|shopping", + "订单查询": "lifeCopilot|productAgent", + "订座排号": "lifeCopilot", + "前车识别": "dialogCopilot|imageLU|qabot", + "频道": "contentCopilot", + "频道#tvV2": "contentCopilot|controlCopilot|smartMiot|iotCopilot", + "频道#soundbox": "controlCopilot|smartMiot|iotCopilot", + "地图|搜索": "mapCopilot|controlCopilot|smartApp", + "生活服务|搜索": "lifeCopilot|lifeAgent|controlCopilot|smartApp", + "应用|购买": "controlCopilot|smartApp", + "地图控制": "mapCopilot|soundboxControl|controlCopilot|toolsCopilot|todolist", + "限行": "mapCopilot", + "音乐播放": "contentCopilot", + "音乐播放#tvV2": "contentCopilot|video", + "音乐播放#miCar": "contentCopilot|station", + "音乐播放#screenSoundbox": "contentCopilot|music", + "音乐播放#glass": "contentCopilot|ancientPoem", + "电台播放": "contentCopilot", + "电台播放#tvV2": "contentCopilot|video", + "视频播放": "contentCopilot", + "电视频道": "contentCopilot", + "电视频道#tvV2": "contentCopilot|controlCopilot|smartMiot|iotCopilot", + "电视频道#soundbox": "controlCopilot|smartMiot|iotCopilot", + "媒体资源问答": "contentCopilot|qabot|dialogCopilot", + "媒体资源问答#tvV2": "contentCopilot|qabot|dialogCopilot|video", + "媒体资源问答#soundbox": "contentCopilot|qabot|dialogCopilot|michat", + "媒体应用播放": "contentCopilot|controlCopilot|soundboxControl|smartApp", + "媒体资源切换": "contentCopilot|soundboxControl|controlCopilot", + "播放器控制": "contentCopilot|soundboxControl|controlCopilot", + "体育赛事播放": "contentCopilot|qabot|dialogCopilot|sports", + "体育赛事预约": "contentCopilot|qabot|dialogCopilot|sports", + "古诗词播放": "contentCopilot", + "古诗词问答": "contentCopilot", + "讲笑话": "contentCopilot", + "声音博物馆": "contentCopilot", + "闹钟计时器": "toolsCopilot", + "简单数学问题": "toolsCopilot", + "打电话": "toolsCopilot", + "发短信": "toolsCopilot", + "通讯录": "toolsCopilot", + "电话号码查询": "toolsCopilot|dialogCopilot|qabot", + "外语翻译": "toolsCopilot", + "翻译控制": "toolsCopilot", + "外语词典查询": "toolsCopilot", + "外语问答": "toolsCopilot", + "外语翻译#glass": "toolsCopilot|dialogCopilot|qabot", + "外语问答#glass": "toolsCopilot|dialogCopilot|qabot", + "外语词典查询#glass": "toolsCopilot|dialogCopilot|qabot", + "翻译控制#glass": "toolsCopilot|dialogCopilot|qabot", + "提醒": "toolsCopilot|dialogCopilot|michat", + "信息记忆": "toolsCopilot", + "记忆查询": "toolsCopilot|dialogCopilot|michat", + "个人信息": "toolsCopilot|dialogCopilot|michat", + "声纹设置": "controlCopilot|voiceprint", + "相机控制": "controlCopilot", + "系统状态查询": "controlCopilot|iotCopilot|productAgent", + "小爱帮助": "productAgent|soundboxControl|controlCopilot", + "小米产品帮助": "productAgent", + "商品购买": "productAgent", + "餐饮服务": "lifeCopilot", + "餐饮服务|应用": "lifeCopilot|controlCopilot|smartApp", + "旅游": "lifeCopilot", + "生活服务": "lifeCopilot", + "交通购票": "lifeCopilot", + "电影票购买": "lifeCopilot", + "电影票购买|应用": "lifeCopilot|controlCopilot|smartApp", + "航班信息查询": "lifeCopilot", + "话费流量服务": "lifeCopilot", + "购物订单查询": "lifeCopilot|productAgent", + "餐厅订座排号": "lifeCopilot" +} diff --git a/skills/intervention-data/scripts/analyze_intervention_sources.py b/skills/intervention-data/scripts/analyze_intervention_sources.py new file mode 100644 index 0000000..c5653ae --- /dev/null +++ b/skills/intervention-data/scripts/analyze_intervention_sources.py @@ -0,0 +1,31 @@ +#!/usr/bin/env python3 +"""分析历史干预文件，归纳设备枚举和 code 到下发 domain 的映射。""" + +from __future__ import annotations + +import argparse +from pathlib import Path + +from intervention_common import DEFAULT_EXACT_SOURCE, DEFAULT_REGEX_SOURCE, build_source_summary, json_dumps + + +def main() -> int: + parser = argparse.ArgumentParser(description="分析历史干预 TSV") + parser.add_argument("--exact-source", default=str(DEFAULT_EXACT_SOURCE), help="单句精确干预 TSV 路径") + parser.add_argument("--regex-source", default=str(DEFAULT_REGEX_SOURCE), help="正则干预 TSV 路径") + parser.add_argument("--top-mapping", type=int, default=30, help="输出前 N 个 code-domain 映射") + args = parser.parse_args() + + summary = build_source_summary(Path(args.exact_source), Path(args.regex_source)) + mapping_items = sorted( + summary["code_domain_mapping"].items(), + key=lambda item: item[1]["count"], + reverse=True, + ) + summary["code_domain_mapping"] = dict(mapping_items[: args.top_mapping]) + print(json_dumps(summary)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/skills/intervention-data/scripts/audit_intervention_sources.py b/skills/intervention-data/scripts/audit_intervention_sources.py new file mode 100644 index 0000000..67bcdd8 --- /dev/null +++ b/skills/intervention-data/scripts/audit_intervention_sources.py @@ -0,0 +1,247 @@ +#!/usr/bin/env python3 +"""审计现有干预文件和当前规则的贴合度。 + +这个脚本用于回答两个问题： +1. 现有单句精确干预/正则干预文件自身是否存在明显格式问题。 +2. 当前 label-master 校验、type_match_agent 下发表和历史数据是否匹配。 + +它不会修改任何文件，只输出 JSON 摘要和少量样例。 +""" + +from __future__ import annotations + +import argparse +import re +from collections import Counter, defaultdict +from pathlib import Path +from typing import Any + +from intervention_common import ( + DEFAULT_EXACT_SOURCE, + DEFAULT_REGEX_SOURCE, + build_source_summary, + domain_compatible, + infer_configured_dispatch_domain, + json_dumps, + load_exact_records, + load_label_validator, + load_regex_records, +) + + +def sample_append(bucket: list[dict[str, Any]], item: dict[str, Any], limit: int) -> None: + if len(bucket) < limit: + bucket.append(item) + + +def validate_unique_targets(targets: set[str]) -> dict[str, dict[str, Any]]: + validator = load_label_validator() + return {target: validator.validate(target) for target in sorted(targets)} + + +def audit_exact_records(summary: dict[str, Any], target_results: dict[str, dict[str, Any]], sample_limit: int) -> dict[str, Any]: + records = load_exact_records(Path(summary["source_files"]["exact"])) + empty_fields = Counter() + target_invalid = Counter() + configured_mismatch = Counter() + duplicate_keys: dict[tuple[str, str], list[Any]] = defaultdict(list) + samples = { + "empty_fields": [], + "invalid_targets": [], + "configured_domain_mismatch": [], + "conflicts": [], + } + + for record in records: + if not record.device: + empty_fields["device"] += 1 + if not record.query: + empty_fields["query"] += 1 + if not record.target: + empty_fields["target"] += 1 + if not record.dispatch_domain: + empty_fields["dispatch_domain"] += 1 + if any(not value for value in (record.device, record.query, record.target, record.dispatch_domain)): + sample_append(samples["empty_fields"], record.__dict__, sample_limit) + + result = target_results.get(record.target) + if result and not result.get("valid"): + target_invalid[record.target] += 1 + sample_append( + samples["invalid_targets"], + { + "line_no": record.line_no, + "target": record.target, + "errors": result.get("errors", []), + }, + sample_limit, + ) + + configured = infer_configured_dispatch_domain(record.target, record.device, summary) + if configured and not domain_compatible(record.dispatch_domain, configured["dispatch_domain"]): + configured_mismatch[configured["matched_key"]] += 1 + sample_append( + samples["configured_domain_mismatch"], + { + "line_no": record.line_no, + "device": record.device, + "query": record.query, + "target": record.target, + "dispatch_domain": record.dispatch_domain, + "configured": configured, + }, + sample_limit, + ) + + duplicate_keys[(record.device, record.query)].append(record) + + conflicts = [] + duplicate_count = 0 + for (device, query), items in duplicate_keys.items(): + if len(items) <= 1: + continue + duplicate_count += len(items) + variants = {(item.target, item.dispatch_domain) for item in items} + if len(variants) > 1: + conflict = { + "device": device, + "query": query, + "lines": [item.line_no for item in items[:10]], + "variants": sorted([{"target": target, "dispatch_domain": domain} for target, domain in variants], key=str), + } + conflicts.append(conflict) + sample_append(samples["conflicts"], conflict, sample_limit) + + return { + "total": len(records), + "empty_fields": dict(empty_fields), + "invalid_target_total": sum(target_invalid.values()), + "invalid_target_top": dict(target_invalid.most_common(20)), + "configured_domain_mismatch_total": sum(configured_mismatch.values()), + "configured_domain_mismatch_top": dict(configured_mismatch.most_common(20)), + "duplicate_row_count_by_device_query": duplicate_count, + "conflict_key_count": len(conflicts), + "samples": samples, + } + + +def audit_regex_records(target_results: dict[str, dict[str, Any]], summary: dict[str, Any], sample_limit: int) -> dict[str, Any]: + records = load_regex_records(Path(summary["source_files"]["regex"])) + empty_fields = Counter() + regex_errors = Counter() + anchor_warnings = 0 + missing_query_fragment = 0 + target_invalid = Counter() + duplicate_keys: dict[tuple[str, str], list[Any]] = defaultdict(list) + samples = { + "empty_fields": [], + "regex_errors": [], + "missing_query_fragment": [], + "invalid_targets": [], + "conflicts": [], + } + + for record in records: + if not record.device: + empty_fields["device"] += 1 + if not record.pattern: + empty_fields["pattern"] += 1 + if not record.target: + empty_fields["target"] += 1 + if any(not value for value in (record.device, record.pattern, record.target)): + sample_append(samples["empty_fields"], record.__dict__, sample_limit) + + if record.pattern: + try: + re.compile(record.pattern) + except re.error as exc: + regex_errors[str(exc)] += 1 + sample_append( + samples["regex_errors"], + {"line_no": record.line_no, "pattern": record.pattern, "error": str(exc)}, + sample_limit, + ) + if "query#" not in record.pattern: + missing_query_fragment += 1 + sample_append(samples["missing_query_fragment"], record.__dict__, sample_limit) + if not record.pattern.startswith("^") or not record.pattern.endswith("$"): + anchor_warnings += 1 + + result = target_results.get(record.target) + if result and not result.get("valid"): + target_invalid[record.target] += 1 + sample_append( + samples["invalid_targets"], + { + "line_no": record.line_no, + "target": record.target, + "errors": result.get("errors", []), + }, + sample_limit, + ) + + duplicate_keys[(record.device, record.pattern)].append(record) + + conflicts = [] + duplicate_count = 0 + for (device, pattern), items in duplicate_keys.items(): + if len(items) <= 1: + continue + duplicate_count += len(items) + variants = {item.target for item in items} + if len(variants) > 1: + conflict = { + "device": device, + "pattern": pattern, + "lines": [item.line_no for item in items[:10]], + "variants": sorted(variants), + } + conflicts.append(conflict) + sample_append(samples["conflicts"], conflict, sample_limit) + + return { + "total": len(records), + "empty_fields": dict(empty_fields), + "regex_error_total": sum(regex_errors.values()), + "regex_error_top": dict(regex_errors.most_common(20)), + "missing_query_fragment": missing_query_fragment, + "anchor_warning_count": anchor_warnings, + "invalid_target_total": sum(target_invalid.values()), + "invalid_target_top": dict(target_invalid.most_common(20)), + "duplicate_row_count_by_device_pattern": duplicate_count, + "conflict_key_count": len(conflicts), + "samples": samples, + } + + +def main() -> int: + parser = argparse.ArgumentParser(description="审计现有干预文件") + parser.add_argument("--exact-source", default=str(DEFAULT_EXACT_SOURCE), help="单句精确干预 TSV 路径") + parser.add_argument("--regex-source", default=str(DEFAULT_REGEX_SOURCE), help="正则干预 TSV 路径") + parser.add_argument("--sample-limit", type=int, default=5, help="每类问题最多输出样例数") + args = parser.parse_args() + + summary = build_source_summary(Path(args.exact_source), Path(args.regex_source)) + exact_records = load_exact_records(Path(args.exact_source)) + regex_records = load_regex_records(Path(args.regex_source)) + target_results = validate_unique_targets({record.target for record in exact_records + regex_records if record.target}) + + payload = { + "source_files": summary["source_files"], + "devices": summary["devices"], + "counts": summary["counts"], + "target_validation": { + "unique_targets": len(target_results), + "valid_unique_targets": sum(1 for item in target_results.values() if item.get("valid")), + "invalid_unique_targets": sum(1 for item in target_results.values() if not item.get("valid")), + }, + "exact": audit_exact_records(summary, target_results, args.sample_limit), + "regex": audit_regex_records(target_results, summary, args.sample_limit), + } + print(json_dumps(payload)) + has_hard_errors = bool(payload["exact"]["empty_fields"] or payload["regex"]["regex_error_total"]) + return 1 if has_hard_errors else 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/skills/intervention-data/scripts/generate_intervention_candidate.py b/skills/intervention-data/scripts/generate_intervention_candidate.py new file mode 100644 index 0000000..9d2cae7 --- /dev/null +++ b/skills/intervention-data/scripts/generate_intervention_candidate.py @@ -0,0 +1,141 @@ +#!/usr/bin/env python3 +"""生成单句精确干预或正则干预候选 TSV，并返回校验结果。""" + +from __future__ import annotations + +import argparse +import json +import subprocess +import sys +from pathlib import Path +from typing import Any + +from intervention_common import ( + DEFAULT_EXACT_SOURCE, + DEFAULT_REGEX_SOURCE, + build_query_pattern, + build_source_summary, + exact_tsv_line, + infer_configured_dispatch_domain, + infer_dispatch_domain_for_device, + json_dumps, + normalize_device, + normalize_target, + regex_tsv_line, +) + + +SCRIPT_DIR = Path(__file__).resolve().parent +VALIDATE_SCRIPT = SCRIPT_DIR / "validate_intervention_records.py" + + +def load_payload(path: str | None) -> dict[str, Any]: + text = Path(path).read_text(encoding="utf-8") if path else sys.stdin.read() + if not text.strip(): + raise ValueError("输入 JSON 不能为空") + return json.loads(text) + + +def run_validator(mode: str, tsv: str, exact_source: str, regex_source: str) -> dict[str, Any]: + proc = subprocess.run( + [ + sys.executable, + str(VALIDATE_SCRIPT), + "--mode", + mode, + "--exact-source", + exact_source, + "--regex-source", + regex_source, + ], + input=tsv, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + check=False, + ) + if not proc.stdout.strip(): + return { + "valid": False, + "errors": [proc.stderr.strip() or f"校验脚本退出：{proc.returncode}"], + } + return json.loads(proc.stdout) + + +def generate_exact(payload: dict[str, Any], exact_source: str, regex_source: str) -> dict[str, Any]: + device = normalize_device(str(payload.get("device", ""))) + query = str(payload.get("query", "")).strip() + target = normalize_target(str(payload.get("target") or payload.get("code") or "")) + dispatch_domain = str(payload.get("dispatch_domain") or payload.get("domain") or "").strip() + + summary = build_source_summary(Path(exact_source), Path(regex_source)) + inferred = None + configured = infer_configured_dispatch_domain(target, device, summary) + if not dispatch_domain: + inferred = infer_dispatch_domain_for_device(target, device, summary) + if inferred: + dispatch_domain = inferred["dispatch_domain"] + + tsv = exact_tsv_line(device, query, target, dispatch_domain) + validation = run_validator("exact", tsv, exact_source, regex_source) + return { + "mode": "exact", + "tsv": tsv, + "record": { + "device": device, + "query": query, + "target": target, + "dispatch_domain": dispatch_domain, + }, + "inferred_dispatch_domain": inferred, + "configured_dispatch_domain": configured, + "validation": validation, + } + + +def generate_regex(payload: dict[str, Any], exact_source: str, regex_source: str) -> dict[str, Any]: + device = normalize_device(str(payload.get("device", ""))) + target = normalize_target(str(payload.get("target") or payload.get("label") or "")) + raw_pattern = str(payload.get("pattern") or payload.get("regex") or "").strip() + query = str(payload.get("query") or payload.get("current_query") or "").strip() + before_queries = payload.get("before_queries") or [] + if not isinstance(before_queries, list): + raise ValueError("before_queries 必须是字符串数组") + pattern = raw_pattern or build_query_pattern(query, [str(item) for item in before_queries]) + + tsv = regex_tsv_line(device, pattern, target) + validation = run_validator("regex", tsv, exact_source, regex_source) + return { + "mode": "regex", + "tsv": tsv, + "record": { + "device": device, + "pattern": pattern, + "target": target, + }, + "validation": validation, + } + + +def main() -> int: + parser = argparse.ArgumentParser(description="生成干预候选 TSV") + parser.add_argument("--input", help="输入 JSON 文件；不传则读取 stdin") + parser.add_argument("--exact-source", default=str(DEFAULT_EXACT_SOURCE), help="历史单句精确干预 TSV") + parser.add_argument("--regex-source", default=str(DEFAULT_REGEX_SOURCE), help="历史正则干预 TSV") + args = parser.parse_args() + + payload = load_payload(args.input) + mode = str(payload.get("mode", "")).strip().lower() + if mode == "exact": + result = generate_exact(payload, args.exact_source, args.regex_source) + elif mode == "regex": + result = generate_regex(payload, args.exact_source, args.regex_source) + else: + raise ValueError("mode 必须是 exact 或 regex") + + print(json_dumps(result)) + return 0 if result.get("validation", {}).get("valid") else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/skills/intervention-data/scripts/intervention_common.py b/skills/intervention-data/scripts/intervention_common.py new file mode 100644 index 0000000..38af201 --- /dev/null +++ b/skills/intervention-data/scripts/intervention_common.py @@ -0,0 +1,330 @@ +#!/usr/bin/env python3 +"""干预数据脚本的公共能力。 + +这里尽量只做确定性处理： +- 读取历史干预 TSV。 +- 归纳设备枚举和 code 到下发 domain 的多数映射。 +- 复用 label-master 校验 target 语法。 +- 校验候选干预条目是否和历史数据重复或冲突。 +""" + +from __future__ import annotations + +import csv +import importlib.util +import json +import re +import sys +from collections import Counter, defaultdict +from dataclasses import dataclass +from pathlib import Path +from typing import Any + + +SCRIPT_DIR = Path(__file__).resolve().parent +SKILL_ROOT = SCRIPT_DIR.parent +REPO_ROOT = SKILL_ROOT.parent.parent +DEFAULT_EXACT_SOURCE = REPO_ROOT / "干预skill" / "20260134" +DEFAULT_REGEX_SOURCE = REPO_ROOT / "干预skill" / "20260207" +LABEL_VALIDATOR_PATH = REPO_ROOT / "skills" / "label-master" / "scripts" / "validate_label_output.py" +CONFIG_MAPPING_PATH = SKILL_ROOT / "knowledge" / "type_match_agent.json" +AGENT_TARGET_RE = re.compile(r'^Agent\(\s*tag\s*=\s*["\'](.+?)["\']\s*\)$') + + +@dataclass(frozen=True) +class ExactRecord: + device: str + query: str + target: str + dispatch_domain: str + line_no: int + + +@dataclass(frozen=True) +class RegexRecord: + device: str + pattern: str + target: str + line_no: int + + +def normalize_device(device: str) -> str: + """归一化历史数据里的设备写法，但保留常见大小写。""" + + value = device.strip() + if value.startswith("#"): + value = value[1:].strip() + aliases = { + "micar": "miCar", + "miai": "miai", + "tVV2": "tvV2", + "tvv2": "tvV2", + "unify": "unify", + } + return aliases.get(value, value) + + +def normalize_target(target: str) -> str: + return target.replace("“", '"').replace("”", '"').replace("‘", "'").replace("’", "'").strip() + + +def read_tsv_rows(path: Path) -> list[list[str]]: + with path.open("r", encoding="utf-8-sig", newline="") as handle: + return list(csv.reader(handle, delimiter="\t")) + + +def load_exact_records(path: Path = DEFAULT_EXACT_SOURCE) -> list[ExactRecord]: + rows = read_tsv_rows(path) + records: list[ExactRecord] = [] + for index, row in enumerate(rows, start=1): + if index == 1 and row[:4] == ["#设备名", "query", "code", "下发domain"]: + continue + if not row or all(not cell.strip() for cell in row): + continue + padded = [*row, "", "", "", ""] + records.append( + ExactRecord( + device=normalize_device(padded[0]), + query=padded[1].strip(), + target=normalize_target(padded[2]), + dispatch_domain=padded[3].strip(), + line_no=index, + ) + ) + return records + + +def load_regex_records(path: Path = DEFAULT_REGEX_SOURCE) -> list[RegexRecord]: + rows = read_tsv_rows(path) + records: list[RegexRecord] = [] + for index, row in enumerate(rows, start=1): + if not row or all(not cell.strip() for cell in row): + continue + padded = [*row, "", ""] + records.append( + RegexRecord( + device=normalize_device(padded[0]), + pattern=padded[1].strip(), + target=normalize_target(padded[2]), + line_no=index, + ) + ) + return records + + +def build_source_summary( + exact_path: Path = DEFAULT_EXACT_SOURCE, + regex_path: Path = DEFAULT_REGEX_SOURCE, +) -> dict[str, Any]: + exact_records = load_exact_records(exact_path) if exact_path.exists() else [] + regex_records = load_regex_records(regex_path) if regex_path.exists() else [] + + exact_devices = Counter(record.device for record in exact_records) + regex_devices = Counter(record.device for record in regex_records) + devices = sorted(set(exact_devices) | set(regex_devices)) + + code_domain_counts: dict[str, Counter[str]] = defaultdict(Counter) + for record in exact_records: + if record.target and record.dispatch_domain: + code_domain_counts[record.target][record.dispatch_domain] += 1 + + code_domain_mapping: dict[str, dict[str, Any]] = {} + for target, counts in code_domain_counts.items(): + domain, count = counts.most_common(1)[0] + total = sum(counts.values()) + code_domain_mapping[target] = { + "dispatch_domain": domain, + "count": count, + "total": total, + "confidence": round(count / total, 4) if total else 0, + "alternatives": [ + {"dispatch_domain": item_domain, "count": item_count} + for item_domain, item_count in counts.most_common(5) + ], + } + + config_mapping = load_type_match_agent_mapping() + + return { + "source_files": { + "exact": str(exact_path), + "regex": str(regex_path), + }, + "counts": { + "exact_records": len(exact_records), + "regex_records": len(regex_records), + }, + "devices": devices, + "device_counts": { + "exact": dict(exact_devices.most_common()), + "regex": dict(regex_devices.most_common()), + }, + "code_domain_mapping": code_domain_mapping, + "configured_code_domain_mapping": config_mapping, + "historical_targets": sorted({record.target for record in exact_records + regex_records if record.target}), + "top_exact_targets": dict(Counter(record.target for record in exact_records).most_common(50)), + "top_regex_targets": dict(Counter(record.target for record in regex_records).most_common(50)), + } + + +def historical_targets(summary: dict[str, Any]) -> set[str]: + targets = set(summary.get("top_exact_targets", {})) | set(summary.get("top_regex_targets", {})) + targets.update(summary.get("code_domain_mapping", {})) + targets.update(summary.get("historical_targets", [])) + return {normalize_target(target) for target in targets if target} + + +def infer_dispatch_domain(target: str, summary: dict[str, Any]) -> dict[str, Any] | None: + normalized = normalize_target(target) + historical = summary.get("code_domain_mapping", {}).get(normalized) + if historical: + return {**historical, "source": "historical"} + configured = infer_configured_dispatch_domain(normalized, "unify", summary) + if configured: + return { + "dispatch_domain": configured["dispatch_domain"], + "count": 0, + "total": 0, + "confidence": None, + "alternatives": [], + "source": "type_match_agent", + "matched_key": configured["matched_key"], + } + return None + + +def infer_dispatch_domain_for_device(target: str, device: str, summary: dict[str, Any]) -> dict[str, Any] | None: + """按设备推断推荐下发 domain。 + + 优先级： + 1. `标签#设备` 这种明确配置。 + 2. 历史精确干预表里的全局多数映射。 + 3. `标签` 通用配置。 + """ + + normalized = normalize_target(target) + configured = infer_configured_dispatch_domain(normalized, device, summary) + if configured and "#" in configured["matched_key"]: + return { + "dispatch_domain": configured["dispatch_domain"], + "count": 0, + "total": 0, + "confidence": None, + "alternatives": [], + "source": "type_match_agent", + "matched_key": configured["matched_key"], + } + + historical = summary.get("code_domain_mapping", {}).get(normalized) + if historical: + return {**historical, "source": "historical"} + + if configured: + return { + "dispatch_domain": configured["dispatch_domain"], + "count": 0, + "total": 0, + "confidence": None, + "alternatives": [], + "source": "type_match_agent", + "matched_key": configured["matched_key"], + } + return None + + +def load_type_match_agent_mapping(path: Path = CONFIG_MAPPING_PATH) -> dict[str, str]: + if not path.exists(): + return {} + data = json.loads(path.read_text(encoding="utf-8")) + return {str(key): str(value) for key, value in data.items()} + + +def extract_agent_tag(target: str) -> str | None: + match = AGENT_TARGET_RE.match(normalize_target(target)) + if not match: + return None + return match.group(1) + + +def infer_configured_dispatch_domain(target: str, device: str, summary: dict[str, Any]) -> dict[str, str] | None: + tag = extract_agent_tag(target) + if not tag: + return None + mapping = summary.get("configured_code_domain_mapping", {}) + normalized_device = normalize_device(device) + for key in (f"{tag}#{normalized_device}", tag): + if key in mapping: + return {"matched_key": key, "dispatch_domain": mapping[key]} + return None + + +def split_domain(domain: str) -> list[str]: + return [part for part in domain.split("|") if part] + + +def domain_compatible(actual: str, expected: str) -> bool: + """判断实际下发 domain 是否和配置基线兼容。 + + 下发表里有些是宽泛入口，例如 `音乐 -> contentCopilot`； + 历史干预里常见更细路径，例如 `contentCopilot|music`。 + 只要二者一方的管道 token 是另一方的子集，就认为兼容。 + """ + + actual_parts = set(split_domain(actual)) + expected_parts = set(split_domain(expected)) + if not actual_parts or not expected_parts: + return False + return actual_parts.issubset(expected_parts) or expected_parts.issubset(actual_parts) + + +def load_label_validator() -> Any: + spec = importlib.util.spec_from_file_location("label_master_validate_label_output", LABEL_VALIDATOR_PATH) + if spec is None or spec.loader is None: + raise RuntimeError(f"无法加载 label-master 校验器：{LABEL_VALIDATOR_PATH}") + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + manifest_path = module.DEFAULT_OUTPUT + validator = module.LabelOutputValidator(module.load_manifest(manifest_path)) + return validator + + +def validate_target(target: str) -> dict[str, Any]: + validator = load_label_validator() + # 历史 TSV 中多行 function/Agent 目标常以字面量 `\n` 存在， + # label-master 校验器需要真实换行才能按多行程序解析。 + return validator.validate(normalize_target(target).replace("\\n", "\n")) + + +def escape_regex_literal(text: str) -> str: + return re.escape(text.strip()) + + +def build_query_pattern(query: str, before_queries: list[str] | None = None) -> str: + before_queries = before_queries or [] + parts: list[str] = [] + for before_query in before_queries: + if before_query.strip(): + parts.append(f"beforeQuery#{escape_regex_literal(before_query)}") + parts.append(f"query#{escape_regex_literal(query)}") + return "^" + "#".join(parts) + "$" + + +def exact_tsv_line(device: str, query: str, target: str, dispatch_domain: str) -> str: + return "\t".join([normalize_device(device), query.strip(), normalize_target(target), dispatch_domain.strip()]) + + +def regex_tsv_line(device: str, pattern: str, target: str) -> str: + return "\t".join([normalize_device(device), pattern.strip(), normalize_target(target)]) + + +def parse_candidate_lines(text: str) -> list[list[str]]: + rows = [] + for line in text.splitlines(): + if line.strip(): + rows.append(next(csv.reader([line], delimiter="\t"))) + return rows + + +def json_dumps(payload: Any) -> str: + return json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) diff --git a/skills/intervention-data/scripts/validate_intervention_records.py b/skills/intervention-data/scripts/validate_intervention_records.py new file mode 100644 index 0000000..9e93ba3 --- /dev/null +++ b/skills/intervention-data/scripts/validate_intervention_records.py @@ -0,0 +1,220 @@ +#!/usr/bin/env python3 +"""校验单句精确干预和正则干预候选 TSV。""" + +from __future__ import annotations + +import argparse +import json +import re +import sys +from pathlib import Path +from typing import Any + +from intervention_common import ( + DEFAULT_EXACT_SOURCE, + DEFAULT_REGEX_SOURCE, + build_source_summary, + domain_compatible, + historical_targets, + infer_configured_dispatch_domain, + infer_dispatch_domain_for_device, + load_exact_records, + load_regex_records, + normalize_device, + normalize_target, + parse_candidate_lines, + validate_target, + json_dumps, +) + + +def load_input_text(path: str | None) -> str: + if path: + return Path(path).read_text(encoding="utf-8") + return sys.stdin.read() + + +def validate_exact_rows(rows: list[list[str]], args: argparse.Namespace) -> dict[str, Any]: + summary = build_source_summary(Path(args.exact_source), Path(args.regex_source)) + devices = set(summary["devices"]) + known_historical_targets = historical_targets(summary) + existing: dict[tuple[str, str], list[Any]] = {} + for record in load_exact_records(Path(args.exact_source)): + existing.setdefault((record.device, record.query), []).append(record) + + results = [] + for index, row in enumerate(rows, start=1): + errors: list[str] = [] + warnings: list[str] = [] + if len(row) != 4: + errors.append(f"单句精确干预必须是 4 列，当前 {len(row)} 列") + results.append({"line": index, "valid": False, "errors": errors, "warnings": warnings, "row": row}) + continue + + device, query, target, dispatch_domain = [cell.strip() for cell in row] + device = normalize_device(device) + target = normalize_target(target) + if device not in devices: + errors.append(f"设备不在历史枚举中：{device}，可选：{', '.join(sorted(devices))}") + if not query: + errors.append("query 不能为空") + if "\t" in query or "\n" in query or "\r" in query: + errors.append("query 不能包含制表符或换行") + target_result = validate_target(target) + if not target_result["valid"]: + if target in known_historical_targets: + warnings.append("target 未通过 label-master 知识库校验，但历史干预文件中已存在；建议后续补齐标签知识库") + target_result["normalized_output"] = target + else: + errors.extend([f"target 不合法：{message}" for message in target_result.get("errors", [])]) + if not dispatch_domain: + errors.append("下发domain 不能为空") + + inferred = infer_dispatch_domain_for_device(target, device, summary) + if inferred and dispatch_domain and dispatch_domain != inferred["dispatch_domain"]: + message = ( + f"下发domain 与推断映射不一致：当前 {dispatch_domain}，" + f"推断为 {inferred['dispatch_domain']}，来源 {inferred.get('source', 'historical')}，置信度 {inferred.get('confidence')}" + ) + if args.strict_domain: + errors.append(message) + else: + warnings.append(message) + if not inferred: + warnings.append("未在历史精确干预表中找到该 target 的下发domain 映射，需要人工确认") + + configured = infer_configured_dispatch_domain(target, device, summary) + if configured and dispatch_domain and not domain_compatible(dispatch_domain, configured["dispatch_domain"]): + warnings.append( + f"下发domain 与 type_match_agent 配置不兼容：当前 {dispatch_domain}，" + f"配置 {configured['matched_key']} -> {configured['dispatch_domain']}" + ) + if not configured: + warnings.append("type_match_agent 下发表中未找到该 Agent tag 的设备/通用映射") + + conflicts = existing.get((device, query), []) + for item in conflicts: + if item.target == target and item.dispatch_domain == dispatch_domain: + warnings.append(f"历史文件已存在相同单句干预：line {item.line_no}") + else: + errors.append( + f"历史文件存在同设备同 query 的不同干预：line {item.line_no}，" + f"{item.target} / {item.dispatch_domain}" + ) + + results.append( + { + "line": index, + "valid": not errors, + "errors": errors, + "warnings": warnings, + "normalized": { + "device": device, + "query": query, + "target": target_result.get("normalized_output", target), + "dispatch_domain": dispatch_domain, + }, + "configured_dispatch_domain": configured, + } + ) + return summarize_results(results) + + +def validate_regex_rows(rows: list[list[str]], args: argparse.Namespace) -> dict[str, Any]: + summary = build_source_summary(Path(args.exact_source), Path(args.regex_source)) + devices = set(summary["devices"]) + known_historical_targets = historical_targets(summary) + existing: dict[tuple[str, str], list[Any]] = {} + for record in load_regex_records(Path(args.regex_source)): + existing.setdefault((record.device, record.pattern), []).append(record) + + results = [] + for index, row in enumerate(rows, start=1): + errors: list[str] = [] + warnings: list[str] = [] + if len(row) != 3: + errors.append(f"正则干预必须是 3 列，当前 {len(row)} 列") + results.append({"line": index, "valid": False, "errors": errors, "warnings": warnings, "row": row}) + continue + + device, pattern, target = [cell.strip() for cell in row] + device = normalize_device(device) + target = normalize_target(target) + if device not in devices: + errors.append(f"设备不在历史枚举中：{device}，可选：{', '.join(sorted(devices))}") + if not pattern: + errors.append("正则不能为空") + else: + try: + re.compile(pattern) + except re.error as exc: + errors.append(f"正则无法编译：{exc}") + if "query#" not in pattern: + errors.append("正则必须包含 query# 片段") + if not pattern.startswith("^") or not pattern.endswith("$"): + warnings.append("历史常见正则通常使用 ^...$ 完整锚定，建议确认是否需要锚定") + + target_result = validate_target(target) + if not target_result["valid"]: + if target in known_historical_targets: + warnings.append("target 未通过 label-master 知识库校验，但历史干预文件中已存在；建议后续补齐标签知识库") + target_result["normalized_output"] = target + else: + errors.extend([f"target 不合法：{message}" for message in target_result.get("errors", [])]) + + conflicts = existing.get((device, pattern), []) + for item in conflicts: + if item.target == target: + warnings.append(f"历史文件已存在相同正则干预：line {item.line_no}") + else: + errors.append( + f"历史文件存在同设备同正则的不同干预：line {item.line_no}，{item.target}" + ) + + results.append( + { + "line": index, + "valid": not errors, + "errors": errors, + "warnings": warnings, + "normalized": { + "device": device, + "pattern": pattern, + "target": target_result.get("normalized_output", target), + }, + } + ) + return summarize_results(results) + + +def summarize_results(results: list[dict[str, Any]]) -> dict[str, Any]: + return { + "valid": all(item["valid"] for item in results), + "total": len(results), + "invalid": sum(1 for item in results if not item["valid"]), + "warning_count": sum(len(item.get("warnings", [])) for item in results), + "results": results, + } + + +def main() -> int: + parser = argparse.ArgumentParser(description="校验干预 TSV 候选") + parser.add_argument("--mode", choices=["exact", "regex"], required=True, help="干预类型") + parser.add_argument("--input", help="候选 TSV 文件路径；不传则读取 stdin") + parser.add_argument("--exact-source", default=str(DEFAULT_EXACT_SOURCE), help="历史单句精确干预 TSV") + parser.add_argument("--regex-source", default=str(DEFAULT_REGEX_SOURCE), help="历史正则干预 TSV") + parser.add_argument("--strict-domain", action="store_true", help="下发domain 和历史多数映射不一致时直接报错") + args = parser.parse_args() + + text = load_input_text(args.input) + rows = parse_candidate_lines(text) + if args.mode == "exact": + payload = validate_exact_rows(rows, args) + else: + payload = validate_regex_rows(rows, args) + print(json_dumps(payload)) + return 0 if payload["valid"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main())