wuyang6
339 lines
12 KiB
Python
Executable File
339 lines
12 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""从 Markdown 标签知识库生成机器可读索引。
|
||
|
||
这个脚本只做确定性整理，不做语义判断：
|
||
- 人继续维护 knowledge/ 下的中文 Markdown。
|
||
- 脚本把标签、intent、function、object、边界文件整理成 JSON manifest。
|
||
- Agent 和校验脚本优先读取 manifest，避免每次全量扫描知识库。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import re
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
|
||
SCRIPT_PATH = Path(__file__).resolve()
|
||
SKILL_ROOT = SCRIPT_PATH.parents[1]
|
||
KNOWLEDGE_ROOT = SKILL_ROOT / "knowledge"
|
||
DEFAULT_OUTPUT = KNOWLEDGE_ROOT / "索引" / "label_manifest.json"
|
||
|
||
|
||
def read_text(path: Path) -> str:
|
||
return path.read_text(encoding="utf-8")
|
||
|
||
|
||
def rel(path: Path) -> str:
|
||
return path.relative_to(SKILL_ROOT).as_posix()
|
||
|
||
|
||
def first_heading(text: str, fallback: str) -> str:
|
||
match = re.search(r"^#\s+(.+?)\s*$", text, re.MULTILINE)
|
||
return match.group(1).strip() if match else fallback
|
||
|
||
|
||
def extract_section(text: str, heading: str) -> str:
|
||
pattern = rf"^##\s+{re.escape(heading)}\s*$"
|
||
match = re.search(pattern, text, re.MULTILINE)
|
||
if not match:
|
||
return ""
|
||
start = match.end()
|
||
next_match = re.search(r"^##\s+", text[start:], re.MULTILINE)
|
||
end = start + next_match.start() if next_match else len(text)
|
||
return text[start:end].strip()
|
||
|
||
|
||
def extract_bullet_value(text: str, label: str) -> str:
|
||
match = re.search(rf"^-\s*{re.escape(label)}[：:]\s*(.+?)\s*$", text, re.MULTILINE)
|
||
return match.group(1).strip() if match else ""
|
||
|
||
|
||
def parse_markdown_table(text: str) -> list[dict[str, str]]:
|
||
rows: list[dict[str, str]] = []
|
||
table_lines = [line.strip() for line in text.splitlines() if line.strip().startswith("|")]
|
||
if len(table_lines) < 2:
|
||
return rows
|
||
header = [cell.strip() for cell in table_lines[0].strip("|").split("|")]
|
||
for line in table_lines[2:]:
|
||
cells = [cell.strip() for cell in line.strip("|").split("|")]
|
||
if len(cells) != len(header):
|
||
continue
|
||
rows.append(dict(zip(header, cells)))
|
||
return rows
|
||
|
||
|
||
def parse_enum_values(description: str) -> list[str]:
|
||
if "枚举" not in description:
|
||
return []
|
||
desc = re.sub(r"[，。；;].*$", "", description.split("枚举", 1)[-1])
|
||
desc = desc.replace("：", ":")
|
||
if ":" in desc:
|
||
desc = desc.split(":", 1)[1]
|
||
parts = re.split(r"[、,，/| 或 和]+", desc)
|
||
values: list[str] = []
|
||
for part in parts:
|
||
cleaned = part.strip().strip("`").strip()
|
||
cleaned = re.sub(r"[（(].*?[）)]", "", cleaned).strip()
|
||
if cleaned and re.fullmatch(r"[A-Za-z0-9_]+", cleaned):
|
||
values.append(cleaned)
|
||
return values
|
||
|
||
|
||
def parse_label_cards() -> list[dict[str, Any]]:
|
||
labels_root = KNOWLEDGE_ROOT / "标签"
|
||
labels: list[dict[str, Any]] = []
|
||
if not labels_root.exists():
|
||
return labels
|
||
|
||
for path in sorted(labels_root.rglob("*.md")):
|
||
text = read_text(path)
|
||
name = first_heading(text, path.stem)
|
||
domain = path.parent.name
|
||
label = {
|
||
"name": name,
|
||
"domain": domain,
|
||
"path": rel(path),
|
||
"old_tag": extract_bullet_value(text, "旧 tag 标签"),
|
||
"agent_candidate": extract_bullet_value(text, "Agent 包装候选（不代表最终）"),
|
||
"function_candidate": extract_bullet_value(text, "Function 输出候选"),
|
||
"recommended_output_shape": extract_bullet_value(text, "推荐输出形态"),
|
||
"scope": extract_section(text, "适用范围"),
|
||
"examples": extract_section(text, "典型 Query"),
|
||
"confusing_labels": extract_section(text, "易混淆标签"),
|
||
"principles": extract_section(text, "划分原则"),
|
||
}
|
||
labels.append(label)
|
||
return labels
|
||
|
||
|
||
def parse_function_catalog() -> list[dict[str, Any]]:
|
||
path = KNOWLEDGE_ROOT / "输出能力" / "函数目录.md"
|
||
if not path.exists():
|
||
return []
|
||
text = read_text(path)
|
||
functions: list[dict[str, Any]] = []
|
||
sections = list(re.finditer(r"^##\s+(.+?)\s*$", text, re.MULTILINE))
|
||
for index, match in enumerate(sections):
|
||
name = match.group(1).strip()
|
||
start = match.end()
|
||
end = sections[index + 1].start() if index + 1 < len(sections) else len(text)
|
||
body = text[start:end].strip()
|
||
params: dict[str, dict[str, Any]] = {}
|
||
param_block = ""
|
||
if "参数：" in body:
|
||
param_block = body.split("参数：", 1)[1]
|
||
if "示例：" in param_block:
|
||
param_block = param_block.split("示例：", 1)[0]
|
||
for row in parse_markdown_table(param_block):
|
||
param_name = row.get("参数", "").strip()
|
||
description = row.get("说明", "").strip()
|
||
if not param_name:
|
||
continue
|
||
params[param_name] = {
|
||
"description": description,
|
||
"enum_values": parse_enum_values(description),
|
||
}
|
||
functions.append(
|
||
{
|
||
"name": name,
|
||
"path": rel(path),
|
||
"params": params,
|
||
"allow_unknown_params": False,
|
||
"summary": re.sub(r"\s+", " ", extract_summary(body)).strip(),
|
||
}
|
||
)
|
||
return functions
|
||
|
||
|
||
def parse_object_catalog() -> list[dict[str, Any]]:
|
||
path = KNOWLEDGE_ROOT / "输出能力" / "对象目录.md"
|
||
if not path.exists():
|
||
return []
|
||
text = read_text(path)
|
||
objects: list[dict[str, Any]] = []
|
||
|
||
sections = list(re.finditer(r"^##\s+(.+?)\s*$", text, re.MULTILINE))
|
||
for index, match in enumerate(sections):
|
||
name = match.group(1).strip()
|
||
start = match.end()
|
||
end = sections[index + 1].start() if index + 1 < len(sections) else len(text)
|
||
body = text[start:end].strip()
|
||
|
||
if name in {"VisionQA 辅助对象", "生成类辅助对象"}:
|
||
for row in parse_markdown_table(body):
|
||
object_name = row.get("对象", "").strip()
|
||
if object_name:
|
||
objects.append(
|
||
{
|
||
"name": object_name,
|
||
"path": rel(path),
|
||
"params": {},
|
||
"allow_unknown_params": True,
|
||
"summary": row.get("说明", "").strip(),
|
||
}
|
||
)
|
||
continue
|
||
|
||
params: dict[str, dict[str, Any]] = {}
|
||
param_block = body.split("参数：", 1)[1] if "参数：" in body else ""
|
||
if "示例：" in param_block:
|
||
param_block = param_block.split("示例：", 1)[0]
|
||
for row in parse_markdown_table(param_block):
|
||
param_name = row.get("参数", "").strip()
|
||
description = row.get("说明", "").strip()
|
||
if not param_name:
|
||
continue
|
||
params[param_name] = {
|
||
"description": description,
|
||
"enum_values": parse_enum_values(description),
|
||
}
|
||
objects.append(
|
||
{
|
||
"name": name,
|
||
"path": rel(path),
|
||
"params": params,
|
||
"allow_unknown_params": False,
|
||
"summary": re.sub(r"\s+", " ", extract_summary(body)).strip(),
|
||
}
|
||
)
|
||
return objects
|
||
|
||
|
||
def parse_intent_catalog() -> list[dict[str, Any]]:
|
||
path = KNOWLEDGE_ROOT / "输出能力" / "意图目录.md"
|
||
if not path.exists():
|
||
return []
|
||
text = read_text(path)
|
||
intents: list[dict[str, Any]] = []
|
||
for row in parse_markdown_table(text):
|
||
name = row.get("intent", "").replace("\\|", "|").strip()
|
||
if not name:
|
||
continue
|
||
intents.append(
|
||
{
|
||
"name": name,
|
||
"agent": row.get("Agent", "").strip(),
|
||
"instruction_type": row.get("指令类型", "").strip(),
|
||
"scope": row.get("简要范围", "").strip(),
|
||
"path": rel(path),
|
||
}
|
||
)
|
||
return intents
|
||
|
||
|
||
def extract_summary(body: str) -> str:
|
||
match = re.search(r"功能范围[：:]\s*(.+?)(?:\n\n|$)", body, re.DOTALL)
|
||
return match.group(1).strip() if match else ""
|
||
|
||
|
||
def parse_boundary_files() -> list[dict[str, str]]:
|
||
boundary_root = KNOWLEDGE_ROOT / "边界"
|
||
if not boundary_root.exists():
|
||
return []
|
||
files: list[dict[str, str]] = []
|
||
for path in sorted(boundary_root.rglob("*.md")):
|
||
if path.name == "README.md":
|
||
continue
|
||
text = read_text(path)
|
||
parts = path.relative_to(boundary_root).parts
|
||
category = parts[0] if len(parts) > 1 else "root"
|
||
files.append({"name": first_heading(text, path.stem), "path": rel(path), "category": category})
|
||
return files
|
||
|
||
|
||
def parse_dimension_files() -> list[dict[str, str]]:
|
||
root = KNOWLEDGE_ROOT / "判断维度"
|
||
if not root.exists():
|
||
return []
|
||
files: list[dict[str, str]] = []
|
||
for path in sorted(root.rglob("*.md")):
|
||
if path == root / "README.md":
|
||
continue
|
||
text = read_text(path)
|
||
item = {"name": first_heading(text, path.stem), "path": rel(path)}
|
||
parent_dimension = extract_bullet_value(text, "父维度")
|
||
output_field = extract_bullet_value(text, "输出字段")
|
||
if parent_dimension:
|
||
item["parent_dimension"] = parent_dimension
|
||
if output_field:
|
||
item["output_field"] = output_field
|
||
files.append(item)
|
||
return files
|
||
|
||
|
||
def build_manifest() -> dict[str, Any]:
|
||
labels = parse_label_cards()
|
||
functions = parse_function_catalog()
|
||
objects = parse_object_catalog()
|
||
intents = parse_intent_catalog()
|
||
|
||
old_tags = sorted({item["old_tag"] for item in labels if item.get("old_tag") and item["old_tag"] != "待确认。"})
|
||
label_names = sorted({item["name"] for item in labels})
|
||
agent_tags = sorted(set(old_tags) | set(label_names))
|
||
|
||
return {
|
||
"schema_version": 1,
|
||
"skill": "label-master",
|
||
"display_name": "标签大师",
|
||
"description": "标签大师的中控标签知识库机器索引；由 scripts/build_label_manifest.py 从 Markdown 生成。",
|
||
"counts": {
|
||
"labels": len(labels),
|
||
"agent_tags": len(agent_tags),
|
||
"functions": len(functions),
|
||
"objects": len(objects),
|
||
"intents": len(intents),
|
||
},
|
||
"agent_tags": agent_tags,
|
||
"labels": labels,
|
||
"functions": functions,
|
||
"objects": objects,
|
||
"intents": intents,
|
||
"boundaries": parse_boundary_files(),
|
||
"dimensions": parse_dimension_files(),
|
||
}
|
||
|
||
|
||
def write_manifest(path: Path, manifest: dict[str, Any]) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
text = json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
|
||
path.write_text(text, encoding="utf-8")
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
parser = argparse.ArgumentParser(description="生成 label-master 标签大师机器索引")
|
||
parser.add_argument("--output", default=str(DEFAULT_OUTPUT), help="输出 JSON 路径")
|
||
parser.add_argument("--check", action="store_true", help="只检查输出文件是否为最新，不写入")
|
||
args = parser.parse_args(argv)
|
||
|
||
output = Path(args.output)
|
||
if not output.is_absolute():
|
||
output = SKILL_ROOT / output
|
||
|
||
manifest = build_manifest()
|
||
new_text = json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
|
||
|
||
if args.check:
|
||
if not output.exists():
|
||
print(f"ERROR: manifest 不存在：{output}", file=sys.stderr)
|
||
return 1
|
||
old_text = output.read_text(encoding="utf-8")
|
||
if old_text != new_text:
|
||
print(f"ERROR: manifest 不是最新：{output}", file=sys.stderr)
|
||
return 1
|
||
print(f"OK: manifest 已是最新：{output}")
|
||
return 0
|
||
|
||
output.parent.mkdir(parents=True, exist_ok=True)
|
||
output.write_text(new_text, encoding="utf-8")
|
||
print(f"Wrote {output}")
|
||
print(json.dumps(manifest["counts"], ensure_ascii=False, sort_keys=True))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|