Spaces:
Sleeping
Sleeping
File size: 2,833 Bytes
0e66264 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 | from typing import Dict, List, Optional
from datetime import datetime, timedelta
import numpy as np
from .gemini_analyzer import GeminiAnalyzer
from .database import DatabaseManager
class PredictionEngine:
def __init__(self, gemini_analyzer: GeminiAnalyzer, db_manager: DatabaseManager):
self.gemini_analyzer = gemini_analyzer
self.db_manager = db_manager
def generate_predictions(self) -> List[Dict]:
"""Generate predictions for current narratives"""
try:
# Get active high-risk narratives
narratives = self.db_manager.get_high_risk_narratives(threshold=60)
predictions = []
for narrative in narratives:
scenarios = self.gemini_analyzer.generate_prediction_scenarios(narrative['content'])
for scenario in scenarios:
prediction = {
'narrative_id': narrative['id'],
'narrative_title': narrative['title'][:50] + "..." if len(narrative['title']) > 50 else narrative['title'],
'risk_score': narrative['risk_score'],
'probability': scenario.get('probability', 50),
'predicted_reach': scenario.get('predicted_reach', 10000),
'timeline': f"{scenario.get('timeline_hours', 24)} hours",
'best_case_scenario': scenario.get('description', 'No description available')
}
predictions.append(prediction)
return predictions[:10] # Limit to top 10
except Exception as e:
print(f"❌ Prediction generation failed: {e}")
return []
def calculate_spread_probability(self, narrative_content: str, historical_data: List[Dict] = None) -> float:
"""Calculate probability of viral spread"""
# This would use ML models in production
# For now, return a calculated estimate
base_probability = 0.3
# Adjust based on content characteristics
if any(term in narrative_content.lower() for term in ['breaking', 'urgent', 'exclusive']):
base_probability += 0.2
if any(term in narrative_content.lower() for term in ['government', 'conspiracy', 'cover-up']):
base_probability += 0.15
return min(base_probability, 0.95)
def predict_timeline(self, narrative_content: str) -> Dict:
"""Predict spread timeline"""
return {
'initial_detection': 0, # hours
'community_spread': 6, # hours
'cross_platform': 12, # hours
'mainstream_pickup': 24, # hours
'peak_virality': 48 # hours
} |