| |
| import json |
| import os |
| from datetime import datetime |
| from typing import Dict, Any, Optional |
|
|
| class MemoryManager: |
| """Manages session data and persistent storage for the LinkedIn enhancer""" |
| |
| def __init__(self, storage_dir: str = "data"): |
| self.storage_dir = storage_dir |
| self.session_data = {} |
| self.persistent_file = os.path.join(storage_dir, "persistent_data.json") |
| |
| |
| os.makedirs(storage_dir, exist_ok=True) |
| |
| |
| self.persistent_data = self._load_persistent_data() |
| |
| def store_session(self, profile_url: str, data: Dict[str, Any]) -> None: |
| """ |
| Store session data for a specific profile |
| |
| Args: |
| profile_url (str): LinkedIn profile URL as key |
| data (Dict[str, Any]): Session data to store |
| """ |
| session_key = self._create_session_key(profile_url) |
| |
| self.session_data[session_key] = { |
| 'timestamp': datetime.now().isoformat(), |
| 'profile_url': profile_url, |
| 'data': data |
| } |
| |
| def get_session(self, profile_url: str) -> Optional[Dict[str, Any]]: |
| """ |
| Retrieve session data for a specific profile |
| |
| Args: |
| profile_url (str): LinkedIn profile URL |
| |
| Returns: |
| Optional[Dict[str, Any]]: Session data if exists |
| """ |
| session_key = self._create_session_key(profile_url) |
| return self.session_data.get(session_key) |
| |
| def store_persistent(self, key: str, data: Any) -> None: |
| """ |
| Store data persistently to disk |
| |
| Args: |
| key (str): Storage key |
| data (Any): Data to store |
| """ |
| self.persistent_data[key] = { |
| 'timestamp': datetime.now().isoformat(), |
| 'data': data |
| } |
| |
| self._save_persistent_data() |
| |
| def get_persistent(self, key: str) -> Optional[Any]: |
| """ |
| Retrieve persistent data |
| |
| Args: |
| key (str): Storage key |
| |
| Returns: |
| Optional[Any]: Stored data if exists |
| """ |
| stored_item = self.persistent_data.get(key) |
| return stored_item['data'] if stored_item else None |
| |
| def store_user_preferences(self, user_id: str, preferences: Dict[str, Any]) -> None: |
| """ |
| Store user preferences |
| |
| Args: |
| user_id (str): User identifier |
| preferences (Dict[str, Any]): User preferences |
| """ |
| pref_key = f"user_preferences_{user_id}" |
| self.store_persistent(pref_key, preferences) |
| |
| def get_user_preferences(self, user_id: str) -> Dict[str, Any]: |
| """ |
| Retrieve user preferences |
| |
| Args: |
| user_id (str): User identifier |
| |
| Returns: |
| Dict[str, Any]: User preferences |
| """ |
| pref_key = f"user_preferences_{user_id}" |
| preferences = self.get_persistent(pref_key) |
| return preferences if preferences else {} |
| |
| def store_analysis_history(self, profile_url: str, analysis: Dict[str, Any]) -> None: |
| """ |
| Store analysis history for tracking improvements |
| |
| Args: |
| profile_url (str): LinkedIn profile URL |
| analysis (Dict[str, Any]): Analysis results |
| """ |
| history_key = f"analysis_history_{self._create_session_key(profile_url)}" |
| |
| |
| history = self.get_persistent(history_key) or [] |
| |
| |
| history.append({ |
| 'timestamp': datetime.now().isoformat(), |
| 'analysis': analysis |
| }) |
| |
| |
| history = history[-10:] |
| |
| self.store_persistent(history_key, history) |
| |
| def get_analysis_history(self, profile_url: str) -> list: |
| """ |
| Retrieve analysis history for a profile |
| |
| Args: |
| profile_url (str): LinkedIn profile URL |
| |
| Returns: |
| list: Analysis history |
| """ |
| history_key = f"analysis_history_{self._create_session_key(profile_url)}" |
| return self.get_persistent(history_key) or [] |
| |
| def clear_session(self, profile_url: str = None) -> None: |
| """ |
| Clear session data |
| |
| Args: |
| profile_url (str, optional): Specific profile to clear, or all if None |
| """ |
| if profile_url: |
| session_key = self._create_session_key(profile_url) |
| self.session_data.pop(session_key, None) |
| else: |
| self.session_data.clear() |
| |
| def clear_session_cache(self, profile_url: str = None) -> None: |
| """ |
| Clear session cache for a specific profile or all profiles |
| |
| Args: |
| profile_url (str, optional): URL to clear cache for. If None, clears all. |
| """ |
| if profile_url: |
| session_key = self._create_session_key(profile_url) |
| if session_key in self.session_data: |
| del self.session_data[session_key] |
| print(f"🗑️ Cleared session cache for: {profile_url}") |
| else: |
| self.session_data.clear() |
| print("🗑️ Cleared all session cache") |
| |
| def force_refresh_session(self, profile_url: str) -> None: |
| """ |
| Force refresh by clearing cache for a specific profile |
| |
| Args: |
| profile_url (str): LinkedIn profile URL |
| """ |
| self.clear_session_cache(profile_url) |
| print(f"🔄 Forced refresh for: {profile_url}") |
|
|
| def get_session_summary(self) -> Dict[str, Any]: |
| """ |
| Get summary of current session data |
| |
| Returns: |
| Dict[str, Any]: Session summary |
| """ |
| return { |
| 'active_sessions': len(self.session_data), |
| 'sessions': list(self.session_data.keys()), |
| 'storage_location': self.storage_dir |
| } |
| |
| def _create_session_key(self, profile_url: str) -> str: |
| """Create a clean session key from profile URL""" |
| |
| import hashlib |
| return hashlib.md5(profile_url.encode()).hexdigest()[:16] |
| |
| def _load_persistent_data(self) -> Dict[str, Any]: |
| """Load persistent data from disk""" |
| if os.path.exists(self.persistent_file): |
| try: |
| with open(self.persistent_file, 'r', encoding='utf-8') as f: |
| return json.load(f) |
| except (json.JSONDecodeError, IOError): |
| return {} |
| return {} |
| |
| def _save_persistent_data(self) -> None: |
| """Save persistent data to disk""" |
| try: |
| with open(self.persistent_file, 'w', encoding='utf-8') as f: |
| json.dump(self.persistent_data, f, indent=2, ensure_ascii=False) |
| except IOError as e: |
| print(f"Warning: Could not save persistent data: {e}") |
| |
| def export_data(self, filename: str = None) -> str: |
| """ |
| Export all data to a JSON file |
| |
| Args: |
| filename (str, optional): Custom filename |
| |
| Returns: |
| str: Path to exported file |
| """ |
| if not filename: |
| timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") |
| filename = f"linkedin_enhancer_export_{timestamp}.json" |
| |
| export_path = os.path.join(self.storage_dir, filename) |
| |
| export_data = { |
| 'session_data': self.session_data, |
| 'persistent_data': self.persistent_data, |
| 'export_timestamp': datetime.now().isoformat() |
| } |
| |
| with open(export_path, 'w', encoding='utf-8') as f: |
| json.dump(export_data, f, indent=2, ensure_ascii=False) |
| |
| return export_path |
|
|
