#!/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()