# app.py """ AI Skill Swap — Unified App (Backend + Module 2 + Module 3 UI) Features: - JSON-backed ProfileStore (create / read / update / delete) - Groq LLM integration with robust error handling - Streamlit UI with animated success messages - Single-file, drop-in replacement for your previous app.py """ from __future__ import annotations import os import json import uuid import re import time from dataclasses import dataclass, asdict from pathlib import Path from typing import List, Dict, Any, Optional, Tuple import streamlit as st # Lazy import Groq so app runs without the package if not installed. try: from groq import Groq except Exception: Groq = None # type: ignore # ---------- Config ---------- DATA_FILE = Path("users.json") MODEL = "llama-3.3-70b-versatile" if not DATA_FILE.exists(): DATA_FILE.write_text("[]", encoding="utf-8") # ---------- Data model ---------- @dataclass class Profile: id: str username: str offers: List[str] wants: List[str] availability: str preferences: str @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=list(d.get("offers") or []), wants=list(d.get("wants") or []), availability=str(d.get("availability") or ""), preferences=str(d.get("preferences") or ""), ) def to_dict(self) -> Dict[str, Any]: return asdict(self) # ---------- Storage & Validation ---------- class ProfileStore: """JSON file-backed profile store.""" 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: # Corrupted file -> return empty list return [] def save_all(self, profiles: List[Profile]) -> None: data = [p.to_dict() for p in profiles] self.path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8") def find_by_username(self, username: str) -> Optional[Profile]: username = (username or "").strip() if not username: return None 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 = 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 self.save_all(profiles) return True, "Profile updated." else: if not profile.id: profile.id = str(uuid.uuid4()) profiles.append(profile) self.save_all(profiles) return True, "Profile created." def delete(self, username: str) -> Tuple[bool, str]: profiles = self.load_all() new = [p for p in profiles if p.username.lower() != username.lower()] if len(new) == len(profiles): return False, "Profile not found." self.save_all(new) return True, "Profile deleted." def validate_profile(profile: Profile) -> Tuple[bool, Optional[str]]: if not profile.username or not profile.username.strip(): return False, "Username is required." if len(profile.username.strip()) > 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 isinstance(s, str) or not s.strip(): return False, "Offers and wants must be non-empty strings." if len(s) > 120: return False, "Individual skill entries must be 120 characters or fewer." return True, None # ---------- Utilities ---------- def normalize_skill_list(text: Optional[str]) -> List[str]: if not text: return [] for sep in ["\n", ",", ";"]: text = text.replace(sep, "|") items = [i.strip().lower() for i in text.split("|") if i.strip()] seen = set() out = [] for it in items: if it not in seen: seen.add(it) out.append(it) return out def make_prompt_for_matching(current_user: Profile, all_users: List[Profile], top_k: int = 5) -> Tuple[str, str]: users_desc = [] for u in all_users: if u.id == current_user.id: continue users_desc.append({ "id": u.id, "username": u.username, "offers": u.offers, "wants": u.wants, "availability": u.availability, "preferences": u.preferences, }) system_instructions = ( "You are a matchmaking assistant for a free skill-exchange platform. " "Users list skills they can teach (offers) and skills they want to learn (wants). " "Recommend the best matches for the current user based on mutual complementarity, " "overlap in availability, and stated preferences. Provide a short explanation for each match " "and a compatibility score (0-100). Return results as a JSON array of objects " "with fields: id, username, score, reason." ) user_message = json.dumps({ "current_user": current_user.to_dict(), "candidates": users_desc, "top_k": top_k }, ensure_ascii=False) return system_instructions, user_message # ---------- Groq LLM helper ---------- def init_groq_client(): api_key = os.environ.get("GROQ_API_KEY") if not api_key: return None if Groq is None: return None try: return Groq(api_key=api_key) except Exception: return None def ask_groq_for_matches(system_instructions: str, user_message: str, model: str = MODEL) -> List[Dict[str, Any]]: client = init_groq_client() if client is None: raise RuntimeError("Groq client is not initialized. Set GROQ_API_KEY and install groq.") messages = [ {"role": "system", "content": system_instructions}, {"role": "user", "content": user_message}, ] try: resp = client.chat.completions.create(messages=messages, model=model) except Exception as e: raise RuntimeError(f"Groq call failed: {e}") content = (resp.choices[0].message.content or "") # Try to extract JSON array from model response json_match = re.search(r"(\[\s*\{[\s\S]*?\}\s*\])", content) if not json_match: raise RuntimeError(f"No JSON array found in LLM response. Raw output:\n{content[:1000]}") try: parsed = json.loads(json_match.group(1)) if not isinstance(parsed, list): raise RuntimeError("Parsed LLM output is not a list.") return parsed except json.JSONDecodeError as e: raise RuntimeError(f"Failed to parse JSON from LLM output: {e}\nRaw:\n{content[:1000]}") # ---------- UI (Streamlit) ---------- st.set_page_config(page_title="AI Skill Swap", page_icon="🤝", layout="wide") st.title("AI Skill Swap — Match & Exchange Skills") # Insert CSS for animations only (dark mode removed) CSS = """ """ st.markdown(CSS, unsafe_allow_html=True) # Layout: sidebar for profiles, main area for matches store = ProfileStore() with st.sidebar: st.header("Your profile") with st.form("profile_form"): username = st.text_input("Username", value=st.session_state.get("username", "")) offers_text = st.text_area("Skills you can teach (one per line or comma-separated)", value=st.session_state.get("offers_text", "")) wants_text = st.text_area("Skills you want to learn (one per line or comma-separated)", value=st.session_state.get("wants_text", "")) availability = st.text_input("Availability (e.g., Weekends)", value=st.session_state.get("availability", "")) preferences = st.text_input("Preferences (e.g., language, online)", value=st.session_state.get("preferences", "")) save = st.form_submit_button("Save / Update profile", use_container_width=True) if save: offers = normalize_skill_list(offers_text) wants = normalize_skill_list(wants_text) profile = Profile(id=str(uuid.uuid4()), username=username.strip(), offers=offers, wants=wants, availability=availability.strip(), preferences=preferences.strip()) ok, msg = store.add_or_update(profile) if ok: st.session_state["username"] = username st.session_state["offers_text"] = offers_text st.session_state["wants_text"] = wants_text st.session_state["availability"] = availability st.session_state["preferences"] = preferences st.success(msg) st.markdown(f"
Saved
{username} saved to local storage.
", unsafe_allow_html=True) else: st.error(msg) st.markdown("---") st.header("Load / Delete") profiles = store.load_all() options = ["-- new profile --"] + [p.username for p in profiles] selected = st.selectbox("Choose profile", options, index=0) if st.button("Load profile") and selected != "-- new profile --": p = store.find_by_username(selected) if p: st.session_state["username"] = p.username st.session_state["offers_text"] = "\n".join(p.offers) st.session_state["wants_text"] = "\n".join(p.wants) st.session_state["availability"] = p.availability st.session_state["preferences"] = p.preferences st.experimental_rerun() else: st.warning("Profile not found.") if st.button("Delete profile") and selected != "-- new profile --": ok, m = store.delete(selected) if ok: st.success(m) # clear session state for form st.session_state["username"] = "" st.session_state["offers_text"] = "" st.session_state["wants_text"] = "" st.session_state["availability"] = "" st.session_state["preferences"] = "" time.sleep(0.2) st.experimental_rerun() else: st.error(m) st.markdown("---") col1, col2 = st.columns([2, 3]) with col1: st.subheader("Community profiles") profiles = store.load_all() if not profiles: st.info("No profiles yet — create your profile in the sidebar.") else: for p in profiles: st.markdown(f"**{p.username}** — offers: {', '.join(p.offers) or '—'}; wants: {', '.join(p.wants) or '—'}") with st.expander("Details"): st.write(p.to_dict()) with col2: st.subheader("Find Matches (AI)") profiles = store.load_all() if not profiles: st.info("Add some profiles to test matchmaking.") else: pick = st.selectbox("Match for profile", [p.username for p in profiles]) top_k = st.slider("Top K matches", 1, 10, 3) if st.button("Run AI matchmaking"): # animated spinner + success card with st.spinner("Generating matches via Groq LLM..."): time.sleep(0.6) current = store.find_by_username(pick) if not current: st.error("Profile not found.") else: try: sys_ins, user_msg = make_prompt_for_matching(current, profiles, top_k=top_k) # ask groq matches = ask_groq_for_matches(sys_ins, user_msg) st.markdown("
Matches Found
Below are the top matches returned by the AI.
", unsafe_allow_html=True) st.json(matches) except Exception as e: st.error(str(e)) st.markdown("---")