File size: 4,300 Bytes
4fd18a2 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 |
"""Profile service for managing user profiles and data."""
import json
import os
from datetime import datetime
from typing import Any, Dict, Optional
from pydantic import BaseModel, Field
from ..config import get_settings
class UserProfile(BaseModel):
"""User profile data model."""
user_id: str
resume: str
skills: list[str]
salary_wish: str
career_goals: str
experience_level: Optional[str] = None
location: Optional[str] = None
education: Optional[str] = None
certifications: Optional[list[str]] = None
created_at: datetime = Field(default_factory=datetime.now)
updated_at: datetime = Field(default_factory=datetime.now)
class ProfileService:
"""Service for managing user profiles."""
def __init__(self):
self.settings = get_settings()
self.profiles_path = self.settings.profiles_db_path
self._ensure_data_dir()
def _ensure_data_dir(self):
"""Ensure data directory exists."""
os.makedirs(os.path.dirname(self.profiles_path), exist_ok=True)
def _load_profiles(self) -> Dict[str, Dict[str, Any]]:
"""Load profiles from file."""
if not os.path.exists(self.profiles_path):
return {}
try:
with open(self.profiles_path, "r", encoding="utf-8") as f:
return json.load(f)
except (json.JSONDecodeError, FileNotFoundError):
return {}
def _save_profiles(self, profiles: Dict[str, Dict[str, Any]]):
"""Save profiles to file."""
with open(self.profiles_path, "w", encoding="utf-8") as f:
json.dump(profiles, f, indent=2, default=str)
def upsert_profile(self, user_id: str, profile_data: str) -> Dict[str, Any]:
"""
Store or update user profile.
Args:
user_id: Unique user identifier
profile_data: JSON string containing profile information
Returns:
Dict with operation result
"""
try:
# Parse profile data
profile_dict = json.loads(profile_data)
# Validate required fields
required_fields = ["resume", "skills", "salary_wish", "career_goals"]
missing_fields = [
field for field in required_fields if field not in profile_dict
]
if missing_fields:
return {
"success": False,
"message": f"Missing required fields: {', '.join(missing_fields)}",
}
# Create profile model
profile_dict["user_id"] = user_id
profile = UserProfile(**profile_dict)
# Load existing profiles
profiles = self._load_profiles()
# Update timestamp if profile exists
if user_id in profiles:
profile.updated_at = datetime.now()
# Store profile
profiles[user_id] = profile.dict()
self._save_profiles(profiles)
return {
"success": True,
"message": "Profile updated successfully",
"user_id": user_id,
"profile": {
"skills_count": len(profile.skills),
"updated_at": profile.updated_at.isoformat(),
},
}
except json.JSONDecodeError:
return {"success": False, "message": "Invalid JSON format in profile data"}
except Exception as e:
return {"success": False, "message": f"Error updating profile: {str(e)}"}
def get_profile(self, user_id: str) -> Optional[UserProfile]:
"""Get user profile by ID."""
profiles = self._load_profiles()
profile_data = profiles.get(user_id)
if profile_data:
return UserProfile(**profile_data)
return None
def delete_profile(self, user_id: str) -> bool:
"""Delete user profile."""
profiles = self._load_profiles()
if user_id in profiles:
del profiles[user_id]
self._save_profiles(profiles)
return True
return False
def list_profiles(self) -> list[str]:
"""List all user IDs."""
profiles = self._load_profiles()
return list(profiles.keys())
|