Spaces:
Sleeping
Sleeping
| 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 | |
| } |