""" pipeline_analytics.py - Internal pipeline performance analytics for AutoDub HerStory. Tracks INTERNAL pipeline performance metrics (NOT YouTube video analytics, which are handled by analytics.py). This module monitors: 1. Pipeline Success/Failure Rates: End-to-end success vs per-stage failures 2. Per-Stage Timing: Duration of each pipeline stage in seconds 3. Error Frequency & Patterns: Most common errors, which stages fail most 4. TTS Quality Metrics: Provider used, speed ratios, truncation events 5. Resource Usage: FFmpeg timing, total processing time per video 6. Daily/Weekly Trends: Success rates and processing time over time Architecture: - Metrics stored in StateManager._state under "pipeline_analytics" key - Non-blocking: all writes are in-memory with periodic persist - Thread-safe via threading.Lock - Integration: pipeline.py calls record_stage_timing(), record_error(), record_success() """ import threading import time from collections import defaultdict from datetime import datetime, timedelta from typing import Any, Dict, List, Optional, Tuple # Pipeline stage names (matching pipeline.py steps) VALID_STAGES = frozenset({ "download", "extract_audio", "transcribe", "translate", "validate", "tts", "compose", "upload", }) # How many recent video records to keep in-memory (ring buffer) MAX_RECENT_VIDEOS = 200 # Auto-persist to StateManager every N record calls PERSIST_INTERVAL = 10 class PipelineAnalytics: """Track and report internal pipeline performance metrics. All writes are in-memory and thread-safe. Periodic persist calls flush data to StateManager for durability across restarts. Usage in pipeline.py:: from pipeline_analytics import PipelineAnalytics # In TranslationPipeline.__init__: self.pipeline_analytics = PipelineAnalytics(state=self.state) # Before each stage: t0 = time.time() ... do stage work ... self.pipeline_analytics.record_stage_timing( video_id, "transcribe", time.time() - t0 ) # On error: self.pipeline_analytics.record_error(video_id, "transcribe", str(e)) # On overall success: self.pipeline_analytics.record_success( video_id, time.time() - pipeline_start, metadata={"segments": len(segments), "tts_provider": "supertonic"} ) """ def __init__(self, state=None): """Initialize PipelineAnalytics. Args: state: StateManager instance for persistence. If None, analytics operate in-memory only (useful for testing). """ self.state = state self._lock = threading.Lock() self._record_counter = 0 self._local_analytics: Optional[Dict[str, Any]] = None # Fallback when no state self._ensure_analytics_state() # ================================================================ # Public Recording API # ================================================================ def record_stage_timing( self, video_id: str, stage_name: str, duration_seconds: float, metadata: Optional[Dict[str, Any]] = None, ) -> None: """Record the duration of a completed pipeline stage. Args: video_id: YouTube video ID being processed. stage_name: One of VALID_STAGES (download, transcribe, etc.). duration_seconds: Wall-clock time of the stage in seconds. metadata: Optional extra info (e.g. file sizes, API response codes). Note: This is non-blocking. The write is in-memory and will be persisted periodically or on explicit persist() call. """ stage_name = self._normalize_stage(stage_name) if stage_name is None: return with self._lock: analytics = self._get_analytics() # Update per-stage aggregate timing stage_key = stage_name stage_agg = analytics["stage_timing"].setdefault(stage_key, { "total_seconds": 0.0, "count": 0, "min_seconds": float("inf"), "max_seconds": 0.0, "last_duration": 0.0, }) stage_agg["total_seconds"] += duration_seconds stage_agg["count"] += 1 stage_agg["min_seconds"] = min(stage_agg["min_seconds"], duration_seconds) stage_agg["max_seconds"] = max(stage_agg["max_seconds"], duration_seconds) stage_agg["last_duration"] = round(duration_seconds, 3) # Update per-video session tracking video_session = analytics["video_sessions"].setdefault(video_id, { "stages": {}, "errors": [], "started_at": datetime.now().isoformat(), "status": "in_progress", }) video_session["stages"][stage_name] = { "duration_seconds": round(duration_seconds, 3), "metadata": metadata or {}, "recorded_at": datetime.now().isoformat(), } # Track FFmpeg timing specifically if metadata and metadata.get("ffmpeg_duration"): ffmpeg_dur = metadata["ffmpeg_duration"] ffmpeg_agg = analytics.setdefault("resource_usage", {}).setdefault( "ffmpeg", { "total_seconds": 0.0, "count": 0, "avg_seconds": 0.0, } ) ffmpeg_agg["total_seconds"] += ffmpeg_dur ffmpeg_agg["count"] += 1 ffmpeg_agg["avg_seconds"] = round( ffmpeg_agg["total_seconds"] / ffmpeg_agg["count"], 3 ) self._maybe_persist() def record_error( self, video_id: str, stage_name: str, error_message: str, metadata: Optional[Dict[str, Any]] = None, ) -> None: """Record a pipeline error at a specific stage. Args: video_id: YouTube video ID that failed. stage_name: Stage where the error occurred. error_message: The error string (will be normalized for pattern matching). metadata: Optional context (e.g. HTTP status code, retry number). """ stage_name = self._normalize_stage(stage_name) if stage_name is None: # Still record even if stage name is unrecognized stage_name = "unknown" error_pattern = self._extract_error_pattern(error_message) with self._lock: analytics = self._get_analytics() # Per-stage error count stage_errors = analytics["errors_by_stage"].setdefault(stage_name, 0) analytics["errors_by_stage"][stage_name] = stage_errors + 1 # Error pattern frequency pattern_count = analytics["error_patterns"].setdefault(error_pattern, 0) analytics["error_patterns"][error_pattern] = pattern_count + 1 # Per-video session video_session = analytics["video_sessions"].setdefault(video_id, { "stages": {}, "errors": [], "started_at": datetime.now().isoformat(), "status": "in_progress", }) video_session["errors"].append({ "stage": stage_name, "error": error_message[:500], # Cap error message length "pattern": error_pattern, "metadata": metadata or {}, "occurred_at": datetime.now().isoformat(), }) video_session["status"] = "failed" # Update daily error counts today = datetime.now().strftime("%Y-%m-%d") daily = analytics["daily_errors"].setdefault(today, {}) daily[stage_name] = daily.get(stage_name, 0) + 1 self._maybe_persist() def record_success( self, video_id: str, total_duration_seconds: float, metadata: Optional[Dict[str, Any]] = None, ) -> None: """Record a successful end-to-end pipeline run. Args: video_id: YouTube video ID that completed successfully. total_duration_seconds: Total wall-clock time from start to finish. metadata: Optional info (segments count, TTS provider, etc.). """ with self._lock: analytics = self._get_analytics() today = datetime.now().strftime("%Y-%m-%d") # Global success counters analytics["total_successes"] = analytics.get("total_successes", 0) + 1 analytics["total_attempts"] = analytics.get("total_attempts", 0) + 1 # Per-video session video_session = analytics["video_sessions"].setdefault(video_id, { "stages": {}, "errors": [], "started_at": datetime.now().isoformat(), "status": "in_progress", }) video_session["status"] = "success" video_session["total_duration_seconds"] = round(total_duration_seconds, 3) video_session["completed_at"] = datetime.now().isoformat() if metadata: video_session["metadata"] = metadata # Track TTS quality metrics from metadata if metadata: self._record_tts_metrics(analytics, metadata) # Daily success tracking daily_success = analytics["daily_successes"].setdefault(today, 0) analytics["daily_successes"][today] = daily_success + 1 # Daily total duration for averages daily_duration = analytics["daily_durations"].setdefault(today, { "total_seconds": 0.0, "count": 0, }) daily_duration["total_seconds"] += total_duration_seconds daily_duration["count"] += 1 # Add to recent videos ring buffer self._add_recent_video(analytics, { "video_id": video_id, "status": "success", "total_duration_seconds": round(total_duration_seconds, 3), "stages_completed": list(video_session.get("stages", {}).keys()), "tts_provider": (metadata or {}).get("tts_provider", "unknown"), "completed_at": datetime.now().isoformat(), }) self._maybe_persist() def record_failure( self, video_id: str, total_duration_seconds: float, failed_stage: str, error_message: str, metadata: Optional[Dict[str, Any]] = None, ) -> None: """Record a failed end-to-end pipeline run. Args: video_id: YouTube video ID that failed. total_duration_seconds: Time spent before failure. failed_stage: Stage where the failure occurred. error_message: The error that caused the failure. metadata: Optional additional context. """ # First record the error self.record_error(video_id, failed_stage, error_message, metadata) with self._lock: analytics = self._get_analytics() today = datetime.now().strftime("%Y-%m-%d") # Global failure counters analytics["total_failures"] = analytics.get("total_failures", 0) + 1 analytics["total_attempts"] = analytics.get("total_attempts", 0) + 1 # Daily failure tracking daily_fail = analytics["daily_failures"].setdefault(today, 0) analytics["daily_failures"][today] = daily_fail + 1 # Per-video session (already updated by record_error, add duration) video_session = analytics["video_sessions"].get(video_id, {}) video_session["total_duration_seconds"] = round(total_duration_seconds, 3) video_session["failed_stage"] = failed_stage # Add to recent videos self._add_recent_video(analytics, { "video_id": video_id, "status": "failed", "total_duration_seconds": round(total_duration_seconds, 3), "failed_stage": failed_stage, "error": error_message[:200], "completed_at": datetime.now().isoformat(), }) self._maybe_persist() # ================================================================ # TTS Quality Metrics # ================================================================ def record_tts_segment( self, video_id: str, segment_index: int, provider: str, speed_ratio: float, truncated: bool, target_duration: float, actual_duration: float, ) -> None: """Record TTS quality metrics for a single segment. Call this per-segment during the voice generation stage to track speed adjustment ratios, truncation events, and provider breakdown. Args: video_id: Video being processed. segment_index: Zero-based segment index. provider: TTS provider name (e.g. "supertonic", "gtts"). speed_ratio: The atempo ratio applied (1.0 = no adjustment). truncated: Whether the segment was truncated to fit. target_duration: Expected segment duration in seconds. actual_duration: Generated audio duration before adjustment. """ with self._lock: analytics = self._get_analytics() tts = analytics.setdefault("tts_metrics", self._default_tts_metrics()) # Provider breakdown provider_stats = tts["providers"].setdefault(provider, { "segments": 0, "total_speed_ratio": 0.0, "truncations": 0, "avg_speed_ratio": 0.0, }) provider_stats["segments"] += 1 provider_stats["total_speed_ratio"] += abs(speed_ratio) provider_stats["avg_speed_ratio"] = round( provider_stats["total_speed_ratio"] / provider_stats["segments"], 4 ) if truncated: provider_stats["truncations"] += 1 # Global TTS metrics tts["total_segments"] += 1 tts["total_speed_ratio"] += abs(speed_ratio) tts["avg_speed_ratio"] = round( tts["total_speed_ratio"] / tts["total_segments"], 4 ) if truncated: tts["total_truncations"] += 1 # Speed ratio distribution buckets bucket = self._speed_ratio_bucket(speed_ratio) tts["speed_distribution"][bucket] = tts["speed_distribution"].get(bucket, 0) + 1 self._maybe_persist() # ================================================================ # Query / Reporting API # ================================================================ def get_summary(self) -> Dict[str, Any]: """Get a comprehensive summary of pipeline performance. Returns: Dict with success rates, per-stage timing, top errors, TTS quality overview, and recent processing stats. """ with self._lock: analytics = self._get_analytics() total_attempts = analytics.get("total_attempts", 0) total_successes = analytics.get("total_successes", 0) total_failures = analytics.get("total_failures", 0) success_rate = ( round((total_successes / total_attempts) * 100, 1) if total_attempts > 0 else 0.0 ) # Per-stage average timing stage_averages = {} for stage, data in analytics.get("stage_timing", {}).items(): count = data.get("count", 0) stage_averages[stage] = { "avg_seconds": round(data["total_seconds"] / count, 2) if count else 0, "min_seconds": round(data.get("min_seconds", 0), 2), "max_seconds": round(data.get("max_seconds", 0), 2), "count": count, } # Top errors error_patterns = analytics.get("error_patterns", {}) top_errors = sorted( error_patterns.items(), key=lambda x: x[1], reverse=True )[:10] # Most failing stages errors_by_stage = analytics.get("errors_by_stage", {}) failing_stages = sorted( errors_by_stage.items(), key=lambda x: x[1], reverse=True ) # TTS summary tts = analytics.get("tts_metrics", self._default_tts_metrics()) tts_summary = { "total_segments": tts["total_segments"], "total_truncations": tts["total_truncations"], "truncation_rate": round( (tts["total_truncations"] / tts["total_segments"]) * 100, 1 ) if tts["total_segments"] > 0 else 0.0, "avg_speed_ratio": tts["avg_speed_ratio"], "providers": { p: { "segments": d["segments"], "avg_speed_ratio": d["avg_speed_ratio"], "truncations": d["truncations"], } for p, d in tts.get("providers", {}).items() }, } # FFmpeg resource usage resource = analytics.get("resource_usage", {}) ffmpeg = resource.get("ffmpeg", {}) # Average total processing time daily_durations = analytics.get("daily_durations", {}) total_dur = sum(d.get("total_seconds", 0) for d in daily_durations.values()) total_dur_count = sum(d.get("count", 0) for d in daily_durations.values()) avg_processing_time = ( round(total_dur / total_dur_count, 1) if total_dur_count else 0.0 ) return { "total_attempts": total_attempts, "total_successes": total_successes, "total_failures": total_failures, "success_rate_pct": success_rate, "avg_processing_time_seconds": avg_processing_time, "stage_timing": stage_averages, "top_errors": [ {"pattern": p, "count": c} for p, c in top_errors ], "failing_stages": [ {"stage": s, "error_count": c} for s, c in failing_stages ], "tts_quality": tts_summary, "ffmpeg_avg_seconds": ffmpeg.get("avg_seconds", 0), "recent_videos_count": len(analytics.get("recent_videos", [])), } def get_trends(self, days: int = 7) -> Dict[str, Any]: """Get daily trends for the last N days. Args: days: Number of days to look back (default 7). Returns: Dict with daily success rates, processing times, and error counts per day. """ with self._lock: analytics = self._get_analytics() daily_data = {} for i in range(days): date = (datetime.now() - timedelta(days=i)).strftime("%Y-%m-%d") successes = analytics.get("daily_successes", {}).get(date, 0) failures = analytics.get("daily_failures", {}).get(date, 0) total = successes + failures dur_data = analytics.get("daily_durations", {}).get(date, {}) dur_count = dur_data.get("count", 0) avg_time = ( round(dur_data["total_seconds"] / dur_count, 1) if dur_count else 0.0 ) stage_errors = analytics.get("daily_errors", {}).get(date, {}) daily_data[date] = { "successes": successes, "failures": failures, "total_attempts": total, "success_rate_pct": round( (successes / total) * 100, 1 ) if total > 0 else 0.0, "avg_processing_time_seconds": avg_time, "errors_by_stage": stage_errors, } # Calculate overall period averages all_successes = sum(d["successes"] for d in daily_data.values()) all_total = sum(d["total_attempts"] for d in daily_data.values()) all_times = [d["avg_processing_time_seconds"] for d in daily_data.values() if d["avg_processing_time_seconds"] > 0] return { "period_days": days, "period_success_rate_pct": round( (all_successes / all_total) * 100, 1 ) if all_total > 0 else 0.0, "period_total_attempts": all_total, "period_avg_processing_time": round( sum(all_times) / len(all_times), 1 ) if all_times else 0.0, "daily": daily_data, } def get_video_report(self, video_id: str) -> Optional[Dict[str, Any]]: """Get the full analytics report for a specific video. Args: video_id: YouTube video ID. Returns: Dict with stage timings, errors, and metadata for this video, or None if the video has no recorded analytics. """ with self._lock: analytics = self._get_analytics() session = analytics.get("video_sessions", {}).get(video_id) if session is None: return None return { "video_id": video_id, **session, } def get_error_patterns(self, min_count: int = 2) -> List[Dict[str, Any]]: """Get error patterns that have occurred at least min_count times. Args: min_count: Minimum occurrence count to include. Returns: List of dicts with 'pattern' and 'count', sorted by frequency. """ with self._lock: analytics = self._get_analytics() patterns = analytics.get("error_patterns", {}) return [ {"pattern": p, "count": c} for p, c in sorted(patterns.items(), key=lambda x: x[1], reverse=True) if c >= min_count ] def get_stage_bottlenecks(self) -> List[Dict[str, Any]]: """Identify pipeline stages that are bottlenecks. Returns stages sorted by average duration (slowest first), including failure rate per stage. Returns: List of dicts with stage name, avg duration, and failure count. """ with self._lock: analytics = self._get_analytics() bottlenecks = [] for stage, data in analytics.get("stage_timing", {}).items(): count = data.get("count", 0) avg = (data["total_seconds"] / count) if count else 0 failure_count = analytics.get("errors_by_stage", {}).get(stage, 0) failure_rate = ( round((failure_count / (count + failure_count)) * 100, 1) if (count + failure_count) > 0 else 0.0 ) bottlenecks.append({ "stage": stage, "avg_seconds": round(avg, 2), "max_seconds": round(data.get("max_seconds", 0), 2), "failure_count": failure_count, "failure_rate_pct": failure_rate, "invocations": count, }) return sorted(bottlenecks, key=lambda x: x["avg_seconds"], reverse=True) # ================================================================ # Persistence # ================================================================ def persist(self) -> None: """Force-persist current analytics to StateManager. Normally called automatically every PERSIST_INTERVAL records. Call this explicitly when you need immediate durability (e.g. after recording a critical error). """ if not self.state or not hasattr(self.state, "_state"): return try: with self._lock: self.state._state["pipeline_analytics"] = self._get_analytics() self.state.save() except Exception as e: print(f"[PIPELINE-ANALYTICS] Warning: could not persist: {e}") # ================================================================ # Internal Methods # ================================================================ def _ensure_analytics_state(self) -> None: """Initialize the pipeline_analytics key in state if missing.""" if self.state and hasattr(self.state, "_state"): if "pipeline_analytics" not in self.state._state: self.state._state["pipeline_analytics"] = self._default_analytics() def _get_analytics(self) -> Dict[str, Any]: """Get analytics dict from state (or in-memory default).""" if self.state and hasattr(self.state, "_state"): analytics = self.state._state.get("pipeline_analytics") if analytics is not None: return analytics # Return a locally-cached default if state isn't available if self._local_analytics is None: self._local_analytics = self._default_analytics() return self._local_analytics @staticmethod def _default_analytics() -> Dict[str, Any]: """Return the default analytics data structure.""" return { "total_attempts": 0, "total_successes": 0, "total_failures": 0, "stage_timing": {}, # stage -> {total_seconds, count, min, max} "errors_by_stage": {}, # stage -> error_count "error_patterns": {}, # normalized_error -> count "daily_successes": {}, # date -> count "daily_failures": {}, # date -> count "daily_errors": {}, # date -> {stage -> count} "daily_durations": {}, # date -> {total_seconds, count} "video_sessions": {}, # video_id -> session data "recent_videos": [], # ring buffer of last N videos "tts_metrics": PipelineAnalytics._default_tts_metrics(), "resource_usage": { "ffmpeg": { "total_seconds": 0.0, "count": 0, "avg_seconds": 0.0, }, }, } @staticmethod def _default_tts_metrics() -> Dict[str, Any]: """Return the default TTS metrics data structure.""" return { "total_segments": 0, "total_truncations": 0, "total_speed_ratio": 0.0, "avg_speed_ratio": 0.0, "providers": {}, # provider -> {segments, total_speed_ratio, truncations} "speed_distribution": {}, # bucket -> count } def _maybe_persist(self) -> None: """Auto-persist every PERSIST_INTERVAL records.""" self._record_counter += 1 if self._record_counter % PERSIST_INTERVAL == 0: try: self.persist() except Exception: pass # Non-blocking: persist failure is not fatal @staticmethod def _normalize_stage(stage_name: str) -> Optional[str]: """Normalize a stage name to a known stage, or return None.""" if not stage_name: return None stage_lower = stage_name.lower().strip() if stage_lower in VALID_STAGES: return stage_lower # Map common aliases aliases = { "extract_audio": "extract_audio", "voice": "tts", "tts_generation": "tts", "composition": "compose", "video_composition": "compose", "youtube_upload": "upload", } return aliases.get(stage_lower, stage_lower) @staticmethod def _extract_error_pattern(error_message: str) -> str: """Extract a normalized error pattern from an error message. Replaces variable parts (IDs, paths, numbers) with placeholders so similar errors are grouped together for pattern analysis. Examples: "Download failed for abc123" -> "Download failed for " "FFmpeg error code 1" -> "FFmpeg error code " """ import re as _re pattern = error_message[:300] # Truncate long errors # Replace common variable parts # YouTube video IDs (11 chars, alphanumeric + dash/underscore) pattern = _re.sub( r'[a-zA-Z0-9_-]{11}', '', pattern ) # File paths pattern = _re.sub( r'/[\w/.-]+\.\w{2,4}', '', pattern ) # Numbers (including decimals) pattern = _re.sub(r'\b\d+\.?\d*\b', '', pattern) # URLs pattern = _re.sub(r'https?://\S+', '', pattern) # Hex strings pattern = _re.sub(r'0x[a-fA-F0-9]+', '', pattern) return pattern[:200] # Final cap @staticmethod def _speed_ratio_bucket(ratio: float) -> str: """Categorize a speed ratio into a named bucket.""" if ratio < 0.5: return "very_slow_<0.5x" elif ratio < 0.8: return "slow_0.5-0.8x" elif ratio < 0.95: return "slight_slow_0.8-0.95x" elif ratio <= 1.05: return "normal_0.95-1.05x" elif ratio <= 1.2: return "slight_fast_1.05-1.2x" elif ratio <= 1.5: return "fast_1.2-1.5x" elif ratio <= 2.0: return "very_fast_1.5-2.0x" else: return "extreme_>2.0x" @staticmethod def _add_recent_video(analytics: Dict[str, Any], entry: Dict[str, Any]) -> None: """Add a video entry to the recent_videos ring buffer.""" recent = analytics.setdefault("recent_videos", []) recent.append(entry) # Trim to max size if len(recent) > MAX_RECENT_VIDEOS: analytics["recent_videos"] = recent[-MAX_RECENT_VIDEOS:] @staticmethod def _record_tts_metrics(analytics: Dict[str, Any], metadata: Dict[str, Any]) -> None: """Extract and record TTS metrics from success metadata.""" tts = analytics.setdefault("tts_metrics", PipelineAnalytics._default_tts_metrics()) provider = metadata.get("tts_provider", "unknown") if provider not in tts["providers"]: tts["providers"][provider] = { "segments": 0, "total_speed_ratio": 0.0, "truncations": 0, "avg_speed_ratio": 0.0, } # Record overall TTS stats from the metadata summary segments_count = metadata.get("segments_count", 0) truncations = metadata.get("truncations", 0) avg_speed = metadata.get("avg_speed_ratio", 1.0) provider_stats = tts["providers"][provider] provider_stats["segments"] += segments_count provider_stats["truncations"] += truncations if segments_count > 0: # Running average total = provider_stats["total_speed_ratio"] + (avg_speed * segments_count) provider_stats["total_speed_ratio"] = total provider_stats["avg_speed_ratio"] = round( total / provider_stats["segments"], 4 ) tts["total_segments"] += segments_count tts["total_truncations"] += truncations