""" Historical data tracking for Google Alerts Reporter Stores and retrieves analysis history for trending """ import json import logging from datetime import datetime, timedelta from pathlib import Path from typing import List, Dict, Optional from .config import settings logger = logging.getLogger(__name__) class HistoricalData: """Manages historical analysis data.""" def __init__(self): """Initialize historical data manager.""" self.data_dir = Path(settings.historical_dir) self.data_dir.mkdir(parents=True, exist_ok=True) self.retention_days = settings.historical_days def save_analysis( self, sentiments: Dict, total_items: int, timestamp: Optional[datetime] = None ) -> bool: """ Save analysis results to historical data. Args: sentiments: Sentiment counts total_items: Total items analyzed timestamp: Optional timestamp (defaults to now) Returns: True if saved successfully """ if timestamp is None: timestamp = datetime.now() try: # Create daily file date_key = timestamp.strftime('%Y-%m-%d') file_path = self.data_dir / f"{date_key}.json" # Load existing data for the day if file_path.exists(): with open(file_path, 'r') as f: daily_data = json.load(f) else: daily_data = { 'date': date_key, 'analyses': [] } # Add new analysis analysis = { 'timestamp': timestamp.isoformat(), 'sentiments': sentiments, 'total_items': total_items, 'percent_negative': (sentiments['neg'] / total_items * 100) if total_items > 0 else 0, 'percent_positive': (sentiments['pos'] / total_items * 100) if total_items > 0 else 0, 'percent_neutral': (sentiments['neu'] / total_items * 100) if total_items > 0 else 0 } daily_data['analyses'].append(analysis) # Save updated data with open(file_path, 'w') as f: json.dump(daily_data, f, indent=2) logger.info(f"Saved historical data for {date_key}") return True except Exception as e: logger.error(f"Failed to save historical data: {e}") return False def get_history(self, days: Optional[int] = None) -> List[Dict]: """ Get historical data for the last N days. Args: days: Number of days to retrieve (defaults to settings) Returns: List of historical data entries """ if days is None: days = self.retention_days try: history = [] start_date = datetime.now() - timedelta(days=days) # Iterate through date range current_date = start_date while current_date <= datetime.now(): date_key = current_date.strftime('%Y-%m-%d') file_path = self.data_dir / f"{date_key}.json" if file_path.exists(): with open(file_path, 'r') as f: daily_data = json.load(f) history.append(daily_data) current_date += timedelta(days=1) logger.info(f"Retrieved {len(history)} days of historical data") return history except Exception as e: logger.error(f"Failed to retrieve historical data: {e}") return [] def get_summary(self, days: Optional[int] = None) -> Dict: """ Get summary statistics from historical data. Args: days: Number of days to analyze Returns: Summary statistics dictionary """ history = self.get_history(days) if not history: return { 'days': 0, 'total_analyses': 0, 'avg_negative': 0, 'avg_positive': 0, 'avg_neutral': 0, 'trend': 'neutral' } # Aggregate all analyses all_analyses = [] for day in history: all_analyses.extend(day['analyses']) if not all_analyses: return { 'days': len(history), 'total_analyses': 0, 'avg_negative': 0, 'avg_positive': 0, 'avg_neutral': 0, 'trend': 'neutral' } # Calculate averages avg_negative = sum(a['percent_negative'] for a in all_analyses) / len(all_analyses) avg_positive = sum(a['percent_positive'] for a in all_analyses) / len(all_analyses) avg_neutral = sum(a['percent_neutral'] for a in all_analyses) / len(all_analyses) # Determine trend (comparing first half vs second half) mid_point = len(all_analyses) // 2 if mid_point > 0: first_half_neg = sum(a['percent_negative'] for a in all_analyses[:mid_point]) / mid_point second_half_neg = sum(a['percent_negative'] for a in all_analyses[mid_point:]) / (len(all_analyses) - mid_point) if second_half_neg > first_half_neg + 5: trend = 'worsening' elif second_half_neg < first_half_neg - 5: trend = 'improving' else: trend = 'stable' else: trend = 'neutral' return { 'days': len(history), 'total_analyses': len(all_analyses), 'avg_negative': round(avg_negative, 1), 'avg_positive': round(avg_positive, 1), 'avg_neutral': round(avg_neutral, 1), 'trend': trend, 'first_date': all_analyses[0]['timestamp'] if all_analyses else None, 'last_date': all_analyses[-1]['timestamp'] if all_analyses else None } def get_chart_data(self, days: Optional[int] = None) -> Dict: """ Get data formatted for Chart.js. Args: days: Number of days Returns: Chart.js compatible data structure """ history = self.get_history(days) labels = [] negative_data = [] positive_data = [] neutral_data = [] for day in history: # Use last analysis of the day if day['analyses']: last_analysis = day['analyses'][-1] labels.append(day['date']) negative_data.append(round(last_analysis['percent_negative'], 1)) positive_data.append(round(last_analysis['percent_positive'], 1)) neutral_data.append(round(last_analysis['percent_neutral'], 1)) return { 'labels': labels, 'datasets': [ { 'label': 'Negativos', 'data': negative_data, 'borderColor': '#f5576c', 'backgroundColor': 'rgba(245, 87, 108, 0.1)', 'tension': 0.4 }, { 'label': 'Positivos', 'data': positive_data, 'borderColor': '#00f2fe', 'backgroundColor': 'rgba(0, 242, 254, 0.1)', 'tension': 0.4 }, { 'label': 'Neutros', 'data': neutral_data, 'borderColor': '#fee140', 'backgroundColor': 'rgba(254, 225, 64, 0.1)', 'tension': 0.4 } ] } def cleanup_old_data(self) -> int: """ Delete historical data older than retention period. Returns: Number of files deleted """ try: cutoff_date = datetime.now() - timedelta(days=self.retention_days) deleted_count = 0 for file_path in self.data_dir.glob('*.json'): # Parse date from filename (YYYY-MM-DD.json) try: date_str = file_path.stem file_date = datetime.strptime(date_str, '%Y-%m-%d') if file_date < cutoff_date: file_path.unlink() deleted_count += 1 logger.info(f"Deleted old historical data: {file_path.name}") except ValueError: # Skip files that don't match date format continue logger.info(f"Cleanup complete: {deleted_count} files deleted") return deleted_count except Exception as e: logger.error(f"Failed to cleanup historical data: {e}") return 0