"""Роутер текущего пользователя.""" from __future__ import annotations from fastapi import APIRouter, Depends, File, Query, Request, UploadFile from fastapi.responses import FileResponse from sqlmodel import Session from app.auth.deps import get_current_user from app.auth.password import check_current_password from app.auth.telegram import TelegramProvider from app.core.security import client_ip from app.core.errors import NotFoundError from app.db.session import get_session from app.models import User from app.schemas import api as s from app.services import audit_service, group_service, stats_service, user_service router = APIRouter(prefix="/users", tags=["users"]) _AVATAR_MAX_BYTES = 2 * 1024 * 1024 # 2 МБ def build_me(session: Session, user: User) -> s.MeRead: groups = [ s.GroupBrief(id=g.id, name=g.name, role=role) for g, role in group_service.list_user_groups(session, user.id) # type: ignore[arg-type] ] return s.MeRead( id=user.id, # type: ignore[arg-type] nickname=user.nickname, role=user.role, auth_provider=user.auth_provider, telegram_id=user.telegram_id, active_group_id=user.active_group_id, bio=user.bio, avatar_url=user_service.avatar_url_for(user.id, user.avatar_path, user.updated_at), # type: ignore[arg-type] favorite_faction_id=user.favorite_faction_id, history_mode=user.history_mode, history_detail=user.history_detail, groups=groups, has_password=user.password_hash is not None, ) @router.get("/me", response_model=s.MeRead) def get_me( session: Session = Depends(get_session), user: User = Depends(get_current_user), ) -> s.MeRead: return build_me(session, user) @router.patch("/me", response_model=s.UserRead) def update_me( body: s.NicknameUpdate, request: Request, session: Session = Depends(get_session), user: User = Depends(get_current_user), ) -> s.UserRead: user_service.update_nickname(session, user, body.nickname) audit_service.record( session, actor_id=user.id, action="update", entity_type="user", entity_id=user.id, payload={"nickname": user.nickname}, ip=client_ip(request), user_agent=request.headers.get("user-agent"), ) session.commit() return s.UserRead( id=user.id, # type: ignore[arg-type] nickname=user.nickname, role=user.role, auth_provider=user.auth_provider, telegram_id=user.telegram_id, active_group_id=user.active_group_id, bio=user.bio, avatar_url=user_service.avatar_url_for(user.id, user.avatar_path, user.updated_at), # type: ignore[arg-type] favorite_faction_id=user.favorite_faction_id, history_mode=user.history_mode, history_detail=user.history_detail, ) @router.patch("/me/profile", response_model=s.MeRead) def update_my_profile( body: s.ProfileUpdate, request: Request, session: Session = Depends(get_session), user: User = Depends(get_current_user), ) -> s.MeRead: # Обновляем только реально переданные поля: «О себе» и любимая фракция правятся # по отдельности, и частичный запрос не должен обнулять соседнее поле. changed = body.model_dump(exclude_unset=True) if "bio" in changed: user_service.update_bio(session, user, changed["bio"]) if "favorite_faction_id" in changed: user_service.update_favorite_faction(session, user, changed["favorite_faction_id"]) if "history_mode" in changed or "history_detail" in changed: user_service.update_history_prefs( session, user, mode=changed.get("history_mode"), detail=changed.get("history_detail"), ) audit_service.record( session, actor_id=user.id, action="update", entity_type="user", entity_id=user.id, payload={key: True for key in changed}, ip=client_ip(request), ) session.commit() return build_me(session, user) @router.put("/me/password", response_model=s.MeRead) def change_my_password( body: s.PasswordChange, request: Request, session: Session = Depends(get_session), user: User = Depends(get_current_user), ) -> s.MeRead: """Задать пароль (первый раз — без текущего) или сменить его (нужен текущий).""" had_password = user.password_hash is not None if had_password: check_current_password(user, body.current_password) user_service.set_password(session, user, body.new_password) audit_service.record( session, actor_id=user.id, action="update", entity_type="user", entity_id=user.id, payload={"password": "changed" if had_password else "set"}, ip=client_ip(request), user_agent=request.headers.get("user-agent"), ) session.commit() return build_me(session, user) @router.post("/me/telegram", response_model=s.MeRead) def link_my_telegram( body: s.TelegramAuthPayload, request: Request, session: Session = Depends(get_session), user: User = Depends(get_current_user), ) -> s.MeRead: """Привязать Telegram к аккаунту. Подпись виджета проверяется так же, как при входе.""" identity = TelegramProvider().authenticate(body.model_dump()) user_service.link_telegram(session, user, identity) audit_service.record( session, actor_id=user.id, action="update", entity_type="user", entity_id=user.id, payload={"telegram": "linked"}, ip=client_ip(request), user_agent=request.headers.get("user-agent"), ) session.commit() return build_me(session, user) @router.put("/me/avatar", response_model=s.MeRead) def upload_my_avatar( file: UploadFile = File(...), session: Session = Depends(get_session), user: User = Depends(get_current_user), ) -> s.MeRead: content, ext = user_service.read_capped_image( file, _AVATAR_MAX_BYTES, "Файл слишком большой (макс. 2 МБ)." ) user_service.set_avatar(session, user, content, ext) return build_me(session, user) @router.delete("/me/avatar", response_model=s.MeRead) def delete_my_avatar( session: Session = Depends(get_session), user: User = Depends(get_current_user), ) -> s.MeRead: user_service.clear_avatar(session, user) return build_me(session, user) @router.get("/{user_id}/avatar") def get_user_avatar( user_id: int, session: Session = Depends(get_session), _user: User = Depends(get_current_user), ) -> FileResponse: target = session.get(User, user_id) path = user_service.avatar_abs_path(target.avatar_path) if target else None if path is None or not path.exists(): raise NotFoundError("Аватар не найден.") ext = path.suffix.lstrip(".") return FileResponse( path, media_type=user_service.avatar_media_type(ext), headers={"Cache-Control": "public, max-age=3600"}, ) @router.get("/{user_id}/profile", response_model=s.PublicProfile) def get_user_profile( user_id: int, session: Session = Depends(get_session), _user: User = Depends(get_current_user), ) -> dict: return user_service.public_profile(session, user_id) @router.get("/{user_id}/matches", response_model=s.MatchHistory) def user_matches( user_id: int, limit: int = Query(20, ge=1, le=100), offset: int = Query(0, ge=0), session: Session = Depends(get_session), _user: User = Depends(get_current_user), ) -> dict: """История партий игрока. Режим — витрина владельца профиля: её видят и гости.""" owner = user_service.get_user(session, user_id) data = stats_service.user_match_list( session, user_id, limit=limit, offset=offset, best_only=owner.history_mode == "best", ) return {**data, "mode": owner.history_mode, "detail": owner.history_detail} @router.get("/me/stats", response_model=s.ProfileStats) def my_stats( session: Session = Depends(get_session), user: User = Depends(get_current_user), ) -> dict: return stats_service.profile_stats(session, user.id) # type: ignore[arg-type] @router.put("/me/active-group", response_model=s.MeRead) def set_active_group( body: s.ActiveGroupUpdate, session: Session = Depends(get_session), user: User = Depends(get_current_user), ) -> s.MeRead: user_service.set_active_group(session, user, body.group_id) return build_me(session, user) @router.get("/nickname-available", response_model=s.NicknameAvailable) def nickname_available( value: str = Query(..., min_length=1), session: Session = Depends(get_session), user: User = Depends(get_current_user), ) -> s.NicknameAvailable: available = user_service.nickname_available(session, value.strip(), exclude_user_id=user.id) return s.NicknameAvailable(available=available) @router.get("/search", response_model=list[s.UserSuggestion]) def search_users( q: str = Query("", max_length=64), limit: int = Query(3, ge=1, le=10), session: Session = Depends(get_session), _user: User = Depends(get_current_user), ) -> list[s.UserSuggestion]: """Автокомплит по нику. Пустой q — пустой список (не ошибка): удобнее при наборе.""" users = user_service.search_users(session, q, limit=limit) return [ s.UserSuggestion( user_id=u.id, # type: ignore[arg-type] nickname=u.nickname, avatar_url=user_service.avatar_url_for(u.id, u.avatar_path, u.updated_at), # type: ignore[arg-type] ) for u in users ]