#!/usr/bin/env python3
"""
Integration Example: Lab Dashboard with YTPipe Pipeline
This script demonstrates how to integrate the Lab Dashboard Service
into a complete video processing workflow.
Workflow:
1. Process video through pipeline
2. Run analysis service
3. Generate both standard and lab dashboards
4. Compare outputs
"""
import asyncio
from pathlib import Path
from ytpipe.core.pipeline import Pipeline
from ytpipe.services.intelligence import AnalyzerService
from ytpipe.services.exporters import DashboardService, LabDashboardService
async def process_with_lab_dashboard(url: str, output_dir: str = "./output"):
"""
Process a YouTube video and generate both dashboard styles.
Args:
url: YouTube video URL
output_dir: Output directory for results
Returns:
Dict with paths to both dashboards
"""
print("=" * 60)
print("YTPipe: Lab Dashboard Integration Example")
print("=" * 60)
# Step 1: Process video through pipeline
print("\n[1/4] Processing video through pipeline...")
print(f" URL: {url}")
pipeline = Pipeline(
output_dir=output_dir,
vector_backend="chromadb",
whisper_model="base"
)
result = await pipeline.process(url)
if not result.success:
print(f"❌ Pipeline failed: {result.error}")
return None
print(f"✓ Video processed in {result.processing_time:.1f}s")
print(f" • Chunks: {result.total_chunks()}")
print(f" • Words: {result.metadata.total_words}")
print(f" • Output: {result.output_dir}")
# Step 2: Run analysis service
print("\n[2/4] Running content analysis...")
analyzer = AnalyzerService(
min_keyword_length=3,
max_keywords=20,
max_topics=5
)
analysis = analyzer.analyze(result.metadata, result.chunks)
print(f"✓ Analysis complete in {analysis.processing_time:.2f}s")
print(f" • Keywords: {len(analysis.top_keywords)}")
print(f" • Topics: {len(analysis.topics)}")
print(f" • Avg quality: {analysis.avg_chunk_quality:.1f}/10")
# Step 3: Generate standard dashboard (already done by pipeline)
print("\n[3/4] Standard dashboard (dark mode)...")
if result.dashboard_path:
print(f"✓ Already generated: {result.dashboard_path}")
else:
dashboard_service = DashboardService()
standard_path = dashboard_service.generate_dashboard(
metadata=result.metadata,
chunks=result.chunks,
output_dir=Path(result.output_dir)
)
print(f"✓ Generated: {standard_path}")
# Step 4: Generate lab dashboard (clinical style)
print("\n[4/4] Lab dashboard (clinical style)...")
lab_service = LabDashboardService()
lab_path = lab_service.generate_lab_dashboard(
metadata=result.metadata,
chunks=result.chunks,
analysis=analysis,
output_dir=Path(result.output_dir)
)
print(f"✓ Generated: {lab_path}")
# Summary
print("\n" + "=" * 60)
print("SUCCESS: Both dashboards generated!")
print("=" * 60)
print("\n📊 Dashboard Comparison:")
print(f"\n1. Standard Dashboard (Dark Mode):")
print(f" file://{Path(result.dashboard_path).absolute()}")
print(f" • OKLCH color system")
print(f" • Dark theme optimized")
print(f" • Interactive, modern design")
print(f"\n2. Lab Dashboard (Clinical Style):")
print(f" file://{lab_path.absolute()}")
print(f" • Swiss design principles")
print(f" • White background, minimal")
print(f" • High contrast, professional")
print("\n📈 Analysis Highlights:")
if analysis.topics:
print(f" Topics: {', '.join(analysis.topics[:3])}")
if analysis.top_keywords:
top_5 = [kw['keyword'] for kw in analysis.top_keywords[:5]]
print(f" Keywords: {', '.join(top_5)}")
print(f" Quality: {analysis.high_quality_chunks}/{analysis.total_chunks} high-quality chunks")
print("\n" + "=" * 60)
return {
"standard_dashboard": result.dashboard_path,
"lab_dashboard": str(lab_path),
"analysis": analysis.dict(),
"metadata": result.metadata.dict(),
"output_dir": result.output_dir
}
async def process_existing_data(output_dir: str):
"""
Generate lab dashboard from existing processed data.
Use this when you've already processed a video and want to
generate the lab dashboard without re-processing.
Args:
output_dir: Directory with existing metadata.json and chunks.jsonl
Returns:
Path to generated lab dashboard
"""
import json
from ytpipe.core.models import VideoMetadata, Chunk
print(f"\nGenerating lab dashboard from: {output_dir}")
# Load metadata
metadata_path = Path(output_dir) / "metadata.json"
with open(metadata_path, 'r') as f:
metadata_dict = json.load(f)
metadata = VideoMetadata(**metadata_dict)
# Load chunks
chunks_path = Path(output_dir) / "chunks.jsonl"
chunks = []
with open(chunks_path, 'r') as f:
for line in f:
chunk_dict = json.loads(line)
chunks.append(Chunk(**chunk_dict))
print(f"✓ Loaded: {len(chunks)} chunks")
# Run analysis
analyzer = AnalyzerService()
analysis = analyzer.analyze(metadata, chunks)
print(f"✓ Analysis: {len(analysis.top_keywords)} keywords, {len(analysis.topics)} topics")
# Generate lab dashboard
lab_service = LabDashboardService()
lab_path = lab_service.generate_lab_dashboard(
metadata=metadata,
chunks=chunks,
analysis=analysis,
output_dir=Path(output_dir)
)
print(f"✓ Dashboard: {lab_path}")
print(f"\nOpen: file://{lab_path.absolute()}")
return lab_path
# ============================================================================
# CLI Interface
# ============================================================================
def main():
"""Main entry point with CLI argument handling."""
import sys
if len(sys.argv) < 2:
print("Usage:")
print(" python INTEGRATION_EXAMPLE.py <youtube_url>")
print(" python INTEGRATION_EXAMPLE.py --existing <output_dir>")
print("\nExamples:")
print(" python INTEGRATION_EXAMPLE.py https://youtube.com/watch?v=dQw4w9WgXcQ")
print(" python INTEGRATION_EXAMPLE.py --existing ./output/dQw4w9WgXcQ")
return
if sys.argv[1] == "--existing":
# Process existing data
if len(sys.argv) < 3:
print("Error: Missing output directory")
return
output_dir = sys.argv[2]
asyncio.run(process_existing_data(output_dir))
else:
# Process new video
url = sys.argv[1]
asyncio.run(process_with_lab_dashboard(url))
if __name__ == "__main__":
main()
