#!/usr/bin/env python3 """AI Deploy Script — Build, Test, Deploy with automatic rollback. Usage: python scripts/ai_deploy.py python scripts/ai_deploy.py --skip-tests python scripts/ai_deploy.py --dry-run python scripts/ai_deploy.py --skip-build --skip-tests Pipeline phases: 1. Build — docker build (or npm run build for frontend-only) 2. Tests — pytest (backend) + vitest (frontend) 3. Deploy — Coolify API call to redeploy 4. Rollback — on failure, restore previous deployment Environment variables: COOLIFY_API_TOKEN — Coolify API token for deployment COOLIFY_APP_UUID — Coolify application UUID COOLIFY_BASE_URL — Coolify base URL (default: https://server.media-on.de) Exit codes: 0 — deployment successful 1 — deployment failed (rollback attempted) 2 — configuration error """ from __future__ import annotations import argparse import json import os import shutil import subprocess import sys import time from dataclasses import dataclass, field from typing import Any # Add project root to path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import httpx # ─── Data Structures ───────────────────────────────────────────────── @dataclass class PhaseResult: """Result of a single pipeline phase.""" name: str success: bool = False duration_s: float = 0.0 output: str = "" error: str = "" metadata: dict[str, Any] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: return { "phase": self.name, "success": self.success, "duration_s": round(self.duration_s, 2), "output": self.output[:500] if self.output else "", "error": self.error[:500] if self.error else "", "metadata": self.metadata, } @dataclass class DeployReport: """Full deployment report.""" started_at: str = "" finished_at: str = "" phases: list[PhaseResult] = field(default_factory=list) overall_success: bool = False rollback_performed: bool = False rollback_success: bool = False def to_dict(self) -> dict[str, Any]: return { "started_at": self.started_at, "finished_at": self.finished_at, "overall_success": self.overall_success, "rollback_performed": self.rollback_performed, "rollback_success": self.rollback_success, "phases": [p.to_dict() for p in self.phases], } # ─── Phase Implementations ─────────────────────────────────────────── def run_command(cmd: list[str], cwd: str | None = None, timeout: int = 600) -> tuple[bool, str, str]: """Run a shell command and return (success, stdout, stderr).""" try: result = subprocess.run( cmd, cwd=cwd, capture_output=True, text=True, timeout=timeout, ) return result.returncode == 0, result.stdout, result.stderr except subprocess.TimeoutExpired: return False, "", f"Command timed out after {timeout}s" except FileNotFoundError as exc: return False, "", str(exc) except Exception as exc: return False, "", str(exc) def phase_build(skip_build: bool, dry_run: bool) -> PhaseResult: """Build phase: docker build or npm run build.""" start = time.perf_counter() result = PhaseResult(name="build") if skip_build: result.success = True result.output = "Build skipped (--skip-build)" result.duration_s = time.perf_counter() - start return result project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if dry_run: result.success = True result.output = "[DRY-RUN] Would run: docker build -t leocrm:latest ." result.duration_s = time.perf_counter() - start return result # Try docker build first docker_available = shutil.which("docker") is not None if docker_available: success, stdout, stderr = run_command( ["docker", "build", "-t", "leocrm:latest", "."], cwd=project_root, timeout=900, ) result.success = success result.output = stdout result.error = stderr result.metadata["build_tool"] = "docker" else: # Fallback: frontend-only build with npm frontend_dir = os.path.join(project_root, "frontend") npm_available = shutil.which("npm") is not None if npm_available and os.path.isdir(frontend_dir): success, stdout, stderr = run_command( ["npm", "run", "build"], cwd=frontend_dir, timeout=300, ) result.success = success result.output = stdout result.error = stderr result.metadata["build_tool"] = "npm" else: result.success = False result.error = "Neither docker nor npm available for build" result.duration_s = time.perf_counter() - start return result def phase_tests(skip_tests: bool, dry_run: bool) -> PhaseResult: """Test phase: pytest (backend) + vitest (frontend).""" start = time.perf_counter() result = PhaseResult(name="tests") if skip_tests: result.success = True result.output = "Tests skipped (--skip-tests)" result.duration_s = time.perf_counter() - start return result if dry_run: result.success = True result.output = "[DRY-RUN] Would run: pytest + vitest" result.duration_s = time.perf_counter() - start return result project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) python_bin = os.environ.get("LEOCRM_PYTHON", "/opt/venv/bin/python") test_outputs: list[str] = [] test_errors: list[str] = [] all_success = True # Backend tests: pytest if os.path.isfile(os.path.join(project_root, "pyproject.toml")): success, stdout, stderr = run_command( [python_bin, "-m", "pytest", "tests/", "-x", "--tb=short", "-q"], cwd=project_root, timeout=300, ) test_outputs.append(f"=== pytest ===\n{stdout}") if stderr: test_errors.append(f"=== pytest stderr ===\n{stderr}") if not success: all_success = False result.metadata["pytest_passed"] = False else: result.metadata["pytest_passed"] = True # Frontend tests: vitest frontend_dir = os.path.join(project_root, "frontend") npm_available = shutil.which("npm") is not None if npm_available and os.path.isdir(frontend_dir): success, stdout, stderr = run_command( ["npm", "run", "test"], cwd=frontend_dir, timeout=300, ) test_outputs.append(f"=== vitest ===\n{stdout}") if stderr: test_errors.append(f"=== vitest stderr ===\n{stderr}") if not success: all_success = False result.metadata["vitest_passed"] = False else: result.metadata["vitest_passed"] = True result.success = all_success result.output = "\n".join(test_outputs) result.error = "\n".join(test_errors) result.duration_s = time.perf_counter() - start return result def phase_deploy(dry_run: bool) -> PhaseResult: """Deploy phase: Coolify API call to redeploy.""" start = time.perf_counter() result = PhaseResult(name="deploy") api_token = os.environ.get("COOLIFY_API_TOKEN", "") app_uuid = os.environ.get("COOLIFY_APP_UUID", "") base_url = os.environ.get("COOLIFY_BASE_URL", "https://server.media-on.de") if not api_token or not app_uuid: result.success = False result.error = "Missing COOLIFY_API_TOKEN or COOLIFY_APP_UUID environment variables" result.duration_s = time.perf_counter() - start return result if dry_run: result.success = True result.output = f"[DRY-RUN] Would call: POST {base_url}/api/v1/applications/{app_uuid}/deploy" result.metadata["api_url"] = f"{base_url}/api/v1/applications/{app_uuid}/deploy" result.duration_s = time.perf_counter() - start return result deploy_url = f"{base_url}/api/v1/applications/{app_uuid}/deploy" headers = {"Authorization": f"Bearer {api_token}"} try: response = httpx.post(deploy_url, headers=headers, timeout=120.0) if response.status_code < 400: result.success = True result.output = f"Deploy triggered: HTTP {response.status_code}" try: result.metadata["response"] = response.json() except Exception: result.metadata["response_text"] = response.text[:200] else: result.success = False result.error = f"Deploy failed: HTTP {response.status_code} - {response.text[:300]}" except httpx.TimeoutException: result.success = False result.error = "Deploy request timed out" except httpx.ConnectError as exc: result.success = False result.error = f"Connection error: {exc}" except Exception as exc: result.success = False result.error = str(exc) result.duration_s = time.perf_counter() - start return result def phase_rollback(dry_run: bool) -> PhaseResult: """Rollback phase: restore previous deployment via Coolify API.""" start = time.perf_counter() result = PhaseResult(name="rollback") api_token = os.environ.get("COOLIFY_API_TOKEN", "") app_uuid = os.environ.get("COOLIFY_APP_UUID", "") base_url = os.environ.get("COOLIFY_BASE_URL", "https://server.media-on.de") if not api_token or not app_uuid: result.success = False result.error = "Missing COOLIFY_API_TOKEN or COOLIFY_APP_UUID for rollback" result.duration_s = time.perf_counter() - start return result if dry_run: result.success = True result.output = f"[DRY-RUN] Would call: POST {base_url}/api/v1/applications/{app_uuid}/rollback" result.duration_s = time.perf_counter() - start return result # Coolify rollback: restart with previous image rollback_url = f"{base_url}/api/v1/applications/{app_uuid}/restart" headers = {"Authorization": f"Bearer {api_token}"} try: response = httpx.post(rollback_url, headers=headers, timeout=120.0) if response.status_code < 400: result.success = True result.output = f"Rollback triggered: HTTP {response.status_code}" else: result.success = False result.error = f"Rollback failed: HTTP {response.status_code} - {response.text[:300]}" except Exception as exc: result.success = False result.error = str(exc) result.duration_s = time.perf_counter() - start return result # ─── Pipeline Orchestration ────────────────────────────────────────── def run_pipeline(skip_build: bool, skip_tests: bool, dry_run: bool) -> DeployReport: """Run the full deployment pipeline.""" from datetime import datetime, timezone report = DeployReport() report.started_at = datetime.now(timezone.utc).isoformat() print(f"\n{'='*60}") print(f" LeoCRM AI Deploy Pipeline") print(f" {'[DRY-RUN]' if dry_run else '[LIVE]'}") print(f"{'='*60}\n") # Phase 1: Build print("▶ Phase 1: Build") build_result = phase_build(skip_build, dry_run) report.phases.append(build_result) _print_phase_result(build_result) if not build_result.success: print("\n✗ Build failed — aborting pipeline.") report.overall_success = False report.finished_at = datetime.now(timezone.utc).isoformat() return report # Phase 2: Tests print("\n▶ Phase 2: Tests") test_result = phase_tests(skip_tests, dry_run) report.phases.append(test_result) _print_phase_result(test_result) if not test_result.success: print("\n✗ Tests failed — aborting pipeline (no deploy).") report.overall_success = False report.finished_at = datetime.now(timezone.utc).isoformat() return report # Phase 3: Deploy print("\n▶ Phase 3: Deploy") deploy_result = phase_deploy(dry_run) report.phases.append(deploy_result) _print_phase_result(deploy_result) if not deploy_result.success: # Phase 4: Rollback print("\n✗ Deploy failed — initiating rollback...") rollback_result = phase_rollback(dry_run) report.phases.append(rollback_result) report.rollback_performed = True report.rollback_success = rollback_result.success _print_phase_result(rollback_result) report.overall_success = False else: report.overall_success = True print("\n✓ Deployment successful!") report.finished_at = datetime.now(timezone.utc).isoformat() return report def _print_phase_result(result: PhaseResult) -> None: """Print a phase result summary.""" status = "✓" if result.success else "✗" print(f" {status} {result.name}: {'PASS' if result.success else 'FAIL'} ({result.duration_s:.1f}s)") if result.output: for line in result.output.strip().split("\n")[-5:]: print(f" │ {line}") if result.error: for line in result.error.strip().split("\n")[-3:]: print(f" │ ERROR: {line}", file=sys.stderr) # ─── CLI Entry Point ───────────────────────────────────────────────── def main() -> int: parser = argparse.ArgumentParser( description="LeoCRM AI Deploy — Build, Test, Deploy with automatic rollback." ) parser.add_argument( "--skip-tests", action="store_true", help="Skip the test phase", ) parser.add_argument( "--skip-build", action="store_true", help="Skip the build phase", ) parser.add_argument( "--dry-run", action="store_true", help="Simulate the pipeline without making changes", ) parser.add_argument( "--output", default="-", help="Write structured JSON report to file (default: stdout)", ) args = parser.parse_args() report = run_pipeline( skip_build=args.skip_build, skip_tests=args.skip_tests, dry_run=args.dry_run, ) # Output structured report report_json = json.dumps(report.to_dict(), indent=2, ensure_ascii=False) if args.output == "-": print(f"\n{'='*60}") print(" Deployment Report") print(f"{'='*60}") print(report_json) else: with open(args.output, "w", encoding="utf-8") as f: f.write(report_json) print(f"\nReport written to {args.output}", file=sys.stderr) return 0 if report.overall_success else 1 if __name__ == "__main__": sys.exit(main())