"""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()