MobileModels/tools/sync_upstream_mobilemodels.py

#!/usr/bin/env python3
"""Sync selected upstream MobileModels data into this repository."""

from __future__ import annotations

import argparse
import filecmp
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path

from project_layout import PROJECT_ROOT, WORKSPACE_ROOT

DEFAULT_REPO_URL = "https://github.com/KHwang9883/MobileModels.git"
DEFAULT_BRANCH = "master"
SYNC_PATHS = [
    "brands",
    "misc",
    "CHANGELOG.md",
    "CHANGELOG_en.md",
    "LICENSE.txt",
]

LOCAL_TEXT_REPLACEMENTS = {
    "brands/oneplus.md": [
        (
            """**一加 Ace 3 / 一加 12R (`aston`) / 一加 Ace 3 原神刻晴定制机 / 一加 12R Genshin Impact Edition (`martin`):**

`PJE110`: 一加 Ace 3 国行版 / 原神刻晴定制机

`CPH2585`: 一加 12R 印度版 / 原神刻晴定制机""",
            """**一加 Ace 3 (`aston`):**

`PJE110`: 一加 Ace 3 国行版 / 原神刻晴定制机

**一加 12R (`aston`):**

`CPH2585`: 一加 12R 印度版 / 原神刻晴定制机""",
        ),
    ],
    "brands/oneplus_en.md": [
        (
            """**OnePlus Ace 3 / OnePlus 12R (`aston`) / OnePlus Ace 3 Genshin Impact Edition / OnePlus 12R Genshin Impact Edition (`martin`):**

`PJE110`: OnePlus Ace 3 China / Genshin Impact Edition

`CPH2585`: OnePlus 12R India / Genshin Impact Edition""",
            """**OnePlus Ace 3 (`aston`):**

`PJE110`: OnePlus Ace 3 China / Genshin Impact Edition

**OnePlus 12R (`aston`):**

`CPH2585`: OnePlus 12R India / Genshin Impact Edition""",
        ),
    ],
}


def run(cmd: list[str], cwd: Path | None = None) -> None:
    subprocess.run(cmd, cwd=cwd or PROJECT_ROOT, check=True)


def remove_path(path: Path) -> None:
    if path.is_dir():
        shutil.rmtree(path)
    elif path.exists():
        path.unlink()


def sync_path(src: Path, dst: Path) -> None:
    if src.is_dir():
        dst.mkdir(parents=True, exist_ok=True)
        source_children = {child.name for child in src.iterdir()}

        for existing in dst.iterdir():
            if existing.name not in source_children:
                remove_path(existing)

        for child in src.iterdir():
            sync_path(child, dst / child.name)
        return

    dst.parent.mkdir(parents=True, exist_ok=True)
    if dst.exists() and filecmp.cmp(src, dst, shallow=False):
        return
    shutil.copy2(src, dst)


def sync_selected_paths(upstream_root: Path) -> None:
    for relative_path in SYNC_PATHS:
        src = upstream_root / relative_path
        dst = WORKSPACE_ROOT / relative_path
        if not src.exists():
            raise FileNotFoundError(f"Missing upstream path: {relative_path}")
        sync_path(src, dst)

    apply_local_corrections()


def apply_local_corrections() -> None:
    for relative_path, replacements in LOCAL_TEXT_REPLACEMENTS.items():
        path = WORKSPACE_ROOT / relative_path
        if not path.exists():
            continue
        text = path.read_text(encoding="utf-8")
        updated = text
        for old, new in replacements:
            updated = updated.replace(old, new)
        if updated != text:
            path.write_text(updated, encoding="utf-8")


def build_index(output_path: str) -> None:
    run(
        [
            sys.executable,
            str(PROJECT_ROOT / "tools/device_mapper.py"),
            "--repo-root",
            str(WORKSPACE_ROOT),
            "build",
            "--output",
            output_path,
        ]
    )


def export_mysql_seed(output_path: str) -> None:
    run(
        [
            sys.executable,
            str(PROJECT_ROOT / "tools/export_mysql_seed.py"),
            "--output",
            output_path,
            "--repo-root",
            str(WORKSPACE_ROOT),
        ]
    )


def load_mysql_seed(seed_path: str) -> None:
    run(
        [
            sys.executable,
            str(PROJECT_ROOT / "tools/load_mysql_seed.py"),
            "--seed",
            seed_path,
        ]
    )


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Sync upstream MobileModels raw data and optionally rebuild the device index."
    )
    parser.add_argument("--repo-url", default=DEFAULT_REPO_URL, help="Upstream git repository URL")
    parser.add_argument("--branch", default=DEFAULT_BRANCH, help="Upstream branch to sync from")
    parser.add_argument(
        "--build-index",
        action="store_true",
        help="Rebuild dist/device_index.json after syncing upstream data",
    )
    parser.add_argument(
        "--index-output",
        default="dist/device_index.json",
        help="Output path for the rebuilt device index",
    )
    parser.add_argument(
        "--export-mysql-seed",
        action="store_true",
        help="Export MySQL seed SQL after syncing upstream data",
    )
    parser.add_argument(
        "--mysql-seed-output",
        default="dist/mobilemodels_mysql_seed.sql",
        help="Output path for the exported MySQL seed SQL",
    )
    parser.add_argument(
        "--load-mysql",
        action="store_true",
        help="Load schema and seed data into MySQL after exporting seed SQL",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()

    with tempfile.TemporaryDirectory(prefix="mobilemodels-upstream-") as tmpdir:
        upstream_root = Path(tmpdir) / "upstream"
        run(
            [
                "git",
                "clone",
                "--depth",
                "1",
                "--branch",
                args.branch,
                args.repo_url,
                str(upstream_root),
            ]
        )
        sync_selected_paths(upstream_root)

    if args.build_index:
        build_index(args.index_output)

    if args.export_mysql_seed or args.load_mysql:
        export_mysql_seed(args.mysql_seed_output)

    if args.load_mysql:
        load_mysql_seed(args.mysql_seed_output)

    return 0


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