zk-data-agent/skills/label-master/scripts/build_label_manifest.py

#!/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())