cloud_utility-deploy_cloud_run_job.py

"""
Google Cloud Run Job Deployment Script for Milhas WhatsApp Agent

This script automates deployment of the agent as a Cloud Run Job (batch),
staying within the free tier.
"""

import os
import subprocess
import sys
from pathlib import Path
from typing import Dict, List, Tuple

GOOGLE_CLOUD_PROJECT_ID = "GOOGLE_CLOUD_PROJECT_ID"  #! Set your project ID
JOB_NAME = "job-name" #! Set your job name
REGION = "us-central1"
MEMORY = "512Mi"
CPU = "1"
TIMEOUT = "600s"
TASKS = 1
MAX_RETRIES = 3

def run_command(command: List[str], cwd: str = None) -> Tuple[bool, str]:
    try:
        result = subprocess.run(
            command,
            check=True,
            cwd=cwd,
            capture_output=True,
            text=True,
            shell=sys.platform == "win32",
        )
        return True, result.stdout
    except subprocess.CalledProcessError as e:
        print(f"Command failed: {' '.join(e.cmd)}")
        print(f"Error: {e.stderr}")
        return False, f"Error: {e.stderr}"

def check_gcloud_installed() -> bool:
    success, _ = run_command(["gcloud", "--version"])

    if not success:
        print("gcloud CLI is not installed or not in PATH.")
        return False

    success, output = run_command(["gcloud", "auth", "list", "--filter=status:ACTIVE", "--format=value(account)"])
    if not success or not output.strip():
        print("You need to authenticate with Google Cloud (gcloud auth login).")
        return False

    return True

def build_and_push_image(project_id: str, job_name: str) -> bool:
    image_name = f"gcr.io/{project_id}/{job_name}"

    print(f"Building Docker image: {image_name}")
    success, output = run_command(["docker", "build", "-t", job_name, "."])
    if not success:
        print(f"Failed to build Docker image: {output}")
        return False

    print("✓ Docker image built successfully")

    print("Tagging image...")
    success, output = run_command(["docker", "tag", job_name, image_name])
    if not success:
        print(f"Failed to tag image: {output}")
        return False

    print("✓ Image tagged successfully")

    print("Pushing image to Google Container Registry...")
    success, output = run_command(["docker", "push", image_name])
    if not success:
        print(f"Failed to push image: {output}")
        return False

    print("✓ Image pushed successfully")
    return True

def deploy_cloud_run_job(
        project_id: str,
        job_name: str,
        region: str,
        memory: str,
        cpu: str,
        timeout: str,
        tasks: int,
        max_retries: int,
        env_file: str = ".env"
) -> bool:
    from dotenv import dotenv_values

    image_name = f"gcr.io/{project_id}/{job_name}"
    env_vars = {}
    if os.path.exists(env_file):
        env_vars = dotenv_values(env_file)
        env_vars = {k: v for k, v in env_vars.items() if v is not None}
    env_vars_str = ",".join(f"{k}={v}" for k, v in env_vars.items())

    cmd = [
        "gcloud", "beta", "run", "jobs", "deploy", job_name,
        f"--image={image_name}",
        f"--region={region}",
        f"--memory={memory}",
        f"--cpu={cpu}",
        f"--max-retries={max_retries}",
        f"--task-timeout={timeout}",
        f"--tasks={tasks}",
        # "--platform=managed"
    ]

    if env_vars:
        cmd.append(f"--set-env-vars={env_vars_str}")

    print("Deploying Cloud Run Job...")
    success, output = run_command(cmd)
    if success:
        print("\n✓ Job deployed successfully!")
        print(output)
        return True
    else:
        print("\n✗ Job deployment failed!")
        print(output)
        return False

def main():
    print("=== Milhas WhatsApp Agent Cloud Run Job Deployment ===\n")

    if not check_gcloud_installed():
        sys.exit(1)

    if not build_and_push_image(GOOGLE_CLOUD_PROJECT_ID, JOB_NAME):
        print("\n✗ Failed to build and push Docker image")
        sys.exit(1)

    print("\n=== Deploying Cloud Run Job ===")
    success = deploy_cloud_run_job(
        project_id=GOOGLE_CLOUD_PROJECT_ID,
        job_name=JOB_NAME,
        region=REGION,
        memory=MEMORY,
        cpu=CPU,
        timeout=TIMEOUT,
        tasks=TASKS,
        max_retries=MAX_RETRIES
    )

    if success:
        print(f"\n✓ Job deployment completed successfully!")
        print(f"Job URL: https://console.cloud.google.com/run/jobs/details/{REGION}/{JOB_NAME}?project={GOOGLE_CLOUD_PROJECT_ID}")
    else:
        print("\n✗ Job deployment failed. Check the error messages above.")
        sys.exit(1)

if __name__ == "__main__":
    main()