platform-codebase/@packages/@infrastructure/egirl-infra/integrations/gitlab/monitor_pipeline.py

#!/usr/bin/env python3
"""Monitor GitLab CI pipeline until completion."""

import sys
import os
import time
import json
from urllib.request import Request, urlopen
from datetime import datetime

PIPELINE_ID = os.getenv("GITLAB_PIPELINE_ID")
PROJECT_ID = os.getenv("GITLAB_PROJECT_ID", "transftw%2Flilith-platform")
TOKEN = os.getenv("GITLAB_TOKEN")
BASE_URL = f"https://gitlab.com/api/v4/projects/{PROJECT_ID}"

if not TOKEN:
    print("Error: GITLAB_TOKEN environment variable not set")
    print("Load from .env or export GITLAB_TOKEN=your-token")
    sys.exit(1)

if not PIPELINE_ID:
    print("Error: GITLAB_PIPELINE_ID environment variable not set")
    print("Usage: export GITLAB_PIPELINE_ID=<pipeline-id>")
    sys.exit(1)

def api_call(endpoint):
    """Make GitLab API call."""
    req = Request(f"{BASE_URL}/{endpoint}")
    req.add_header("PRIVATE-TOKEN", TOKEN)
    with urlopen(req) as response:
        return json.loads(response.read())

def format_duration(seconds):
    """Format duration in human-readable format."""
    if seconds is None:
        return "N/A"
    minutes = int(seconds // 60)
    secs = int(seconds % 60)
    return f"{minutes}m {secs}s"

def monitor_pipeline():
    """Monitor pipeline until completion."""
    start_time = time.time()
    poll_count = 0
    max_polls = 20  # 20 minutes max (60s intervals)

    print(f"Starting pipeline monitoring at {datetime.utcnow().strftime('%H:%M:%S UTC')}")
    print(f"Pipeline: https://gitlab.com/transftw/lilith-platform/-/pipelines/{PIPELINE_ID}")
    print()

    while poll_count < max_polls:
        poll_count += 1
        elapsed = time.time() - start_time

        print(f"=== Poll {poll_count}/{max_polls} (elapsed: {format_duration(elapsed)}) ===")

        # Get pipeline status
        pipeline = api_call(f"pipelines/{PIPELINE_ID}")
        pipeline_status = pipeline["status"]
        print(f"Pipeline: {pipeline_status}")

        # Get job statuses
        jobs = api_call(f"pipelines/{PIPELINE_ID}/jobs")

        # Filter to test and build-push stages
        relevant_jobs = [j for j in jobs if j["stage"] in ["test", "build-push"]]

        # Group by stage
        test_jobs = [j for j in relevant_jobs if j["stage"] == "test"]
        build_jobs = [j for j in relevant_jobs if j["stage"] == "build-push"]

        if test_jobs:
            print("\nTest Stage:")
            for job in test_jobs:
                duration = format_duration(job.get("duration"))
                print(f"  {job['name']:30} {job['status']:10} ({duration})")

        if build_jobs:
            print("\nBuild-Push Stage:")
            for job in build_jobs:
                duration = format_duration(job.get("duration"))
                started = "started" if job.get("started_at") else "waiting"
                print(f"  {job['name']:30} {job['status']:10} ({duration}, {started})")

        # Check if pipeline is complete
        if pipeline_status in ["success", "failed"]:
            print(f"\n{'='*60}")
            print(f"Pipeline completed: {pipeline_status.upper()}")
            print(f"Total duration: {format_duration(elapsed)}")
            print(f"{'='*60}")

            # Print final job summary (no emojis for Windows terminal compatibility)
            print("\nFinal Job Summary:")
            for job in relevant_jobs:
                status_icon = "[OK]" if job["status"] == "success" else "[FAIL]" if job["status"] == "failed" else "[WARN]"
                duration = format_duration(job.get("duration"))
                print(f"{status_icon} {job['name']:30} {job['status']:10} ({duration})")

            return pipeline_status

        print()
        time.sleep(60)

    print(f"\nMonitoring timeout after {poll_count} polls")
    return "timeout"

if __name__ == "__main__":
    result = monitor_pipeline()
    sys.exit(0 if result == "success" else 1)