105 lines
3.2 KiB
Python
105 lines
3.2 KiB
Python
from fastapi import APIRouter, HTTPException, status, UploadFile, File
|
|
from sqlalchemy import select
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
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
|
|
|
|
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}",
|
|
)
|
|
|
|
# Save file
|
|
filename = f"{current_user.id}_{uuid.uuid4().hex}.{ext}"
|
|
filepath = Path(settings.UPLOAD_DIR) / "avatars" / filename
|
|
filepath.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
with open(filepath, "wb") as f:
|
|
f.write(contents)
|
|
|
|
# Update user
|
|
current_user.avatar_path = str(filepath)
|
|
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")
|