AI-Skill-Connector / storage.py
BasitAliii's picture
Update storage.py
bb83c5b verified
from dataclasses import dataclass, asdict
from typing import List, Dict, Any, Optional, Tuple
import uuid
import json
from pathlib import Path
from config import DATA_FILE
from utils import normalize_skill_list
# -----------------------------
# Profile Data Class
# -----------------------------
@dataclass
class Profile:
id: str
username: str
offers: List[str]
wants: List[str]
availability: str
preferences: str
avatar: Optional[str] = None
@staticmethod
def from_dict(d: Dict[str, Any]) -> "Profile":
return Profile(
id=str(d.get("id") or uuid.uuid4()),
username=str(d.get("username") or "").strip(),
offers=normalize_skill_list(
",".join(d.get("offers", []))
if isinstance(d.get("offers"), list)
else d.get("offers")
),
wants=normalize_skill_list(
",".join(d.get("wants", []))
if isinstance(d.get("wants"), list)
else d.get("wants")
),
availability=str(d.get("availability") or ""),
preferences=str(d.get("preferences") or ""),
avatar=d.get("avatar"),
)
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
# -----------------------------
# ProfileStore
# -----------------------------
class ProfileStore:
def __init__(self, path: Path = DATA_FILE) -> None:
self.path = path
self._ensure_file()
def _ensure_file(self) -> None:
if not self.path.exists():
self.path.write_text("[]", encoding="utf-8")
def load_all(self) -> List[Profile]:
try:
data = json.loads(self.path.read_text(encoding="utf-8"))
return [Profile.from_dict(d) for d in data if isinstance(d, dict)]
except json.JSONDecodeError:
return []
def save_all(self, profiles: List[Profile]) -> None:
self.path.write_text(
json.dumps([p.to_dict() for p in profiles], indent=2, ensure_ascii=False),
encoding="utf-8",
)
def find_by_username(self, username: str) -> Optional[Profile]:
username = (username or "").strip()
for p in self.load_all():
if p.username.lower() == username.lower():
return p
return None
def add_or_update(self, profile: Profile) -> Tuple[bool, str]:
ok, err = self.validate_profile(profile)
if not ok:
return False, f"Validation failed: {err}"
profiles = self.load_all()
existing = next((p for p in profiles if p.username.lower() == profile.username.lower()), None)
if existing:
existing.offers = profile.offers
existing.wants = profile.wants
existing.availability = profile.availability
existing.preferences = profile.preferences
existing.avatar = profile.avatar
self.save_all(profiles)
return True, "Profile updated."
profile.id = profile.id or str(uuid.uuid4())
profiles.append(profile)
self.save_all(profiles)
return True, "Profile created."
def delete(self, username: str) -> Tuple[bool, str]:
username = (username or "").strip()
profiles = self.load_all()
profiles_new = [p for p in profiles if p.username.lower() != username.lower()]
if len(profiles_new) == len(profiles):
return False, "Profile not found."
self.save_all(profiles_new)
return True, "Profile deleted."
@staticmethod
def validate_profile(profile: Profile) -> Tuple[bool, Optional[str]]:
if not profile.username:
return False, "Username is required."
if len(profile.username) > 60:
return False, "Username must be 60 characters or fewer."
if not profile.offers and not profile.wants:
return False, "At least one offer or want is required."
for s in profile.offers + profile.wants:
if not s or len(s) > 120:
return False, "Invalid skill entry."
return True, None