yolov26_3d/tools/feishu_project/recover_downloaded_issue_calibs.py

#!/usr/bin/env python3
"""Recover missing camera4.json files for downloaded standard-path issue data."""

from __future__ import annotations

import argparse
import json
import sys
from datetime import datetime
from pathlib import Path

FILE = Path(__file__).resolve()
ROOT = FILE.parents[2]
if str(ROOT) not in sys.path:
    sys.path.append(str(ROOT))

from tools.feishu_project.case_calib_recovery import CalibRecoveryResult, recover_camera4_json


DEFAULT_DOWNLOAD_ROOT = Path("/data1/dongying/Mono3d/G1Q3/feishu_project/downloaded_issue_data")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--download-root",
        default=str(DEFAULT_DOWNLOAD_ROOT),
        help="Root directory containing downloaded issue data.",
    )
    parser.add_argument(
        "--issue-id",
        action="append",
        dest="issue_ids",
        type=int,
        help="Only process the specified issue id. Can be repeated.",
    )
    parser.add_argument(
        "--path-root",
        action="append",
        default=[],
        help=(
            "Optional explicit copied target root to recover. Can be repeated. "
            "Example: downloaded_issue_data/issue_x/path_01/case_dir"
        ),
    )
    parser.add_argument(
        "--manifest-path",
        default="",
        help="Optional explicit manifest path. Defaults to <download-root>/calib_recovery_manifest.json",
    )
    parser.add_argument("--dry-run", action="store_true")
    return parser.parse_args()


def log_progress(message: str) -> None:
    timestamp = datetime.now().astimezone().strftime("%Y-%m-%d %H:%M:%S")
    print(f"[recover_downloaded_issue_calibs {timestamp}] {message}", flush=True)


def discover_target_roots(download_root: Path, issue_ids: list[int], explicit_roots: list[str]) -> list[Path]:
    if explicit_roots:
        return sorted({Path(path).resolve() for path in explicit_roots})

    issue_dirs: list[Path]
    if issue_ids:
        issue_dirs = [download_root / f"issue_{issue_id}" for issue_id in sorted(set(issue_ids))]
    else:
        issue_dirs = sorted(path for path in download_root.glob("issue_*") if path.is_dir())

    targets: list[Path] = []
    seen: set[Path] = set()
    for issue_dir in issue_dirs:
        if not issue_dir.is_dir():
            continue
        for path_dir in sorted(path for path in issue_dir.glob("path_*") if path.is_dir()):
            direct_case = path_dir / "sigmastar.1" / "camera4.bin"
            if direct_case.is_file() and path_dir not in seen:
                seen.add(path_dir)
                targets.append(path_dir)

            for child in sorted(path_dir.iterdir()):
                if not child.is_dir() or child.name == "test_data":
                    continue
                if child in seen:
                    continue
                seen.add(child)
                targets.append(child)
    return targets


def result_to_dict(target_root: Path, result: CalibRecoveryResult) -> dict:
    return {
        "target_root": str(target_root),
        "target_path": str(result.target_path),
        "status": result.status,
        "detail": result.detail,
        "source_path": None if result.source_path is None else str(result.source_path),
        "source_kind": result.source_kind,
    }


def build_manifest(
    download_root: Path,
    dry_run: bool,
    issue_ids: list[int],
    target_roots: list[Path],
    results: list[dict],
) -> dict:
    summary: dict[str, int] = {}
    for result in results:
        status = result["status"]
        summary[status] = summary.get(status, 0) + 1

    return {
        "generated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
        "download_root": str(download_root),
        "dry_run": dry_run,
        "issue_filter": issue_ids,
        "target_roots": [str(path) for path in target_roots],
        "summary": summary,
        "results": results,
    }


def print_summary(manifest: dict) -> None:
    print(f"download_root: {manifest['download_root']}")
    print(f"dry_run: {manifest['dry_run']}")
    print(f"target_roots: {len(manifest['target_roots'])}")
    for status, count in sorted(manifest["summary"].items()):
        print(f"{status}: {count}")


def main() -> int:
    args = parse_args()
    download_root = Path(args.download_root).resolve()
    manifest_path = (
        Path(args.manifest_path).resolve()
        if args.manifest_path
        else download_root / "calib_recovery_manifest.json"
    )
    target_roots = discover_target_roots(download_root, args.issue_ids or [], args.path_root)
    log_progress(f"targets_to_process: {len(target_roots)}")

    results: list[dict] = []
    for index, target_root in enumerate(target_roots, start=1):
        log_progress(f"[{index}/{len(target_roots)}] start: {target_root}")
        recovery = recover_camera4_json(
            source_root=target_root,
            target_root=target_root,
            dry_run=args.dry_run,
        )
        results.append(result_to_dict(target_root, recovery))
        log_progress(f"[{index}/{len(target_roots)}] done: {recovery.status}")

    manifest = build_manifest(
        download_root=download_root,
        dry_run=args.dry_run,
        issue_ids=args.issue_ids or [],
        target_roots=target_roots,
        results=results,
    )
    if not args.dry_run:
        manifest_path.parent.mkdir(parents=True, exist_ok=True)
        manifest_path.write_text(
            json.dumps(manifest, ensure_ascii=False, indent=2) + "\n",
            encoding="utf-8",
        )

    print_summary(manifest)
    if args.dry_run:
        print(f"manifest (not written in dry-run): {manifest_path}")
    else:
        print(f"manifest: {manifest_path}")
    return 0


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