xiaobaifupan/app/tools/build_preservation_manifest.py

from __future__ import annotations

import argparse
import hashlib
import json
from datetime import datetime
from pathlib import Path


REPOSITORY_ROOT = Path(__file__).resolve().parents[2]

DIRECTORIES = (
    "backend",
    "config",
    "static",
    "tests",
    "tools",
    "vendor",
    "游资skills",
)

ROOT_FILES = (
    ".dockerignore",
    ".env.example",
    ".gitignore",
    "advanced_strategies.py",
    "alert_service.py",
    "api_access.py",
    "app_config.py",
    "ARCHITECTURE.md",
    "assistant_agent.py",
    "chart_data_provider.py",
    "compose.yaml",
    "database.py",
    "demo_data.py",
    "Dockerfile",
    "DOCKER_DEPLOY.md",
    "heaven_agent.py",
    "heaven_engine.py",
    "ifind_client.py",
    "llm_strategy.py",
    "llm_stream.py",
    "market_insights.py",
    "mentor_agent.py",
    "package-lock.json",
    "package.json",
    "playwright.config.js",
    "README.md",
    "realtime_aggregator.py",
    "requirements.txt",
    "screener.py",
    "security.py",
    "sentiment_engine.py",
    "server.py",
    "strategy_tracking.py",
    "sync_data.py",
    "THIRD_PARTY_NOTICES.md",
    "trade_journal.py",
    "tushare_client.py",
)


def digest(path: Path) -> str:
    checksum = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(1024 * 1024), b""):
            checksum.update(block)
    return checksum.hexdigest()


def display_path(path: Path, base: Path) -> str:
    try:
        return path.relative_to(base).as_posix() or "."
    except ValueError:
        return path.as_posix()


def source_files(source_root: Path) -> list[Path]:
    files = [source_root / name for name in ROOT_FILES]
    for directory in DIRECTORIES:
        root = source_root / directory
        if not root.is_dir():
            continue
        files.extend(
            path
            for path in root.rglob("*")
            if path.is_file() and "__pycache__" not in path.parts
        )
    return sorted(
        set(files), key=lambda path: path.relative_to(source_root).as_posix()
    )


def build_manifest(
    source_root: Path,
    target_root: Path,
    source_commit: str,
) -> dict[str, object]:
    assets = []
    mismatches = []
    for source in source_files(source_root):
        relative = source.relative_to(source_root)
        target = target_root / relative
        source_exists = source.is_file()
        source_hash = digest(source) if source_exists else ""
        target_hash = digest(target) if target.is_file() else ""
        if not source_exists:
            status = "missing_source"
        elif not target.is_file():
            status = "missing_target"
        elif source_hash == target_hash:
            status = "identical"
        else:
            status = "mismatch"
        if status != "identical":
            mismatches.append(relative.as_posix())
        assets.append(
            {
                "source": relative.as_posix(),
                "target": display_path(target, REPOSITORY_ROOT),
                "bytes": source.stat().st_size if source_exists else 0,
                "sha256": source_hash,
                "disposition": "original_copy_pending_move",
                "status": status,
            }
        )
    return {
        "schema_version": 1,
        "generated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
        "source_commit": source_commit,
        "source_root": display_path(source_root, REPOSITORY_ROOT),
        "target_root": display_path(target_root, REPOSITORY_ROOT),
        "excluded": [
            "next/",
            "data/review.db and SQLite sidecars",
            "data/private-mentor-skills/",
            ".env",
            "node_modules/",
            "logs, caches and generated test results",
        ],
        "asset_count": len(assets),
        "mismatch_count": len(mismatches),
        "mismatches": mismatches,
        "assets": assets,
    }


def main() -> int:
    parser = argparse.ArgumentParser(
        description=(
            "Build an exact-copy manifest for a preservation migration stage. "
            "This is a migration-only tool; it does not describe the final moved layout."
        )
    )
    parser.add_argument("--source-root", type=Path, default=REPOSITORY_ROOT)
    parser.add_argument("--target-root", type=Path, default=REPOSITORY_ROOT / "app")
    parser.add_argument("--source-commit", default="")
    parser.add_argument(
        "--output",
        type=Path,
        required=True,
        help="write to a new audit path; do not overwrite committed slice evidence",
    )
    args = parser.parse_args()
    source_root = args.source_root.resolve()
    target_root = args.target_root.resolve()
    output = args.output.resolve()
    manifest = build_manifest(source_root, target_root, args.source_commit)
    output.parent.mkdir(parents=True, exist_ok=True)
    output.write_text(
        json.dumps(manifest, ensure_ascii=False, indent=2) + "\n",
        encoding="utf-8",
    )
    print(
        f"assets={manifest['asset_count']} mismatches={manifest['mismatch_count']} "
        f"output={output}"
    )
    return 1 if manifest["mismatch_count"] else 0


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