QuantEngineByItz/tools/deploy_gas.py

"""deploy_gas.py -- WBS-5.2 GAS auto-deploy script

Bundles src/gas_adapter_parts/ + src/gas/ -> Temp/gas_deploy/ then clasp push.
Usage: python tools/deploy_gas.py [--dry-run] [--skip-push]
"""
import os
import shutil
import json
import argparse
import subprocess
import urllib.request
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
SRC_PARTS = ROOT / "src" / "gas_adapter_parts"
SRC_GAS   = ROOT / "src" / "gas"
DEPLOY_DIR = ROOT / "Temp" / "gas_deploy"

# Resolve a file from multiple candidate directories
def _find(filename: str) -> Path | None:
    candidates = [
        ROOT / filename,
        SRC_PARTS / filename,
        SRC_GAS / "core" / filename,
        SRC_GAS / "collection" / filename,
        SRC_GAS / "engines" / filename,
        SRC_GAS / "reports" / filename,
    ]
    for c in candidates:
        if c.exists():
            return c
    return None

# dst_name -> [src_file, ...] (concatenated in order)
BUNDLE_MAP: dict[str, list[str]] = {
    "appsscript.json":         ["appsscript.json"],
    "gas_lib.gs":              ["gas_lib.gs"],
    "data_feed_base.gs":       ["data_feed_base.gs"],
    "gas_apex_runtime_core.gs":["gas_apex_runtime_core.gs"],
    "gas_data_collect.gs":     ["gas_data_collect.gs"],
    "gas_data_feed.gs":        ["gas_data_feed.gs"],
    "gas_harness_rows.gs":     ["gas_harness_rows.gs"],
    "gas_report.gs":           ["gas_report.gs"],
    "gas_event_calendar.gs":   ["gas_event_calendar.gs"],
    "gas_apex_alpha_watch.gs": ["gas_apex_alpha_watch.gs"],
}

SCRIPT_ID = "1xfeBAeeknmnBtSvrIqWXO_2hc3ByeriLUOSuOOB4YxLLHhN3zdnL7tVh"
PROJECT_ID = "1072944905499"
DEPLOYMENT_ID = "AKfycbzq1XM53XafyCNYurnF9TAQHT3FHBDsBd36rCbCoWSmJD3SaZ1BHCPDYZYhclG9qD5Y"
DEFAULT_WEBAPP_URL = f"https://script.google.com/macros/s/{DEPLOYMENT_ID}/exec"
SECTOR_TREND_JSON = ROOT / "Temp" / "sector_trend_analysis_v1.json"
ETF_REP_JSON = ROOT / "Temp" / "etf_representative_monitor_v1.json"
SECTOR_INSIGHT_BUNDLE = DEPLOY_DIR / "gas_sector_insight_payload.gs"


def get_now_kst() -> str:
    from datetime import datetime, timezone, timedelta
    kst = timezone(timedelta(hours=9))
    return datetime.now(kst).strftime("%Y-%m-%d %H:%M:%S KST")


def build_deploy(dry_run: bool = False) -> bool:
    import re
    print("[deploy_gas] src_parts=" + str(SRC_PARTS))
    print("[deploy_gas] src_gas=  " + str(SRC_GAS))
    print("[deploy_gas] dst=      " + str(DEPLOY_DIR))
    if not dry_run:
        if DEPLOY_DIR.exists():
            shutil.rmtree(DEPLOY_DIR)
        DEPLOY_DIR.mkdir(parents=True, exist_ok=True)

    now_kst = get_now_kst()
    ok = True
    for dst_name, src_files in BUNDLE_MAP.items():
        dst_path = DEPLOY_DIR / dst_name
        parts: list[str] = []
        for sf in src_files:
            src_path = _find(sf)
            if src_path is None:
                print("  WARN: " + sf + " not found")
                ok = False
                continue

            # Update Last Updated timestamp for gas_lib.gs in place before copying
            if sf == "gas_lib.gs" and not dry_run:
                orig_content = src_path.read_text(encoding="utf-8")
                updated_orig = re.sub(
                    r"//\s*Last\s+Updated:\s*\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2}\s+KST",
                    f"// Last Updated: {now_kst}",
                    orig_content
                )
                if updated_orig != orig_content:
                    src_path.write_text(updated_orig, encoding="utf-8")
                    print(f"  [gas_lib.gs] Updated source file 'Last Updated' timestamp to: {now_kst}")
                    parts.append(updated_orig)
                else:
                    parts.append(orig_content)
            else:
                parts.append(src_path.read_text(encoding="utf-8"))

        if not parts:
            continue
        content = "\n".join(parts)
        tag = "[DRY]" if dry_run else "write"
        print("  " + tag + " " + dst_name + " (" + str(len(content)) + " chars)")
        if not dry_run:
            dst_path.write_text(content, encoding="utf-8")

    if not dry_run:
        clasp_cfg = {
            "scriptId": SCRIPT_ID,
            "projectId": PROJECT_ID,
            "rootDir": str(DEPLOY_DIR.relative_to(ROOT)).replace("\\", "/"),
        }
        (ROOT / ".clasp.json").write_text(
            json.dumps(clasp_cfg, ensure_ascii=False, indent=2), encoding="utf-8"
        )
        print("  write .clasp.json -> rootDir=" + clasp_cfg["rootDir"])
    return ok


def clasp_push() -> bool:
    print("[deploy_gas] clasp push -f ...")
    res = subprocess.run(
        ["npx", "@google/clasp", "push", "-f"],
        cwd=str(ROOT),
        shell=True,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )
    print(res.stdout)
    if res.stderr:
        print("STDERR: " + res.stderr[:500])
    if res.returncode == 0:
        print("[deploy_gas] clasp push OK")
        return True
    print("[deploy_gas] clasp push FAILED rc=" + str(res.returncode))
    return False


def clasp_deploy() -> bool:
    print(f"[deploy_gas] clasp deploy -i {DEPLOYMENT_ID} ...")
    now_kst = get_now_kst()
    desc = f"Auto-deployed on {now_kst}"
    res = subprocess.run(
        ["npx", "@google/clasp", "deploy", "-i", DEPLOYMENT_ID, "-d", desc],
        cwd=str(ROOT),
        shell=True,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )
    print(res.stdout)
    if res.stderr:
        print("STDERR: " + res.stderr[:500])
    if res.returncode == 0:
        print("[deploy_gas] clasp deploy OK")
        return True
    print("[deploy_gas] clasp deploy FAILED rc=" + str(res.returncode))
    return False


def _sector_insight_payload() -> dict:
    if not SECTOR_TREND_JSON.exists():
        raise FileNotFoundError(SECTOR_TREND_JSON)
    if not ETF_REP_JSON.exists():
        raise FileNotFoundError(ETF_REP_JSON)
    return {
        "action": "sync_sector_insights",
        "sector_trend_analysis": json.loads(SECTOR_TREND_JSON.read_text(encoding="utf-8")),
        "etf_representative_monitor": json.loads(ETF_REP_JSON.read_text(encoding="utf-8")),
    }


def write_sector_insight_bundle() -> bool:
    try:
        payload = _sector_insight_payload()
    except Exception as exc:
        print("[deploy_gas] cannot build sector insight payload: " + str(exc))
        return False
    bundle = (
        "// Auto-generated by tools/deploy_gas.py\n"
        "// Contains the latest sector insight payload for clasp run fallback.\n"
        "const __SECTOR_INSIGHT_PAYLOAD__ = "
        + json.dumps(payload, ensure_ascii=False, indent=2)
        + ";\n"
        "function syncSectorInsightSheetsFromBundle_() {\n"
        "  return syncSectorInsightSheets(__SECTOR_INSIGHT_PAYLOAD__);\n"
        "}\n"
    )
    SECTOR_INSIGHT_BUNDLE.write_text(bundle, encoding="utf-8")
    print("[deploy_gas] write " + str(SECTOR_INSIGHT_BUNDLE))
    return True


def sync_sector_insights(webapp_url: str) -> bool:
    if not webapp_url:
        print("[deploy_gas] sync-sector-insights requires --webapp-url")
        return False
    try:
        payload = _sector_insight_payload()
    except Exception as exc:
        print("[deploy_gas] missing sector insight data: " + str(exc))
        return False
    body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(
        webapp_url,
        data=body,
        headers={"Content-Type": "application/json; charset=utf-8"},
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=120) as resp:
            text = resp.read().decode("utf-8", errors="replace")
        print("[deploy_gas] sync_sector_insights OK")
        print(text)
        return True
    except Exception as exc:
        print("[deploy_gas] sync_sector_insights FAILED: " + str(exc))
        return False


def sync_sector_insights_via_clasp_run() -> bool:
    if not SECTOR_INSIGHT_BUNDLE.exists():
        print(f"[deploy_gas] missing {SECTOR_INSIGHT_BUNDLE.name}")
        return False
    print("[deploy_gas] clasp run syncSectorInsightSheetsFromBundle_ ...")
    res = subprocess.run(
        ["npx", "@google/clasp", "run", "syncSectorInsightSheetsFromBundle_", "--nondev"],
        cwd=str(ROOT),
        shell=True,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )
    print(res.stdout)
    if res.stderr:
        print("STDERR: " + res.stderr[:500])
    if res.returncode != 0:
        print("[deploy_gas] clasp run syncSectorInsightSheetsFromBundle_ FAILED rc=" + str(res.returncode))
        return False
    print("[deploy_gas] clasp run syncSectorInsightSheetsFromBundle_ OK")
    return True


def run_pre_deploy_linter() -> bool:
    print("[deploy_gas] Running pre-deploy gas thin-adapter audit...")
    # Run auditor v1
    audit_res = subprocess.run(
        ["python", "tools/audit_gas_thin_adapter_v1.py"],
        cwd=str(ROOT),
        shell=True,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )
    if audit_res.returncode != 0:
        print("[deploy_gas] Error: tools/audit_gas_thin_adapter_v1.py failed")
        print(audit_res.stdout)
        print(audit_res.stderr)
        return False

    # Run validator v2
    validate_res = subprocess.run(
        ["python", "tools/validate_gas_thin_adapter_v2.py"],
        cwd=str(ROOT),
        shell=True,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )
    print(validate_res.stdout)
    if validate_res.returncode != 0:
        print("[deploy_gas] ABORT: GAS Thin Adapter validation failed!")
        if validate_res.stderr:
            print("STDERR: " + validate_res.stderr)
        return False

    print("[deploy_gas] Pre-deploy thin-adapter audit PASS")
    return True


def main() -> None:
    parser = argparse.ArgumentParser(description="GAS auto-deploy")
    parser.add_argument("--dry-run", action="store_true", help="List files without writing")
    parser.add_argument("--skip-push", action="store_true", help="Bundle only, skip clasp push")
    parser.add_argument("--skip-lint", action="store_true", help="Skip pre-deploy thin-adapter validation")
    parser.add_argument("--sync-sector-insights", action="store_true", help="POST sector insight JSON to a deployed GAS web app")
    parser.add_argument("--webapp-url", default=os.environ.get("GAS_WEBAPP_URL", DEFAULT_WEBAPP_URL), help="Apps Script web app URL for sync POST")
    args = parser.parse_args()

    if not args.skip_lint:
        if not run_pre_deploy_linter():
            raise SystemExit(1)

    ok = build_deploy(dry_run=args.dry_run)
    if not ok:
        print("[deploy_gas] Some source files missing -- check warnings above")
        raise SystemExit(1)

    if args.sync_sector_insights and not args.dry_run and not args.skip_push:
        if not write_sector_insight_bundle():
            raise SystemExit(1)

    if args.dry_run or args.skip_push:
        print("[deploy_gas] dry-run/skip-push -- push skipped")
        if args.sync_sector_insights:
            print("[deploy_gas] sync skipped because push/deploy was skipped")
        return

    if not clasp_push():
        raise SystemExit(1)

    if not clasp_deploy():
        raise SystemExit(1)

    if args.sync_sector_insights:
        if not sync_sector_insights(args.webapp_url):
            print("[deploy_gas] webapp sync failed; falling back to clasp run")
            if not sync_sector_insights_via_clasp_run():
                raise SystemExit(1)

    print("[deploy_gas] Done. To run_all: python tools/automate_routine.py")


if __name__ == "__main__":
    main()