426 lines
15 KiB
Python
426 lines
15 KiB
Python
from fastapi import HTTPException, Request
|
||
from fastapi.security import OAuth2PasswordBearer
|
||
from passlib.context import CryptContext
|
||
import jwt
|
||
from datetime import datetime, timedelta, timezone, date
|
||
import hashlib
|
||
from typing import List, Dict, Optional, Set, Any
|
||
from datetime import timedelta
|
||
from db import redis_client
|
||
from roles import *
|
||
from config import *
|
||
from bson import ObjectId
|
||
from collections.abc import Mapping, Sequence
|
||
from io import BytesIO
|
||
from PIL import Image, ImageDraw, ImageFilter
|
||
import httpx
|
||
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
||
import os
|
||
import base64
|
||
import json
|
||
|
||
|
||
pwd_context = CryptContext(schemes=["argon2"], deprecated="auto")
|
||
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")
|
||
|
||
OP_MAP = {"like_in": "like"}
|
||
ENC_KEY = base64.b64decode(ENC_KEY_B64)
|
||
|
||
|
||
class MongoJSONEncoder(json.JSONEncoder):
|
||
def default(self, o):
|
||
if isinstance(o, ObjectId):
|
||
return str(o)
|
||
if isinstance(o, (datetime, date)):
|
||
return o.isoformat()
|
||
return super().default(o)
|
||
|
||
|
||
def encrypt(plaintext: bytes) -> str:
|
||
aesgcm = AESGCM(ENC_KEY)
|
||
|
||
nonce = os.urandom(12)
|
||
ciphertext = aesgcm.encrypt(nonce, plaintext, None)
|
||
|
||
return base64.b64encode(nonce + ciphertext).decode("utf-8")
|
||
|
||
|
||
def decrypt(token_b64: str) -> bytes:
|
||
aesgcm = AESGCM(ENC_KEY)
|
||
data = base64.b64decode(token_b64.encode("utf-8"))
|
||
|
||
nonce = data[:12]
|
||
ciphertext = data[12:]
|
||
return aesgcm.decrypt(nonce, ciphertext, None)
|
||
|
||
|
||
async def convert_bio_to_webp(bio: BytesIO, quality: int = 95) -> BytesIO:
|
||
bio.seek(0)
|
||
im = Image.open(bio).convert("RGB")
|
||
out = BytesIO()
|
||
im.save(out, format="WEBP", quality=quality, method=6)
|
||
out.seek(0)
|
||
return out
|
||
|
||
|
||
def clause_to_mql(clause: Dict[str, Any]) -> Dict[str, Any]:
|
||
field = clause["field"]
|
||
op = OP_MAP.get(clause.get("op"), clause.get("op"))
|
||
value = clause.get("value")
|
||
if isinstance(value, list):
|
||
return {"or": [{"field": field, "op": op, "value": v} for v in value]}
|
||
return {"field": field, "op": op, "value": value}
|
||
|
||
|
||
def convert_to_mql(filter_obj: Dict[str, Any]) -> Dict[str, Any]:
|
||
if "and" in filter_obj:
|
||
return {"and": [clause_to_mql(c) for c in filter_obj["and"]]}
|
||
if "or" in filter_obj:
|
||
return {"or": [clause_to_mql(c) for c in filter_obj["or"]]}
|
||
return clause_to_mql(filter_obj)
|
||
|
||
|
||
def hash_token(token: str) -> str:
|
||
return hashlib.sha256(token.encode()).hexdigest()
|
||
|
||
|
||
def pil_to_jpeg_bytesio(img: Image.Image, quality=100) -> BytesIO:
|
||
bio = BytesIO()
|
||
img.save(bio, format="JPEG", quality=quality)
|
||
bio.seek(0)
|
||
return bio
|
||
|
||
|
||
async def fetch_bytes(client: httpx.AsyncClient, url: str) -> bytes:
|
||
async with client.stream("GET", url, timeout=60000) as resp:
|
||
resp.raise_for_status()
|
||
chunks = []
|
||
async for chunk in resp.aiter_bytes():
|
||
if chunk:
|
||
chunks.append(chunk)
|
||
return b"".join(chunks)
|
||
|
||
|
||
def safe_strip(value: Optional[str], default: str = "") -> str:
|
||
return value.strip() if value else default
|
||
|
||
|
||
def parse_competition_text(competition):
|
||
parts = [
|
||
'Приглашаем художников!',
|
||
competition['title'],
|
||
competition['short_desc'],
|
||
f"Тема: {competition['topic']}",
|
||
f"Организатор: {competition['user']['username']}",
|
||
f"Подача заявок до: {competition['date_finish'].strftime('%d.%m.%Y')}",
|
||
"",
|
||
f"https://avtopost.kultmost.ru/competition/{competition['id']}" # TODO
|
||
]
|
||
return '\n'.join(parts)
|
||
|
||
|
||
def parse_text_exhibition(exhibition):
|
||
paths = ["Выставка",
|
||
exhibition.get('title'),
|
||
f"Тема: {exhibition.get('topic')}",
|
||
f"Организатор: {exhibition['user']['username']}",
|
||
f"С: {exhibition['datetime_start'].date().strftime('%d.%m.%Y')}",
|
||
f"По: {exhibition['date_finish'].date().strftime('%d.%m.%Y')}",
|
||
# '',
|
||
# f"avtopost.kultmost.ru/exhibition/{exhibition.get('_id')}"
|
||
]
|
||
return '\n'.join(paths)
|
||
|
||
|
||
def parse_text_competition(competition):
|
||
paths = ["Приглашаем художников!",
|
||
competition.get('title'),
|
||
competition.get('short_desc'),
|
||
|
||
f"Тема: {competition.get('topic')}",
|
||
f"Организатор: {competition['user']['username']}",
|
||
f"Подача заявок до: {competition['date_finish'].date().strftime('%d.%m.%Y')}",
|
||
'',
|
||
f"avtopost.kultmost.ru/competition/{competition.get('_id')}"]
|
||
return '\n'.join(paths)
|
||
|
||
|
||
def parse_posts_text(post: dict, social_type: str = None, service: bool = False):
|
||
def get_description(text):
|
||
if text and len(text)>400:
|
||
return ' '.join(text[:400].split(' ')[:-1:]) + f"..." # \n<a href=\"https://kultmost.ru/post/{id}\"><b><u>Читать дальше</u></b></a>" # TODO
|
||
else:
|
||
return text
|
||
description = get_description(post['description'])
|
||
user_name = (
|
||
post['author']
|
||
or post['user']['gallery_name']
|
||
or post['user']['name']
|
||
)
|
||
grade = post['grade'] or post['user']['grade']
|
||
grade = grade if grade != "Без звания" else None
|
||
city = f"Город: {post['city']}" if post['city'] else (
|
||
f"Город: {post['user']['living_place']}"
|
||
if post['user']['living_place']
|
||
else None
|
||
)
|
||
match social_type:
|
||
case 'tg':
|
||
user_name = f"<b>{user_name}</b>"
|
||
case _:
|
||
pass
|
||
parts = [
|
||
user_name,
|
||
city or '',
|
||
grade or '',
|
||
post['title'],
|
||
f"Размер: {post['size']}",
|
||
f"Материал: {post['material']}",
|
||
f"Жанр: {post['genre']}",
|
||
f"Стоимость: {post['price']}",
|
||
description or '',
|
||
"",
|
||
'Культмост - Автопост avtopost.kultmost.ru - новый, оригинальный формат размещения и продвижения произведений искусства. Маркетплейс Культмост kultmost.ru/post', # TODO
|
||
]
|
||
if service:
|
||
match social_type:
|
||
case 'vk':
|
||
parts.append('По вопросам приобретения https://vk.com/magekon') # TODO
|
||
case 'tg':
|
||
parts.append('По вопросам приобретения @gennadiy0605') # TODO
|
||
case _:
|
||
pass
|
||
return '\n'.join(parts)
|
||
|
||
|
||
def insert_logo(
|
||
background,
|
||
logo,
|
||
scale_factor=0.095,
|
||
padding=8,
|
||
radius=15,
|
||
):
|
||
logo_w, logo_h = logo.size
|
||
back_w, back_h = background.size
|
||
new_w = max(1, int(back_w * scale_factor))
|
||
ratio = logo_w / logo_h if logo_h > 0 else 1
|
||
new_h = max(1, int(new_w / ratio))
|
||
logo = logo.resize((new_w, new_h), Image.Resampling.LANCZOS)
|
||
mask = Image.new("L", logo.size, 0)
|
||
draw = ImageDraw.Draw(mask)
|
||
draw.rounded_rectangle(
|
||
[(0, 0), logo.size],
|
||
radius=radius,
|
||
fill=255,
|
||
)
|
||
logo.putalpha(mask)
|
||
positionX = padding
|
||
positionY = back_h - new_h - padding
|
||
background.paste(logo, (positionX, 0 if positionY < 0 else positionY ), logo)
|
||
out = BytesIO()
|
||
background.save(out, format="JPEG")
|
||
out.seek(0)
|
||
return out
|
||
|
||
|
||
def _convert_objids(value):
|
||
if isinstance(value, ObjectId):
|
||
return str(value)
|
||
|
||
if isinstance(value, Mapping):
|
||
for k, v in list(value.items()):
|
||
value[k] = _convert_objids(v)
|
||
return value
|
||
|
||
if isinstance(value, (list, tuple, set)):
|
||
converted = [_convert_objids(v) for v in value]
|
||
return type(value)(converted)
|
||
|
||
return value
|
||
|
||
|
||
def should_sign_avatar(item, social):
|
||
return item.get('sign_avatar', False) or social.get('sign_avatar', True)
|
||
|
||
|
||
def bson_clear(doc):
|
||
return _convert_objids(doc)
|
||
|
||
|
||
def display_user(user):
|
||
return {
|
||
"username": user.get('username'),
|
||
"img": user.get('img'),
|
||
"phone": user.get('phone'),
|
||
"id": user.get('id'),
|
||
"_id": str(user.get('_id')) or user.get('id'),
|
||
"roles": user.get('roles'),
|
||
"email": user.get('email'),
|
||
"living_place": user.get('living_place'),
|
||
"grade": user.get('grade'),
|
||
}
|
||
|
||
|
||
def normalize_phone(phone: Optional[str]) -> Optional[str]:
|
||
if not phone or len(phone) < 6:
|
||
return None
|
||
|
||
cleaned = ''.join(char for char in phone.strip() if char.isdigit() or char == '+')
|
||
|
||
if cleaned.startswith('+7'):
|
||
return cleaned
|
||
|
||
if cleaned.startswith('7'):
|
||
return '+' + cleaned
|
||
|
||
if cleaned.startswith('8') and len(cleaned) == 11:
|
||
return '+7' + cleaned[1:]
|
||
|
||
if len(cleaned) >= 10:
|
||
return '+7' + cleaned
|
||
|
||
|
||
def get_bearer_token(request: Request):
|
||
auth = request.headers.get("Authorization")
|
||
if auth and auth.startswith("Bearer "):
|
||
return auth.split(" ", 1)[1]
|
||
|
||
|
||
async def check_bruteforce(username: str):
|
||
key = f"login_attempts:{hash(username)}"
|
||
attempts = await redis_client.get(key)
|
||
if attempts is None:
|
||
return
|
||
|
||
attempts = int(attempts)
|
||
if attempts >= 8:
|
||
ttl = await redis_client.ttl(key)
|
||
minutes = (ttl + 59) // 60 if ttl > 0 else 15
|
||
raise HTTPException(
|
||
status_code=429,
|
||
detail=f"Too many failed login attempts. Try again in ~{minutes} minutes."
|
||
)
|
||
|
||
|
||
async def increment_failed_attempt(username: str):
|
||
key = f"login_attempts:{hash(username)}"
|
||
pipe = await redis_client.pipeline()
|
||
pipe.incr(key)
|
||
pipe.expire(key, 20 * 60)
|
||
pipe.execute()
|
||
|
||
|
||
async def reset_failed_attempts(username: str):
|
||
key = f"login_attempts:{hash(username)}"
|
||
await redis_client.delete(key)
|
||
|
||
|
||
def verify_password(plain: str, hashed: str) -> bool:
|
||
return pwd_context.verify(plain, hashed)
|
||
|
||
|
||
def get_password_hash(password: str) -> str:
|
||
return pwd_context.hash(password)
|
||
|
||
|
||
def create_jwt_token(data: Dict, expires_delta: timedelta) -> str:
|
||
to_encode = data.copy()
|
||
expire = datetime.now(timezone.utc) + expires_delta
|
||
to_encode.update({"exp": expire})
|
||
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
|
||
|
||
|
||
def get_all_user_scopes(roles: List[str], extra_scopes: List[str]) -> List[str]:
|
||
scopes: Set[str] = set(extra_scopes)
|
||
for role in roles:
|
||
if role in ROLE_SCOPES:
|
||
scopes.update(ROLE_SCOPES[role])
|
||
return list(scopes)
|
||
|
||
def is_safe_redirect(next_url: str) -> bool:
|
||
if not next_url:
|
||
return False
|
||
if next_url.startswith('/') and not next_url.startswith('//'):
|
||
return True
|
||
return False
|
||
|
||
|
||
async def build_composite(data, scale: float = 1.5, new_w: int = 500):
|
||
src_img = Image.open(BytesIO(data)).convert('RGB')
|
||
|
||
# --------------------------------------------------------------
|
||
# 2️⃣ Параметры конечного результата
|
||
# --------------------------------------------------------------
|
||
new_h = int(new_w / scale) # высота для соотношения 3:2 (≈333)
|
||
canvas_sz = (new_w, new_h) # размер «фонового» холста
|
||
|
||
# --------------------------------------------------------------
|
||
# 3️⃣ Фоновое изображение – cover + blur(3)
|
||
# --------------------------------------------------------------
|
||
# - «cover» = заполняем весь холст, обрезая лишнее,
|
||
# сохраняем пропорции оригинала.
|
||
# - Затем делаем размытие радиусом 3.
|
||
# --------------------------------------------------------------
|
||
# Сначала масштабируем так, чтобы одна из сторон полностью покрывала холст
|
||
src_ratio = src_img.width / src_img.height
|
||
canvas_ratio = new_w / new_h
|
||
|
||
if src_ratio > canvas_ratio: # изображение более «широкое»
|
||
# масштабируем по высоте, ширина будет больше холста → обрезаем по горизонтали
|
||
scale_h = new_h
|
||
scale_w = int(scale_h * src_ratio)
|
||
else: # изображение более «высокое» или точное соотношение
|
||
scale_w = new_w
|
||
scale_h = int(scale_w / src_ratio)
|
||
|
||
# масштабируем
|
||
bg_resized = src_img.resize((scale_w, scale_h), Image.LANCZOS)
|
||
|
||
# обрезаем центр, чтобы получить точно new_w × new_h
|
||
left = (scale_w - new_w) // 2
|
||
top = (scale_h - new_h) // 2
|
||
right = left + new_w
|
||
bottom = top + new_h
|
||
bg_cropped = bg_resized.crop((left, top, right, bottom))
|
||
|
||
# размытие
|
||
bg_blurred = bg_cropped.filter(ImageFilter.GaussianBlur(radius=3))
|
||
|
||
# --------------------------------------------------------------
|
||
# 4️⃣ Переднее изображение – contain + центрировать
|
||
# --------------------------------------------------------------
|
||
# - «contain» = вписать изображение в границы canvas,
|
||
# полностью показывая его, оставляя пустые поля (прозрачные)
|
||
# --------------------------------------------------------------
|
||
# Находим масштаб, чтобы изображение полностью поместилось
|
||
if src_ratio > canvas_ratio:
|
||
# ограничиваем ширину
|
||
fore_w = new_w
|
||
fore_h = int(fore_w / src_ratio)
|
||
else:
|
||
# ограничиваем высоту
|
||
fore_h = new_h
|
||
fore_w = int(fore_h * src_ratio)
|
||
|
||
fore_resized = src_img.resize((fore_w, fore_h), Image.LANCZOS)
|
||
|
||
# Создаём прозрачный холст того же размера, что и фон
|
||
final_img = Image.new('RGB', canvas_sz, (0, 0, 0))
|
||
|
||
# Сначала помещаем размытый фон
|
||
final_img.paste(bg_blurred, (0, 0))
|
||
|
||
# Затем — переднее изображение, центрированное
|
||
offset_x = (new_w - fore_w) // 2
|
||
offset_y = (new_h - fore_h) // 2
|
||
final_img.paste(fore_resized, (offset_x, offset_y))
|
||
|
||
# --------------------------------------------------------------
|
||
# 5️⃣ Возврат / сохранение
|
||
# --------------------------------------------------------------
|
||
out_bytes = BytesIO()
|
||
# final_img.save(out_bytes, format='JPEG', quality=85)
|
||
final_img.save(out_bytes, format="WEBP", quality=85, method=6)
|
||
out_bytes.seek(0)
|
||
return out_bytes # JPEG‑байты готового композитного изображения
|