71 lines
2.8 KiB
Python
71 lines
2.8 KiB
Python
from app.db.database import users_collection
|
||
from app.core.config import FILES_URL
|
||
from fastapi import HTTPException, UploadFile
|
||
from datetime import datetime
|
||
|
||
class CapeService:
|
||
async def set_cape(self, username: str, cape_file: UploadFile):
|
||
"""Установка или замена плаща через загрузку файла (PNG или GIF)"""
|
||
if not cape_file.content_type.startswith('image/'):
|
||
raise HTTPException(status_code=400, detail="File must be an image")
|
||
|
||
# Определяем расширение
|
||
ext = None
|
||
if cape_file.content_type == "image/png":
|
||
ext = "png"
|
||
elif cape_file.content_type == "image/gif":
|
||
ext = "gif"
|
||
else:
|
||
raise HTTPException(status_code=400, detail="Only PNG and GIF capes are supported")
|
||
|
||
max_size = 2 * 1024 * 1024 # 2MB
|
||
contents = await cape_file.read()
|
||
if len(contents) > max_size:
|
||
raise HTTPException(status_code=400, detail="File too large (max 2MB)")
|
||
|
||
# Удаляем старый плащ, если есть
|
||
user = await users_collection.find_one({"username": username})
|
||
if user and user.get("cloak_url"):
|
||
from urllib.parse import urlparse
|
||
import os
|
||
old_url = user["cloak_url"]
|
||
old_filename = os.path.basename(urlparse(old_url).path)
|
||
old_path = os.path.join("app/static/capes", old_filename)
|
||
if os.path.exists(old_path):
|
||
try:
|
||
os.remove(old_path)
|
||
except Exception:
|
||
pass
|
||
|
||
# Создаем папку для плащей, если ее нет
|
||
from pathlib import Path
|
||
cape_dir = Path("app/static/capes")
|
||
cape_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
cape_filename = f"{username}_{int(datetime.now().timestamp())}.{ext}"
|
||
cape_path = cape_dir / cape_filename
|
||
|
||
with open(cape_path, "wb") as f:
|
||
f.write(contents)
|
||
|
||
result = await users_collection.update_one(
|
||
{"username": username},
|
||
{"$set": {
|
||
"cloak_url": f"{FILES_URL}/capes/{cape_filename}"
|
||
}}
|
||
)
|
||
|
||
if result.modified_count == 0:
|
||
raise HTTPException(status_code=404, detail="User not found")
|
||
return {"status": "success"}
|
||
|
||
async def remove_cape(self, username: str):
|
||
"""Удаление плаща"""
|
||
result = await users_collection.update_one(
|
||
{"username": username},
|
||
{"$unset": {"cloak_url": ""}}
|
||
)
|
||
if result.modified_count == 0:
|
||
raise HTTPException(status_code=404, detail="User not found")
|
||
return {"status": "success"}
|