496 lines
18 KiB
Python
496 lines
18 KiB
Python
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, PublicProfileRead, SocialHandles, HeadlinerBand
|
|
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
|
|
# Social handles
|
|
bluesky_handle: Optional[str] = None
|
|
mastodon_handle: Optional[str] = None
|
|
instagram_handle: Optional[str] = None
|
|
location: Optional[str] = None
|
|
|
|
# Preset avatar colors - Jewel Tones (Primary Set)
|
|
AVATAR_COLORS = [
|
|
"#0F4C81", # Sapphire
|
|
"#9B111E", # Ruby
|
|
"#50C878", # Emerald
|
|
"#9966CC", # Amethyst
|
|
"#0D98BA", # Topaz
|
|
"#E0115F", # Rose Quartz
|
|
"#082567", # Lapis
|
|
"#FF7518", # Carnelian
|
|
"#006B3C", # Jade
|
|
"#1C1C1C", # Onyx
|
|
"#E6E200", # Citrine
|
|
"#702963", # Garnet
|
|
]
|
|
|
|
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, social handles, 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
|
|
|
|
# Social handles (simple sanitization)
|
|
if update.bluesky_handle is not None:
|
|
current_user.bluesky_handle = update.bluesky_handle.strip() or None
|
|
if update.mastodon_handle is not None:
|
|
current_user.mastodon_handle = update.mastodon_handle.strip() or None
|
|
if update.instagram_handle is not None:
|
|
# Remove @ prefix if user includes it
|
|
handle = update.instagram_handle.strip().lstrip('@')
|
|
current_user.instagram_handle = handle or None
|
|
if update.location is not None:
|
|
current_user.location = update.location.strip() or 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}
|
|
|
|
class PrivacyUpdate(BaseModel):
|
|
profile_public: Optional[bool] = None
|
|
show_attendance_public: Optional[bool] = None
|
|
appear_in_leaderboards: Optional[bool] = None
|
|
|
|
@router.patch("/me/privacy")
|
|
def update_privacy(
|
|
update: PrivacyUpdate,
|
|
current_user: User = Depends(get_current_user),
|
|
session: Session = Depends(get_session)
|
|
):
|
|
"""Update privacy settings"""
|
|
if update.profile_public is not None:
|
|
current_user.profile_public = update.profile_public
|
|
if update.show_attendance_public is not None:
|
|
current_user.show_attendance_public = update.show_attendance_public
|
|
if update.appear_in_leaderboards is not None:
|
|
current_user.appear_in_leaderboards = update.appear_in_leaderboards
|
|
|
|
session.add(current_user)
|
|
session.commit()
|
|
session.refresh(current_user)
|
|
|
|
return {
|
|
"profile_public": current_user.profile_public,
|
|
"show_attendance_public": current_user.show_attendance_public,
|
|
"appear_in_leaderboards": current_user.appear_in_leaderboards
|
|
}
|
|
|
|
@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
|
|
|
|
# --- GDPR Data Export ---
|
|
|
|
@router.get("/me/export")
|
|
def export_my_data(
|
|
current_user: User = Depends(get_current_user),
|
|
session: Session = Depends(get_session)
|
|
):
|
|
"""Export all user data for GDPR compliance"""
|
|
from models import Rating, Comment, Notification, Badge
|
|
|
|
# Get profile
|
|
profile = session.exec(select(Profile).where(Profile.user_id == current_user.id)).first()
|
|
|
|
# Get preferences
|
|
prefs = current_user.preferences
|
|
|
|
# Get attendance
|
|
attendance = session.exec(
|
|
select(Attendance, Show)
|
|
.join(Show, Attendance.show_id == Show.id)
|
|
.where(Attendance.user_id == current_user.id)
|
|
).all()
|
|
|
|
# Get reviews
|
|
reviews = session.exec(
|
|
select(Review).where(Review.user_id == current_user.id)
|
|
).all()
|
|
|
|
# Get ratings
|
|
ratings = session.exec(
|
|
select(Rating).where(Rating.user_id == current_user.id)
|
|
).all()
|
|
|
|
# Get groups
|
|
groups = session.exec(
|
|
select(Group)
|
|
.join(GroupMember, Group.id == GroupMember.group_id)
|
|
.where(GroupMember.user_id == current_user.id)
|
|
).all()
|
|
|
|
return {
|
|
"account": {
|
|
"id": current_user.id,
|
|
"email": current_user.email,
|
|
"joined_at": current_user.joined_at.isoformat() if current_user.joined_at else None,
|
|
"xp": current_user.xp,
|
|
"level": current_user.level,
|
|
},
|
|
"profile": {
|
|
"username": profile.username if profile else None,
|
|
"display_name": profile.display_name if profile else None,
|
|
"bio": current_user.bio,
|
|
"avatar_bg_color": current_user.avatar_bg_color,
|
|
"avatar_text": current_user.avatar_text,
|
|
},
|
|
"preferences": {
|
|
"wiki_mode": prefs.wiki_mode if prefs else False,
|
|
"show_ratings": prefs.show_ratings if prefs else True,
|
|
"show_comments": prefs.show_comments if prefs else True,
|
|
"theme": prefs.theme if prefs else "system",
|
|
},
|
|
"privacy": {
|
|
"profile_public": current_user.profile_public,
|
|
"show_attendance_public": current_user.show_attendance_public,
|
|
"appear_in_leaderboards": current_user.appear_in_leaderboards,
|
|
},
|
|
"attendance": [
|
|
{"show_id": a.show_id, "date": s.date.isoformat() if s.date else None}
|
|
for a, s in attendance
|
|
],
|
|
"reviews": [
|
|
{"id": r.id, "score": r.score, "blurb": r.blurb, "content": r.content, "created_at": r.created_at.isoformat() if r.created_at else None}
|
|
for r in reviews
|
|
],
|
|
"ratings": [
|
|
{"id": r.id, "score": r.score, "performance_id": r.performance_id, "show_id": r.show_id}
|
|
for r in ratings
|
|
],
|
|
"groups": [
|
|
{"id": g.id, "name": g.name}
|
|
for g in groups
|
|
],
|
|
}
|
|
|
|
# --- Account Deletion ---
|
|
|
|
class DeleteAccountRequest(BaseModel):
|
|
confirm_email: str
|
|
|
|
@router.delete("/me")
|
|
def delete_my_account(
|
|
request: DeleteAccountRequest,
|
|
current_user: User = Depends(get_current_user),
|
|
session: Session = Depends(get_session)
|
|
):
|
|
"""Permanently delete user account and all associated data"""
|
|
from models import Rating, Comment, Notification, Badge
|
|
|
|
# Verify email confirmation matches
|
|
if request.confirm_email.lower() != current_user.email.lower():
|
|
raise HTTPException(status_code=400, detail="Email confirmation doesn't match")
|
|
|
|
user_id = current_user.id
|
|
|
|
# Delete related data in order (foreign key constraints)
|
|
session.exec(select(Review).where(Review.user_id == user_id)).all()
|
|
for review in session.exec(select(Review).where(Review.user_id == user_id)).all():
|
|
session.delete(review)
|
|
|
|
for rating in session.exec(select(Rating).where(Rating.user_id == user_id)).all():
|
|
session.delete(rating)
|
|
|
|
for attendance in session.exec(select(Attendance).where(Attendance.user_id == user_id)).all():
|
|
session.delete(attendance)
|
|
|
|
for membership in session.exec(select(GroupMember).where(GroupMember.user_id == user_id)).all():
|
|
session.delete(membership)
|
|
|
|
for profile in session.exec(select(Profile).where(Profile.user_id == user_id)).all():
|
|
session.delete(profile)
|
|
|
|
if current_user.preferences:
|
|
session.delete(current_user.preferences)
|
|
|
|
# Finally delete the user
|
|
session.delete(current_user)
|
|
session.commit()
|
|
|
|
return {"message": "Account deleted successfully"}
|
|
|
|
# --- Dynamic ID Routes (must be last to avoid conflicts with /me, /avatar) ---
|
|
|
|
@router.get("/profile/{username}", response_model=PublicProfileRead)
|
|
def get_public_profile(username: str, session: Session = Depends(get_session)):
|
|
"""Get rich public profile for poster view"""
|
|
# 1. Find profile by username
|
|
profile = session.exec(select(Profile).where(Profile.username == username)).first()
|
|
|
|
# Fallback: check if username matches a user email prefix (legacy/fallback)
|
|
# or just 404. Let's start with strict Profile lookup.
|
|
if not profile:
|
|
# Try to find by User ID if it looks like an int? No, username is string.
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
user = session.get(User, profile.user_id)
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
# 2. Get Stats
|
|
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()
|
|
unique_venues = session.exec(select(func.count(func.distinct(Show.venue_id))).join(Attendance).where(Attendance.user_id == user.id)).one()
|
|
|
|
# 3. Get Headliners (Preferences)
|
|
headliners = []
|
|
supporting = []
|
|
|
|
# Sort prefs by priority/tier
|
|
# We need to eager load vertical
|
|
if user.vertical_preferences:
|
|
for pref in user.vertical_preferences:
|
|
band_data = HeadlinerBand(
|
|
name=pref.vertical.name,
|
|
slug=pref.vertical.slug,
|
|
tier=pref.tier,
|
|
logo_url=pref.vertical.logo_url
|
|
)
|
|
if pref.tier == "headliner":
|
|
headliners.append(band_data)
|
|
else:
|
|
supporting.append(band_data)
|
|
|
|
# Social Handles
|
|
socials = SocialHandles(
|
|
bluesky=user.bluesky_handle,
|
|
mastodon=user.mastodon_handle,
|
|
instagram=user.instagram_handle
|
|
)
|
|
|
|
return PublicProfileRead(
|
|
id=user.id,
|
|
username=profile.username,
|
|
display_name=profile.display_name or profile.username,
|
|
bio=user.bio,
|
|
avatar=user.avatar,
|
|
avatar_bg_color=user.avatar_bg_color,
|
|
avatar_text=user.avatar_text,
|
|
location=user.location,
|
|
social_handles=socials,
|
|
headliners=headliners,
|
|
supporting_acts=supporting,
|
|
stats={
|
|
"shows_attended": attendance_count,
|
|
"reviews_written": review_count,
|
|
"venues_visited": unique_venues
|
|
},
|
|
joined_at=user.joined_at
|
|
)
|
|
|
|
|
|
@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
|