from typing import List, Optional from fastapi import APIRouter, Depends, HTTPException, Query from sqlmodel import Session, select, func from pydantic import BaseModel from database import get_session from models import User, Review, Attendance, Group, GroupMember, Show, UserPreferences, Profile from schemas import UserRead, ReviewRead, ShowRead, GroupRead, UserPreferencesUpdate from auth import get_current_user router = APIRouter(prefix="/users", tags=["users"]) class UserProfileUpdate(BaseModel): bio: Optional[str] = None avatar: Optional[str] = None username: Optional[str] = None display_name: Optional[str] = None avatar_bg_color: Optional[str] = None avatar_text: Optional[str] = None # Preset avatar colors - more distinct palette AVATAR_COLORS = [ "#1E3A8A", # Deep Blue "#DC2626", # Bright Red "#059669", # Emerald "#D97706", # Amber/Orange "#7C3AED", # Vibrant Purple "#DB2777", # Magenta Pink "#0F766E", # Teal "#1F2937", # Dark Gray "#B45309", # Burnt Orange "#4338CA", # Indigo "#0891B2", # Cyan "#65A30D", # Lime Green ] class AvatarUpdate(BaseModel): bg_color: Optional[str] = None text: Optional[str] = None # 1-3 alphanumeric chars # Note: Dynamic routes like /{user_id} are placed at the END of this file # to avoid conflicts with static routes like /me and /avatar @router.patch("/me", response_model=UserRead) def update_my_profile( update: UserProfileUpdate, current_user: User = Depends(get_current_user), session: Session = Depends(get_session) ): """Update current user's bio, avatar, and primary profile""" if update.bio is not None: current_user.bio = update.bio if update.avatar is not None: current_user.avatar = update.avatar if update.avatar_bg_color is not None: # Validate it's a valid hex color or preset if update.avatar_bg_color in AVATAR_COLORS or (update.avatar_bg_color.startswith('#') and len(update.avatar_bg_color) == 7): current_user.avatar_bg_color = update.avatar_bg_color if update.avatar_text is not None: # Validate 1-3 alphanumeric characters import re if len(update.avatar_text) <= 3 and re.match(r'^[A-Za-z0-9]*$', update.avatar_text): current_user.avatar_text = update.avatar_text if update.avatar_text else None if update.username or update.display_name: # Find or create primary profile query = select(Profile).where(Profile.user_id == current_user.id) profile = session.exec(query).first() if not profile: if not update.username: raise HTTPException(status_code=400, detail="Username required for new profile") # Check uniqueness (naive check) existing = session.exec(select(Profile).where(Profile.username == update.username)).first() if existing: raise HTTPException(status_code=400, detail="Username taken") profile = Profile( user_id=current_user.id, username=update.username, display_name=update.display_name or update.username ) session.add(profile) else: if update.username: # Check uniqueness if changing if update.username != profile.username: existing = session.exec(select(Profile).where(Profile.username == update.username)).first() if existing: raise HTTPException(status_code=400, detail="Username taken") profile.username = update.username if update.display_name: profile.display_name = update.display_name session.add(profile) session.add(current_user) session.commit() session.refresh(current_user) return current_user @router.patch("/me/avatar") def update_avatar( update: AvatarUpdate, current_user: User = Depends(get_current_user), session: Session = Depends(get_session) ): """Update avatar customization""" import re if update.bg_color is not None: if update.bg_color in AVATAR_COLORS or (update.bg_color.startswith('#') and len(update.bg_color) == 7): current_user.avatar_bg_color = update.bg_color else: raise HTTPException(status_code=400, detail="Invalid color. Use a preset color or valid hex.") if update.text is not None: if len(update.text) > 3: raise HTTPException(status_code=400, detail="Avatar text must be 3 characters or less") if update.text and not re.match(r'^[A-Za-z0-9]*$', update.text): raise HTTPException(status_code=400, detail="Avatar text must be alphanumeric") current_user.avatar_text = update.text.upper() if update.text else None session.add(current_user) session.commit() session.refresh(current_user) return { "avatar_bg_color": current_user.avatar_bg_color, "avatar_text": current_user.avatar_text } @router.get("/avatar/colors") def get_avatar_colors(): """Get available avatar preset colors""" return {"colors": AVATAR_COLORS} @router.patch("/me/preferences", response_model=UserPreferencesUpdate) def update_preferences( prefs: UserPreferencesUpdate, current_user: User = Depends(get_current_user), session: Session = Depends(get_session) ): # Find or create if not current_user.preferences: # Need to create db_prefs = UserPreferences(user_id=current_user.id) current_user.preferences = db_prefs # Link it? # Actually, if relation is set up, adding to session should work. # But safest to create explicitly. db_prefs = UserPreferences( user_id=current_user.id, wiki_mode=prefs.wiki_mode if prefs.wiki_mode is not None else False, show_ratings=prefs.show_ratings if prefs.show_ratings is not None else True, show_comments=prefs.show_comments if prefs.show_comments is not None else True ) session.add(db_prefs) else: db_prefs = current_user.preferences if prefs.wiki_mode is not None: db_prefs.wiki_mode = prefs.wiki_mode if prefs.show_ratings is not None: db_prefs.show_ratings = prefs.show_ratings if prefs.show_comments is not None: db_prefs.show_comments = prefs.show_comments session.add(db_prefs) session.commit() session.refresh(db_prefs) return db_prefs # --- User Stats --- @router.get("/{user_id}/stats") def get_user_stats(user_id: int, session: Session = Depends(get_session)): # Check if user exists user = session.get(User, user_id) if not user: raise HTTPException(status_code=404, detail="User not found") attendance_count = session.exec(select(func.count(Attendance.id)).where(Attendance.user_id == user_id)).one() review_count = session.exec(select(func.count(Review.id)).where(Review.user_id == user_id)).one() group_count = session.exec(select(func.count(GroupMember.id)).where(GroupMember.user_id == user_id)).one() return { "attendance_count": attendance_count, "review_count": review_count, "group_count": group_count } # --- User Data Lists --- @router.get("/{user_id}/attendance", response_model=List[ShowRead]) def get_user_attendance( user_id: int, offset: int = 0, limit: int = Query(default=50, le=100), session: Session = Depends(get_session) ): # Return shows the user attended shows = session.exec( select(Show) .join(Attendance, Show.id == Attendance.show_id) .where(Attendance.user_id == user_id) .order_by(Show.date.desc()) .offset(offset) .limit(limit) ).all() return shows @router.get("/{user_id}/reviews", response_model=List[ReviewRead]) def get_user_reviews( user_id: int, offset: int = 0, limit: int = Query(default=50, le=100), session: Session = Depends(get_session) ): reviews = session.exec( select(Review) .where(Review.user_id == user_id) .order_by(Review.created_at.desc()) .offset(offset) .limit(limit) ).all() return reviews @router.get("/{user_id}/groups", response_model=List[GroupRead]) def get_user_groups( user_id: int, offset: int = 0, limit: int = Query(default=50, le=100), session: Session = Depends(get_session) ): groups = session.exec( select(Group) .join(GroupMember, Group.id == GroupMember.group_id) .where(GroupMember.user_id == user_id) .offset(offset) .limit(limit) ).all() return groups # --- Dynamic ID Routes (must be last to avoid conflicts with /me, /avatar) --- @router.get("/{user_id}", response_model=UserRead) def get_user_public(user_id: int, session: Session = Depends(get_session)): """Get public user profile""" user = session.get(User, user_id) if not user: raise HTTPException(status_code=404, detail="User not found") return user