From 4a031794d49a63e41deacd35194e15b444d368be286d31c7794bf636a9ffe6ca Mon Sep 17 00:00:00 2001 From: Vladislav Date: Tue, 30 Jun 2026 19:13:54 -0400 Subject: [PATCH] =?UTF-8?q?Bot=20commit:=20=D0=94=D0=BE=D0=B1=D0=B0=D0=B2?= =?UTF-8?q?=D0=BB=D0=B5=D0=BD=20=D1=84=D0=B0=D0=B9=D0=BB=20utils.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- utils.py | 414 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 414 insertions(+) create mode 100644 utils.py diff --git a/utils.py b/utils.py new file mode 100644 index 0000000..4c9d8cd --- /dev/null +++ b/utils.py @@ -0,0 +1,414 @@ +from fastapi import HTTPException, Request +from fastapi.security import OAuth2PasswordBearer +from passlib.context import CryptContext +import jwt +from datetime import datetime, timedelta, timezone +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 + +pwd_context = CryptContext(schemes=["argon2"], deprecated="auto") +oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login") + +OP_MAP = {"like_in": "like"} +ENC_KEY = base64.b64decode(ENC_KEY_B64) + + +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Читать дальше" # 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"{user_name}" + 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‑байты готового композитного изображения