Files
QuantEngineByItz/tools/deploy_gas.py
T

196 lines
6.5 KiB
Python

"""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
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 = [
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"],
# gdc_01 + gdc_02 bundled as single file (GAS project legacy name)
"gas_data_collect.gs": [
"gdc_01_fetch_fundamentals.gs",
"gdc_02_account_satellite.gs",
],
"gdf_01_price_metrics.gs": ["gdf_01_price_metrics.gs"],
"gdf_02_harness_assembly.gs": ["gdf_02_harness_assembly.gs"],
"gdf_03_portfolio_gates.gs": ["gdf_03_portfolio_gates.gs"],
"gdf_04_execution_quality.gs":["gdf_04_execution_quality.gs"],
"gdf_05_alpha_engines.gs": ["gdf_05_alpha_engines.gs"],
"gdf_06_rebalance.gs": ["gdf_06_rebalance.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"
DEPLOYMENT_ID = "AKfycbzq1XM53XafyCNYurnF9TAQHT3FHBDsBd36rCbCoWSmJD3SaZ1BHCPDYZYhclG9qD5Y"
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,
"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 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")
args = parser.parse_args()
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.dry_run or args.skip_push:
print("[deploy_gas] dry-run/skip-push -- push skipped")
return
if not clasp_push():
raise SystemExit(1)
if not clasp_deploy():
raise SystemExit(1)
print("[deploy_gas] Done. To run_all: python tools/automate_routine.py")
if __name__ == "__main__":
main()