from __future__ import annotations from pathlib import Path from tempfile import TemporaryDirectory from typing import Optional import argparse import os import tarfile import time import urllib.request try: import paramiko except ImportError as exc: # pragma: no cover - local tooling guard raise SystemExit("paramiko_missing: install paramiko in the local venv before deploying") from exc ROOT = Path(__file__).resolve().parents[1] DEFAULT_HOST = "192.168.3.90" DEFAULT_USER = "root" DEFAULT_REMOTE_PATH = "/root/deploy/ai-ict" DEFAULT_REMOTE_ARCHIVE = "/root/ai-ict-deploy.tgz" DEFAULT_SMOKE_BASE_URL = "http://192.168.3.90:8000" DEPLOY_PATHS = [ "Dockerfile", ".dockerignore", "docker-compose.yml", "pyproject.toml", "README.md", ".env.example", "docs", "SESSION.md", "TASKS.md", "src", "scripts", "migrations", "data", ] def main(argv: Optional[list[str]] = None, env=None) -> int: env = os.environ if env is None else env args = _parse_args(argv, env) if not args.password: raise SystemExit("missing_password: set AI_ICT_DEPLOY_PASSWORD or pass --password") with TemporaryDirectory() as tmp_dir: archive_path = Path(tmp_dir) / "ai-ict-deploy.tgz" create_archive(archive_path) client = connect_ssh(host=args.host, user=args.user, password=args.password, port=args.port) try: run_remote(client, f"mkdir -p {args.remote_path}") upload_file(client, archive_path, args.remote_archive) run_remote( client, build_deploy_command( remote_path=args.remote_path, remote_archive=args.remote_archive, ), timeout=600, ) finally: client.close() smoke = run_http_smoke(args.smoke_base_url, attempts=args.smoke_attempts, sleep_seconds=args.smoke_sleep_seconds) print(f"deploy_complete host={args.host} remote_path={args.remote_path}") print(f"smoke_complete endpoints={len(smoke)}") return 0 def create_archive(archive_path: Path) -> None: with tarfile.open(archive_path, "w:gz") as archive: for relative in DEPLOY_PATHS: path = ROOT / relative if not path.exists(): continue archive.add(path, arcname=relative, filter=_tar_filter) def connect_ssh(*, host: str, user: str, password: str, port: int): client = paramiko.SSHClient() client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) client.connect( hostname=host, port=port, username=user, password=password, look_for_keys=False, allow_agent=False, timeout=20, banner_timeout=20, auth_timeout=20, ) return client def upload_file(client, local_path: Path, remote_path: str) -> None: with client.open_sftp() as sftp: sftp.put(str(local_path), remote_path) def run_remote(client, command: str, *, timeout: int = 120) -> str: stdin, stdout, stderr = client.exec_command(command, timeout=timeout) exit_status = stdout.channel.recv_exit_status() out = stdout.read().decode("utf-8", errors="replace") err = stderr.read().decode("utf-8", errors="replace") if exit_status != 0: raise RuntimeError(f"remote_command_failed:{exit_status}\n{command}\nSTDOUT:\n{out}\nSTDERR:\n{err}") return out def build_deploy_command(*, remote_path: str, remote_archive: str) -> str: remote_path_q = _sh_quote(remote_path) remote_archive_q = _sh_quote(remote_archive) env_backup = "/root/ai-ict-deploy.env.bak" return " && ".join( [ f"mkdir -p {remote_path_q}", f"if [ -f {remote_path_q}/.env ]; then cp {remote_path_q}/.env {_sh_quote(env_backup)}; fi", f"rm -rf {remote_path_q}/*", f"tar -xzf {remote_archive_q} -C {remote_path_q}", f"if [ -f {_sh_quote(env_backup)} ]; then mv {_sh_quote(env_backup)} {remote_path_q}/.env; fi", f"cd {remote_path_q}", "((docker compose down || docker-compose down) >/dev/null 2>&1 || true)", "((docker compose up -d --build) || (docker-compose up -d --build))", "((docker compose ps) || (docker-compose ps))", ] ) def run_http_smoke(base_url: str, *, attempts: int = 10, sleep_seconds: float = 3.0) -> list[dict]: endpoints = [ "/health", "/trading/runtime-status", "/trading/research-council/optimization-cycle/latest?instrument_id=1", "/trading/research-council/optimization-cycles?instrument_id=1&limit=1", "/trading/research-council/optimization-review?instrument_id=1", "/trading/review-packet?instrument_id=1", ] results = [] for endpoint in endpoints: url = f"{base_url.rstrip('/')}{endpoint}" last_error = None for attempt in range(1, attempts + 1): try: with urllib.request.urlopen(url, timeout=20) as response: body = response.read(4096) results.append({"url": url, "status": response.status, "bytes": len(body), "attempt": attempt}) break except Exception as exc: # pragma: no cover - exercised by live deploys last_error = exc if attempt < attempts: time.sleep(sleep_seconds) else: raise RuntimeError(f"http_smoke_failed:{url}:{last_error}") return results def _tar_filter(info: tarfile.TarInfo): normalized = info.name.replace("\\", "/") ignored_parts = {".git", ".venv", "__pycache__", ".pytest_cache", "node_modules"} if any(part in ignored_parts for part in normalized.split("/")): return None return info def _parse_args(argv: Optional[list[str]], env) -> argparse.Namespace: parser = argparse.ArgumentParser(description="Deploy ai-exchange over password SSH without expect.") parser.add_argument("--host", default=env.get("AI_ICT_DEPLOY_HOST", DEFAULT_HOST)) parser.add_argument("--port", type=int, default=int(env.get("AI_ICT_DEPLOY_PORT", "22"))) parser.add_argument("--user", default=env.get("AI_ICT_DEPLOY_USER", DEFAULT_USER)) parser.add_argument("--password", default=env.get("AI_ICT_DEPLOY_PASSWORD", "")) parser.add_argument("--remote-path", default=env.get("AI_ICT_DEPLOY_REMOTE_PATH", DEFAULT_REMOTE_PATH)) parser.add_argument("--remote-archive", default=env.get("AI_ICT_DEPLOY_REMOTE_ARCHIVE", DEFAULT_REMOTE_ARCHIVE)) parser.add_argument("--smoke-base-url", default=env.get("AI_ICT_DEPLOY_SMOKE_BASE_URL", DEFAULT_SMOKE_BASE_URL)) parser.add_argument("--smoke-attempts", type=int, default=int(env.get("AI_ICT_DEPLOY_SMOKE_ATTEMPTS", "10"))) parser.add_argument("--smoke-sleep-seconds", type=float, default=float(env.get("AI_ICT_DEPLOY_SMOKE_SLEEP_SECONDS", "3"))) return parser.parse_args(argv) def _sh_quote(value: str) -> str: return "'" + value.replace("'", "'\"'\"'") + "'" if __name__ == "__main__": raise SystemExit(main())