the-mindscape-of-bro-tsong/scripts/rebuild_indexes.py

import argparse
import json
import sys
from dataclasses import dataclass
from datetime import date
from pathlib import Path


IGNORED_CARD_NAMES = {"README.md", "card_index.md"}


@dataclass
class RebuildResult:
    drift_found: bool
    differences: list
    report_path: Path


def read_json(path):
    return json.loads(path.read_text(encoding="utf-8"))


def write_text(path, text):
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(text, encoding="utf-8")


def regression_case_counts(root):
    path = root / "tests" / "regression_cases.json"
    if not path.exists():
        return {}
    data = read_json(path)
    counts = {}
    for case in data.get("regression_cases", []):
        model_id = case.get("model_id")
        if model_id:
            counts[model_id] = counts.get(model_id, 0) + 1
    return counts


def card_path_for_model(root, model_id):
    path = root / "cards" / f"{model_id}.md"
    if path.exists():
        return path.relative_to(root).as_posix()
    return f"cards/{model_id}.md"


def existing_index_version(root):
    path = root / "models" / "model_index.json"
    if not path.exists():
        return "0.1"
    try:
        data = read_json(path)
    except json.JSONDecodeError:
        return "0.1"
    return data.get("index_version", "0.1")


def build_model_index(root):
    counts = regression_case_counts(root)
    models = []
    for model_path in sorted((root / "models").glob("*.model.json")):
        model = read_json(model_path)
        model_id = model["model_id"]
        models.append({
            "model_id": model_id,
            "model_name": model["model_name"],
            "model_type": model["model_type"],
            "pipeline_position": model["pipeline_position"],
            "model_file": model_path.relative_to(root).as_posix(),
            "card_file": card_path_for_model(root, model_id),
            "source_article_count": len(model.get("source_articles", [])),
            "source_evidence_count": len(model.get("source_evidence", [])),
            "regression_case_count": counts.get(model_id, 0),
            "stability_level": model.get("stability_profile", {}).get("stability_level"),
            "regression_status": model.get("regression_status"),
            "status": model.get("status"),
            "last_updated": model.get("last_updated")
        })

    return {
        "index_version": existing_index_version(root),
        "last_updated": date.today().isoformat(),
        "models": sorted(models, key=lambda item: item["model_id"])
    }


def render_json(data):
    return json.dumps(data, ensure_ascii=False, indent=2) + "\n"


def render_card_index(model_index):
    lines = [
        "# Card Index",
        "",
        "| Model ID | 模型名称 | 类型 | 流程位置 | Card | Model JSON | 稳固性 | 回归状态 | 状态 |",
        "| --- | --- | --- | --- | --- | --- | --- | --- | --- |",
    ]
    for entry in model_index["models"]:
        lines.append(
            "| {model_id} | {model_name} | {model_type} | {pipeline_position} | "
            "`{card_file}` | `{model_file}` | {stability_level} | {regression_status} | {status} |".format(**entry)
        )
    return "\n".join(lines) + "\n"


def collect_differences(root, expected_model_index_text, expected_card_index_text):
    checks = [
        (root / "models" / "model_index.json", expected_model_index_text, "models/model_index.json"),
        (root / "cards" / "card_index.md", expected_card_index_text, "cards/card_index.md"),
    ]
    differences = []
    for path, expected_text, label in checks:
        if not path.exists():
            differences.append(f"{label} is missing")
            continue
        actual_text = path.read_text(encoding="utf-8")
        if actual_text != expected_text:
            differences.append(f"{label} differs from generated index")
    return differences


def write_report(root, mode, differences):
    report_path = root / "reports" / "index_rebuild_report.md"
    lines = [
        "# Index Rebuild Report",
        "",
        f"Mode: `{mode}`",
        f"Status: `{'DRIFT' if differences else 'PASS'}`",
        "",
    ]
    if differences:
        lines.append("## Differences")
        lines.append("")
        lines.extend(f"- {difference}" for difference in differences)
    else:
        lines.append("## Result")
        lines.append("")
        lines.append("- models/model_index.json matches generated model registry.")
        lines.append("- cards/card_index.md matches generated card index.")
    write_text(report_path, "\n".join(lines) + "\n")
    return report_path


def rebuild_indexes(root, write=False):
    root = Path(root)
    model_index = build_model_index(root)
    model_index_text = render_json(model_index)
    card_index_text = render_card_index(model_index)

    if write:
        write_text(root / "models" / "model_index.json", model_index_text)
        write_text(root / "cards" / "card_index.md", card_index_text)
        differences = []
        mode = "write"
    else:
        differences = collect_differences(root, model_index_text, card_index_text)
        mode = "check"

    report_path = write_report(root, mode, differences)
    return RebuildResult(bool(differences), differences, report_path)


def main():
    parser = argparse.ArgumentParser(description="Rebuild or check model and card indexes.")
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("--check", action="store_true", help="Check generated indexes against files without writing them.")
    group.add_argument("--write", action="store_true", help="Write generated indexes to disk.")
    args = parser.parse_args()

    root = Path(__file__).resolve().parents[1]
    result = rebuild_indexes(root, write=args.write)
    print(f"index rebuild report written to {result.report_path}")
    if result.drift_found:
        for difference in result.differences:
            print(f"DRIFT: {difference}")
        return 1
    print("index check passed")
    return 0


if __name__ == "__main__":
    sys.exit(main())