scripts/build_project_resource_map.py

#!/usr/bin/env python3
"""Build a complete navigable resource map for scripts, results, docs, and HF bundles."""

from __future__ import annotations

import argparse
import json
import subprocess
from collections import Counter, defaultdict
from datetime import datetime, timezone
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
HF_ROOT = ROOT.parent / "hf_publish"
JSON_OUTPUT = ROOT / "docs/data/project_resource_map.json"
MD_OUTPUT = ROOT / "PROJECT_RESOURCE_MAP.md"
GITHUB_BLOB = "https://github.com/ChaoYue0307/ropedia-xperience-10m-task-suite/blob/main"
GITHUB_RAW = "https://raw.githubusercontent.com/ChaoYue0307/ropedia-xperience-10m-task-suite/main"
PAGES_BASE = "https://chaoyue0307.github.io/ropedia-xperience-10m-task-suite"

SCAN_ROOTS = [
    "scripts",
    "docs",
    "docs/data",
    "docs/assets",
    "results",
    "configs",
    "notes",
]

TOP_LEVEL_PATTERNS = [
    "*.md",
    "*.json",
    "*.toml",
    "*.txt",
    "Dockerfile",
    "requirements.txt",
    "package.json",
]

SKIP_PARTS = {
    ".git",
    ".venv",
    "__pycache__",
    ".pytest_cache",
    "node_modules",
    "test-results",
}

HF_BUNDLES = {
    "space": {
        "label": "HF Space bundle",
        "repo_type": "space",
        "repo_id": "cy0307/ropedia-xperience-10m-task-suite",
        "raw_base": "https://huggingface.co/spaces/cy0307/ropedia-xperience-10m-task-suite/raw/main",
    },
    "artifacts": {
        "label": "HF artifact dataset bundle",
        "repo_type": "dataset",
        "repo_id": "cy0307/ropedia-xperience-10m-task-suite-artifacts",
        "raw_base": "https://huggingface.co/datasets/cy0307/ropedia-xperience-10m-task-suite-artifacts/raw/main",
    },
    "model": {
        "label": "HF baseline model bundle",
        "repo_type": "model",
        "repo_id": "cy0307/ropedia-xperience-10m-task-baselines",
        "raw_base": "https://huggingface.co/cy0307/ropedia-xperience-10m-task-baselines/raw/main",
    },
    "weights_results": {
        "label": "HF weights/results bundle",
        "repo_type": "model",
        "repo_id": "cy0307/ropedia-xperience-10m-weights-results",
        "raw_base": "https://huggingface.co/cy0307/ropedia-xperience-10m-weights-results/raw/main",
    },
    "qwen3_lora_128ep": {
        "label": "Qwen3-Omni LoRA bundle",
        "repo_type": "model",
        "repo_id": "cy0307/ropedia-qwen3-omni-lora-128ep",
        "raw_base": "https://huggingface.co/cy0307/ropedia-qwen3-omni-lora-128ep/raw/main",
    },
    "cosmos3_super_forward_dynamics_lora_128ep": {
        "label": "Cosmos3-Super LoRA bundle",
        "repo_type": "model",
        "repo_id": "cy0307/ropedia-cosmos3-super-forward-dynamics-lora-128ep",
        "raw_base": "https://huggingface.co/cy0307/ropedia-cosmos3-super-forward-dynamics-lora-128ep/raw/main",
    },
}


def run_git_lines(*args: str) -> set[str]:
    try:
        output = subprocess.check_output(["git", "-C", str(ROOT), *args], text=True)
    except (subprocess.CalledProcessError, FileNotFoundError):
        return set()
    return {line.strip() for line in output.splitlines() if line.strip()}


def is_skipped(path: Path) -> bool:
    return any(part in SKIP_PARTS for part in path.parts)


def repo_files() -> list[Path]:
    files: set[Path] = set()
    for root_name in SCAN_ROOTS:
        root = ROOT / root_name
        if root.exists():
            files.update(path for path in root.rglob("*") if path.is_file() and not is_skipped(path))
    for pattern in TOP_LEVEL_PATTERNS:
        files.update(path for path in ROOT.glob(pattern) if path.is_file())
    return sorted(files, key=lambda path: path.relative_to(ROOT).as_posix())


def hf_files(hf_root: Path) -> list[tuple[str, Path]]:
    if not hf_root.exists():
        return []
    files: list[tuple[str, Path]] = []
    for bundle_dir in sorted(hf_root.iterdir()):
        if not bundle_dir.is_dir() or is_skipped(bundle_dir):
            continue
        bundle = bundle_dir.name
        for path in bundle_dir.rglob("*"):
            if path.is_file() and not is_skipped(path):
                files.append((bundle, path))
    return sorted(files, key=lambda item: (item[0], item[1].relative_to(hf_root / item[0]).as_posix()))


def file_kind(path: str, source_root: str) -> str:
    if source_root.startswith("hf_publish"):
        return "hf_bundle"
    if path.startswith("scripts/omni/"):
        return "omni_script"
    if path.startswith("scripts/"):
        return "script"
    if path.startswith("results/omni_finetune/"):
        return "omni_result"
    if path.startswith("results/"):
        return "result"
    if path.startswith("docs/data/"):
        return "structured_data"
    if path.startswith("docs/assets/"):
        return "visual_asset"
    if path.startswith("configs/"):
        return "config"
    if path.startswith("notes/"):
        return "note"
    if path.endswith(".md"):
        return "documentation"
    return "resource"


def script_role(path: str) -> str:
    name = Path(path).name
    stem = Path(path).stem
    if name.endswith(".md"):
        return "runbook"
    if stem.startswith("validate_"):
        return "validator"
    if stem.startswith("build_"):
        return "builder"
    if stem.startswith(("render_", "generate_")):
        return "renderer"
    if stem.startswith(("sync_", "publish_", "upload_")):
        return "publisher"
    if stem.startswith(("train_", "run_train", "run_")):
        return "runner"
    if stem.startswith(("eval_", "score_")):
        return "evaluator"
    if stem.startswith(("collect_", "merge_", "package_", "prepare_")):
        return "packager"
    if stem.startswith(("watch_", "monitor_", "defer_", "launch_", "auto_start_")):
        return "orchestrator"
    if stem.startswith(("analyze_", "audit_", "probe_", "diagnose_")):
        return "auditor"
    if stem.startswith(("export_", "extract_", "download_", "stage_", "transfer_")):
        return "data-prep"
    return "utility"


def purpose_for(path: str, kind: str) -> str:
    name = Path(path).name
    stem = Path(path).stem.replace("_", " ")
    if kind in {"script", "omni_script"}:
        return f"{script_role(path).replace('-', ' ').title()} script for {stem}."
    if kind in {"result", "omni_result"}:
        return "Committed or local result artifact: metrics, predictions, model weights, logs, manifests, or generated analysis."
    if kind == "structured_data":
        return "Website and Hugging Face structured data mirror."
    if kind == "visual_asset":
        return "Website figure, chart, icon, preview, or generated visual asset."
    if kind == "hf_bundle":
        return "Prepared Hugging Face upload bundle file; publish with scripts/publish_hf_bundles.py."
    if kind == "config":
        return "Configuration used by training, evaluation, packaging, or public-surface generation."
    if name.endswith(".md"):
        return "Human-readable project note or public documentation."
    return "Project resource."


def repo_access(path: str, tracked: bool) -> dict:
    access: dict[str, str | bool] = {
        "local_path": f"repo:{path}",
        "tracked_in_git": tracked,
    }
    if tracked:
        access["github"] = f"{GITHUB_BLOB}/{path}"
        access["raw"] = f"{GITHUB_RAW}/{path}"
    if path.startswith("docs/") and tracked:
        site_path = path.removeprefix("docs/")
        access["site"] = f"{PAGES_BASE}/{site_path}"
    return access


def hf_access(bundle: str, relative_path: str, path: Path) -> dict:
    meta = HF_BUNDLES.get(bundle, {})
    access: dict[str, str | bool] = {
        "local_path": f"hf_publish/{bundle}:{relative_path}",
        "bundle": bundle,
        "published_repo": meta.get("repo_id", ""),
    }
    raw_base = meta.get("raw_base")
    if raw_base:
        access["hf_raw"] = f"{raw_base}/{relative_path}"
    return access


def file_record(path: Path, rel: str, *, source_root: str, tracked: bool = False, bundle: str | None = None) -> dict:
    stat = path.stat()
    kind = file_kind(rel, source_root)
    record = {
        "path": rel,
        "name": path.name,
        "kind": kind,
        "role": script_role(rel) if kind in {"script", "omni_script"} else kind.replace("_", " "),
        "source_root": source_root,
        "purpose": purpose_for(rel, kind),
        "size_bytes": stat.st_size,
        "size_human": human_size(stat.st_size),
        "modified_utc": datetime.fromtimestamp(stat.st_mtime, timezone.utc).replace(microsecond=0).isoformat(),
    }
    if bundle:
        record["bundle"] = bundle
        record["access"] = hf_access(bundle, rel, path)
    else:
        record["access"] = repo_access(rel, tracked)
    return record


def human_size(size: int) -> str:
    value = float(size)
    for unit in ("B", "KB", "MB", "GB"):
        if value < 1024 or unit == "GB":
            if unit == "B":
                return f"{int(value)} {unit}"
            return f"{value:.2f} {unit}"
        value /= 1024
    return f"{size} B"


def build_payload(hf_root: Path) -> dict:
    tracked = run_git_lines("ls-files")
    entries: list[dict] = []
    for path in repo_files():
        rel = path.relative_to(ROOT).as_posix()
        entries.append(file_record(path, rel, source_root="repo", tracked=rel in tracked))
    for bundle, path in hf_files(hf_root):
        rel = path.relative_to(hf_root / bundle).as_posix()
        entries.append(file_record(path, rel, source_root=f"hf_publish/{bundle}", bundle=bundle))

    counts_by_kind = Counter(entry["kind"] for entry in entries)
    counts_by_root = Counter(entry["source_root"] for entry in entries)
    counts_by_script_role = Counter(entry["role"] for entry in entries if entry["kind"] in {"script", "omni_script"})
    public_repo_files = sum(1 for entry in entries if entry.get("access", {}).get("github"))
    site_files = sum(1 for entry in entries if entry.get("access", {}).get("site"))
    hf_bundle_files = sum(1 for entry in entries if entry["kind"] == "hf_bundle")
    local_untracked_repo_files = sum(1 for entry in entries if entry["source_root"] == "repo" and not entry.get("access", {}).get("tracked_in_git"))

    return {
        "generated_at_utc": datetime.now(timezone.utc).replace(microsecond=0).isoformat(),
        "repo_root": ".",
        "hf_publish_root": "../hf_publish",
        "summary": {
            "total_files": len(entries),
            "script_files": counts_by_kind["script"] + counts_by_kind["omni_script"],
            "result_files": counts_by_kind["result"] + counts_by_kind["omni_result"],
            "structured_data_files": counts_by_kind["structured_data"],
            "visual_asset_files": counts_by_kind["visual_asset"],
            "hf_bundle_files": hf_bundle_files,
            "github_linked_files": public_repo_files,
            "site_linked_files": site_files,
            "local_untracked_repo_files": local_untracked_repo_files,
        },
        "counts_by_kind": dict(sorted(counts_by_kind.items())),
        "counts_by_source_root": dict(sorted(counts_by_root.items())),
        "counts_by_script_role": dict(sorted(counts_by_script_role.items())),
        "quick_routes": [
            {
                "question": "Which script builds or validates a public artifact?",
                "filter": "kind=script or kind=omni_script",
                "best_view": "Use script role filters: builder, validator, renderer, publisher, runner, evaluator, packager.",
            },
            {
                "question": "Where are the current metrics, predictions, and model-output files?",
                "filter": "kind=result or kind=omni_result",
                "best_view": "Use the result rows and open the GitHub link when tracked, or local path when a file is still local-only.",
            },
            {
                "question": "Which public files feed the website and Hugging Face mirrors?",
                "filter": "kind=structured_data or kind=visual_asset or kind=hf_bundle",
                "best_view": "Open site links for docs files and HF raw links for staged bundle files.",
            },
        ],
        "entries": entries,
    }


def md_link(label: str, href: str | None) -> str:
    if not href:
        return label
    return f"[{label}]({href})"


def entry_link(entry: dict) -> str:
    access = entry.get("access", {})
    for key in ("site", "github", "hf_raw", "raw"):
        href = access.get(key)
        if href:
            return md_link(entry["path"], href)
    return f"`{entry['path']}`"


def table_rows(entries: list[dict], *, kinds: set[str] | None = None, limit: int | None = None) -> list[str]:
    rows = []
    selected = [entry for entry in entries if kinds is None or entry["kind"] in kinds]
    for entry in selected[:limit]:
        access = entry.get("access", {})
        public = []
        if access.get("site"):
            public.append(md_link("site", str(access["site"])))
        if access.get("github"):
            public.append(md_link("repo", str(access["github"])))
        if access.get("hf_raw"):
            public.append(md_link("HF raw", str(access["hf_raw"])))
        if not public:
            public.append("local/staged")
        rows.append(
            "| "
            + " | ".join(
                [
                    entry_link(entry),
                    entry["kind"],
                    entry.get("role", ""),
                    entry["size_human"],
                    ", ".join(public),
                ]
            )
            + " |"
        )
    return rows


def write_markdown(payload: dict, path: Path) -> None:
    entries = payload["entries"]
    lines = [
        "# Project Resource Map",
        "",
        "Generated inventory for scripts, results, website data, visual assets, and prepared Hugging Face bundles.",
        "",
        "This file does not move or rewrite result artifacts. It indexes them in place so readers and maintainers can open the right file directly.",
        "",
        "## Summary",
        "",
    ]
    for key, value in payload["summary"].items():
        lines.append(f"- **{key.replace('_', ' ').title()}**: {value}")
    lines.extend(
        [
            "",
            "## Script Role Counts",
            "",
            "| Role | Files |",
            "|---|---:|",
        ]
    )
    for role, count in payload["counts_by_script_role"].items():
        lines.append(f"| {role} | {count} |")
    lines.extend(
        [
            "",
            "## Complete Script Inventory",
            "",
            "| File | Kind | Role | Size | Direct access |",
            "|---|---|---|---:|---|",
        ]
    )
    lines.extend(table_rows(entries, kinds={"script", "omni_script"}))
    lines.extend(
        [
            "",
            "## Complete Result Inventory",
            "",
            "| File | Kind | Role | Size | Direct access |",
            "|---|---|---|---:|---|",
        ]
    )
    lines.extend(table_rows(entries, kinds={"result", "omni_result"}))
    lines.extend(
        [
            "",
            "## Website Data And Assets",
            "",
            "| File | Kind | Role | Size | Direct access |",
            "|---|---|---|---:|---|",
        ]
    )
    lines.extend(table_rows(entries, kinds={"structured_data", "visual_asset"}))
    lines.extend(
        [
            "",
            "## Prepared Hugging Face Bundles",
            "",
            "| File | Kind | Role | Size | Direct access |",
            "|---|---|---|---:|---|",
        ]
    )
    lines.extend(table_rows(entries, kinds={"hf_bundle"}))
    lines.append("")
    path.write_text("\n".join(lines), encoding="utf-8")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--hf-root", type=Path, default=HF_ROOT)
    parser.add_argument("--json-output", type=Path, default=JSON_OUTPUT)
    parser.add_argument("--md-output", type=Path, default=MD_OUTPUT)
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    payload = build_payload(args.hf_root)
    args.json_output.parent.mkdir(parents=True, exist_ok=True)
    args.json_output.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    write_markdown(payload, args.md_output)
    print(
        "wrote "
        f"{args.json_output.relative_to(ROOT)} and {args.md_output.relative_to(ROOT)} "
        f"({payload['summary']['total_files']} files indexed)"
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())