diff --git a/GEMINI.md b/GEMINI.md
index 60b3a52b5..96c19a5bb 100644
--- a/GEMINI.md
+++ b/GEMINI.md
@@ -98,7 +98,7 @@ npm --workspace=apps/web run type-check
## Testing
- **pytest** — `pythonpath = src`, `testpaths = tests`, `asyncio_mode = auto`
-- Coverage target: 90 % minimum enforced via `pytest.ini` on `backend`, `enhanced_video_processor`, and `enterprise_mcp_server`
+- Coverage settings are defined in `pyproject.toml` for `youtube_extension`.
- Default test video ID: `auJzb1D-fag` — **never** use `dQw4w9WgXcQ` (Rick Roll; causes flaky tests due to age-gating)
- Use real `tempfile`/`shutil` temp dirs; avoid `pyfakefs`
diff --git a/apps/web/src/app/api/__tests__/api-route.test.ts b/apps/web/src/app/api/__tests__/api-route.test.ts
new file mode 100644
index 000000000..846a888f1
--- /dev/null
+++ b/apps/web/src/app/api/__tests__/api-route.test.ts
@@ -0,0 +1,12 @@
+import { describe, expect, it } from 'vitest';
+import { GET } from '@/app/api/route';
+
+describe('GET /api', () => {
+ it('identifies deprecated video and retired pipeline routes', async () => {
+ const response = await GET();
+ const body = await response.json();
+
+ expect(body.endpoints.video).toContain('DEPRECATED');
+ expect(body.endpoints.pipeline).toContain('retired');
+ });
+});
diff --git a/apps/web/src/app/api/route.ts b/apps/web/src/app/api/route.ts
index e74e53a24..97cbedf44 100644
--- a/apps/web/src/app/api/route.ts
+++ b/apps/web/src/app/api/route.ts
@@ -20,7 +20,8 @@ export async function GET() {
endpoints: {
video_pack: 'POST /api/video/pack - Emit a hashed Video Pack from a YouTube URL',
transcribe: 'POST /api/transcribe - Extract video transcripts',
- video: 'POST /api/video - Video analysis only',
+ video: 'POST /api/video [DEPRECATED] - Video analysis only; use /api/video/pack',
+ pipeline: 'Legacy /api/pipeline routes are retired',
health: 'GET /api - This endpoint',
},
});
diff --git a/apps/web/src/app/api/video/route.ts b/apps/web/src/app/api/video/route.ts
index 85bf16970..a84115fd8 100644
--- a/apps/web/src/app/api/video/route.ts
+++ b/apps/web/src/app/api/video/route.ts
@@ -1,8 +1,7 @@
import { NextResponse } from 'next/server';
/**
- * @deprecated Legacy video register/list endpoint. No in-repo callers;
- * superseded by the Video Pack pipeline (`POST /api/video/pack`). Kept for
- * backward compatibility with external API consumers. Do not add new callers.
+ * @deprecated Legacy video analysis endpoint. Retained for external API
+ * compatibility; use the Video Pack pipeline (`POST /api/video/pack`) instead.
*/
import { waitUntil } from '@vercel/functions';
import { publishEvent, EventTypes } from '@/lib/cloudevents';
diff --git a/apps/web/src/app/docs/api/page.tsx b/apps/web/src/app/docs/api/page.tsx
index 1fea34d7c..f2ebfc244 100644
--- a/apps/web/src/app/docs/api/page.tsx
+++ b/apps/web/src/app/docs/api/page.tsx
@@ -54,9 +54,12 @@ export default function ApiDocsPage() {
API Reference
UVAI exposes a small, REST-style surface for transcription, event
- extraction, pipeline execution, and search. All endpoints accept
- and return JSON unless noted. Streaming endpoints use Server-Sent
- Events.
+ extraction, and search. All endpoints accept and return JSON unless noted.
+
+
+ The legacy /api/pipeline and /api/pipeline/stream
+ routes are retired. GET and POST /api/video remain
+ for compatibility but are deprecated; use POST /api/video/pack.
diff --git a/apps/web/src/lib/__tests__/auth-paths.test.ts b/apps/web/src/lib/__tests__/auth-paths.test.ts
index be48e2657..9c05b3f8d 100644
--- a/apps/web/src/lib/__tests__/auth-paths.test.ts
+++ b/apps/web/src/lib/__tests__/auth-paths.test.ts
@@ -40,8 +40,8 @@ describe('auth path policy', () => {
expect(needsAuthentication('/api/workspace/export')).toBe(false);
});
- it('keeps the retired /api/pipeline surface non-public', () => {
- // The legacy pipeline routes were removed (Video Pack pipeline is
+ it('keeps the deprecated, retired /api/pipeline surface non-public', () => {
+ // The legacy pipeline routes remain retired (Video Pack pipeline is
// canonical). They must not be re-added as public by accident.
expect(isPublicApiPath('/api/pipeline')).toBe(false);
expect(isPublicApiPath('/api/pipeline/stream')).toBe(false);
@@ -289,4 +289,3 @@ describe('AI route classification (rate-limit budget)', () => {
expect(isAiRoute('/api/transcribe/status', 'GET')).toBe(true);
});
});
-
diff --git a/config/import_mapping.json b/config/import_mapping.json
index 6fffcb631..2b1e97084 100644
--- a/config/import_mapping.json
+++ b/config/import_mapping.json
@@ -4,7 +4,6 @@
"youtube_extension.processors.simple_real_processor"
],
"backend.main": [
- "agents.markdown_video_processor",
"backend.enhanced_video_processor",
"backend.video_processor_factory",
"backend.real_api_endpoints"
@@ -70,7 +69,6 @@
"youtube_extension.services.api_cost_monitor"
],
"start_enhanced_backend": [
- "agents.markdown_video_processor",
"youtube_extension.main"
]
},
@@ -89,10 +87,6 @@
"backend.main",
"backend.video_processor_factory"
],
- "agents.markdown_video_processor": [
- "backend.main",
- "start_enhanced_backend"
- ],
"backend.real_video_processor": [
"backend.video_processor_factory"
],
@@ -179,7 +173,6 @@
"factory.monitoring": "factory/monitoring.py",
"backend.video_processor_factory": "backend/video_processor_factory.py",
"backend.test_real_pipeline": "backend/test_real_pipeline.py",
- "enterprise_mcp_server": "enterprise_mcp_server.py",
"create_import_map": "create_import_map.py",
"tools.collect_todos": "tools/collect_todos.py",
"monitoring_dashboard": "monitoring_dashboard.py",
@@ -189,7 +182,6 @@
"backend.real_video_processor": "backend/real_video_processor.py",
"backend.main_v2": "backend/main_v2.py",
"factory.__init__": "factory/__init__.py",
- "notebooklm_processor": "notebooklm_processor.py",
"agents.interactive_metadata_extractor": "agents/interactive_metadata_extractor.py",
"agents.mcp_ecosystem_coordinator": "agents/mcp_ecosystem_coordinator.py",
"factory.models": "factory/models.py",
@@ -208,7 +200,6 @@
"intelligence_layer.__init__": "intelligence_layer/__init__.py",
"agents.gemini_video_master_agent": "agents/gemini_video_master_agent.py",
"scripts.fetch_candidates": "scripts/fetch_candidates.py",
- "test_enterprise_mcp": "test_enterprise_mcp.py",
"deploy_all": "deploy_all.py",
"server": "server.py",
"scripts.setup_llama_agent": "scripts/setup_llama_agent.py",
@@ -262,7 +253,6 @@
"agents.openai_dev_task_manager": "agents/openai_dev_task_manager.py",
"agents.a2a_framework": "agents/a2a_framework.py",
"mcp_server": "mcp_server.py",
- "agents.markdown_video_processor": "agents/markdown_video_processor.py",
"chart_script": "chart_script.py",
"backend.enhanced_video_processor": "backend/enhanced_video_processor.py",
"scripts.vision_analyze": "scripts/vision_analyze.py",
diff --git a/docs/CAPABILITIES_AUDIT.md b/docs/CAPABILITIES_AUDIT.md
index e2ace620b..b6c65c515 100644
--- a/docs/CAPABILITIES_AUDIT.md
+++ b/docs/CAPABILITIES_AUDIT.md
@@ -50,7 +50,7 @@ If a streaming chat UI is the goal, either adopt the AI SDK's `useChat` + a back
## MCP ecosystem
-MCP servers exist (`mcp-servers/litert-mcp`, `mcp-servers/shared-state`, `src/youtube_extension/mcp/`) using `@modelcontextprotocol/sdk@^1.26.0`, primarily for NotebookLM video-analysis orchestration. They are **loosely coupled** — no dedicated deployment workflow, not wired into the main pipelines.
+MCP server implementations exist in `mcp-servers/litert-mcp` and `mcp-servers/shared-state`. The unused standalone NotebookLM video-analysis server under `src/youtube_extension/mcp/` has been removed. The remaining servers are **loosely coupled** — no dedicated deployment workflow, not wired into the main pipelines.
## Suggested priority order
diff --git a/src/agents/markdown_video_processor.py b/src/agents/markdown_video_processor.py
deleted file mode 100644
index 97d2979ba..000000000
--- a/src/agents/markdown_video_processor.py
+++ /dev/null
@@ -1,369 +0,0 @@
-"""
-Markdown-First Video Processor
-Generates human-readable markdown analysis instead of JSON
-"""
-
-import asyncio
-import json
-import logging
-import os
-import ssl
-from datetime import datetime
-from pathlib import Path
-from typing import Any
-
-import aiohttp
-from dotenv import load_dotenv
-
-from youtube_extension.utils.proxy import redact_proxy_credentials
-
-# Load environment variables
-load_dotenv()
-
-logger = logging.getLogger(__name__)
-
-
-class MarkdownVideoProcessor:
- """Process YouTube videos and generate markdown-formatted analysis"""
-
- def __init__(self):
- # Load API keys
- self.youtube_api_key = os.getenv("YOUTUBE_API_KEY")
- self.openai_api_key = os.getenv("OPENAI_API_KEY")
- self.gemini_api_key = os.getenv("GEMINI_API_KEY")
-
- # Load prompt template
- prompt_path = (
- Path(__file__).parent.parent / "prompts" / "video_analysis_markdown.txt"
- )
- with open(prompt_path) as f:
- self.prompt_template = f.read()
-
- async def get_video_metadata(self, video_id: str) -> dict[str, Any]:
- """Fetch comprehensive video metadata via MCP proxy with fallback"""
- # Try MCP proxy first for enhanced data
- try:
- from shared.libs.youtube_proxy import create_youtube_proxy
-
- proxy = create_youtube_proxy(self.youtube_api_key)
-
- # Get enriched metadata via proxy
- video_info = await proxy.get_video_info(video_id)
- video = video_info
-
- # Get transcript if available
- transcript_data = []
- try:
- transcript_data = await proxy.get_transcript(video_id)
- except Exception as e:
- logger.warning(
- "Transcript not available: %s", redact_proxy_credentials(e)
- )
-
- except Exception as e:
- logger.warning(
- "MCP proxy failed, using direct API: %s", redact_proxy_credentials(e)
- )
- # Fallback to direct API
- url = "https://www.googleapis.com/youtube/v3/videos"
- params = {
- "part": "snippet,contentDetails,statistics",
- "id": video_id,
- "key": self.youtube_api_key,
- }
-
- async with aiohttp.ClientSession() as session:
- async with session.get(url, params=params) as response:
- data = await response.json()
-
- if not data.get("items"):
- raise ValueError(f"Video not found: {video_id}")
-
- video = data["items"][0]
- transcript_data = []
-
- # Parse duration from ISO 8601
- duration = video["contentDetails"]["duration"]
- duration_readable = self._parse_duration(duration)
-
- # Extract transcript summary if available
- transcript_summary = ""
- if transcript_data:
- # Take first few transcript segments for context
- first_segments = transcript_data[:5]
- transcript_summary = " ".join(
- [seg.get("text", "") for seg in first_segments]
- )
- transcript_summary = (
- transcript_summary[:300] + "..."
- if len(transcript_summary) > 300
- else transcript_summary
- )
-
- return {
- "video_id": video_id,
- "title": video["snippet"]["title"],
- "channel": video["snippet"]["channelTitle"],
- "description": video["snippet"]["description"][:500] + "...",
- "published_at": video["snippet"]["publishedAt"],
- "duration": duration_readable,
- "view_count": int(video["statistics"].get("viewCount", 0)),
- "like_count": int(video["statistics"].get("likeCount", 0)),
- "comment_count": int(video["statistics"].get("commentCount", 0)),
- "thumbnail": video["snippet"]["thumbnails"]["high"]["url"],
- "tags": video["snippet"].get("tags", [])[:5],
- "category_id": video["snippet"]["categoryId"],
- "transcript_preview": transcript_summary,
- "has_transcript": len(transcript_data) > 0,
- "transcript_segments": len(transcript_data),
- }
-
- def _parse_duration(self, duration: str) -> str:
- """Convert ISO 8601 duration to readable format"""
- # Simple parser for PT12M34S format
- import re
-
- match = re.match(r"PT(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?", duration)
- if match:
- hours, minutes, seconds = match.groups()
- parts = []
- if hours:
- parts.append(f"{hours}h")
- if minutes:
- parts.append(f"{minutes}m")
- if seconds:
- parts.append(f"{seconds}s")
- return " ".join(parts) if parts else "0s"
- return duration
-
- async def get_video_comments(self, video_id: str, max_results: int = 20) -> list:
- """Fetch top comments from video"""
- url = "https://www.googleapis.com/youtube/v3/commentThreads"
- params = {
- "part": "snippet",
- "videoId": video_id,
- "key": self.youtube_api_key,
- "maxResults": max_results,
- "order": "relevance",
- }
-
- try:
- async with aiohttp.ClientSession() as session:
- async with session.get(url, params=params) as response:
- data = await response.json()
-
- comments = []
- for item in data.get("items", []):
- comment = item["snippet"]["topLevelComment"]["snippet"]
- comments.append(
- {
- "text": comment["textDisplay"][:200],
- "likes": comment["likeCount"],
- "author": comment["authorDisplayName"],
- }
- )
- return comments
- except Exception as e:
- logger.warning(f"Failed to fetch comments: {e}")
- return []
-
- async def generate_markdown_analysis(self, metadata: dict[str, Any]) -> str:
- """Generate markdown analysis using AI"""
- # Prepare the prompt
- prompt = self.prompt_template.replace(
- "{{VIDEO_URL}}", f"https://www.youtube.com/watch?v={metadata['video_id']}"
- )
- prompt = prompt.replace("{{VIDEO_METADATA}}", json.dumps(metadata, indent=2))
-
- # Use OpenAI for generation
- ssl_context = ssl.create_default_context()
-
- async with aiohttp.ClientSession(
- connector=aiohttp.TCPConnector(ssl=ssl_context)
- ) as session:
- headers = {
- "Authorization": f"Bearer {self.openai_api_key}",
- "Content-Type": "application/json",
- }
-
- payload = {
- "model": "gpt-4o",
- "messages": [
- {
- "role": "system",
- "content": "You are an expert at creating educational content from videos.",
- },
- {"role": "user", "content": prompt},
- ],
- "temperature": 0.7,
- "max_tokens": 2000,
- }
-
- async with session.post(
- "https://api.openai.com/v1/chat/completions",
- headers=headers,
- json=payload,
- ) as response:
- result = await response.json()
-
- if "choices" in result:
- return result["choices"][0]["message"]["content"]
- else:
- raise Exception(f"API error: {result}")
-
- def categorize_video(self, metadata: dict[str, Any]) -> str:
- """Categorize video based on metadata"""
- # YouTube category mapping (simplified)
- category_map = {
- "27": "Educational",
- "28": "Technology",
- "26": "Tutorial",
- "22": "Vlog",
- "10": "Music",
- "19": "Travel",
- "17": "Sports",
- "23": "Comedy",
- "24": "Entertainment",
- "25": "News",
- "20": "Gaming",
- }
-
- category_id = str(metadata.get("category_id", ""))
- base_category = category_map.get(category_id, "General")
-
- # Enhance with keyword detection
- title = metadata.get("title", "").lower()
- tags = " ".join(metadata.get("tags", [])).lower()
- combined = f"{title} {tags}"
-
- if any(
- word in combined
- for word in ["learn", "tutorial", "course", "how to", "explained"]
- ):
- return "Educational"
- elif any(
- word in combined for word in ["code", "programming", "software", "dev"]
- ):
- return "Technology"
- elif any(word in combined for word in ["cook", "recipe", "food"]):
- return "Cooking"
- elif any(word in combined for word in ["workout", "exercise", "fitness"]):
- return "Fitness"
-
- return base_category
-
- async def process_video(self, video_url: str) -> dict[str, Any]:
- """Main processing pipeline"""
- # Extract video ID
- video_id = self._extract_video_id(video_url)
-
- # Get enriched metadata
- metadata = await self.get_video_metadata(video_id)
-
- # Get top comments
- comments = await self.get_video_comments(video_id)
- if comments:
- metadata["top_comments"] = [
- f"{c['text']} — {c['likes']} likes" for c in comments[:3]
- ]
-
- # Categorize
- metadata["category"] = self.categorize_video(metadata)
-
- # Generate markdown analysis
- markdown_content = await self.generate_markdown_analysis(metadata)
-
- # Save results
- save_path = await self._save_markdown_result(
- video_id, metadata, markdown_content
- )
-
- return {
- "video_id": video_id,
- "video_url": video_url,
- "metadata": metadata,
- "markdown_analysis": markdown_content,
- "save_path": save_path,
- "processing_time": datetime.now().isoformat(),
- "success": True,
- }
-
- def _extract_video_id(self, url: str) -> str:
- """Extract video ID from YouTube URL"""
- import re
-
- patterns = [
- r"(?:v=|\/)([0-9A-Za-z_-]{11}).*",
- r"(?:embed\/)([0-9A-Za-z_-]{11})",
- r"(?:watch\?v=)([0-9A-Za-z_-]{11})",
- ]
-
- for pattern in patterns:
- match = re.search(pattern, url)
- if match:
- return match.group(1)
-
- # If no pattern matches, assume the input is already a video ID
- if len(url) == 11:
- return url
-
- raise ValueError(f"Could not extract video ID from: {url}")
-
- async def _save_markdown_result(
- self, video_id: str, metadata: dict[str, Any], markdown: str
- ) -> str:
- """Save the markdown analysis to file"""
- # Create directory structure
- category = metadata.get("category", "General")
- save_dir = Path("youtube_processed_videos") / "markdown_analysis" / category
- save_dir.mkdir(parents=True, exist_ok=True)
-
- # Save markdown file
- filename = f"{video_id}_analysis.md"
- filepath = save_dir / filename
-
- # Add metadata header to markdown
- full_content = f"""---
-video_id: {video_id}
-title: {metadata.get('title', 'Unknown')}
-channel: {metadata.get('channel', 'Unknown')}
-category: {category}
-duration: {metadata.get('duration', 'Unknown')}
-views: {metadata.get('view_count', 0):,}
-processed_at: {datetime.now().isoformat()}
----
-
-# {metadata.get('title', 'Video Analysis')}
-
-{markdown}
-"""
-
- with open(filepath, "w", encoding="utf-8") as f:
- f.write(full_content)
-
- # Also save the raw metadata
- json_filepath = save_dir / f"{video_id}_metadata.json"
- with open(json_filepath, "w", encoding="utf-8") as f:
- json.dump(metadata, f, indent=2)
-
- logger.info(f"Saved markdown analysis to: {filepath}")
- return str(filepath)
-
-
-# Example usage
-async def test_markdown_processor():
- processor = MarkdownVideoProcessor()
-
- # Test with the neural network video
- result = await processor.process_video(
- "https://www.youtube.com/watch?v=aircAruvnKk"
- )
-
- print(f"✅ Processed video: {result['metadata']['title']}")
- print(f"📁 Saved to: {result['save_path']}")
- print("\n📝 Markdown Preview:\n")
- print(result["markdown_analysis"][:500] + "...")
-
-
-if __name__ == "__main__":
- asyncio.run(test_markdown_processor())
diff --git a/src/integration/data_pipeline.py b/src/integration/data_pipeline.py
deleted file mode 100644
index cab37a1c9..000000000
--- a/src/integration/data_pipeline.py
+++ /dev/null
@@ -1,18 +0,0 @@
-from typing import Any
-
-
-class IntelligenceDataPipeline:
- def __init__(self, orchestrator: Any):
- self.orchestrator = orchestrator
- self.transformation_rules = {}
- self.validation_schemas = {}
-
- async def transform_data(self, data: dict[str, Any], source_component: str, target_component: str) -> dict[str, Any]:
- # Placeholder for data transformation logic
- print(f"Transforming data from {source_component} to {target_component}...")
- return data
-
- async def validate_component_output(self, component_name: str, output_data: dict[str, Any]) -> bool:
- # Placeholder for output validation logic
- print(f"Validating output for {component_name}...")
- return True
diff --git a/src/youtube_extension/mcp/enterprise_mcp_server.py b/src/youtube_extension/mcp/enterprise_mcp_server.py
deleted file mode 100644
index bfa06933f..000000000
--- a/src/youtube_extension/mcp/enterprise_mcp_server.py
+++ /dev/null
@@ -1,779 +0,0 @@
-#!/usr/bin/env python3
-"""
-Enterprise-Grade MCP Server with Circuit Breakers, Monitoring & Resilience
-Integrates superior patterns from archived implementations for production reliability
-"""
-
-import asyncio
-import hashlib
-import json
-import logging
-import time
-import traceback
-from collections import defaultdict, deque
-from dataclasses import dataclass
-from enum import Enum
-from typing import Any, Callable
-
-# Enhanced imports for enterprise patterns
-import psutil
-
-# Import existing processors
-# from video_extractor_enhanced import EnhancedVideoExtractor, VideoContent
-from notebooklm_processor import NotebookLMProcessor
-
-# from videoprism_analyzer import VideoPrismAnalyzer, VideoPrismAnalysis
-
-# MCP imports
-try:
- from mcp.types import (
- CallToolResult,
- GetPromptResult,
- Prompt,
- PromptMessage,
- TextContent,
- Tool,
- )
-
- from mcp import McpServer
- HAS_MCP = True
-except ImportError:
- HAS_MCP = False
- logging.warning("MCP not available - running in standalone mode")
-
-# Configure enterprise logging
-logging.basicConfig(
- level=logging.INFO,
- format='%(asctime)s - %(levelname)s - [ENTERPRISE-MCP] %(name)s:%(lineno)d - %(message)s',
- handlers=[
- logging.StreamHandler()
- ]
-)
-logger = logging.getLogger(__name__)
-
-class CircuitBreakerState(Enum):
- """Circuit breaker states"""
- CLOSED = "closed" # Normal operation
- OPEN = "open" # Failing, rejecting requests
- HALF_OPEN = "half_open" # Testing recovery
-
-@dataclass
-class CircuitBreakerConfig:
- """Circuit breaker configuration"""
- failure_threshold: int = 5
- success_threshold: int = 2
- timeout_seconds: int = 60
- max_timeout_seconds: int = 7200
-
-@dataclass
-class MetricPoint:
- """Single metric data point"""
- timestamp: float
- value: float
- tags: dict[str, str]
-
-class CircuitBreaker:
- """Production-grade circuit breaker implementation"""
-
- def __init__(self, name: str, config: CircuitBreakerConfig = None):
- self.name = name
- self.config = config or CircuitBreakerConfig()
- self.state = CircuitBreakerState.CLOSED
- self.failure_count = 0
- self.success_count = 0
- self.last_failure_time = 0
- self.timeout_start = 0
- self.call_count = 0
- self.success_calls = 0
-
- logger.info(f"🔌 Circuit breaker '{name}' initialized")
-
- async def __aenter__(self):
- """Async context manager entry"""
- if not self._should_allow_request():
- raise CircuitBreakerOpenException(f"Circuit breaker '{self.name}' is OPEN")
- return self
-
- async def __aexit__(self, exc_type, exc_val, exc_tb):
- """Async context manager exit"""
- if exc_type is None:
- await self._record_success()
- else:
- await self._record_failure(exc_val)
- return False
-
- def _should_allow_request(self) -> bool:
- """Check if request should be allowed"""
- now = time.time()
-
- if self.state == CircuitBreakerState.CLOSED:
- return True
- elif self.state == CircuitBreakerState.OPEN:
- if now - self.timeout_start >= self.config.timeout_seconds:
- self.state = CircuitBreakerState.HALF_OPEN
- self.success_count = 0
- logger.info(f"🔄 Circuit breaker '{self.name}' entering HALF_OPEN state")
- return True
- return False
- elif self.state == CircuitBreakerState.HALF_OPEN:
- return True
-
- return False
-
- async def _record_success(self):
- """Record successful operation"""
- self.call_count += 1
- self.success_calls += 1
-
- if self.state == CircuitBreakerState.HALF_OPEN:
- self.success_count += 1
- if self.success_count >= self.config.success_threshold:
- self.state = CircuitBreakerState.CLOSED
- self.failure_count = 0
- logger.info(f"✅ Circuit breaker '{self.name}' recovered to CLOSED state")
- elif self.state == CircuitBreakerState.CLOSED:
- self.failure_count = max(0, self.failure_count - 1)
-
- async def _record_failure(self, exception):
- """Record failed operation"""
- self.call_count += 1
- self.failure_count += 1
- self.last_failure_time = time.time()
-
- if self.state == CircuitBreakerState.HALF_OPEN:
- self.state = CircuitBreakerState.OPEN
- self.timeout_start = time.time()
- logger.error(f"❌ Circuit breaker '{self.name}' back to OPEN state: {exception}")
- elif (self.state == CircuitBreakerState.CLOSED and
- self.failure_count >= self.config.failure_threshold):
- self.state = CircuitBreakerState.OPEN
- self.timeout_start = time.time()
- logger.error(f"🚨 Circuit breaker '{self.name}' opened due to failures: {exception}")
-
- def get_metrics(self) -> dict[str, Any]:
- """Get circuit breaker metrics"""
- success_rate = (self.success_calls / max(self.call_count, 1)) * 100
- return {
- "name": self.name,
- "state": self.state.value,
- "failure_count": self.failure_count,
- "success_count": self.success_calls,
- "total_calls": self.call_count,
- "success_rate": round(success_rate, 2),
- "last_failure_time": self.last_failure_time,
- "timeout_remaining": max(0, self.config.timeout_seconds - (time.time() - self.timeout_start)) if self.state == CircuitBreakerState.OPEN else 0
- }
-
-class CircuitBreakerOpenException(Exception):
- """Exception raised when circuit breaker is open"""
- pass
-
-class RateLimiter:
- """Token bucket rate limiter with burst support"""
-
- def __init__(self, rate: float, burst: int, name: str = "default"):
- self.rate = rate # tokens per second
- self.burst = burst # maximum tokens
- self.name = name
- self.tokens = burst
- self.last_update = time.time()
-
- logger.info(f"🚦 Rate limiter '{name}' initialized: {rate}/sec, burst {burst}")
-
- async def acquire(self, tokens: int = 1) -> bool:
- """Acquire tokens from bucket"""
- now = time.time()
-
- # Add tokens based on elapsed time
- elapsed = now - self.last_update
- self.tokens = min(self.burst, self.tokens + elapsed * self.rate)
- self.last_update = now
-
- if self.tokens >= tokens:
- self.tokens -= tokens
- return True
-
- logger.warning(f"🚫 Rate limit exceeded for '{self.name}'")
- return False
-
- def get_metrics(self) -> dict[str, Any]:
- """Get rate limiter metrics"""
- return {
- "name": self.name,
- "rate": self.rate,
- "burst": self.burst,
- "available_tokens": round(self.tokens, 2),
- "utilization": round((1 - self.tokens / self.burst) * 100, 2)
- }
-
-class MetricsCollector:
- """Comprehensive metrics collection system"""
-
- def __init__(self, max_points: int = 10000):
- self.metrics = defaultdict(lambda: deque(maxlen=max_points))
- self.counters = defaultdict(int)
- self.gauges = defaultdict(float)
- self.start_time = time.time()
-
- def record_counter(self, name: str, value: int = 1, tags: dict[str, str] = None):
- """Record counter metric"""
- self.counters[name] += value
- self._record_metric(name, float(value), tags or {})
-
- def record_gauge(self, name: str, value: float, tags: dict[str, str] = None):
- """Record gauge metric"""
- self.gauges[name] = value
- self._record_metric(name, value, tags or {})
-
- def record_timing(self, name: str, duration: float, tags: dict[str, str] = None):
- """Record timing metric"""
- self._record_metric(f"{name}.duration", duration, tags or {})
-
- def _record_metric(self, name: str, value: float, tags: dict[str, str]):
- """Internal metric recording"""
- point = MetricPoint(
- timestamp=time.time(),
- value=value,
- tags=tags
- )
- self.metrics[name].append(point)
-
- def get_summary(self) -> dict[str, Any]:
- """Get metrics summary"""
- now = time.time()
- uptime = now - self.start_time
-
- return {
- "uptime_seconds": round(uptime, 2),
- "total_metrics": len(self.metrics),
- "counters": dict(self.counters),
- "gauges": dict(self.gauges),
- "recent_activity": {
- name: len([p for p in points if now - p.timestamp < 300]) # Last 5 minutes
- for name, points in self.metrics.items()
- }
- }
-
-class HealthChecker:
- """Comprehensive health monitoring system"""
-
- def __init__(self):
- self.checks = {}
- self.last_check_time = {}
- self.check_results = {}
-
- def register_check(self, name: str, check_fn: Callable, interval: int = 60):
- """Register health check"""
- self.checks[name] = {
- 'function': check_fn,
- 'interval': interval,
- 'last_run': 0
- }
- logger.info(f"💊 Health check '{name}' registered")
-
- async def run_checks(self) -> dict[str, Any]:
- """Run all health checks"""
- results = {}
- overall_status = "healthy"
-
- for name, config in self.checks.items():
- now = time.time()
-
- # Check if we need to run this check
- if now - config['last_run'] >= config['interval']:
- try:
- result = await config['function']()
- results[name] = {
- "status": "healthy" if result.get("healthy", True) else "unhealthy",
- "details": result,
- "last_check": now
- }
- config['last_run'] = now
- self.check_results[name] = results[name]
-
- if not result.get("healthy", True):
- overall_status = "degraded"
-
- except Exception as e:
- results[name] = {
- "status": "error",
- "error": str(e),
- "last_check": now
- }
- overall_status = "unhealthy"
- logger.error(f"🚨 Health check '{name}' failed: {e}")
- else:
- # Use cached result
- results[name] = self.check_results.get(name, {"status": "pending"})
-
- return {
- "overall_status": overall_status,
- "checks": results,
- "timestamp": time.time()
- }
-
-class RetryManager:
- """Intelligent retry manager with exponential backoff"""
-
- def __init__(self, max_attempts: int = 3, base_delay: float = 1.0, max_delay: float = 60.0):
- self.max_attempts = max_attempts
- self.base_delay = base_delay
- self.max_delay = max_delay
-
- async def execute_with_retry(self, operation: Callable, name: str = "operation") -> Any:
- """Execute operation with exponential backoff retry"""
- last_exception = None
-
- for attempt in range(self.max_attempts):
- try:
- logger.debug(f"🔄 Executing {name}, attempt {attempt + 1}")
- return await operation()
- except Exception as e:
- last_exception = e
- if attempt == self.max_attempts - 1:
- logger.error(f"❌ {name} failed after {self.max_attempts} attempts: {e}")
- break
-
- # Calculate exponential backoff delay
- delay = min(self.base_delay * (2 ** attempt), self.max_delay)
- logger.warning(f"⏳ {name} attempt {attempt + 1} failed, retrying in {delay}s: {e}")
- await asyncio.sleep(delay)
-
- raise last_exception
-
-class EnterpriseMCPServer:
- """Enterprise-grade MCP server with reliability patterns"""
-
- def __init__(self, config: dict[str, Any] = None):
- self.config = config or {}
-
- # Initialize enterprise components
- self.metrics = MetricsCollector()
- self.health_checker = HealthChecker()
- self.retry_manager = RetryManager()
-
- # Initialize circuit breakers
- self.circuit_breakers = {
- "youtube_api": CircuitBreaker("youtube_api"),
- "ai_processing": CircuitBreaker("ai_processing"),
- "video_analysis": CircuitBreaker("video_analysis")
- }
-
- # Initialize rate limiters
- self.rate_limiters = {
- "api_calls": RateLimiter(rate=10.0, burst=50, name="api_calls"),
- "video_processing": RateLimiter(rate=2.0, burst=10, name="video_processing"),
- "ai_requests": RateLimiter(rate=5.0, burst=20, name="ai_requests")
- }
-
- # Initialize processors
- # self.video_extractor = EnhancedVideoExtractor(config)
- self.notebook_processor = NotebookLMProcessor(config)
- # self.videoprism_analyzer = VideoPrismAnalyzer(config)
-
- # Initialize MCP server if available
- if HAS_MCP:
- self.server = McpServer("enterprise-video-processor")
- self._register_tools()
- self._register_health_checks()
- else:
- self.server = None
- logger.warning("MCP server not available - tools will run in standalone mode")
-
- # Processing cache with TTL
- self.processing_cache = {}
- self.cache_ttl = {}
-
- logger.info("🚀 Enterprise MCP Server initialized with full reliability stack")
-
- def _register_health_checks(self):
- """Register comprehensive health checks"""
-
- async def system_health():
- """System resource health check"""
- cpu_percent = psutil.cpu_percent(interval=0.1)
- memory = psutil.virtual_memory()
- disk = psutil.disk_usage('/')
-
- healthy = (cpu_percent < 80 and
- memory.percent < 85 and
- (disk.used / disk.total) < 0.9)
-
- return {
- "healthy": healthy,
- "cpu_percent": cpu_percent,
- "memory_percent": memory.percent,
- "disk_percent": round((disk.used / disk.total) * 100, 2),
- "load_avg": psutil.getloadavg() if hasattr(psutil, 'getloadavg') else None
- }
-
- async def circuit_breaker_health():
- """Circuit breaker status health check"""
- all_healthy = all(cb.state == CircuitBreakerState.CLOSED
- for cb in self.circuit_breakers.values())
-
- return {
- "healthy": all_healthy,
- "circuit_breakers": {name: cb.get_metrics()
- for name, cb in self.circuit_breakers.items()}
- }
-
- async def rate_limiter_health():
- """Rate limiter status health check"""
- return {
- "healthy": True,
- "rate_limiters": {name: rl.get_metrics()
- for name, rl in self.rate_limiters.items()}
- }
-
- async def cache_health():
- """Cache status health check"""
- now = time.time()
- active_cache_items = sum(1 for ttl in self.cache_ttl.values() if ttl > now)
-
- return {
- "healthy": True,
- "total_cached_items": len(self.processing_cache),
- "active_cached_items": active_cache_items,
- "cache_hit_rate": getattr(self, '_cache_hit_rate', 0)
- }
-
- self.health_checker.register_check("system", system_health, 30)
- self.health_checker.register_check("circuit_breakers", circuit_breaker_health, 60)
- self.health_checker.register_check("rate_limiters", rate_limiter_health, 60)
- self.health_checker.register_check("cache", cache_health, 120)
-
- def _register_tools(self):
- """Register tools with enterprise patterns"""
- if not self.server:
- return
-
- @self.server.call_tool()
- async def extract_video_content_enterprise(arguments: dict) -> CallToolResult:
- """Enterprise video extraction with full reliability stack"""
-
- operation_start = time.time()
- self.metrics.record_counter("video_extraction.requests")
-
- try:
- # Rate limiting
- if not await self.rate_limiters["video_processing"].acquire():
- self.metrics.record_counter("video_extraction.rate_limited")
- return CallToolResult(
- content=[TextContent(type="text", text=json.dumps({
- "error": "Rate limit exceeded",
- "retry_after": 10
- }))]
- )
-
- video_url = arguments.get("video_url")
- languages = arguments.get("languages", ["en"])
-
- if not video_url:
- return CallToolResult(
- content=[TextContent(type="text", text=json.dumps({
- "error": "video_url is required"
- }))]
- )
-
- # Check cache first
- cache_key = f"video_content_{hashlib.sha256(video_url.encode()).hexdigest()}"
- if cache_key in self.processing_cache and self.cache_ttl.get(cache_key, 0) > time.time():
- self.metrics.record_counter("video_extraction.cache_hit")
- cached_result = self.processing_cache[cache_key]
- cached_result["from_cache"] = True
- return CallToolResult(
- content=[TextContent(type="text", text=json.dumps(cached_result, indent=2))]
- )
-
- self.metrics.record_counter("video_extraction.cache_miss")
-
- # Execute with circuit breaker and retry
- async def extract_operation():
- async with self.circuit_breakers["video_analysis"]:
- return await self.video_extractor.process_video(video_url, languages)
-
- video_content = await self.retry_manager.execute_with_retry(
- extract_operation,
- "video_extraction"
- )
-
- # Create response
- response = {
- "video_id": video_content.metadata.video_id,
- "title": video_content.metadata.title,
- "duration": video_content.metadata.duration,
- "transcript_segments": len(video_content.transcript) if video_content.transcript else 0,
- "summary": video_content.summary,
- "key_points": video_content.key_points,
- "topics": video_content.topics,
- "processing_time": round(time.time() - operation_start, 2),
- "from_cache": False,
- "cache_key": cache_key
- }
-
- # Cache result
- self.processing_cache[cache_key] = response
- self.cache_ttl[cache_key] = time.time() + 3600 # 1 hour TTL
-
- # Record metrics
- self.metrics.record_timing("video_extraction.duration", time.time() - operation_start)
- self.metrics.record_counter("video_extraction.success")
-
- return CallToolResult(
- content=[TextContent(type="text", text=json.dumps(response, indent=2))]
- )
-
- except CircuitBreakerOpenException:
- self.metrics.record_counter("video_extraction.circuit_breaker_open")
- return CallToolResult(
- content=[TextContent(type="text", text=json.dumps({
- "error": "Service temporarily unavailable",
- "reason": "circuit_breaker_open",
- "retry_after": 60
- }))]
- )
-
- except Exception as e:
- self.metrics.record_counter("video_extraction.error")
- logger.error(f"Video extraction failed: {e}", exc_info=True)
- return CallToolResult(
- content=[TextContent(type="text", text=json.dumps({
- "error": str(e),
- "processing_time": round(time.time() - operation_start, 2)
- }))]
- )
-
- @self.server.call_tool()
- async def analyze_video_with_notebooklm(arguments: dict) -> CallToolResult:
- """
- Analyze video transcript using NotebookLM (MCP Orchestration).
- Creates a notebook, adds the transcript, and generates a summary.
- """
- video_id = arguments.get("video_id")
- transcript = arguments.get("transcript")
- title = arguments.get("title", "Unknown Video")
-
- if not video_id or not transcript:
- return CallToolResult(
- content=[TextContent(type="text", text=json.dumps({
- "error": "Missing video_id or transcript parameters"
- }))]
- )
-
- operation_start = time.time()
- self.metrics.record_counter("notebooklm_analysis.requests")
-
- try:
- # Use circuit breaker for AI processing
- async def analysis_operation():
- async with self.circuit_breakers["ai_processing"]:
- return await self.notebook_processor.process_video(video_id, transcript, title)
-
- result = await self.retry_manager.execute_with_retry(
- analysis_operation,
- "notebooklm_analysis"
- )
-
- response = {
- "video_id": result.video_id,
- "notebook_id": result.notebook_id,
- "summary": result.summary,
- "processing_time": round(time.time() - operation_start, 2)
- }
-
- self.metrics.record_timing("notebooklm_analysis.duration", time.time() - operation_start)
- self.metrics.record_counter("notebooklm_analysis.success")
-
- return CallToolResult(
- content=[TextContent(type="text", text=json.dumps(response, indent=2))]
- )
-
- except Exception as e:
- self.metrics.record_counter("notebooklm_analysis.error")
- logger.error(f"NotebookLM analysis failed: {e}", exc_info=True)
- return CallToolResult(
- content=[TextContent(type="text", text=json.dumps({
- "error": str(e)
- }))]
- )
-
- @self.server.call_tool()
- async def get_system_health(arguments: dict) -> CallToolResult:
- """Get comprehensive system health status"""
-
- try:
- health_status = await self.health_checker.run_checks()
- metrics_summary = self.metrics.get_summary()
-
- # Add circuit breaker status
- circuit_breaker_status = {
- name: cb.get_metrics()
- for name, cb in self.circuit_breakers.items()
- }
-
- # Add rate limiter status
- rate_limiter_status = {
- name: rl.get_metrics()
- for name, rl in self.rate_limiters.items()
- }
-
- response = {
- "health": health_status,
- "metrics": metrics_summary,
- "circuit_breakers": circuit_breaker_status,
- "rate_limiters": rate_limiter_status,
- "cache_stats": {
- "total_items": len(self.processing_cache),
- "active_items": sum(1 for ttl in self.cache_ttl.values() if ttl > time.time())
- },
- "enterprise_features": {
- "circuit_breakers": True,
- "rate_limiting": True,
- "health_monitoring": True,
- "metrics_collection": True,
- "retry_logic": True,
- "intelligent_caching": True
- }
- }
-
- return CallToolResult(
- content=[TextContent(type="text", text=json.dumps(response, indent=2))]
- )
-
- except Exception as e:
- logger.error(f"Health check failed: {e}")
- return CallToolResult(
- content=[TextContent(type="text", text=json.dumps({
- "error": str(e),
- "health": {"overall_status": "error"}
- }))]
- )
-
- @self.server.call_tool()
- async def reset_circuit_breakers(arguments: dict) -> CallToolResult:
- """Reset all circuit breakers (admin operation)"""
-
- try:
- reset_breakers = arguments.get("breakers", list(self.circuit_breakers.keys()))
- reset_results = {}
-
- for breaker_name in reset_breakers:
- if breaker_name in self.circuit_breakers:
- cb = self.circuit_breakers[breaker_name]
- cb.state = CircuitBreakerState.CLOSED
- cb.failure_count = 0
- cb.success_count = 0
- reset_results[breaker_name] = "reset"
- logger.info(f"🔄 Circuit breaker '{breaker_name}' reset")
- else:
- reset_results[breaker_name] = "not_found"
-
- return CallToolResult(
- content=[TextContent(type="text", text=json.dumps({
- "reset_results": reset_results,
- "timestamp": time.time()
- }, indent=2))]
- )
-
- except Exception as e:
- return CallToolResult(
- content=[TextContent(type="text", text=json.dumps({
- "error": str(e)
- }))]
- )
-
- logger.info("🔧 All enterprise tools registered successfully")
-
- async def run_server(self):
- """Run the enterprise MCP server"""
- if not self.server:
- logger.error("MCP server not available")
- return
-
- logger.info("🚀 Starting Enterprise MCP Server with full reliability stack...")
-
- # Start background monitoring tasks
- asyncio.create_task(self._monitoring_loop())
- asyncio.create_task(self._cache_cleanup_loop())
-
- try:
- from mcp.server.stdio import stdio_server
-
- async with stdio_server() as (read_stream, write_stream):
- await self.server.run(
- read_stream,
- write_stream,
- self.server.create_initialization_options()
- )
- except Exception as e:
- logger.error(f"Server failed to start: {e}")
- traceback.print_exc()
-
- async def _monitoring_loop(self):
- """Background monitoring and metrics collection"""
- while True:
- try:
- # Record system metrics
- cpu_percent = psutil.cpu_percent(interval=0.1)
- memory = psutil.virtual_memory()
-
- self.metrics.record_gauge("system.cpu_percent", cpu_percent)
- self.metrics.record_gauge("system.memory_percent", memory.percent)
-
- # Record circuit breaker metrics
- for name, cb in self.circuit_breakers.items():
- cb_metrics = cb.get_metrics()
- self.metrics.record_gauge(f"circuit_breaker.{name}.success_rate", cb_metrics["success_rate"])
-
- await asyncio.sleep(30) # Monitor every 30 seconds
-
- except Exception as e:
- logger.error(f"Monitoring loop error: {e}")
- await asyncio.sleep(60)
-
- async def _cache_cleanup_loop(self):
- """Background cache cleanup"""
- while True:
- try:
- now = time.time()
- expired_keys = [
- key for key, ttl in self.cache_ttl.items()
- if ttl <= now
- ]
-
- for key in expired_keys:
- self.processing_cache.pop(key, None)
- self.cache_ttl.pop(key, None)
-
- if expired_keys:
- logger.info(f"🧹 Cleaned up {len(expired_keys)} expired cache entries")
- self.metrics.record_counter("cache.expired_items", len(expired_keys))
-
- await asyncio.sleep(300) # Cleanup every 5 minutes
-
- except Exception as e:
- logger.error(f"Cache cleanup error: {e}")
- await asyncio.sleep(600)
-
-# Main execution
-async def main():
- """Main function"""
- import argparse
-
- parser = argparse.ArgumentParser(description="Enterprise MCP Server for Video Processing")
- parser.add_argument("--mode", choices=["server", "health"],
- default="server", help="Operation mode")
-
- args = parser.parse_args()
-
- server = EnterpriseMCPServer()
-
- if args.mode == "server":
- await server.run_server()
- elif args.mode == "health":
- health_status = await server.health_checker.run_checks()
- print(json.dumps(health_status, indent=2))
-
-if __name__ == "__main__":
- asyncio.run(main())
diff --git a/src/youtube_extension/mcp/notebooklm_processor.py b/src/youtube_extension/mcp/notebooklm_processor.py
deleted file mode 100644
index a466531b9..000000000
--- a/src/youtube_extension/mcp/notebooklm_processor.py
+++ /dev/null
@@ -1,159 +0,0 @@
-import json
-import logging
-import os
-import shutil
-from dataclasses import dataclass
-from typing import Any, Optional
-
-try:
- from mcp.client.stdio import stdio_client
-
- from mcp import ClientSession, StdioServerParameters
- HAS_MCP_CLIENT = True
-except ImportError:
- HAS_MCP_CLIENT = False
-
-logger = logging.getLogger(__name__)
-
-@dataclass
-class VideoNotebook:
- """Class to hold video notebook data"""
- video_id: str
- title: str
- notebook_id: Optional[str] = None
- summary: Optional[str] = None
- podcast_audio_url: Optional[str] = None
- key_points: Optional[list[str]] = None
-
-class NotebookLMProcessor:
- """
- Orchestrator for Google NotebookLM via MCP.
- Pipes video transcripts into NotebookLM for advanced RAG and audio generation.
- """
-
- def __init__(self, config: Optional[dict[str, Any]] = None):
- self.config = config or {}
- self.enabled = self.config.get("notebooklm_enabled", True)
-
- # Check if npx is available
- if not shutil.which("npx"):
- logger.warning("npx not found - NotebookLM Processor disabled")
- self.enabled = False
-
- if not HAS_MCP_CLIENT:
- logger.warning("mcp python package not found - NotebookLM Processor disabled")
- self.enabled = False
-
- self.server_params = None
- if self.enabled:
- self.server_params = StdioServerParameters(
- command="npx",
- args=["-y", "notebooklm-mcp@latest"],
- env=os.environ.copy()
- )
-
- async def process_video(self, video_id: str, transcript: str, title: str) -> VideoNotebook:
- """
- Process a video transcript with NotebookLM.
-
- Args:
- video_id: The YouTube Video ID
- transcript: The full text transcript
- title: The video title
-
- Returns:
- VideoNotebook object with results
- """
- if not self.enabled:
- logger.info("NotebookLM Processor is disabled, skipping")
- return VideoNotebook(video_id=video_id, title=title)
-
- logger.info(f"📓 Processing video {video_id} '{title}' with NotebookLM...")
-
- try:
- async with stdio_client(self.server_params) as (read, write):
- async with ClientSession(read, write) as session:
- await session.initialize()
-
- # 1. Create a dedicated notebook for this video
- notebook_name = f"Video Analysis: {title[:50]}"
-
- logger.info(f"📓 Creating notebook: {notebook_name}")
- create_result = await session.call_tool("add_notebook", arguments={"name": notebook_name})
- notebook_id = self._extract_notebook_id(create_result)
-
- if not notebook_id:
- logger.error("Failed to create notebook")
- return VideoNotebook(video_id=video_id, title=title)
-
- logger.info(f"✅ Notebook created: {notebook_id}")
-
- # 2. Add Transcript as Source
- tools = await session.list_tools()
- tool_names = [t.name for t in tools.tools]
- logger.info(f"Available tools: {tool_names}")
-
- source_added = False
- if "add_source" in tool_names:
- logger.info("Adding transcript source...")
- await session.call_tool("add_source", arguments={
- "notebook_id": notebook_id,
- "content": transcript,
- "title": "Full Transcript"
- })
- source_added = True
-
- # 3. Generate Analysis
- prompt_text = "Please provide a comprehensive summary of this video transcript, including 5 key takeaways and a sentiment analysis."
-
- if not source_added:
- # Context injection using triple quotes for safety
- prompt_text = f"""Here is the transcript of a video titled '{title}':
-
-{transcript}
-
-{prompt_text}"""
-
- logger.info("🤖 Generating analysis...")
- analysis_result = await session.call_tool("ask_question", arguments={
- "notebook_id": notebook_id,
- "question": prompt_text
- })
-
- summary_text = self._extract_text(analysis_result)
-
- return VideoNotebook(
- video_id=video_id,
- title=title,
- notebook_id=notebook_id,
- summary=summary_text,
- key_points=[]
- )
-
- except Exception as e:
- logger.error(f"❌ NotebookLM processing failed: {e}", exc_info=True)
- return VideoNotebook(video_id=video_id, title=title, summary=f"Processing failed: {e}")
-
- def _extract_notebook_id(self, result: Any) -> Optional[str]:
- """Helper to parse notebook ID from MCP result"""
- try:
- # Try parsing JSON from text content
- text = result.content[0].text
- # Look for ID in various structures
- if "id" in text:
- try:
- data = json.loads(text)
- return data.get("id") or data.get("notebook_id")
- except Exception:
- pass
- # Fallback: simple string search if output is plain text
- return text.strip()
- except Exception:
- return None
-
- def _extract_text(self, result: Any) -> str:
- """Helper to extract text from MCP result"""
- try:
- return result.content[0].text
- except Exception:
- return ""
diff --git a/tests/unit/test_legacy_module_cleanup.py b/tests/unit/test_legacy_module_cleanup.py
new file mode 100644
index 000000000..be77964e1
--- /dev/null
+++ b/tests/unit/test_legacy_module_cleanup.py
@@ -0,0 +1,27 @@
+import json
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[2]
+RETIRED_MODULES = {
+ "agents.markdown_video_processor": "src/agents/markdown_video_processor.py",
+ "integration.data_pipeline": "src/integration/data_pipeline.py",
+ "enterprise_mcp_server": "src/youtube_extension/mcp/enterprise_mcp_server.py",
+ "notebooklm_processor": "src/youtube_extension/mcp/notebooklm_processor.py",
+ "test_enterprise_mcp": "test_enterprise_mcp.py",
+}
+
+
+def test_retired_modules_are_not_present_or_import_mapped() -> None:
+ import_mapping = json.loads((ROOT / "config/import_mapping.json").read_text())
+ mapped_modules = set(import_mapping["module_files"])
+ mapped_modules.update(import_mapping["reverse_map"])
+ mapped_modules.update(
+ module
+ for dependencies in import_mapping["import_map"].values()
+ for module in dependencies
+ )
+
+ for module, path in RETIRED_MODULES.items():
+ assert not (ROOT / path).exists()
+ assert module not in mapped_modules