deployment_monitor.py

"""Real-time deployment status monitoring."""

from __future__ import annotations

import time
from typing import Any, Dict, List, Optional


class DeploymentMonitor:
    """Monitors deployment status in real-time."""

    def __init__(self):
        self.deployments: Dict[str, Dict[str, Any]] = {}

    def track_deployment(
        self,
        deployment_id: str,
        platform: str,
        repo: str,
        status: str = "initiated"
    ) -> Dict[str, Any]:
        """Track a new deployment."""
        deployment = {
            "id": deployment_id,
            "platform": platform,
            "repo": repo,
            "status": status,
            "started_at": time.time(),
            "updated_at": time.time(),
            "stages": [],
            "logs": []
        }
        self.deployments[deployment_id] = deployment
        return deployment

    def update_deployment_status(
        self,
        deployment_id: str,
        status: str,
        stage: Optional[str] = None,
        log_message: Optional[str] = None
    ) -> Dict[str, Any]:
        """Update deployment status."""
        if deployment_id not in self.deployments:
            return {"error": "Deployment not found"}

        deployment = self.deployments[deployment_id]
        deployment["status"] = status
        deployment["updated_at"] = time.time()

        if stage:
            deployment["stages"].append({
                "name": stage,
                "status": status,
                "timestamp": time.time()
            })

        if log_message:
            deployment["logs"].append({
                "message": log_message,
                "timestamp": time.time(),
                "level": "info" if status == "success" else "warning" if status == "pending" else "error"
            })

        return deployment

    def get_deployment_status(self, deployment_id: str) -> Dict[str, Any]:
        """Get current deployment status."""
        if deployment_id not in self.deployments:
            return {"error": "Deployment not found"}

        deployment = self.deployments[deployment_id]
        elapsed = time.time() - deployment["started_at"]

        return {
            **deployment,
            "elapsed_time": f"{elapsed:.1f}s",
            "is_complete": deployment["status"] in ["success", "failed", "cancelled"],
            "is_running": deployment["status"] in ["initiated", "building", "deploying"]
        }

    def get_status_summary(self) -> Dict[str, Any]:
        """Get summary of all deployments."""
        total = len(self.deployments)
        running = len([d for d in self.deployments.values() if d["status"] in ["initiated", "building", "deploying"]])
        successful = len([d for d in self.deployments.values() if d["status"] == "success"])
        failed = len([d for d in self.deployments.values() if d["status"] == "failed"])

        return {
            "total_deployments": total,
            "running": running,
            "successful": successful,
            "failed": failed,
            "success_rate": (successful / total * 100) if total > 0 else 0
        }