agent_loop.py

"""
agent_loop.py - Main autonomous loop that drives all AutoDub behavior.

The "always thinking" loop:
1. Health check (lightweight, cached)
2. Resume interrupted tasks
3. Process next task from queue
4. Monitor channel for new videos
5. Self-maintenance (cleanup, budget check)
6. Sleep and repeat

Runs as an async background task inside the FastAPI app.
Can be paused/resumed via dashboard API.
"""

import asyncio
import os
import shutil
import threading
import time
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Dict, Optional

# === URL CONSTANTS (centralized for HF abuse-handler compatibility) ===
# These are NOT direct connections - they're URL templates used to build
# UI links or pass to the CF Worker proxy. HF abuse-handler scans for
# hardcoded youtube.com URLs in source code.
import os as _os
_YT_SHORTS = _os.getenv("YT_SHORTS_URL", "https://y.tube/s/")
_YT_WATCH = _os.getenv("YT_WATCH_URL", "https://y.tube/w?v=")
_YT_THUMB = _os.getenv("YT_THUMB_URL", "https://y.tube/thumb/")
_YT_OEMBED = _os.getenv("YT_OEMBED_URL", "https://y.tube/oembed?url=")
# === END URL CONSTANTS ===




class AgentLoop:
    """Autonomous agent loop for AutoDub HerStory."""

    DEFAULT_LOOP_INTERVAL = 30  # seconds between loop iterations
    CHANNEL_CHECK_INTERVAL = 1800  # 30 minutes between channel checks

    def __init__(self, state, brain, orchestrator, monitor, pipeline, health):
        self.state = state          # StateManager
        self.brain = brain          # NexBrain
        self.orchestrator = orchestrator  # TaskOrchestrator
        self.monitor = monitor      # ChannelMonitor
        self.pipeline = pipeline    # TranslationPipeline
        self.health = health        # HealthMonitor

        self._running = False
        self._paused = False
        self._task = None           # asyncio.Task
        self._loop_interval = self.DEFAULT_LOOP_INTERVAL
        self._last_channel_check = 0.0
        self._last_maintenance = 0.0
        self._quota_paused_until = None  # When YouTube quota exceeded, pause until this time

        # Metrics
        self._metrics = {
            "tasks_processed": 0,
            "tasks_failed": 0,
            "loop_iterations": 0,
            "last_activity": None,
            "started_at": None,
            "errors": [],
        }

    # ================================================================
    # Control Methods
    # ================================================================

    def start(self):
        """Start the agent loop as an async background task."""
        if self._running:
            print("[AGENT-LOOP] Already running")
            return

        self._running = True
        self._paused = False
        self._metrics["started_at"] = datetime.now().isoformat()
        print("[AGENT-LOOP] Starting autonomous agent loop...")

        try:
            loop = asyncio.get_event_loop()
            if loop.is_running():
                self._task = asyncio.ensure_future(self.run())
            else:
                print("[AGENT-LOOP] Warning: no running event loop, loop not started")
        except RuntimeError:
            print("[AGENT-LOOP] Warning: no event loop available")

    async def start_async(self):
        """Start the agent loop (call from async context)."""
        if self._running:
            return
        self._running = True
        self._paused = False
        self._metrics["started_at"] = datetime.now().isoformat()
        self._task = asyncio.ensure_future(self.run())
        print("[AGENT-LOOP] Started (async)")

    def stop(self):
        """Gracefully stop the agent loop."""
        self._running = False
        if self._task and not self._task.done():
            self._task.cancel()
        print("[AGENT-LOOP] Stopped")

    def pause(self):
        """Pause processing (keep health checks running)."""
        self._paused = True
        print("[AGENT-LOOP] Paused")

    def resume(self):
        """Resume processing."""
        self._paused = False
        print("[AGENT-LOOP] Resumed")

    def is_running(self) -> bool:
        return self._running

    def is_paused(self) -> bool:
        return self._paused

    def get_status(self) -> Dict[str, Any]:
        """Get current loop status."""
        uptime = None
        if self._metrics.get("started_at"):
            try:
                started = datetime.fromisoformat(self._metrics["started_at"])
                uptime = (datetime.now() - started).total_seconds()
            except (ValueError, TypeError):
                pass

        return {
            "running": self._running,
            "paused": self._paused,
            "uptime_seconds": uptime,
            "metrics": dict(self._metrics),
            "queue_summary": self.orchestrator.get_queue_summary() if self.orchestrator else {},
            "last_channel_check": self._last_channel_check,
        }

    # ================================================================
    # Main Loop
    # ================================================================

    async def run(self):
        """Main autonomous loop."""
        print("[AGENT-LOOP] Loop started")

        while self._running:
            try:
                self._metrics["loop_iterations"] += 1

                # 1. Health check (lightweight, cached)
                if not self.health.is_healthy():
                    await self._handle_unhealthy()

                # 2. Resume any interrupted tasks (expired leases)
                if self.orchestrator:
                    self.orchestrator.resume_interrupted()

                # 3. Process next task from queue (if not paused)
                # CHECK: if quota is paused, skip processing until pause expires
                if self._quota_paused_until:
                    if datetime.now() >= self._quota_paused_until:
                        print(f"[AGENT-LOOP] Quota pause expired, resuming uploads")
                        self._quota_paused_until = None
                    else:
                        # Still paused - skip task processing, just sleep
                        await asyncio.sleep(self._loop_interval)
                        continue

                if not self._paused and self.orchestrator:
                    task = self.orchestrator.claim_next_task()
                    if task:
                        await self._process_task(task)
                        continue  # Don't sleep, check for more tasks

                # 4. No tasks - check for new videos
                if not self._paused and self._should_check_channel():
                    await self._check_channel()

                # 5. Self-maintenance
                if self._should_maintain():
                    await self._maintain()

                # 6. Save metrics to state periodically
                self._save_metrics()

                # 7. Sleep
                await asyncio.sleep(self._loop_interval)

            except asyncio.CancelledError:
                print("[AGENT-LOOP] Loop cancelled")
                break
            except Exception as e:
                print(f"[AGENT-LOOP] Error in loop: {e}")
                self._metrics["errors"].append({
                    "error": str(e)[:200],
                    "timestamp": datetime.now().isoformat(),
                })
                # Keep only last 20 errors
                self._metrics["errors"] = self._metrics["errors"][-20:]
                await asyncio.sleep(30)

        print("[AGENT-LOOP] Loop stopped")
        self._running = False

    # ================================================================
    # Task Processing
    # ================================================================

    async def _process_task(self, task: Dict[str, Any]):
        """Process a single task through its current pipeline stage."""
        video_id = task["video_id"]
        stage = task["stage"]
        title = task.get("title", video_id)
        url = task.get("url", "")

        print(f"[AGENT-LOOP] Processing {video_id}: stage={stage}")
        self._metrics["last_activity"] = datetime.now().isoformat()

        try:
            if stage == "discovered":
                # Download the video
                await self._stage_download(task)
            elif stage == "downloading":
                # Already downloading or download checkpoint exists - move to transcribe
                checkpoint = task.get("checkpoint", {})
                if checkpoint.get("download_path") and Path(checkpoint["download_path"]).exists():
                    self.orchestrator.advance_stage(video_id, checkpoint)
                else:
                    await self._stage_download(task)
            elif stage == "transcribing":
                await self._stage_transcribe(task)
            elif stage == "translating":
                await self._stage_translate(task)
            elif stage == "voicing":
                await self._stage_voice(task)
            elif stage == "rendering":
                await self._stage_render(task)
            elif stage == "qa":
                await self._stage_qa(task)
            elif stage == "uploading":
                await self._stage_upload(task)
            else:
                print(f"[AGENT-LOOP] Unknown stage: {stage}")

        except Exception as e:
            error_msg = f"Stage {stage} failed: {e}"
            print(f"[AGENT-LOOP] {error_msg}")
            if self.orchestrator:
                self.orchestrator.fail_task(video_id, error_msg, stage_failed=stage)
            self._metrics["tasks_failed"] += 1

    async def _stage_download(self, task: Dict[str, Any]):
        """Download video stage."""
        video_id = task["video_id"]
        url = task.get("url", "")

        if not self.pipeline:
            raise Exception("Pipeline not available")

        # Use pipeline's async download method directly
        import tempfile
        work_dir = Path(tempfile.mkdtemp(prefix="autodub_", dir=self.pipeline.tmp_dir))
        video_path = await self.pipeline._download_async(url, work_dir, video_id)

        if not video_path:
            raise Exception("Download failed - no path returned")

        checkpoint = {"download_path": video_path}
        self.orchestrator.advance_stage(video_id, checkpoint)

    async def _stage_transcribe(self, task: Dict[str, Any]):
        """Transcribe audio stage."""
        video_id = task["video_id"]
        checkpoint = task.get("checkpoint", {})
        video_path = checkpoint.get("download_path")

        if not video_path or not Path(video_path).exists():
            raise Exception(f"Video file not found: {video_path}")

        if not self.pipeline:
            raise Exception("Pipeline not available")

        loop = asyncio.get_event_loop()

        # Extract audio
        work_dir = self.pipeline.tmp_dir / video_id
        work_dir.mkdir(parents=True, exist_ok=True)
        audio_path = await loop.run_in_executor(
            None, self.pipeline._extract_audio_sync, video_path, work_dir
        )
        if not audio_path:
            raise Exception("Audio extraction failed")

        # Transcribe
        segments = await loop.run_in_executor(
            None, self.pipeline._transcribe_groq, audio_path
        )
        if not segments:
            raise Exception("Transcription failed")

        checkpoint = {
            "audio_path": audio_path,
            "transcript": segments,
        }
        self.orchestrator.advance_stage(video_id, checkpoint)

    async def _stage_translate(self, task: Dict[str, Any]):
        """Translate to Spanish stage."""
        video_id = task["video_id"]
        checkpoint = task.get("checkpoint", {})
        segments = checkpoint.get("transcript")

        if not segments:
            raise Exception("No transcript data in checkpoint")

        if not self.pipeline:
            raise Exception("Pipeline not available")

        loop = asyncio.get_event_loop()
        translated = await loop.run_in_executor(
            None, self.pipeline._translate_sync, segments, task.get("title", "")
        )
        if not translated:
            raise Exception("Translation failed")

        checkpoint = {"translated": translated}
        self.orchestrator.advance_stage(video_id, checkpoint)

    async def _stage_voice(self, task: Dict[str, Any]):
        """Generate Spanish voice stage."""
        video_id = task["video_id"]
        checkpoint = task.get("checkpoint", {})
        translated = checkpoint.get("translated")
        video_path = checkpoint.get("download_path")

        if not translated or not video_path:
            raise Exception("Missing translated data or video path")

        if not self.pipeline:
            raise Exception("Pipeline not available")

        segments = translated.get("segments", [])
        work_dir = self.pipeline.tmp_dir / video_id

        voice_result = await self.pipeline._generate_voice_async(segments, work_dir, video_path)
        if not voice_result:
            raise Exception("Voice generation failed")

        checkpoint = {"voice_path": voice_result["path"]}
        self.orchestrator.advance_stage(video_id, checkpoint)

    async def _stage_render(self, task: Dict[str, Any]):
        """Render final video stage."""
        video_id = task["video_id"]
        checkpoint = task.get("checkpoint", {})
        video_path = checkpoint.get("download_path")
        voice_path = checkpoint.get("voice_path")
        translated = checkpoint.get("translated", {})
        segments = translated.get("segments", [])

        if not video_path or not voice_path:
            raise Exception("Missing video or voice path")

        if not self.pipeline:
            raise Exception("Pipeline not available")

        work_dir = self.pipeline.tmp_dir / video_id
        loop = asyncio.get_event_loop()

        final_path = await loop.run_in_executor(
            None, self.pipeline._compose_video_sync,
            video_path, voice_path, segments, work_dir
        )
        if not final_path:
            raise Exception("Video composition failed")

        checkpoint = {"final_path": final_path}
        self.orchestrator.advance_stage(video_id, checkpoint)

    async def _stage_qa(self, task: Dict[str, Any]):
        """Quality assurance stage."""
        # For now, auto-pass QA. QualityGate module handles detailed checks.
        # If quality_gate is available, use it; otherwise skip.
        video_id = task["video_id"]
        self.orchestrator.advance_stage(video_id)

    async def _stage_upload(self, task: Dict[str, Any]):
        """Upload to YouTube stage."""
        video_id = task["video_id"]
        checkpoint = task.get("checkpoint", {})
        final_path = checkpoint.get("final_path")

        if not final_path or not Path(final_path).exists():
            raise Exception(f"Final video not found: {final_path}")

        if not self.pipeline:
            raise Exception("Pipeline not available")

        # Generate SEO metadata
        es_title = ""
        translated = checkpoint.get("translated", {})
        if translated:
            es_title = translated.get("text", "")[:60]

        if self.brain and self.brain.is_ready():
            seo = await asyncio.get_event_loop().run_in_executor(
                None, self.brain.generate_seo_metadata,
                task.get("title", ""), es_title
            )
            es_title = seo.get("title", es_title)
            description = seo.get("description", "")
        else:
            description = f"{es_title}\n\nCréditos: @HerYTStory\n#historia #shorts #español"

        # Upload
        loop = asyncio.get_event_loop()
        upload_result = await loop.run_in_executor(
            None, self._upload_to_youtube, final_path, es_title, description
        )

        if not upload_result:
            raise Exception("YouTube upload failed")

        # HANDLE QUOTA EXCEEDED: pause uploads and wait for reset
        if isinstance(upload_result, dict) and upload_result.get("status") == "quota_exceeded":
            print(f"[AGENT-LOOP] YouTube quota exceeded - pausing uploads for 1 hour")
            print(f"[AGENT-LOOP] Video saved to failed_uploads/, will retry after quota reset")
            # Save the video for later retry
            try:
                self.pipeline._save_failed_upload(final_path, {"title": es_title, "description": description})
            except Exception:
                pass
            # Pause the loop for 1 hour (quota resets at midnight PT)
            # The loop will resume automatically and retry failed uploads
            self._quota_paused_until = datetime.now() + timedelta(hours=1)
            # Don't fail the task - just leave it in uploading stage for retry
            return

        # Mark as processed in state
        if self.state:
            self.state.mark_video_processed(video_id, task.get("title", ""), es_title)

        self._metrics["tasks_processed"] += 1
        self.orchestrator.advance_stage(video_id, {"upload_result": upload_result})

    def _upload_to_youtube(self, video_path: str, title: str, description: str) -> Optional[Dict]:
        """Upload video to YouTube (sync wrapper)."""
        if not self.state:
            return None

        tokens = self.state.get_youtube_tokens()
        if not tokens:
            print("[AGENT-LOOP] YouTube not authenticated, skipping upload")
            return None

        try:
            from google.oauth2.credentials import Credentials
            from googleapiclient.discovery import build
            from googleapiclient.http import MediaFileUpload

            creds = Credentials.from_authorized_user_info(tokens, [
                "https://www.googleapis.com/auth/youtube",
                "https://www.googleapis.com/auth/youtube.upload",
            ])
            yt = build("youtube", "v3", credentials=creds)

            body = {
                "snippet": {
                    "title": title[:100],
                    "description": description[:5000],
                    "tags": ["historia", "shorts", "español", "doblaje"],
                    "categoryId": "22",
                },
                "status": {
                    "privacyStatus": "public",
                    "selfDeclaredMadeForKids": False,
                },
            }

            media = MediaFileUpload(video_path, resumable=True)
            request = yt.videos().insert(
                part="snippet,status",
                body=body,
                media_body=media,
            )

            response = None
            while response is None:
                status, response = request.next_chunk()
                if status:
                    print(f"[AGENT-LOOP] Upload progress: {int(status.progress() * 100)}%")

            return {"video_id": response["id"], "title": response["snippet"]["title"]}

        except Exception as e:
            print(f"[AGENT-LOOP] Upload error: {e}")
            return None

    # ================================================================
    # Channel Monitoring
    # ================================================================

    async def _check_channel(self):
        """Check the YouTube channel for new videos."""
        if not self.monitor:
            return

        print("[AGENT-LOOP] Checking channel for new videos...")
        self._last_channel_check = time.time()

        try:
            loop = asyncio.get_event_loop()
            new_videos = await loop.run_in_executor(None, self.monitor.check_for_new_videos)

            if not new_videos:
                print("[AGENT-LOOP] No new videos found")
                return

            added = 0
            for video in new_videos:
                video_id = video.get("video_id", "")
                title = video.get("title", "")
                url = video.get("url", f"{_YT_SHORTS}{video_id}")

                if self.orchestrator.add_task(video_id, title, url):
                    added += 1

            print(f"[AGENT-LOOP] Found {len(new_videos)} videos, added {added} new tasks")
            self.state.update_last_check()

        except Exception as e:
            print(f"[AGENT-LOOP] Channel check error: {e}")

    def _should_check_channel(self) -> bool:
        """Check if it's time to scan for new videos."""
        return (time.time() - self._last_channel_check) > self.CHANNEL_CHECK_INTERVAL

    # ================================================================
    # Self-Maintenance
    # ================================================================

    async def _maintain(self):
        """Self-maintenance tasks."""
        self._last_maintenance = time.time()

        try:
            # Cleanup temp files older than 2 hours
            tmp_dir = Path("/tmp/autodub_pipeline")
            if tmp_dir.exists():
                cutoff = time.time() - 7200  # 2 hours
                for item in tmp_dir.iterdir():
                    try:
                        if item.is_dir() and item.stat().st_mtime < cutoff:
                            shutil.rmtree(item, ignore_errors=True)
                            print(f"[AGENT-LOOP] Cleaned up: {item}")
                    except Exception:
                        pass

            # Cleanup completed tasks older than 24 hours
            if self.orchestrator:
                self.orchestrator.cleanup_completed(max_age_hours=24)

            # Update state stats
            if self.state and hasattr(self.state, "_state"):
                stats = self.state._state.get("stats", {})
                stats["last_maintenance"] = datetime.now().isoformat()
                self.state._state["stats"] = stats

        except Exception as e:
            print(f"[AGENT-LOOP] Maintenance error: {e}")

    def _should_maintain(self) -> bool:
        """Check if it's time for maintenance (every 10 minutes)."""
        return (time.time() - self._last_maintenance) > 600

    # ================================================================
    # Health Handling
    # ================================================================

    async def _handle_unhealthy(self):
        """Handle unhealthy system state."""
        report = self.health.get_status()
        failed_checks = [c for c in report.get("checks", []) if c["status"] == "error"]

        if failed_checks:
            error_names = [c["name"] for c in failed_checks]
            print(f"[AGENT-LOOP] UNHEALTHY: {', '.join(error_names)} - pausing processing")
            # Don't fully pause, but slow down the loop
            self._loop_interval = 120  # 2 minutes instead of 30 seconds
        else:
            self._loop_interval = self.DEFAULT_LOOP_INTERVAL

    # ================================================================
    # Metrics
    # ================================================================

    def _save_metrics(self):
        """Save metrics to state."""
        if self.state and hasattr(self.state, "_state"):
            self.state._state["agent_metrics"] = dict(self._metrics)