109 lines
3.4 KiB
Python
109 lines
3.4 KiB
Python
from fastapi import APIRouter, HTTPException, status, UploadFile, File
|
|
from sqlalchemy import select
|
|
|
|
from app.api.deps import DbSession, CurrentUser
|
|
from app.core.config import settings
|
|
from app.models import User
|
|
from app.schemas import UserPublic, UserUpdate, TelegramLink, MessageResponse
|
|
from app.services.storage import storage_service
|
|
|
|
router = APIRouter(prefix="/users", tags=["users"])
|
|
|
|
|
|
@router.get("/{user_id}", response_model=UserPublic)
|
|
async def get_user(user_id: int, db: DbSession):
|
|
result = await db.execute(select(User).where(User.id == user_id))
|
|
user = result.scalar_one_or_none()
|
|
|
|
if not user:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="User not found",
|
|
)
|
|
|
|
return UserPublic.model_validate(user)
|
|
|
|
|
|
@router.patch("/me", response_model=UserPublic)
|
|
async def update_me(data: UserUpdate, current_user: CurrentUser, db: DbSession):
|
|
if data.nickname is not None:
|
|
current_user.nickname = data.nickname
|
|
|
|
await db.commit()
|
|
await db.refresh(current_user)
|
|
|
|
return UserPublic.model_validate(current_user)
|
|
|
|
|
|
@router.post("/me/avatar", response_model=UserPublic)
|
|
async def upload_avatar(
|
|
current_user: CurrentUser,
|
|
db: DbSession,
|
|
file: UploadFile = File(...),
|
|
):
|
|
# Validate file
|
|
if not file.content_type.startswith("image/"):
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="File must be an image",
|
|
)
|
|
|
|
contents = await file.read()
|
|
if len(contents) > settings.MAX_UPLOAD_SIZE:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"File too large. Maximum size is {settings.MAX_UPLOAD_SIZE // 1024 // 1024} MB",
|
|
)
|
|
|
|
# Get file extension
|
|
ext = file.filename.split(".")[-1].lower() if file.filename else "jpg"
|
|
if ext not in settings.ALLOWED_IMAGE_EXTENSIONS:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=f"Invalid file type. Allowed: {settings.ALLOWED_IMAGE_EXTENSIONS}",
|
|
)
|
|
|
|
# Delete old avatar if exists
|
|
if current_user.avatar_path:
|
|
await storage_service.delete_file(current_user.avatar_path)
|
|
|
|
# Upload file
|
|
filename = storage_service.generate_filename(current_user.id, file.filename)
|
|
file_path = await storage_service.upload_file(
|
|
content=contents,
|
|
folder="avatars",
|
|
filename=filename,
|
|
content_type=file.content_type or "image/jpeg",
|
|
)
|
|
|
|
# Update user
|
|
current_user.avatar_path = file_path
|
|
await db.commit()
|
|
await db.refresh(current_user)
|
|
|
|
return UserPublic.model_validate(current_user)
|
|
|
|
|
|
@router.post("/me/telegram", response_model=MessageResponse)
|
|
async def link_telegram(
|
|
data: TelegramLink,
|
|
current_user: CurrentUser,
|
|
db: DbSession,
|
|
):
|
|
# Check if telegram_id already linked to another user
|
|
result = await db.execute(
|
|
select(User).where(User.telegram_id == data.telegram_id, User.id != current_user.id)
|
|
)
|
|
if result.scalar_one_or_none():
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail="This Telegram account is already linked to another user",
|
|
)
|
|
|
|
current_user.telegram_id = data.telegram_id
|
|
current_user.telegram_username = data.telegram_username
|
|
|
|
await db.commit()
|
|
|
|
return MessageResponse(message="Telegram account linked successfully")
|