Spaces:
Sleeping
Sleeping
| """ | |
| Utilities for personalization and draft logging. | |
| Manages user preferences and stores draft history for future suggestions. | |
| """ | |
| import json | |
| import os | |
| from typing import Dict, Any, List, Optional | |
| from datetime import datetime | |
| from config.settings import PERSONALIZATION_FILE, DRAFT_LOGS_FILE, DATA_DIR | |
| class PersonalizationManager: | |
| """Manages user personalization data.""" | |
| def __init__(self): | |
| self._ensure_data_dir() | |
| self.data = self._load_data() | |
| def _ensure_data_dir(self): | |
| """Ensure the data directory exists.""" | |
| if not os.path.exists(DATA_DIR): | |
| os.makedirs(DATA_DIR) | |
| def _load_data(self) -> Dict[str, Any]: | |
| """Load personalization data from file.""" | |
| if os.path.exists(PERSONALIZATION_FILE): | |
| try: | |
| with open(PERSONALIZATION_FILE, 'r', encoding='utf-8') as f: | |
| return json.load(f) | |
| except (json.JSONDecodeError, IOError): | |
| return self._get_default_data() | |
| return self._get_default_data() | |
| def _get_default_data(self) -> Dict[str, Any]: | |
| """Get default personalization data.""" | |
| return { | |
| "user_profile": { | |
| "name": "", | |
| "company": "", | |
| "title": "", | |
| "email": "" | |
| }, | |
| "style_preferences": { | |
| "default_tone": "formal", | |
| "signature_style": "professional", | |
| "include_company_in_signature": True | |
| }, | |
| "frequent_recipients": [], | |
| "custom_templates": {} | |
| } | |
| def save(self): | |
| """Save personalization data to file.""" | |
| self._ensure_data_dir() | |
| with open(PERSONALIZATION_FILE, 'w', encoding='utf-8') as f: | |
| json.dump(self.data, f, indent=4) | |
| def get_user_profile(self) -> Dict[str, Any]: | |
| """Get user profile information.""" | |
| return self.data.get("user_profile", {}) | |
| def update_user_profile(self, **kwargs): | |
| """Update user profile fields.""" | |
| if "user_profile" not in self.data: | |
| self.data["user_profile"] = {} | |
| self.data["user_profile"].update(kwargs) | |
| self.save() | |
| def get_default_tone(self) -> str: | |
| """Get the default tone preference.""" | |
| return self.data.get("style_preferences", {}).get("default_tone", "formal") | |
| def set_default_tone(self, tone: str): | |
| """Set the default tone preference.""" | |
| if "style_preferences" not in self.data: | |
| self.data["style_preferences"] = {} | |
| self.data["style_preferences"]["default_tone"] = tone | |
| self.save() | |
| def get_signature(self) -> str: | |
| """Generate a signature based on user profile.""" | |
| profile = self.get_user_profile() | |
| style_prefs = self.data.get("style_preferences", {}) | |
| parts = [] | |
| if profile.get("name"): | |
| parts.append(profile["name"]) | |
| if profile.get("title"): | |
| parts.append(profile["title"]) | |
| if style_prefs.get("include_company_in_signature") and profile.get("company"): | |
| parts.append(profile["company"]) | |
| return "\n".join(parts) | |
| def add_frequent_recipient(self, name: str, email: str, company: str = ""): | |
| """Add a frequent recipient.""" | |
| if "frequent_recipients" not in self.data: | |
| self.data["frequent_recipients"] = [] | |
| # Check if already exists | |
| for recipient in self.data["frequent_recipients"]: | |
| if recipient.get("email") == email: | |
| recipient.update({"name": name, "company": company}) | |
| self.save() | |
| return | |
| self.data["frequent_recipients"].append({ | |
| "name": name, | |
| "email": email, | |
| "company": company | |
| }) | |
| self.save() | |
| def get_frequent_recipients(self) -> List[Dict[str, str]]: | |
| """Get list of frequent recipients.""" | |
| return self.data.get("frequent_recipients", []) | |
| class DraftLogger: | |
| """Manages draft logging for learning and personalization.""" | |
| def __init__(self): | |
| self._ensure_data_dir() | |
| self.data = self._load_data() | |
| def _ensure_data_dir(self): | |
| """Ensure the data directory exists.""" | |
| if not os.path.exists(DATA_DIR): | |
| os.makedirs(DATA_DIR) | |
| def _load_data(self) -> Dict[str, Any]: | |
| """Load draft logs from file.""" | |
| if os.path.exists(DRAFT_LOGS_FILE): | |
| try: | |
| with open(DRAFT_LOGS_FILE, 'r', encoding='utf-8') as f: | |
| return json.load(f) | |
| except (json.JSONDecodeError, IOError): | |
| return {"drafts": [], "edit_history": []} | |
| return {"drafts": [], "edit_history": []} | |
| def save(self): | |
| """Save draft logs to file.""" | |
| self._ensure_data_dir() | |
| with open(DRAFT_LOGS_FILE, 'w', encoding='utf-8') as f: | |
| json.dump(self.data, f, indent=4) | |
| def log_draft(self, prompt: str, tone: str, intent: str, draft: Dict[str, Any], | |
| final_email: str, was_edited: bool = False): | |
| """Log a generated draft.""" | |
| if "drafts" not in self.data: | |
| self.data["drafts"] = [] | |
| self.data["drafts"].append({ | |
| "timestamp": datetime.now().isoformat(), | |
| "prompt": prompt, | |
| "tone": tone, | |
| "intent": intent, | |
| "draft": draft, | |
| "final_email": final_email, | |
| "was_edited": was_edited | |
| }) | |
| # Keep only last 100 drafts | |
| if len(self.data["drafts"]) > 100: | |
| self.data["drafts"] = self.data["drafts"][-100:] | |
| self.save() | |
| def log_edit(self, original_email: str, edited_email: str, tone: str): | |
| """Log an edit made by the user.""" | |
| if "edit_history" not in self.data: | |
| self.data["edit_history"] = [] | |
| self.data["edit_history"].append({ | |
| "timestamp": datetime.now().isoformat(), | |
| "original": original_email, | |
| "edited": edited_email, | |
| "tone": tone | |
| }) | |
| # Keep only last 50 edits | |
| if len(self.data["edit_history"]) > 50: | |
| self.data["edit_history"] = self.data["edit_history"][-50:] | |
| self.save() | |
| def get_recent_drafts(self, limit: int = 10) -> List[Dict[str, Any]]: | |
| """Get recent drafts.""" | |
| drafts = self.data.get("drafts", []) | |
| return drafts[-limit:][::-1] # Most recent first | |
| def get_edit_history(self, limit: int = 10) -> List[Dict[str, Any]]: | |
| """Get recent edit history.""" | |
| edits = self.data.get("edit_history", []) | |
| return edits[-limit:][::-1] | |
| def get_tone_statistics(self) -> Dict[str, int]: | |
| """Get statistics on tone usage.""" | |
| stats = {} | |
| for draft in self.data.get("drafts", []): | |
| tone = draft.get("tone", "unknown") | |
| stats[tone] = stats.get(tone, 0) + 1 | |
| return stats | |
| def get_intent_statistics(self) -> Dict[str, int]: | |
| """Get statistics on intent usage.""" | |
| stats = {} | |
| for draft in self.data.get("drafts", []): | |
| intent = draft.get("intent", "unknown") | |
| stats[intent] = stats.get(intent, 0) + 1 | |
| return stats | |