hvac-kia-content/test_youtube_competitive_enhanced.py

#!/usr/bin/env python3
"""
Test script for enhanced YouTube competitive intelligence scraper system.
Demonstrates Phase 2 features including centralized quota management, 
enhanced analysis, and comprehensive competitive intelligence.
"""

import os
import sys
import json
import logging
from pathlib import Path

# Add src to path
sys.path.append(str(Path(__file__).parent / 'src'))

from competitive_intelligence.youtube_competitive_scraper import (
    create_single_youtube_competitive_scraper,
    create_youtube_competitive_scrapers,
    YouTubeQuotaManager
)

def setup_logging():
    """Setup logging for testing."""
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
        handlers=[
            logging.StreamHandler(),
            logging.FileHandler('test_youtube_competitive.log')
        ]
    )

def test_quota_manager():
    """Test centralized quota management."""
    print("=" * 60)
    print("TESTING CENTRALIZED QUOTA MANAGER")
    print("=" * 60)
    
    # Get quota manager instance
    quota_manager = YouTubeQuotaManager()
    
    # Show initial status
    status = quota_manager.get_quota_status()
    print(f"Initial Quota Status:")
    print(f"  Used: {status['quota_used']}")
    print(f"  Remaining: {status['quota_remaining']}")
    print(f"  Limit: {status['quota_limit']}")
    print(f"  Percentage: {status['quota_percentage']:.1f}%")
    print(f"  Reset Time: {status['quota_reset_time']}")
    
    # Test quota reservation
    print(f"\nTesting quota reservation...")
    operations = ['channels_list', 'playlist_items_list', 'videos_list']
    
    for operation in operations:
        success = quota_manager.check_and_reserve_quota(operation, 1)
        print(f"  Reserve {operation}: {'✓' if success else '✗'}")
        if success:
            status = quota_manager.get_quota_status()
            print(f"    New quota used: {status['quota_used']}")

def test_single_scraper():
    """Test creating and using a single competitive scraper."""
    print("\n" + "=" * 60)
    print("TESTING SINGLE COMPETITOR SCRAPER")
    print("=" * 60)
    
    # Test with AC Service Tech (high priority competitor)
    competitor = 'ac_service_tech'
    data_dir = Path('data')
    logs_dir = Path('logs')
    
    print(f"Creating scraper for: {competitor}")
    
    scraper = create_single_youtube_competitive_scraper(data_dir, logs_dir, competitor)
    
    if not scraper:
        print("❌ Failed to create scraper")
        return
    
    print("✅ Scraper created successfully")
    
    # Get competitor metadata
    metadata = scraper.get_competitor_metadata()
    print(f"\nCompetitor Metadata:")
    print(f"  Name: {metadata['competitor_name']}")
    print(f"  Handle: {metadata['channel_handle']}")
    print(f"  Category: {metadata['competitive_profile']['category']}")
    print(f"  Priority: {metadata['competitive_profile']['competitive_priority']}")
    print(f"  Target Audience: {metadata['competitive_profile']['target_audience']}")
    print(f"  Content Focus: {', '.join(metadata['competitive_profile']['content_focus'])}")
    
    # Test content discovery (limited sample)
    print(f"\nTesting content discovery (5 videos)...")
    try:
        videos = scraper.discover_content_urls(5)
        print(f"✅ Discovered {len(videos)} videos")
        
        if videos:
            sample_video = videos[0]
            print(f"\nSample video analysis:")
            print(f"  Title: {sample_video['title'][:50]}...")
            print(f"  Published: {sample_video['published_at']}")
            print(f"  Content Focus Tags: {sample_video.get('content_focus_tags', [])}")
            print(f"  Days Since Publish: {sample_video.get('days_since_publish', 'Unknown')}")
    
    except Exception as e:
        print(f"❌ Content discovery failed: {e}")
    
    # Test competitive analysis
    print(f"\nTesting competitive analysis...")
    try:
        analysis = scraper.run_competitor_analysis()
        
        if 'error' in analysis:
            print(f"❌ Analysis failed: {analysis['error']}")
        else:
            print(f"✅ Analysis completed successfully")
            print(f"  Sample Size: {analysis['sample_size']}")
            
            # Show key insights
            if 'content_analysis' in analysis:
                content = analysis['content_analysis']
                print(f"  Primary Content Focus: {content.get('primary_content_focus', 'Unknown')}")
                print(f"  Content Diversity Score: {content.get('content_diversity_score', 0)}")
            
            if 'competitive_positioning' in analysis:
                positioning = analysis['competitive_positioning']
                overlap = positioning.get('content_overlap', {})
                print(f"  Content Overlap: {overlap.get('total_overlap_percentage', 0)}%")
                print(f"  Competition Level: {overlap.get('direct_competition_level', 'unknown')}")
            
            if 'content_gaps' in analysis:
                gaps = analysis['content_gaps']
                print(f"  Opportunity Score: {gaps.get('opportunity_score', 0)}")
                opportunities = gaps.get('hkia_opportunities', [])
                if opportunities:
                    print(f"  Key Opportunities:")
                    for opp in opportunities[:3]:
                        print(f"    • {opp}")
    
    except Exception as e:
        print(f"❌ Competitive analysis failed: {e}")

def test_all_scrapers():
    """Test creating all YouTube competitive scrapers."""
    print("\n" + "=" * 60)
    print("TESTING ALL COMPETITIVE SCRAPERS")
    print("=" * 60)
    
    data_dir = Path('data')
    logs_dir = Path('logs')
    
    print("Creating all YouTube competitive scrapers...")
    scrapers = create_youtube_competitive_scrapers(data_dir, logs_dir)
    
    print(f"\nCreated {len(scrapers)} scrapers:")
    for key, scraper in scrapers.items():
        metadata = scraper.get_competitor_metadata()
        print(f"  • {key}: {metadata['competitor_name']} ({metadata['competitive_profile']['competitive_priority']} priority)")
    
    # Test quota status after all scrapers created
    quota_manager = YouTubeQuotaManager()
    final_status = quota_manager.get_quota_status()
    print(f"\nFinal quota status:")
    print(f"  Used: {final_status['quota_used']}/{final_status['quota_limit']} ({final_status['quota_percentage']:.1f}%)")

def main():
    """Main test function."""
    print("YouTube Competitive Intelligence Scraper - Phase 2 Enhanced Testing")
    print("=" * 70)
    
    # Setup logging
    setup_logging()
    
    # Check environment
    if not os.getenv('YOUTUBE_API_KEY'):
        print("❌ YOUTUBE_API_KEY environment variable not set")
        print("Please set YOUTUBE_API_KEY to test the scrapers")
        return
    
    try:
        # Test quota manager
        test_quota_manager()
        
        # Test single scraper
        test_single_scraper()
        
        # Test all scrapers creation
        test_all_scrapers()
        
        print("\n" + "=" * 60)
        print("TESTING COMPLETE")
        print("=" * 60)
        print("✅ All tests completed successfully!")
        print("Check logs for detailed information.")
        
    except Exception as e:
        print(f"\n❌ Testing failed: {e}")
        raise

if __name__ == '__main__':
    main()