mediamtx/dev_scripts/harness.py

"""Project-specific DevHarness entry point for the MediaMTX customization fork."""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

from wiki_docs import WikiClient, WikiDocsError, load_config, parse_mirror, sync_all


ROOT = Path(__file__).resolve().parents[1]
CONFIG = ROOT / "wiki-docs.json"
REQUIRED_FILES = (
    "AGENTS.md",
    "CLAUDE.md",
    ".gitea/issue_template/epic.md",
    ".gitea/issue_template/mvp.md",
    ".gitea/issue_template/task.md",
    "wiki-docs.json",
    "dev_scripts/harness.py",
    "dev_scripts/wiki_docs.py",
    "tests/test_harness.py",
)
EXPECTED_MAPPINGS = {
    "Home": "docs/maintainer/README.md",
    "Project-Profile": "docs/maintainer/1-project-profile.md",
    "Development-Workflow": "docs/maintainer/2-development-workflow.md",
    "Architecture-and-Code-Map": "docs/maintainer/3-architecture-and-code-map.md",
    "Product-Requirements-Overview": "docs/maintainer/4-product-requirements-overview.md",
    "Local-Development-and-Verification": "docs/maintainer/5-local-development-and-verification.md",
    "Common-Changes": "docs/maintainer/6-common-changes.md",
    "Troubleshooting": "docs/maintainer/7-troubleshooting.md",
}


def check(strict: bool) -> list[str]:
    errors: list[str] = []
    for relative in REQUIRED_FILES:
        if not (ROOT / relative).is_file():
            errors.append(f"缺少文件：{relative}")

    if "ip_camera.env" not in (ROOT / ".gitignore").read_text(encoding="utf-8"):
        errors.append(".gitignore 未忽略 ip_camera.env")

    try:
        raw = json.loads(CONFIG.read_text(encoding="utf-8"))
        actual = {item["page"]: item["path"] for item in raw.get("mappings", [])}
        if actual != EXPECTED_MAPPINGS:
            errors.append("wiki-docs.json 的核心映射与项目约定不一致")
    except (OSError, ValueError, KeyError, TypeError) as exc:
        errors.append(f"无法读取 wiki-docs.json：{exc}")

    agents = (ROOT / "AGENTS.md").read_text(encoding="utf-8") if (ROOT / "AGENTS.md").is_file() else ""
    for phrase in ("Gitea 单元工单", "docs/maintainer/", "ip_camera.env", "upstream"):
        if phrase not in agents:
            errors.append(f"AGENTS.md 缺少项目规则：{phrase}")

    if strict:
        for path in EXPECTED_MAPPINGS.values():
            target = ROOT / path
            if not target.is_file():
                errors.append(f"缺少 Wiki 镜像：{path}")
                continue
            try:
                metadata, _ = parse_mirror(target.read_text(encoding="utf-8"))
            except WikiDocsError as exc:
                errors.append(f"Wiki 镜像无效 {path}：{exc}")
                continue
            if not metadata.get("wiki_revision"):
                errors.append(f"Wiki 镜像缺少 revision：{path}")

    return errors


def run_check(args: argparse.Namespace) -> int:
    errors = check(args.strict)
    if errors:
        for error in errors:
            print(f"错误：{error}")
        return 1
    print("MediaMTX DevHarness 检查通过")
    return 0


def run_sync(args: argparse.Namespace) -> int:
    try:
        config = load_config(Path(args.config))
        for message in sync_all(config, WikiClient(config), check=args.check):
            print(message)
    except WikiDocsError as exc:
        print(f"错误：{exc}", file=sys.stderr)
        return 1
    if args.verify:
        errors = check(True)
        if errors:
            for error in errors:
                print(f"错误：{error}")
            return 1
        try:
            for message in sync_all(config, WikiClient(config), check=True):
                print(message)
        except WikiDocsError as exc:
            print(f"错误：{exc}", file=sys.stderr)
            return 1
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(description="MediaMTX 定制版 DevHarness")
    sub = parser.add_subparsers(dest="command", required=True)
    check_parser = sub.add_parser("check", help="检查项目 Harness 结构")
    check_parser.add_argument("--strict", action="store_true")
    check_parser.set_defaults(handler=run_check)
    sync_parser = sub.add_parser("sync", help="从 Gitea Wiki 单向同步维护文档")
    sync_parser.add_argument("--config", default=str(CONFIG))
    sync_parser.add_argument("--check", action="store_true")
    sync_parser.add_argument("--verify", action="store_true")
    sync_parser.set_defaults(handler=run_sync)
    args = parser.parse_args()
    return args.handler(args)


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