Update app.py

app.py

@@ -147,8 +147,186 @@ class DataAnalysisAssistant:
         return f'<div class="analysis-text">{response}</div>'
 
 class AnalysisHistory:
-    """Manages analysis history and persistence."""
-    [Previous AnalysisHistory implementation remains the same]
+    """
+    Manages analysis history and persistence.
+    
+    Attributes:
+        history_file (Path): Path to the JSON file storing analysis history
+        history (List[Dict]): List of historical analysis entries
+        
+    Each history entry is a dictionary containing:
+        - timestamp: ISO format timestamp
+        - query: The user's analysis query
+        - result: The analysis result/response
+    """
+    
+    def __init__(self, history_file: str = HISTORY_FILE):
+        """
+        Initialize the analysis history manager.
+
+        Args:
+            history_file (str): Path to history JSON file. Defaults to HISTORY_FILE.
+        """
+        self.history_file = Path(history_file)
+        self.history = self._load_history()
+
+    def _load_history(self) -> List[Dict]:
+        """
+        Load analysis history from file.
+
+        Returns:
+            List[Dict]: List of historical analysis entries
+        """
+        if self.history_file.exists():
+            try:
+                with self.history_file.open('r') as f:
+                    return json.load(f)
+            except json.JSONDecodeError:
+                logger.error(f"Invalid JSON in history file: {self.history_file}")
+                return []
+            except Exception as e:
+                logger.error(f"Error loading history file: {str(e)}")
+                return []
+        return []
+
+    def add_entry(self, query: str, result: str) -> None:
+        """
+        Add a new analysis entry to history.
+
+        Args:
+            query (str): The analysis query
+            result (str): The analysis result/response
+        """
+        entry = {
+            'timestamp': datetime.now().isoformat(),
+            'query': query,
+            'result': result
+        }
+        self.history.append(entry)
+        self._save_history()
+
+    def get_recent_analyses(self, limit: int = 5) -> List[Dict]:
+        """
+        Get most recent analysis entries.
+
+        Args:
+            limit (int): Maximum number of entries to return. Defaults to 5.
+
+        Returns:
+            List[Dict]: Recent analysis entries, sorted by timestamp (newest first)
+        """
+        return sorted(
+            self.history,
+            key=lambda x: x['timestamp'],
+            reverse=True
+        )[:limit]
+
+    def _save_history(self) -> None:
+        """Save history to file."""
+        try:
+            with self.history_file.open('w') as f:
+                json.dump(self.history, f, indent=2)
+        except Exception as e:
+            logger.error(f"Failed to save history: {str(e)}")
+
+    def clear_history(self) -> None:
+        """Clear all analysis history."""
+        self.history = []
+        self._save_history()
+
+    def get_history_by_date(self, start_date: datetime, end_date: datetime) -> List[Dict]:
+        """
+        Get analysis history within a date range.
+
+        Args:
+            start_date (datetime): Start of date range
+            end_date (datetime): End of date range
+
+        Returns:
+            List[Dict]: Analysis entries within the specified date range
+        """
+        filtered_history = []
+        for entry in self.history:
+            try:
+                entry_date = datetime.fromisoformat(entry['timestamp'])
+                if start_date <= entry_date <= end_date:
+                    filtered_history.append(entry)
+            except Exception as e:
+                logger.error(f"Error parsing entry date: {str(e)}")
+                continue
+        return filtered_history
+
+    def search_history(self, search_term: str) -> List[Dict]:
+        """
+        Search analysis history for a specific term.
+
+        Args:
+            search_term (str): Term to search for in queries and results
+
+        Returns:
+            List[Dict]: Matching analysis entries
+        """
+        search_term = search_term.lower()
+        return [
+            entry for entry in self.history
+            if search_term in entry['query'].lower() 
+            or search_term in entry['result'].lower()
+        ]
+
+    def get_statistics(self) -> Dict[str, Any]:
+        """
+        Get statistics about the analysis history.
+
+        Returns:
+            Dict[str, Any]: Statistics including total entries, date range, etc.
+        """
+        if not self.history:
+            return {
+                "total_entries": 0,
+                "date_range": None,
+                "average_entries_per_day": 0
+            }
+
+        dates = [datetime.fromisoformat(entry['timestamp']) for entry in self.history]
+        first_date = min(dates)
+        last_date = max(dates)
+        days_span = (last_date - first_date).days or 1
+
+        return {
+            "total_entries": len(self.history),
+            "date_range": {
+                "first": first_date.isoformat(),
+                "last": last_date.isoformat()
+            },
+            "average_entries_per_day": len(self.history) / days_span
+        }
+
+    def export_history(self, format: str = 'json') -> str:
+        """
+        Export analysis history in specified format.
+
+        Args:
+            format (str): Export format ('json' or 'csv'). Defaults to 'json'.
+
+        Returns:
+            str: Path to exported file
+
+        Raises:
+            ValueError: If format is not supported
+        """
+        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
+        if format == 'json':
+            export_path = f'analysis_history_{timestamp}.json'
+            with open(export_path, 'w') as f:
+                json.dump(self.history, f, indent=2)
+            return export_path
+        elif format == 'csv':
+            export_path = f'analysis_history_{timestamp}.csv'
+            df = pd.DataFrame(self.history)
+            df.to_csv(export_path, index=False)
+            return export_path
+        else:
+            raise ValueError(f"Unsupported export format: {format}")
 
 def process_file(file: gr.File) -> Optional[pd.DataFrame]:
     """Process uploaded file into DataFrame."""