210 lines
6.6 KiB
Python
210 lines
6.6 KiB
Python
"""
|
||
Test script for AI comment analysis using OpenRouter.
|
||
"""
|
||
import os
|
||
import sys
|
||
import django
|
||
|
||
# Setup Django
|
||
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
django.setup()
|
||
|
||
from apps.social.models import SocialMediaComment
|
||
from apps.social.services.openrouter_service import OpenRouterService
|
||
from apps.social.services.analysis_service import AnalysisService
|
||
from django.utils import timezone
|
||
from datetime import timedelta
|
||
|
||
|
||
def test_openrouter_service():
|
||
"""Test the OpenRouter service."""
|
||
print("\n" + "="*60)
|
||
print("Testing OpenRouter Service")
|
||
print("="*60)
|
||
|
||
service = OpenRouterService()
|
||
|
||
# Check if configured
|
||
if not service.is_configured():
|
||
print("❌ OpenRouter service not configured. Please set OPENROUTER_API_KEY in .env")
|
||
return False
|
||
|
||
print("✅ OpenRouter service configured")
|
||
|
||
# Test single comment analysis
|
||
test_comments = [
|
||
{'id': 1, 'text': 'This is an amazing video! I learned so much.'},
|
||
{'id': 2, 'text': 'Terrible content, waste of time.'},
|
||
{'id': 3, 'text': 'Okay video, could be better.'}
|
||
]
|
||
|
||
print(f"\nAnalyzing {len(test_comments)} test comments...")
|
||
result = service.analyze_comments(test_comments)
|
||
|
||
if result.get('success'):
|
||
print("✅ Analysis successful!")
|
||
for analysis in result.get('analyses', []):
|
||
print(f"\nComment {analysis.get('comment_id')}:")
|
||
print(f" Sentiment: {analysis.get('sentiment')}")
|
||
print(f" Score: {analysis.get('sentiment_score')}")
|
||
print(f" Confidence: {analysis.get('confidence')}")
|
||
print(f" Keywords: {analysis.get('keywords')}")
|
||
print(f" Topics: {analysis.get('topics')}")
|
||
return True
|
||
else:
|
||
print(f"❌ Analysis failed: {result.get('error')}")
|
||
return False
|
||
|
||
|
||
def test_analysis_service():
|
||
"""Test the Analysis service."""
|
||
print("\n" + "="*60)
|
||
print("Testing Analysis Service")
|
||
print("="*60)
|
||
|
||
service = AnalysisService()
|
||
|
||
# Get statistics
|
||
print("\nGetting analysis statistics...")
|
||
stats = service.get_analysis_statistics(days=30)
|
||
|
||
print(f"\n✅ Statistics retrieved:")
|
||
print(f" Total comments: {stats['total_comments']}")
|
||
print(f" Analyzed: {stats['analyzed_comments']}")
|
||
print(f" Unanalyzed: {stats['unanalyzed_comments']}")
|
||
print(f" Analysis rate: {stats['analysis_rate']:.2f}%")
|
||
print(f" Sentiment distribution: {stats['sentiment_distribution']}")
|
||
print(f" Average confidence: {stats['average_confidence']:.4f}")
|
||
|
||
# Get top keywords
|
||
print("\nGetting top keywords...")
|
||
keywords = service.get_top_keywords(limit=10, days=30)
|
||
print(f"✅ Top {len(keywords)} keywords:")
|
||
for i, kw in enumerate(keywords[:5], 1):
|
||
print(f" {i}. {kw['keyword']} ({kw['count']} mentions)")
|
||
|
||
return True
|
||
|
||
|
||
def test_database_queries():
|
||
"""Test database queries for analyzed comments."""
|
||
print("\n" + "="*60)
|
||
print("Testing Database Queries")
|
||
print("="*60)
|
||
|
||
# Count total comments
|
||
total = SocialMediaComment.objects.count()
|
||
print(f"\nTotal comments in database: {total}")
|
||
|
||
# Count analyzed comments (those with ai_analysis populated)
|
||
analyzed_count = 0
|
||
for comment in SocialMediaComment.objects.all():
|
||
if comment.ai_analysis and comment.ai_analysis != {}:
|
||
analyzed_count += 1
|
||
print(f"Analyzed comments: {analyzed_count}")
|
||
|
||
# Count unanalyzed comments
|
||
unanalyzed_count = total - analyzed_count
|
||
print(f"Unanalyzed comments: {unanalyzed_count}")
|
||
|
||
# Get recent analyzed comments
|
||
recent_analyzed = SocialMediaComment.objects.all()
|
||
|
||
# Filter for analyzed comments and sort by scraped_at
|
||
analyzed_list = []
|
||
for comment in recent_analyzed:
|
||
if comment.ai_analysis and comment.ai_analysis != {}:
|
||
analyzed_list.append(comment)
|
||
|
||
analyzed_list.sort(key=lambda x: x.scraped_at or timezone.now(), reverse=True)
|
||
recent_analyzed = analyzed_list[:5]
|
||
|
||
print(f"\nRecent analyzed comments:")
|
||
for comment in recent_analyzed:
|
||
sentiment = comment.ai_analysis.get('sentiment', {}).get('classification', {}).get('en', 'N/A') if comment.ai_analysis else 'N/A'
|
||
confidence = comment.ai_analysis.get('sentiment', {}).get('confidence', 0) if comment.ai_analysis else 0
|
||
print(f" - {comment.platform}: {sentiment} (confidence: {confidence})")
|
||
|
||
return True
|
||
|
||
|
||
def test_pending_analysis():
|
||
"""Test analyzing pending comments."""
|
||
print("\n" + "="*60)
|
||
print("Testing Pending Comment Analysis")
|
||
print("="*60)
|
||
|
||
service = AnalysisService()
|
||
|
||
# Get count of pending comments (using ai_analysis check)
|
||
pending_count = 0
|
||
for comment in SocialMediaComment.objects.all():
|
||
if not comment.ai_analysis or comment.ai_analysis == {}:
|
||
pending_count += 1
|
||
|
||
print(f"\nPending comments to analyze: {pending_count}")
|
||
|
||
if pending_count == 0:
|
||
print("ℹ️ No pending comments to analyze")
|
||
return True
|
||
|
||
# Analyze a small batch (limit to 5 for testing)
|
||
print(f"\nAnalyzing up to 5 pending comments...")
|
||
result = service.analyze_pending_comments(limit=5)
|
||
|
||
if result.get('success'):
|
||
print(f"✅ Analysis complete:")
|
||
print(f" Analyzed: {result['analyzed']}")
|
||
print(f" Failed: {result['failed']}")
|
||
print(f" Skipped: {result['skipped']}")
|
||
return True
|
||
else:
|
||
print(f"❌ Analysis failed: {result.get('error')}")
|
||
return False
|
||
|
||
|
||
def main():
|
||
"""Run all tests."""
|
||
print("\n" + "="*60)
|
||
print("AI Comment Analysis Test Suite")
|
||
print("="*60)
|
||
|
||
results = []
|
||
|
||
# Test 1: OpenRouter Service
|
||
results.append(('OpenRouter Service', test_openrouter_service()))
|
||
|
||
# Test 2: Analysis Service
|
||
results.append(('Analysis Service', test_analysis_service()))
|
||
|
||
# Test 3: Database Queries
|
||
results.append(('Database Queries', test_database_queries()))
|
||
|
||
# Test 4: Pending Analysis
|
||
results.append(('Pending Analysis', test_pending_analysis()))
|
||
|
||
# Summary
|
||
print("\n" + "="*60)
|
||
print("Test Summary")
|
||
print("="*60)
|
||
|
||
for test_name, passed in results:
|
||
status = "✅ PASSED" if passed else "❌ FAILED"
|
||
print(f"{test_name}: {status}")
|
||
|
||
all_passed = all(result[1] for result in results)
|
||
|
||
print("\n" + "="*60)
|
||
if all_passed:
|
||
print("✅ All tests passed!")
|
||
else:
|
||
print("❌ Some tests failed")
|
||
print("="*60 + "\n")
|
||
|
||
return all_passed
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|