#!/usr/bin/env python3
"""
Test Enhanced Analyzer Service
Tests the new analysis capabilities added to AnalyzerService:
- generate_summary()
- extract_entities()
- analyze_sentiment()
- calculate_difficulty()
- extract_action_items()
"""
import json
from pathlib import Path
from ytpipe.services.intelligence.analyzer import AnalyzerService
from ytpipe.core.models import VideoMetadata, Chunk
def load_test_data():
"""Load test data from a processed video."""
# Look for any processed video in data directory
data_dir = Path("data")
if not data_dir.exists():
print("❌ No data directory found. Run ytpipe on a video first.")
return None, None
# Find first video directory
video_dirs = [d for d in data_dir.iterdir() if d.is_dir()]
if not video_dirs:
print("❌ No processed videos found in data/")
return None, None
video_dir = video_dirs[0]
print(f"📂 Using video: {video_dir.name}")
# Load metadata
metadata_path = video_dir / "metadata.json"
if not metadata_path.exists():
print(f"❌ No metadata.json found in {video_dir}")
return None, None
with open(metadata_path) as f:
metadata_dict = json.load(f)
metadata = VideoMetadata(**metadata_dict)
# Load chunks
chunks_path = video_dir / "chunks.jsonl"
if not chunks_path.exists():
print(f"❌ No chunks.jsonl found in {video_dir}")
return None, None
chunks = []
with open(chunks_path) as f:
for line in f:
chunk_dict = json.loads(line)
chunks.append(Chunk(**chunk_dict))
print(f"✅ Loaded {len(chunks)} chunks")
return metadata, chunks
def test_generate_summary(analyzer, metadata, chunks):
"""Test summary generation."""
print("\n" + "="*60)
print("📝 SUMMARY GENERATION")
print("="*60)
summary = analyzer.generate_summary(metadata, chunks, max_bullets=5)
print(f"\nGenerated {len(summary)} bullet points:")
for i, bullet in enumerate(summary, 1):
print(f"{i}. {bullet}")
return summary
def test_extract_entities(analyzer, chunks):
"""Test entity extraction."""
print("\n" + "="*60)
print("🏷️ ENTITY EXTRACTION")
print("="*60)
entities = analyzer.extract_entities(chunks, max_entities=10)
print(f"\nExtracted {len(entities)} entities:")
for entity in entities:
print(f" • {entity['entity']:20s} [{entity['type']:8s}] - {entity['count']} occurrences")
return entities
def test_analyze_sentiment(analyzer, chunks):
"""Test sentiment analysis."""
print("\n" + "="*60)
print("😊 SENTIMENT ANALYSIS")
print("="*60)
sentiment = analyzer.analyze_sentiment(chunks)
print(f"\nOverall Sentiment: {sentiment['sentiment'].upper()}")
print(f"Sentiment Score: {sentiment['score']:.2f} (0=negative, 1=positive)")
print(f"\nDistribution:")
print(f" Positive words: {sentiment['distribution']['positive']}")
print(f" Negative words: {sentiment['distribution']['negative']}")
print(f" Neutral words: {sentiment['distribution']['neutral']}")
return sentiment
def test_calculate_difficulty(analyzer, chunks):
"""Test difficulty calculation."""
print("\n" + "="*60)
print("📊 DIFFICULTY ANALYSIS")
print("="*60)
difficulty = analyzer.calculate_difficulty(chunks)
print(f"\nDifficulty Level: {difficulty['level'].upper()}")
print(f"Difficulty Score: {difficulty['score']:.2f} (0=beginner, 1=expert)")
print(f"\nFactors:")
for factor, value in difficulty['factors'].items():
print(f" {factor:25s}: {value}")
return difficulty
def test_extract_action_items(analyzer, chunks):
"""Test action item extraction."""
print("\n" + "="*60)
print("✓ ACTION ITEMS")
print("="*60)
action_items = analyzer.extract_action_items(chunks, max_items=5)
print(f"\nExtracted {len(action_items)} action items:")
for i, item in enumerate(action_items, 1):
print(f"{i}. {item}")
return action_items
def main():
"""Run all enhanced analyzer tests."""
print("="*60)
print("🧪 ENHANCED ANALYZER SERVICE TEST")
print("="*60)
# Load test data
metadata, chunks = load_test_data()
if not metadata or not chunks:
print("\n❌ Cannot run tests without data. Please process a video first:")
print(" python -m ytpipe.cli.main 'https://youtube.com/watch?v=VIDEO_ID'")
return
# Initialize analyzer
analyzer = AnalyzerService()
# Run all tests
summary = test_generate_summary(analyzer, metadata, chunks)
entities = test_extract_entities(analyzer, chunks)
sentiment = test_analyze_sentiment(analyzer, chunks)
difficulty = test_calculate_difficulty(analyzer, chunks)
action_items = test_extract_action_items(analyzer, chunks)
# Summary
print("\n" + "="*60)
print("✅ ALL TESTS COMPLETED")
print("="*60)
print(f"Summary Points: {len(summary)}")
print(f"Entities Found: {len(entities)}")
print(f"Sentiment: {sentiment['sentiment']} ({sentiment['score']:.2f})")
print(f"Difficulty: {difficulty['level']} ({difficulty['score']:.2f})")
print(f"Action Items: {len(action_items)}")
print("\n🎉 Enhanced analyzer features ready for dashboard integration!")
if __name__ == "__main__":
main()
