# `main.py` generate from `main.ipynb` by `build.ipynb`
# FastAPI
## App
from fastapi import FastAPI, APIRouter, UploadFile, File, Depends, HTTPException, Body, Security, Form, Request, Response, BackgroundTasks, Query
from motor.motor_asyncio import AsyncIOMotorCollection
import jwt
from datetime import datetime, timedelta, timezone, date
import io
import asyncio
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse, StreamingResponse, RedirectResponse
from starlette.datastructures import UploadFile as StarletteUploadFile
from fastapi.staticfiles import StaticFiles
import mimetypes
from urllib.parse import quote
from starlette.middleware.gzip import GZipMiddleware
import uuid
import bcrypt
from typing import List, Any, Dict, Optional, Union
from datetime import timedelta
from urllib.parse import urlencode
from bson import ObjectId
from PIL import Image
from io import BytesIO
from dateutil.parser import isoparse
import httpx
import string
from db import *
import vk as vk
import tg as tg
from delivery_router import router as delivery_router
from auth import get_user, get_current_user, get_bearer_token, get_access_token, get_token
from roles import *
from utils import *
from tasks import *
from models import *
from config import *
import logging
import json
from templates import templates
import email_task as email_task
from pydantic import EmailStr
import secrets
from pymongo import ReturnDocument
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
### Конфигурация
app = FastAPI(title="KULTMOST")
app.mount("/static", StaticFiles(directory="static"), name="static")
app.add_middleware(GZipMiddleware, minimum_size=500)
app.add_middleware(
CORSMiddleware,
allow_origins=['*'],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(delivery_router)
app.include_router(vk.router, prefix='/api/v1/social', tags=["Auth"])
### Хелперы
RUS_TO_LAT = {
'а':'a','б':'b','в':'v','г':'g','д':'d','е':'e','ё':'yo','ж':'zh','з':'z',
'и':'i','й':'y','к':'k','л':'l','м':'m','н':'n','о':'o','п':'p','р':'r',
'с':'s','т':'t','у':'u','ф':'f','х':'kh','ц':'ts','ч':'ch','ш':'sh','щ':'shch',
'ъ':'"','ы':'y','ь':"'",'э':'e','ю':'yu','я':'ya',
' ':'-',
'А':'A','Б':'B','В':'V','Г':'G','Д':'D','Е':'E','Ё':'Yo','Ж':'Zh','З':'Z',
'И':'I','Й':'Y','К':'K','Л':'L','М':'M','Н':'N','О':'O','П':'P','Р':'R',
'С':'S','Т':'T','У':'U','Ф':'F','Х':'Kh','Ц':'Ts','Ч':'Ch','Ш':'Sh','Щ':'Shch',
'Ъ':'"','Ы':'Y','Ь':"'",'Э':'E','Ю':'Yu','Я':'Ya'
}
def transliterate(text: str) -> str:
return ''.join(RUS_TO_LAT.get(ch, '' if ch not in string.ascii_letters else ch) for ch in text)
async def get_latest_news() -> List[dict]:
cursor = news_collection.find({'publish_on_platform': True}).sort("_id", -1).limit(30)
docs: List[dict] = []
async for doc in cursor:
published = doc.get("published_at")
if published is not None and hasattr(published, "isoformat"):
doc["published_at"] = published.isoformat()
docs.append(doc)
return docs
async def logo_user_task(user: dict):
img_id = user.get('img') if user else None
new_id = None
if img_id:
# download original image bytes
file_data = await gridfs_map['users'].open_download_stream(ObjectId(img_id))
original_image_data = await file_data.read()
# store original in GridFS (keep a copy)
original_img_id = await gridfs_map['users'].upload_from_stream(f'original_{img_id}', BytesIO(original_image_data))
# open image with PIL
img = Image.open(BytesIO(original_image_data))
w, h = img.size
# resize if needed
if w > 500:
factor = 500 / w
new_size = (500, int(h * factor))
img = img.resize(new_size, resample=Image.LANCZOS)
# ensure image mode is compatible with JPEG
if img.mode in ('RGBA', 'LA') or (img.mode == 'P' and 'transparency' in img.info):
# composite onto white background to remove alpha
background = Image.new('RGB', img.size, (255, 255, 255))
alpha = img.split()[-1] if img.mode != 'P' else img.convert('RGBA').split()[-1]
background.paste(img.convert('RGBA'), mask=alpha)
img = background
else:
img = img.convert('RGB')
# save to buffer as JPEG
buf = BytesIO()
# img.save(buf, format='JPEG', quality=85)
img.save(buf, format="WEBP", quality=85, method=6)
buf.seek(0)
# upload processed avatar to GridFS
new_id = await gridfs_map['users'].upload_from_stream('500_avatar.jpg', buf)
# update user doc with new image ids (store as strings to match your original code)
await users_collection.update_one(
{'_id': ObjectId(user.get('_id'))},
{'$set': {'img': str(new_id), 'original_img': str(original_img_id)}}
)
async def poster_competition(competition):
img_id = competition.get('poster_orig_id')
try:
stream = await gridfs_map['competition'].open_download_stream(ObjectId(img_id))
data = await stream.read()
buf = await build_composite(data, 0.75, 800)
new_id = await gridfs_map['competition'].upload_from_stream(f"500_{stream.filename}", buf)
except Exception:
new_id = img_id
print(f'Ошибка обработки id {img_id} в выставке {competition["_id"]}')
if new_id:
await competitions_col.update_one(
{'_id': ObjectId(competition.get('_id'))},
{'$set': {
'poster_file_id': new_id
}}
)
async def poster_exhibition(exhibition):
img_id = exhibition.get('poster_orig_id')
try:
stream = await gridfs_map['exhibition'].open_download_stream(ObjectId(img_id))
data = await stream.read()
buf = await build_composite(data, 0.73, 800)
new_id = await gridfs_map['exhibition'].upload_from_stream(f"500_{stream.filename}", buf)
except Exception:
new_id = img_id
print(f'Ошибка обработки id {img_id} в выставке {exhibition["_id"]}')
if new_id:
await exhibition_collection.update_one(
{'_id': ObjectId(exhibition.get('_id'))},
{'$set': {
'poster_file_id': new_id
}}
)
async def replace_new_medias(post, links_media):
img_ids = post.get('medias_original', [])
if not isinstance(img_ids, list):
img_ids = [img_ids]
new_ids = []
for img_id in img_ids:
if img_id in links_media:
new_ids.append(links_media['img_id'])
continue
try:
stream = await gridfs_map['posts'].open_download_stream(ObjectId(img_id))
data = await stream.read()
buf = await build_composite(data)
new_id = await gridfs_map['posts'].upload_from_stream(f"500_{stream.filename}", buf)
new_ids.append(str(new_id))
except Exception:
new_ids.append(str(img_id))
print(f'Ошибка обработки id {img_id} в посте {post["_id"]}')
if new_ids:
await posts_collection.update_one(
{'_id': ObjectId(post.get('_id'))},
{'$set': {
'medias': new_ids
}}
)
async def replace_post_medias(post):
img_ids = post.get('medias_original', [])
if not isinstance(img_ids, list):
img_ids = [img_ids]
new_ids = []
for img_id in img_ids:
try:
stream = await gridfs_map['posts'].open_download_stream(ObjectId(img_id))
data = await stream.read()
buf = await build_composite(data)
new_id = await gridfs_map['posts'].upload_from_stream(f"500_{stream.filename}", buf)
new_ids.append(str(new_id))
except Exception:
new_ids.append(str(img_id))
print(f'Ошибка обработки id {img_id} в посте {post["_id"]}')
if new_ids:
await posts_collection.update_one(
{'_id': ObjectId(post.get('_id'))},
{'$set': {
'medias': new_ids
}}
)
async def replace_news_medias(news):
img_ids = news.get('orig_media', [])
if not isinstance(img_ids, list):
img_ids = [img_ids]
new_ids = []
for img_id in img_ids:
try:
stream = await gridfs_map['news'].open_download_stream(ObjectId(img_id))
data = await stream.read()
buf = await build_composite(data)
new_id = await gridfs_map['news'].upload_from_stream(f"500_{stream.filename}", buf)
new_ids.append(str(new_id))
except Exception:
new_ids.append(str(img_id))
print(f'Ошибка обработки id {img_id} в посте {news["_id"]}')
if new_ids:
await news_collection.update_one(
{'_id': ObjectId(news.get('_id'))},
{'$set': {
'media': new_ids
}}
)
async def save_to_gridfs(upload: UploadFile, bucket_name: str):
if upload is None:
return None
grid = gridfs_map.get(bucket_name)
if grid is None:
return None
filename = upload.filename or str(uuid.uuid4())
contents = await upload.read()
stream = io.BytesIO(contents)
try:
file_id = await grid.upload_from_stream(filename, stream)
return str(file_id)
finally:
await upload.close()
async def posts_temp(user,request):
data=[]
async for i in posts_collection.find({'user._id': user.get('_id')}).sort('_id',-1).limit(1000):
data.append(i)
return templates.TemplateResponse(
request=request, name="profile/posts.html", context={ "user": user, 'cards': data}
)
@app.get("/cron/day")
async def cron_every_day(token: str = Security(get_token)):
res = await competitions_col.update_many(
{
"date_finish": {"$lt": datetime.now(timezone.utc) + timedelta(hours=3)},
"status": {"$lt": 2},
},
{"$set": {"status": 2}},
)
return {}
# await notify_collection.delete_many({'title':'Доставка картины'})
# await notify_collection.update_many({'title':'Доставка картины'},{'$set':{'type':'delivery'}})
# async for existing in delivery_collection.find({
# 'exhibition_id':
# '6a55f0e90154435ba37a7c25'
# ,
# 'status':
# 'Ожидает получения'
# }):
# print(str(existing['receiver'].get('_id')))
# notify = await notify_in_profile(NotifyItem(
# user_id=str(existing['receiver'].get('_id')),
# type_item='delivery',
# title='Доставка картины',
# link=f'/profile/delivery',
# text=f'{str(existing['sender']['username'])} прислал вам картину — {existing['title']}'
# ))
# print(notify)
# await delivery_collection.update_one({'_id': existing.get('_id')},{'$set':{'notify': notify.upserted_id}})
@app.get("/cron/5")
async def cron_every_five_min(token: str = Security(get_token)):
now = datetime.now(timezone.utc)
limit_dt = now + timedelta(hours=3)
prod_filter = {"prod": PROD}
# 1) If there's an exhibition that is going to be finished soon -> move deliveries to "Возврат художнику"
exhibition_finished = await exhibition_collection.find_one_and_update(
{**prod_filter, "status": 1, "date_finish": {"$lt": limit_dt}},
{"$set": {"status": 2}},
return_document=ReturnDocument.AFTER,
)
if exhibition_finished:
exhibition_id_str = str(exhibition_finished["_id"])
async for doc in delivery_collection.find(
{"exhibition_id": exhibition_id_str, "status": "Получено", **prod_filter},
{"_id": 0},
):
sender = doc.get("sender")
post_id = doc.get("post_id")
exhibition_id = doc.get("exhibition_id")
new_doc = {
**doc,
"delivery_type": "",
"track": "",
"status": "Возврат художнику",
"receiver": sender,
"sender": doc.get("receiver"),
}
await delivery_collection.update_one(
{**prod_filter, "status": "Возврат художнику", "post_id": post_id, "exhibition_id": exhibition_id},
{"$set": new_doc},
upsert=True,
)
return {}
tasks = []
async for exhibition_to_start in exhibition_collection.find(
{**prod_filter}
):
exhibition_id_str = str(exhibition_to_start["_id"])
exhibition_title = exhibition_to_start.get("title")
notified_user_id = None
notified_post_user_id = None
async for doc in delivery_collection.find(
{"exhibition_id": exhibition_id_str, "status": "Получено", **prod_filter},
{"_id": 0},
):
post_id = doc.get("post_id")
exhibition_id = doc.get("exhibition_id")
post = await posts_collection.find_one({"_id": ObjectId(post_id)})
# Use user from post; keep them even if multiple deliveries exist
if post and post.get("user") and not notified_post_user_id:
notified_post_user_id = str(post["user"].get("_id"))
new_doc = {
**doc,
"exhibition": exhibition_to_start,
"status": "На выставке",
"user": post.get("user") if post else None,
"post": post,
**prod_filter
}
tasks.append(delivery_collection.update_one(
{**prod_filter, "status": "На выставке", "post_id": post_id, "exhibition_id": exhibition_id},
{"$set": new_doc},
upsert=True,
))
tasks.append(posts_collection.update_one(
{"_id": ObjectId(post_id)},
{"$set": {"status": "На выставке"}}
))
# NOTE: your original code used `post` from inside the loop (could be last item / even undefined).
# This version notifies once using the first found post user.
if notified_post_user_id:
await notify_in_profile(
NotifyItem(
user_id=notified_post_user_id,
type_item="exhibtion",
title=f"Выставка {exhibition_title}",
link="/profile/exhibition",
text="Ваши картины на выставке",
)
)
if tasks:
asyncio.gather(*tasks)
# 2) Otherwise find an exhibition that should start soon -> move deliveries to "На выставке"
exhibition_to_start = await exhibition_collection.find_one_and_update(
{**prod_filter, "status": 0, "datetime_start": {"$lt": limit_dt}},
{"$set": {"status": 1}},
return_document=ReturnDocument.AFTER,
)
if not exhibition_to_start:
return {}
# If vk/tg arrays are empty -> send social invitation
if exhibition_to_start.get("vk") == [] or exhibition_to_start.get("tg") == []:
await service_social_exhibition(bson_clear(exhibition_to_start))
return {}
### Эндпоинты
@app.post("/api/register", response_model=dict)
async def register(
background_tasks: BackgroundTasks,
username: Optional[str] = Form(None),
phone: Optional[str] = Form(None),
email: Optional[str] = Form(None),
password: str = Form(...),
client_id: str = Form(...),
roles: List[str] = Form(...),
birthday: Optional[date] = Form(None),
grade: Optional[str] = Form(None),
Female_Name_Lastame: Optional[str] = Form(None),
living_place: Optional[str] = Form(None),
offer_accepted: Optional[bool] = Form(False),
img: Optional[UploadFile] = File(None),
request: Request = None,
response: Response = None
):
user = UserRegister(
username=username, phone=phone, email=email, password=password,
client_id=client_id, roles=roles, birthday=birthday, grade=grade,
Female_Name_Lastame=Female_Name_Lastame, living_place=living_place,
offer_accepted=offer_accepted
)
# if normalize_phone(user.phone) and await users_collection.find_one({"phone": normalize_phone(user.phone)}):
# raise HTTPException(400, "Телефон уже зарегестрирован")
if user.email and await users_collection.find_one({"email": user.email}):
raise HTTPException(400, "Email уже зарегестрирован")
hashed_password = get_password_hash(user.password)
allowed_roles = [r for r in user.roles if r not in BLOCK_ROLES]
if any(r in REQUIRES_OFFER for r in user.roles):
if not user.offer_accepted:
raise HTTPException(status_code=400, detail="Оферта должна быть принята")
if not user.offer_accepted_at:
user.offer_accepted_at = datetime.utcnow()
if img:
img_id = await save_to_gridfs(img, 'users')
else:
img_id = '/static/painter.svg'
doc = {
"username": user.username,
"phone": normalize_phone(user.phone) if user.phone else None,
"email": user.email,
"img": img_id,
"hashed_password": hashed_password,
"roles": allowed_roles,
"birthday": user.birthday.isoformat() if user.birthday else None,
"grade": user.grade,
"female_name_lastname": user.female_name_lastname,
"living_place": user.living_place,
"client_id": user.client_id,
"extra_scopes": user.extra_scopes,
"offer_accepted": bool(user.offer_accepted),
"offer_accepted_at": user.offer_accepted_at.isoformat() if user.offer_accepted_at else None,
}
result = await users_collection.insert_one(doc)
user_id_str = str(result.inserted_id)
doc_out = {
"_id": user_id_str,
"id": user_id_str,
**{k: v for k, v in doc.items()}
}
background_tasks.add_task(email_new_user, bson_clear(doc_out))
background_tasks.add_task(logo_user_task, bson_clear(doc_out))
scopes = get_all_user_scopes(allowed_roles, [])
client_id = user.client_id
client_ip = request.headers.get('X-Real-IP') or request.client.host
access_exp_delta = timedelta(days=ACCESS_TOKEN_EXPIRE_MINUTES)
refresh_exp_delta = timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
access_token = create_jwt_token(
{"sub": user_id_str, "scopes": scopes, "client_id": client_id, "ip": client_ip},
access_exp_delta
)
refresh_token = create_jwt_token(
{"sub": user_id_str, "type": "refresh", "client_id": client_id, "ip": client_ip},
refresh_exp_delta
)
await redis_client.setex(f"access:{access_token}", int(access_exp_delta.total_seconds()), user_id_str)
await redis_client.setex(f"refresh:{hash_token(refresh_token)}", int(refresh_exp_delta.total_seconds()), user_id_str)
session_key = f"user_sessions:{user_id_str}"
await redis_client.sadd(session_key, f"access:{access_token}")
await redis_client.sadd(session_key, f"refresh:{hash_token(refresh_token)}")
response.set_cookie(
key="access_token",
value=access_token,
httponly=True,
secure=PROD,
samesite="Strict",
max_age=int(access_exp_delta.total_seconds())
)
response.set_cookie(
key="refresh_token",
value=refresh_token,
httponly=True,
secure=PROD,
samesite="Strict",
max_age=int(refresh_exp_delta.total_seconds())
)
return {
"ok": True,
"access_token": access_token,
"refresh_token": refresh_token,
"user_id": user_id_str,
"token_type": "bearer"
}
@app.post("/api/admin/login")
async def login_admin(
request: Request,
response: Response,
login_data: AdminLoginRequest = Body(...),
current_user: Dict = Security(get_current_user, scopes=["admin"])
):
user = await users_collection.find_one({"_id": ObjectId(login_data.user_id)})
user_id_str = str(user["_id"])
scopes = get_all_user_scopes(user.get("roles", []), user.get("extra_scopes", []))
client_ip = request.headers.get('X-Real-IP') or request.headers.get('X-Real-IP') or request.client.host
access_exp_delta = timedelta(days=ACCESS_TOKEN_EXPIRE_MINUTES)
refresh_exp_delta = timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
access_token = create_jwt_token(
{"sub": user_id_str, "scopes": scopes, "client_id": 'admin', "ip": client_ip},
access_exp_delta
)
refresh_token = create_jwt_token(
{"sub": user_id_str, "type": "refresh", "client_id": 'admin', "ip": client_ip},
refresh_exp_delta
)
await redis_client.setex(f"access:{access_token}", int(access_exp_delta.total_seconds()), user_id_str)
await redis_client.setex(f"refresh:{hash_token(refresh_token)}", int(refresh_exp_delta.total_seconds()), user_id_str)
session_key = f"user_sessions:{user_id_str}"
await redis_client.sadd(session_key, f"access:{access_token}")
await redis_client.sadd(session_key, f"refresh:{hash_token(refresh_token)}")
admin_access_token = request.cookies.get("access_token")
admin_refresh_token = request.cookies.get("refresh_token")
response.set_cookie(
key="admin_access_token",
value=admin_access_token,
httponly=True,
secure=PROD,
samesite="Strict",
max_age=int(access_exp_delta.total_seconds())
)
response.set_cookie(
key="admin_refresh_token",
value=admin_refresh_token,
httponly=True,
secure=PROD,
samesite="Strict",
max_age=int(refresh_exp_delta.total_seconds())
)
response.set_cookie(
key="access_token",
value=access_token,
httponly=True,
secure=PROD,
samesite="Strict",
max_age=int(access_exp_delta.total_seconds())
)
response.set_cookie(
key="refresh_token",
value=refresh_token,
httponly=True,
secure=PROD,
samesite="Strict",
max_age=int(refresh_exp_delta.total_seconds())
)
return {
"access_token": access_token,
"refresh_token": refresh_token,
"user_id": user_id_str,
"token_type": "bearer"
}
@app.post("/api/admin/back")
async def login_admin_back(
request: Request,
response: Response,
):
access_exp_delta = timedelta(days=ACCESS_TOKEN_EXPIRE_MINUTES)
refresh_exp_delta = timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
admin_access_token = request.cookies.get("admin_access_token")
admin_refresh_token = request.cookies.get("refreadmin_refresh_tokensh_token")
response.set_cookie(
key="access_token",
value=admin_access_token,
httponly=True,
secure=PROD,
samesite="Strict",
max_age=int(access_exp_delta.total_seconds())
)
response.set_cookie(
key="refresh_token",
value=admin_refresh_token,
httponly=True,
secure=PROD,
samesite="Strict",
max_age=int(refresh_exp_delta.total_seconds())
)
return {'ok': True}
@app.get("/login/{token}")
async def login_token(token: str, request: Request, response: Response):
token_hash = hash_token(token)
log = await users_token_collection.find_one(
{'token_hash': token_hash}
)
if not log:
return RedirectResponse("/login?" + urlencode({"next": '/profile'}))
next_url = log.get('url', '/')
created = log.get("created_at")
if not created:
try:
created = isoparse(ObjectId(log.get("_id")).generation_time.replace(tzinfo=timezone.utc))
except Exception:
return RedirectResponse("/login?" + urlencode({"next": next_url}))
created = created.replace(tzinfo=timezone.utc)
now = datetime.now(timezone.utc)
if created < now - timedelta(days=TOKEN_TTL_DAYS):
await users_token_collection.delete_one({'token_hash': token_hash})
return RedirectResponse("/login?" + urlencode({"next": log.get('url', next_url)}))
user_id = log.get("user_id")
if not user_id:
return RedirectResponse("/login?" + urlencode({"next": log.get('url', next_url)}))
user = await users_collection.find_one({"_id": ObjectId(user_id)})
if not user:
return RedirectResponse("/login?" + urlencode({"next": log.get('url', next_url)}))
user_id_str = str(user["_id"])
scopes = get_all_user_scopes(user.get("roles", []), user.get("extra_scopes", []))
client_ip = request.headers.get('X-Real-IP') or request.client.host
access_exp_delta = timedelta(days=ACCESS_TOKEN_EXPIRE_MINUTES)
refresh_exp_delta = timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
access_token = create_jwt_token(
{"sub": user_id_str, "scopes": scopes, "client_id": 'token', "ip": client_ip},
access_exp_delta
)
refresh_token = create_jwt_token(
{"sub": user_id_str, "type": "refresh", "client_id": 'token', "ip": client_ip},
refresh_exp_delta
)
await redis_client.setex(f"access:{access_token}", int(access_exp_delta.total_seconds()), user_id_str)
await redis_client.setex(f"refresh:{hash_token(refresh_token)}", int(refresh_exp_delta.total_seconds()), user_id_str)
session_key = f"user_sessions:{user_id_str}"
await redis_client.sadd(session_key, f"access:{access_token}")
await redis_client.sadd(session_key, f"refresh:{hash_token(refresh_token)}")
response.set_cookie(
key="access_token",
value=access_token,
httponly=True,
secure=PROD,
samesite="Strict",
max_age=int(access_exp_delta.total_seconds())
)
response.set_cookie(
key="refresh_token",
value=refresh_token,
httponly=True,
secure=PROD,
samesite="Strict",
max_age=int(refresh_exp_delta.total_seconds())
)
response.set_cookie(
key="access_token",
value=access_token,
httponly=True,
secure=PROD,
domain='.kultmost.ru',
samesite="Strict",
max_age=int(access_exp_delta.total_seconds())
)
response.set_cookie(
key="refresh_token",
value=refresh_token,
httponly=True,
secure=PROD,
domain='.kultmost.ru',
samesite="Strict",
max_age=int(refresh_exp_delta.total_seconds())
)
# Validate and sanitize redirect
if not is_safe_redirect(next_url):
next_url = '/'
return RedirectResponse(next_url,headers=response.headers)
@app.post("/api/login")
async def login(
request: Request,
response: Response,
login_data: LoginRequest = Body(...),
):
identifier = safe_strip(login_data.identifier)
await check_bruteforce(identifier)
or_conditions = [
{"email": identifier},
{"email": identifier.lower()},
]
try:
normalized_phone = normalize_phone(identifier)
if normalized_phone:
or_conditions.append({"phone": normalized_phone})
except:
pass
users = [doc async for doc in users_collection.find({"$or": or_conditions})]
if not users:
await increment_failed_attempt(identifier)
raise HTTPException(401, "Incorrect credentials")
# Track if a valid password was found
valid_user_found = False
for user in users:
stored = user.get("hashed_password")
if isinstance(stored, bytes):
stored = stored.decode()
# Check if the password is hashed with Argon2
if stored.startswith("$argon2"):
ok = verify_password(login_data.password, stored)
if ok:
valid_user_found = True
break # Password is valid, exit the loop
# Check if the password is hashed with bcrypt
elif stored.startswith("$2y$"):
stored = stored.replace("$2y$", "$2b$", 1)
if bcrypt.checkpw(login_data.password.encode(), stored.encode()):
valid_user_found = True
new_hash = get_password_hash(login_data.password)
break # Password is valid, exit the loop
# If no valid user was found, increment attempts and raise error
if not valid_user_found:
await increment_failed_attempt(identifier)
raise HTTPException(401, "Incorrect credentials")
# If we reach this point, authentication succeeded
# Proceed with further logic such as generating tokens or returning user details
# await users_collection.update_one({"_id": user["_id"]}, {"$set": {"hashed_password": new_hash}})
await reset_failed_attempts(identifier)
user_id_str = str(user["_id"])
scopes = get_all_user_scopes(user.get("roles", []), user.get("extra_scopes", []))
client_ip = request.headers.get('X-Real-IP') or request.client.host
access_exp_delta = timedelta(days=ACCESS_TOKEN_EXPIRE_MINUTES)
refresh_exp_delta = timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
access_token = create_jwt_token(
{"sub": user_id_str, "scopes": scopes, "client_id": login_data.client_id, "ip": client_ip},
access_exp_delta
)
refresh_token = create_jwt_token(
{"sub": user_id_str, "type": "refresh", "client_id": login_data.client_id, "ip": client_ip},
refresh_exp_delta
)
await redis_client.setex(f"access:{access_token}", int(access_exp_delta.total_seconds()), user_id_str)
await redis_client.setex(f"refresh:{hash_token(refresh_token)}", int(refresh_exp_delta.total_seconds()), user_id_str)
session_key = f"user_sessions:{user_id_str}"
await redis_client.sadd(session_key, f"access:{access_token}")
await redis_client.sadd(session_key, f"refresh:{hash_token(refresh_token)}")
response.set_cookie(
key="access_token",
value=access_token,
httponly=True,
secure=PROD,
samesite="Strict",
max_age=int(access_exp_delta.total_seconds())
)
response.set_cookie(
key="refresh_token",
value=refresh_token,
httponly=True,
secure=PROD,
samesite="Strict",
max_age=int(refresh_exp_delta.total_seconds())
)
return {
"access_token": access_token,
"refresh_token": refresh_token,
"user_id": user_id_str,
"token_type": "bearer"
}
@app.post("/api/old/login")
async def old_login(
request: Request,
response: Response,
):
token_old = request.cookies.get('token') or request.cookies.get('v2token')
access = await get_access_token(request, response)
if token_old and not access:
try:
headers = {'Authorization': f"Bearer {token_old}"}
async with httpx.AsyncClient(http2=True, verify=ctx) as client:
res = await client.get('https://avtopost.kultmost.ru/api/v1/auth/me', headers=headers)
email = res.json()['data']['email']
print(email)
user = await users_collection.find_one({'email': email})
except:
return {'ok': False}
user_id_str = str(user["_id"])
scopes = get_all_user_scopes(user.get("roles", []), user.get("extra_scopes", []))
client_ip = request.headers.get('X-Real-IP') or request.client.host
access_exp_delta = timedelta(days=ACCESS_TOKEN_EXPIRE_MINUTES)
refresh_exp_delta = timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
access_token = create_jwt_token(
{"sub": user_id_str, "scopes": scopes, "client_id": 'old_token', "ip": client_ip},
access_exp_delta
)
refresh_token = create_jwt_token(
{"sub": user_id_str, "type": "refresh", "client_id": 'old_token', "ip": client_ip},
refresh_exp_delta
)
await redis_client.setex(f"access:{access_token}", int(access_exp_delta.total_seconds()), user_id_str)
await redis_client.setex(f"refresh:{hash_token(refresh_token)}", int(refresh_exp_delta.total_seconds()), user_id_str)
session_key = f"user_sessions:{user_id_str}"
await redis_client.sadd(session_key, f"access:{access_token}")
await redis_client.sadd(session_key, f"refresh:{hash_token(refresh_token)}")
response.set_cookie(
key="access_token",
value=access_token,
httponly=True,
secure=PROD,
samesite="Strict",
max_age=int(access_exp_delta.total_seconds())
)
response.set_cookie(
key="refresh_token",
value=refresh_token,
httponly=True,
secure=PROD,
samesite="Strict",
max_age=int(refresh_exp_delta.total_seconds())
)
return {'ok': True}
@app.post("/refresh", response_model=TokenResponse)
async def refresh_token(
request: Request,
response: Response,
req: RefreshRequest
):
refresh_token = (
request.cookies.get("refresh_token") or
get_bearer_token(request) or
(req.refresh_token if req else None)
)
if not refresh_token:
raise HTTPException(401, "Refresh token required")
client_id = None
if req:
client_id = req.client_id
refresh_hash = hash_token(refresh_token)
user_id_str = await redis_client.get(f"refresh:{refresh_hash}")
if not user_id_str:
raise HTTPException(status_code=401, detail="Invalid or expired refresh token")
try:
payload = jwt.decode(refresh_token, SECRET_KEY, algorithms=[ALGORITHM])
if payload.get("type") != "refresh":
raise HTTPException(status_code=401, detail="Not a refresh token")
if payload.get("sub") != user_id_str:
raise HTTPException(status_code=401, detail="Token mismatch")
except jwt.InvalidTokenError as e:
raise HTTPException(status_code=401, detail=f"Invalid refresh token: {str(e)}")
await redis_client.delete(f"refresh:{refresh_hash}")
user = await users_collection.find_one({"_id": ObjectId(user_id_str)})
scopes = get_all_user_scopes(user.get("roles", []), user.get("extra_scopes", []))
client_ip = request.headers.get('X-Real-IP') or request.client.host
access_exp = timedelta(days=ACCESS_TOKEN_EXPIRE_MINUTES)
refresh_exp = timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
new_access = create_jwt_token(
{"sub": user_id_str, "scopes": scopes, "client_id": client_id, "ip": client_ip},
access_exp
)
new_refresh = create_jwt_token(
{"sub": user_id_str, "type": "refresh", "client_id": client_id, "ip": client_ip},
refresh_exp
)
new_refresh_hash = hash_token(new_refresh)
await redis_client.setex(f"access:{new_access}", int(access_exp.total_seconds()), user_id_str)
await redis_client.setex(f"refresh:{new_refresh_hash}", int(refresh_exp.total_seconds()), user_id_str)
session_key = f"user_sessions:{user_id_str}"
await redis_client.sadd(session_key, f"access:{new_access}")
await redis_client.sadd(session_key, f"refresh:{new_refresh_hash}")
response.set_cookie(
key="access_token",
value=new_access,
httponly=True,
secure=PROD,
samesite="strict",
max_age=int(access_exp.total_seconds())
)
response.set_cookie(
key="refresh_token",
value=new_refresh,
httponly=True,
secure=PROD,
samesite="strict",
max_age=int(refresh_exp.total_seconds())
)
return {
"access_token": new_access,
"token_type": "bearer",
"refresh_token": new_refresh
}
@app.post("/logout")
async def logout(
request: Request,
response: Response,
current_user: Dict = Depends(get_current_user)
):
user_id_str = current_user["id"]
session_key = f"user_sessions:{user_id_str}"
tokens = await redis_client.smembers(session_key)
for t in tokens:
await redis_client.delete(t)
await redis_client.delete(session_key)
for cookie_name in request.cookies.keys():
if cookie_name in ['zoom']:
continue
for dom in [None,'.kultmost.ru','www.kultmost.ru','avtopost.kultmost.ru','avtopost.kultmost.ru','kultmost.ru']:
response.delete_cookie(
key=cookie_name,
domain=dom,
path="/",
secure=False,
httponly=False,
)
return {"message": "Logged out from all sessions"}
@app.get("/api/me", response_model=dict)
async def get_me(response: Response, current_user: Dict = Security(get_current_user, scopes=["read:self"])):
user = display_user(current_user)
return bson_clear(user)
@app.get("/healthz")
async def healthz():
try:
await mongo_client.admin.command('ping')
except Exception as e:
logging.error(f"MongoDB is unhealthy: {str(e)}")
raise HTTPException(status_code=500, detail=f"MongoDB is unhealthy: {str(e)}")
try:
await redis_client.ping()
except Exception as e:
logging.error(f"Redis is unhealthy: {str(e)}")
raise HTTPException(status_code=500, detail=f"Redis is unhealthy: {str(e)}")
return {}
@app.get("/registration", response_class=HTMLResponse)
async def index(
request: Request,
next: str | None = None,
):
return templates.TemplateResponse(
request=request, name="registration.html", context={'next': next}
)
async def get_unique_users(competition_request_col: AsyncIOMotorCollection, competition_id: str, finish: bool = False) -> List[Dict[str, Any]]:
# Determine match criteria based on finish parameter
status_match = {"$match": {"posts.status": "Проходит"}} if finish else {"$match": {}}
pipeline = [
{"$match": {"competition_id": competition_id}},
{"$unwind": "$user"},
{"$project": {
"user": 1,
"posts": 1,
"status": 1
}},
{"$unwind": "$posts"},
status_match, # Use the conditional match here
{"$group": {
"_id": "$user._id",
"user": {"$first": "$user"},
"unique_posts": {"$addToSet": "$posts._id"},
"status": {"$first": "$status"}
}},
{"$project": {
"user": 1,
"post_count": {"$size": "$unique_posts"},
"status": { "$cond": {"if": {"$ne": ["$status", None]}, "then": "$status", "else": False } }
}}
]
cursor = competition_request_col.aggregate(pipeline)
results = []
async for doc in cursor:
results.append(doc)
return results
from typing import Any, Dict, List
from bson import ObjectId
from motor.motor_asyncio import AsyncIOMotorCollection
async def get_unique_posts_by_user(
competition_request_col: AsyncIOMotorCollection,
competition_id: str,
user_id: str,
) -> list[dict[str, Any]]:
pipeline = [
{"$match": {"competition_id": competition_id, "user._id": user_id}},
{"$unwind": "$posts"},
{
"$addFields": {
"status_priority": {
"$switch": {
"branches": [
{"case": {"$eq": ["$posts.status", "Проходит"]}, "then": 1},
{"case": {"$eq": ["$posts.status", "На рассмотрении"]}, "then": 2},
],
"default": 3,
}
}
}
},
{"$sort": {"status_priority": 1, "posts._id": 1}}, # tie-breaker if needed
{"$group": {"_id": "$posts._id", "post": {"$first": "$posts"}}},
{"$replaceRoot": {"newRoot": "$post"}},
]
cursor = competition_request_col.aggregate(pipeline)
return await cursor.to_list(length=None)
async def get_unique_posts_с(competition_request_col: AsyncIOMotorCollection, competition_id: str, finish: bool = False) -> List[Any]:
match_criteria = {
"competition_id": competition_id,
}
if finish:
match_criteria['posts.status'] = 'Проходит'
pipeline = [
{"$match": match_criteria},
{"$unwind": "$posts"},
{"$match": {"posts.status": "Проходит"} if finish else {}},
{
"$group": {
"_id": "$posts._id", # Group by unique post ID
"post": {"$first": "$posts"} # Retrieve the first occurrence
}
},
{"$project": {"_id": 0, "post": 1}} # Include the post details
]
cursor = competition_request_col.aggregate(pipeline)
unique_posts = await cursor.to_list(length=None) # Fetch all unique posts
return [unique_post["post"] for unique_post in unique_posts] if unique_posts else []
@app.get("/exhibition/{exhibition_id}", response_class=HTMLResponse)
async def exhibition_members(
request: Request,
exhibition_id: str,
user: Dict = Security(get_user)
):
exhibition = await exhibition_collection.find_one({'_id': ObjectId(exhibition_id)})
if exhibition:
posts = await delivery_collection.find({'status': 'На выставке', "exhibition_id": exhibition_id}).to_list()
return templates.TemplateResponse(
request=request, name="exhibition/index.html", context={"id": 1, "user": user, "exhibition": exhibition, "posts": posts}
)
else:
raise HTTPException(status_code=307, headers={"Location": '/competitions'}, detail="Конкурс не найден")
@app.get("/competition/{competition_id}", response_class=HTMLResponse)
async def competition_members(
request: Request,
competition_id: str,
user: Dict = Security(get_user)
):
competition = await competitions_col.find_one({'_id': ObjectId(competition_id)})
if competition:
members = await get_unique_users(competition_request_col, competition_id, competition['status']==2)
posts = await get_unique_posts_с(competition_request_col, competition_id, competition['status']==2)
return templates.TemplateResponse(
request=request, name="competition/index.html", context={"id": 1, "user": user, "competition": competition, "members": members, "posts": posts}
)
else:
raise HTTPException(status_code=307, headers={"Location": '/competitions'}, detail="Конкурс не найден")
@app.get("/competition/{competition_id}/{member_id}", response_class=HTMLResponse)
async def competition_member(
request: Request,
competition_id: str,
member_id: str,
user: Dict = Security(get_user)
):
competition = await competitions_col.find_one({'_id': ObjectId(competition_id)})
if competition:
member = await users_collection.find_one({'_id':ObjectId(member_id)})
req = await competition_request_col.find_one({'competition_id': competition_id,'user._id': member_id})
member['status'] = req.get('status', False)
if member:
posts = await get_unique_posts_by_user(competition_request_col, competition_id, member_id)
return templates.TemplateResponse(
request=request, name="competition/member/index.html", context={"id": 1, "user": user, "competition": competition, 'member':member, 'posts': posts}
)
else:
raise HTTPException(status_code=307, headers={"Location": f'/competition/{competition._id}'}, detail="Участник не найден")
else:
raise HTTPException(status_code=307, headers={"Location": '/competitions'}, detail="Конкурс не найден")
async def _serialize(doc):
doc = doc.copy()
if "_id" in doc:
doc["_id"] = str(doc["_id"])
if "user" in doc:
doc["user"] = {'img': doc["user"].get('img'), 'username': doc["user"].get('username')}
# if "_id" in doc["user"]:
# doc["user"]["_id"] = str(doc["user"]["_id"])
# doc["user"]["notify"] = []
return doc
@app.get("/api/competitions", response_model=List)
async def get_competition():
competitions = []
current_time = datetime.now(timezone.utc) + timedelta(hours=3)
async for doc in competitions_col.find({
'datetime_start': {'$lt': current_time}
},{'fb':0,'inst':0,'max':0,'tg':0,'vk':0}).sort('_id',-1).limit(32):
competitions.append(bson_clear(doc))
return competitions
@app.get("/exhibitions", response_class=HTMLResponse)
async def exhibitions_all(
request: Request,
user: Dict = Security(get_user)
):
current_time = datetime.now(timezone.utc) + timedelta(hours=3)
find_list = [
{'datetime_start': {'$lt': current_time}},
{'user._id': user.get('id','') if user else ''}
]
if user and 'admin' in user.get('roles',[]):
find_list.append({'datetime_start': {'$gte': current_time}})
exhibitions = await exhibition_collection.find({
'$or': find_list,
'prod': PROD
}).sort({'_id':-1}).limit(1000).to_list()
return templates.TemplateResponse(
request=request, name="exhibitions.html", context={"id": 1, "user": user, "exhibitions": exhibitions}
)
@app.get("/competitions", response_class=HTMLResponse)
async def competition(
request: Request,
user: Dict = Security(get_user)
):
competitions = []
current_time = datetime.now(timezone.utc) + timedelta(hours=3)
find_list = [
{'datetime_start': {'$lt': current_time}},
{'user._id': user.get('id','') if user else ''}
]
if user and 'admin' in user.get('roles',[]):
find_list.append({'datetime_start': {'$gte': current_time}})
async for doc in competitions_col.find({
'$or': find_list
} if PROD else {}).sort({'_id':-1}).limit(1000):
competitions.append(doc)
return templates.TemplateResponse(
request=request, name="competitions.html", context={"id": 1, "user": user, "competitions": competitions}
)
router = APIRouter()
@router.post("/api/restore")
async def restore_api(
request: Request,
restore_data: RestoreRequest = Body(...),
):
# 1. Валидация и очистка входных данных
identifier = restore_data.identifier.strip()
if not identifier or not is_valid_email(identifier):
raise HTTPException(status_code=400, detail="Некорректный формат email")
# 2. Проверка анти-спама / ограничения частоты запросов
client_ip = request.client.host
rate_limit_key = f"rate_limit:restore:{identifier}:{client_ip}"
# Проверяем, запрашивал ли пользователь сброс недавно (например, за последние 60 секунд)
existing_request = await redis_client.get(rate_limit_key)
if existing_request:
# Опционально: получаем TTL, чтобы сообщить пользователю, когда можно повторить попытку
ttl = await redis_client.ttl(rate_limit_key)
raise HTTPException(
status_code=429,
detail=f"Слишком много запросов. Попробуйте через {ttl} секунд."
)
# Устанавливаем флаг ограничения частоты (истекает через 60 секунд)
await redis_client.set(rate_limit_key, "1", ex=60)
# 3. Поиск пользователя
user = await users_collection.find_one({'$or': [
{'email': identifier},
{'email': identifier.lower()}
]})
if not user:
return {'ok': True}
# 4. Генерация токена
token = secrets.token_urlsafe(32)
expires_at = datetime.now(timezone.utc) + timedelta(hours=1)
token_data = {
"user_id": str(user["_id"]),
"email": user["email"],
"next": restore_data.next,
"expires_at": expires_at.isoformat()
}
# 5. Сохранение токена сброса в Redis
reset_token_key = f"reset_token:{token}"
await redis_client.set(reset_token_key, json.dumps(token_data), ex=3600)
# 6. Генерация ссылки для сброса
base_url = str(request.base_url).rstrip('/')
reset_link = f"{base_url}/restore/{token}"
# 7. Асинхронная отправка email
try:
email_task.notify.delay(
to_email=user['email'],
data={
'link': reset_link,
'email': user['email']
},
subject='Восстановление пароля',
template='restore.html'
)
except Exception as e:
# Логирование ошибки
print(f"Не удалось отправить письмо: {e}")
# В зависимости от политики, вы можете захотеть удалить токен, если письмо не отправлено
await redis_client.delete(reset_token_key)
raise HTTPException(status_code=500, detail="Ошибка отправки письма")
return {'ok': True}
@router.get("/restore/{token}", response_class=HTMLResponse)
async def restore_page(request: Request, token: str):
token_key = f"reset_token:{token}"
token_data = await redis_client.get(token_key)
if not token_data:
return HTMLResponse(
"Ссылка недействительна или истекла
",
status_code=400
)
return templates.TemplateResponse(
request=request, name="restore_token.html", context={"token": token}
)
@router.post("/api/reset-password")
async def reset_password(
data: ResetPasswordRequest = Body(...),
):
token_key = f"reset_token:{data.token}"
token_data_str = await redis_client.get(token_key)
if not token_data_str:
raise HTTPException(status_code=400, detail="Токен недействителен или истёк")
token_data = json.loads(token_data_str)
hashed_password = get_password_hash(data.new_password)
await users_collection.update_one(
{"_id": ObjectId(token_data["user_id"])},
{"$set": {"hashed_password": hashed_password}}
)
await redis_client.delete(token_key)
return {"ok": True, "message": "Пароль успешно изменён", "next": token_data.get('next', '/profile')}
@router.get("/restore", response_class=HTMLResponse)
async def restore_init(
request: Request,
next: str | None = None,
):
return templates.TemplateResponse(
request=request, name="restore.html", context={'next': next}
)
app.include_router(router, tags=['Восстановление пароля'])
@app.get("/login", response_class=HTMLResponse)
async def login_(
request: Request,
next_url: str | None = None,
):
if not next_url:
next_url = request.headers.get("Referer")
context={"next": next_url}
print(next_url)
if next_url and 'profile/competition' in next_url:
context.update({'notify':{'text':'Участвовать в конкурсе могут только пользователи сервиса! Пожалуйста, войдите или зарегистрируйтесь.', 'status':'info','duration':8000}})
return templates.TemplateResponse(
request=request,
name="login.html",
context=context,
)
@app.post("/api/switch-profile")
async def switch_profile(request: Request,
response: Response,user: Dict = Security(get_current_user)):
await users_collection.update_one({'_id':ObjectId(user.get('_id'))},{'$set':{'profile':'' if user.get('profile',False) else 'partner'}})
return {'ok': True}
@app.post("/api/post/description")
async def post_edit_description(
data: PostEditDescription = Body(...),
current_user: Dict = Security(get_current_user)
):
await posts_collection.update_one(
{'_id':ObjectId(data.post_id), 'user._id': current_user.get('_id')},
{'$set': {'description': data.description}}
)
return {'ok': True}
@app.delete("/api/partner/exhibition/delete")
async def partner_exhibition_delete(
data: PartnerExhibitionDelete = Body(...),
current_user: Dict = Security(get_current_user, scopes=["create:super_items"])
):
exhibition = await exhibition_collection.find_one_and_delete(
{'_id':ObjectId(data.exhibition_id), 'user._id': current_user.get('_id')},
)
for i in exhibition.get('tg',[]):
tg.delete_from_tg.delay(i.get('result',[{}])[0].get('message_id'), i.get('doc'))
for i in exhibition.get('vk',[]):
vk.delete_wall.delay(i.get('response',{}).get('post_id'), i.get('doc'))
return {'ok': True}
@app.delete("/api/partner/competition/delete")
async def partner_competition_delete(
data: Partnercompetitionstop = Body(...),
current_user: Dict = Security(get_current_user, scopes=["create:super_items"])
):
competition = await competitions_col.delete_one(
{'_id':ObjectId(data.competition_id), 'user._id': current_user.get('_id')},
)
for i in competition.get('tg',[]):
tg.delete_from_tg.delay(i.get('result',[{}])[0].get('message_id'), i.get('doc'))
for i in competition.get('vk',[]):
vk.delete_wall.delay(i.get('response',{}).get('post_id'), i.get('doc'))
return {'ok': True}
@app.post("/api/partner/competition/finish")
async def partner_competition_finish(
data: Partnercompetitionstop = Body(...),
current_user: Dict = Security(get_current_user, scopes=["create:super_items"])
):
await competitions_col.update_one(
{'_id':ObjectId(data.competition_id), 'user._id': current_user.get('_id')},
{'$set': {'status': 2}},
)
return {'ok': True}
@app.post("/api/partner/competition/stop")
async def partner_competition_stop(
data: Partnercompetitionstop = Body(...),
current_user: Dict = Security(get_current_user, scopes=["create:super_items"])
):
await competitions_col.update_one(
{'_id':ObjectId(data.competition_id), 'user._id': current_user.get('_id')},
{'$set': {'status': 1}},
)
return {'ok': True}
@app.post("/api/partner/competition/select")
async def partner_competition_request_artist(
background_tasks: BackgroundTasks,
data: CompetitionArtistRequest = Body(...),
current_user: Dict = Security(get_current_user, scopes=["create:super_items"])
):
print(data)
competition = await competitions_col.find_one({'_id':ObjectId(data.competition_id), 'user._id': current_user.get('_id')})
if competition:
await competition_request_col.update_many(
{'user._id': data.user_id, 'competition_id': data.competition_id},
{'$set': {'status': True,"posts.$[].status": 'Проходит' }},
)
background_tasks.add_task(accept_competition_req, competition, data.user_id)
social_vk = await social_collection.find_one({'user_id':current_user.get('_id'), 'group': {"$exists": False}, 'type':'vk'})
user = display_user(current_user)
user['social_vk'] = social_vk
await invitations_collection.update_one({
"user_id": data.user_id,
"item": competition.get('title'),
"title": "Вы прошли отбор",
"type_": competition.get("type_"),
"user": user,
"prod": PROD,
"type": "Конкурс"
},{'$set':{
"user_id": data.user_id,
"item": competition.get('title'),
"title": "Вы прошли отбор",
"type_": competition.get("type_"),
"user": user,
"prod": PROD,
"type": "Конкурс"
}},upsert=True)
return {'ok': True}
else:
return {'ok': False}
from fastapi import HTTPException, BackgroundTasks, Body, Security
from bson import ObjectId
from typing import Dict
@app.post(
"/api/partner/competition/request",
response_model=dict,
summary="Обновление статуса заявки участника в конкурсе (для партнёра)",
description="""
Позволяет владельцу конкурса обновить статус конкретного поста в заявке.
- Проверяет права владельца конкурса.
- Обновляет статус поста в массиве `posts`.
- Если после обновления все посты в заявке имеют статус "Проходит" или "На рассмотрении" —
заявка помечается `notify: True` и запускается фоновая обработка.
""",
status_code=200,
tags=["Partner", "Competition", "Requests"],
)
async def partner_competition_request(
background_tasks: BackgroundTasks,
data: PartnercompetitionRequest = Body(...),
current_user: Dict = Security(get_current_user, scopes=["create:super_items"])
):
"""
Обработчик обновления статуса поста в заявке на конкурс.
Flow:
1. Проверка владения конкурсом.
2. Массовое обновление статуса поста в массиве заявок.
3. Проверка полностью одобренных заявок → установка `notify` + фоновая задача.
"""
# Проверка прав владельца конкурса
competition = await competitions_col.find_one({
'_id': ObjectId(data.competition_id),
'user._id': current_user.get('_id')
})
if not competition:
raise HTTPException(
status_code=403,
detail="Конкурс не найден или у вас нет прав на его редактирование"
)
# Обновляем статус конкретного поста
await competition_request_col.update_many(
{
'posts._id': ObjectId(data.post_id),
'competition_id': data.competition_id
},
{'$set': {'posts.$[p].status': data.status}},
array_filters=[{'p._id': ObjectId(data.post_id)}]
)
posts_list = []
user = None
async for req in competition_request_col.find({
'posts._id': ObjectId(data.post_id),
'competition_id': data.competition_id,
}):
if not user:
user = req.get('user')
posts_list.extend(req.get('posts', []))
unique_posts = {str(p.get('_id')): p for p in posts_list}.values()
background_tasks.add_task(update_competition_req, competition, user, unique_posts)
return {'ok': True}
async def _update_post_description(post):
if post.get("description"):
await posts_collection.update_one(
{"_id": ObjectId(post["postId"])},
{"$set": {"description": post["description"]}}
)
async def background_request_competition(posts: List[dict]):
coros = [_update_post_description(p) for p in posts]
if coros:
await asyncio.gather(*coros)
@app.post("/api/send_competition_request/{competition_id}")
async def send_competition_request(
request: Request,
competition_id: str,
background_tasks: BackgroundTasks,
user: Dict = Security(get_current_user),
):
try:
data = await request.json()
except Exception:
raise HTTPException(status_code=400, detail="Invalid JSON")
posts_input = data.get("posts") or []
if not posts_input:
raise HTTPException(status_code=400, detail="Поле 'posts' обязательно")
# Build mapping: postId -> nomination (if provided)
nomination_map: dict[str, str | None] = {}
post_ids: list[str] = []
for p in posts_input:
post_id = p.get("postId")
if post_id:
post_ids.append(post_id)
nomination_map[post_id] = p.get("nomination") # can be str or None
unique_post_ids = list(dict.fromkeys(post_ids))
if not unique_post_ids:
raise HTTPException(status_code=400, detail="Нет валидных postId")
# Validate ObjectIds
try:
obj_ids = [ObjectId(pid) for pid in unique_post_ids]
except Exception:
raise HTTPException(status_code=400, detail="Некорректный формат postId")
# Check already used posts
used_post_ids: set[str] = set()
async for req in competition_request_col.find({
"user._id": user["_id"],
"competition_id": competition_id
}):
for post in req.get("posts", []):
if post_id := post.get("_id"):
used_post_ids.add(str(post_id))
new_obj_ids = [oid for oid in obj_ids if str(oid) not in used_post_ids]
excluded_count = len(obj_ids) - len(new_obj_ids)
if not new_obj_ids:
return {"ok": True, "message": "Все посты уже были отправлены ранее"}
# Fetch full posts
cursor = posts_collection.find({"_id": {"$in": new_obj_ids}})
fetched_posts = [doc async for doc in cursor]
# Enrich posts with nomination from input
for post in fetched_posts:
post_id_str = str(post.get("_id"))
if nomination := nomination_map.get(post_id_str):
post["nomination"] = nomination
doc = {
"user": display_user(user),
"competition_id": competition_id,
"posts": fetched_posts,
"created_at": datetime.utcnow(),
}
result = await competition_request_col.insert_one(doc)
# Background tasks
background_tasks.add_task(background_request_competition, posts_input) # pass original input
background_tasks.add_task(new_competiton_req, bson_clear(doc))
response = {
"ok": True,
"inserted_id": str(result.inserted_id)
}
if excluded_count > 0:
response["excluded"] = excluded_count
response["message"] = f"Исключено {excluded_count} уже использованных постов"
return response
@app.get("/profile/news", response_class=HTMLResponse)
async def profile_news(
request: Request,
user: Dict = Security(get_current_user)
):
news = [doc async for doc in news_collection.find({'user._id': user.get('_id')}).sort('_id',-1).limit(1000)]
return templates.TemplateResponse(
request=request, name="profile/news.html", context={ "user": user, 'news': news}
)
@app.get("/profile/form", response_class=HTMLResponse)
async def profile_form(
request: Request,
user: Dict = Security(get_current_user)
):
return templates.TemplateResponse(
request=request, name="profile/form.html", context={ "user": user}
)
async def competitions(user,request):
data=[]
async for i in competitions_col.find({'user._id': user.get('_id')}).sort('_id',-1).limit(1000):
data.append(i)
return templates.TemplateResponse(
request=request, name="profile/partner/competitions.html", context={ "user": user, 'cards': data}
)
@app.get("/profile/partner/competitions", response_class=HTMLResponse)
async def profile_partner_competitions(
request: Request,
user: Dict = Security(get_current_user, scopes=["create:super_items"])
):
await users_collection.update_one({'_id':ObjectId(user.get('_id'))},{'$set':{'profile':'partner'}})
user['profile'] = 'partner'
return await competitions(user,request)
async def get_unique_posts(competition_request_col: AsyncIOMotorCollection, competition_id: str) -> List[Dict[str, Any]]:
pipeline = [
{"$match": {"competition_id": competition_id}},
{"$project": {"posts": 1, "_id": 0}},
{"$unwind": "$posts"},
{"$group": {
"_id": "$posts._id", # group by post _id for uniqueness
"post": {"$first": "$posts"}
}},
{"$addFields": {
"status_order": {
"$switch": {
"branches": [
{"case": {"$eq": ["$post.status", 'Проходит']}, "then": 2},
{"case": {"$eq": ["$post.status", 'На рассмотрении']}, "then": 3}
],
"default": 1
}
}
}},
{"$sort": {"status_order": 1}},
{"$project": {"post": 1, "_id": 0}}
]
cursor = competition_request_col.aggregate(pipeline)
results = []
async for doc in cursor:
results.append(doc['post']) # Adjust based on the added $project stage
return results
async def add_competition_id_to_posts(posts, competition_id):
oid = ObjectId(competition_id)
for post in posts:
post['competition'] = oid
return posts
async def get_unique_users_p(competition_request_col: AsyncIOMotorCollection, competition_id: str) -> List[Dict[str, Any]]:
pipeline = [
{"$match": {"competition_id": competition_id}},
{"$unwind": "$user"},
{"$project": {
"user": 1,
"posts": 1,
"status": 1
}},
{"$unwind": "$posts"},
{"$group": {
"_id": "$user._id",
"user": {"$first": "$user"},
"unique_posts": {"$addToSet": "$posts._id"},
"status": {"$first": "$status"}
}},
{"$project": {
"user": 1,
"post_count": {"$size": "$unique_posts"},
"status": { "$cond": { "if": {"$ne": ["$status", None]}, "then": "$status", "else": False}}
}},
{"$sort": {"status": 1, "_id": 1}}
]
cursor = competition_request_col.aggregate(pipeline)
results = []
async for doc in cursor:
results.append(doc)
return results
@app.get("/profile/partner/competition/{competition_id}", response_class=HTMLResponse)
async def profile_partner_competition(
request: Request,
competition_id: str,
user: Dict = Security(get_current_user)
):
competition = await competitions_col.find_one({'_id': ObjectId(competition_id), 'user._id': user.get('_id')})
if not competition:
raise HTTPException(status_code=403, detail="Not enough permissions")
members = await get_unique_users_p(competition_request_col, competition_id)
posts = await get_unique_posts(competition_request_col, competition_id)
posts = await add_competition_id_to_posts(posts, competition_id)
return templates.TemplateResponse(
request=request, name="profile/partner/competition.html", context={ "user": user, "competition": competition, 'posts': posts, "members": members}
)
@app.get("/profile/partner/competition/{competition_id}/{member_id}", response_class=HTMLResponse)
async def profile_partner_competition_member(
request: Request,
competition_id: str,
member_id: str,
user: Dict = Security(get_current_user)
):
competition = await competitions_col.find_one({'_id': ObjectId(competition_id), 'user._id': user.get('_id')})
if not competition:
raise HTTPException(status_code=403, detail="Not enough permissions")
social_vk = await social_collection.find_one({'user_id':member_id,'type':'vk','group':{"$exists": False}})
member = await users_collection.find_one({'_id':ObjectId(member_id)})
req = await competition_request_col.find_one({'competition_id': competition_id,'user._id': member_id})
member['status'] = req.get('status', False)
posts = await get_unique_posts_by_user(competition_request_col, competition_id, member_id)
posts = await add_competition_id_to_posts(posts, competition_id)
return templates.TemplateResponse(
request=request, name="profile/partner/competition_member.html", context={ "user": user, "social_vk":social_vk, "competition": competition, 'posts': posts, "member": member}
)
@app.get("/profile/competition", response_class=HTMLResponse)
async def profile_competitions(
request: Request,
user: Dict = Security(get_current_user)
):
competitions = []
current_time = datetime.now(timezone.utc) + timedelta(hours=3)
async for i in competitions_col.find({
'$or': [
{'datetime_start': {'$lt': current_time}},
{'user._id': user.get('_id')}
]
} if PROD else {}).sort('_id', -1).limit(1000):
competitions.append(i)
return templates.TemplateResponse(
request=request, name="profile/competition.html", context={ "user": user, 'competitions': competitions}
)
@app.get("/profile/competition/{competition_id}", response_class=HTMLResponse)
async def profile_competition(
request: Request,
competition_id: str,
user: Dict = Security(get_current_user)
):
# required_keys = [
# "biography",
# "messengers",
# "style",
# "material_manufacturer",
# "genres",
# "education", "material_manufacturer",
# "art_types", "material_use",
# "unions", "master_classes",
# "style", "workshop_tour",
# "specialization",
# ]
# if all(user.get(key) for key in required_keys):
competition = await competitions_col.find_one({'_id':ObjectId(competition_id)})
posts=[]
async for i in posts_collection.find({'user._id': user.get('_id')}).sort('_id',-1).limit(1000):
posts.append(i)
return templates.TemplateResponse(
request=request, name="profile/competition/index.html", context={
"user": user, 'competition': competition, 'posts': posts,
'notify':{'text':'Выберите картины на конкурс','status':'info','duration':3000}
}
)
# else:
# return templates.TemplateResponse(
# request=request, name="profile/form.html", context={ "user": user, 'notify':{'text':'В начале заполните анкету под звездочками','status':'error','duration':8000}}
# )
@app.get("/profile", response_class=HTMLResponse)
async def profile(
request: Request,
response: Response,
user: Dict = Security(get_current_user)
):
if user.get('profile',False):
return templates.TemplateResponse(
request=request, name="profile/index.html", context={ "user": user, 'notify': {'text': 'Кабинет партнера', 'status': 'info', 'duration': 1500}}
)
else:
return templates.TemplateResponse(
request=request, name="profile/index.html", context={ "user": user}
)
@app.get("/profile/partner", response_class=HTMLResponse)
async def profile(
request: Request,
response: Response,
user: Dict = Security(get_current_user)
):
if user.get('profile',False):
return await competitions(user,request)
else:
return await posts_temp(user,request)
@app.get("/profile/posts", response_class=HTMLResponse)
async def profile_posts(
request: Request,
user: Dict = Security(get_current_user)
):
return await posts_temp(user,request)
@app.get("/profile/password", response_class=HTMLResponse)
async def profile_password(
request: Request,
user: Dict = Security(get_current_user)
):
return templates.TemplateResponse(
request=request, name="profile/password.html", context={"user": user}
)
@app.get("/profile/exhibition", response_class=HTMLResponse)
async def profile_exhibition(
request: Request,
user: Dict = Security(get_current_user)
):
# data = exhibition_collection.
data = await exhibition_collection.find({}).sort('_id', -1).to_list()
return templates.TemplateResponse(
request=request, name="profile/exhibition.html", context={"user": user, "data": data, "now": datetime.now()}
)
@app.get("/")
async def index(request: Request, user: Dict = Security(get_user)):
news = await get_latest_news()
return templates.TemplateResponse(
request=request, name="index.html", context={"user": user, "news": news}
)
@app.get("/about")
async def about(request: Request, user: Dict = Security(get_user)):
news = await get_latest_news()
return templates.TemplateResponse(
request=request, name="about.html", context={"user": user}
)
@app.get("/news")
async def news(request: Request, user: Dict = Security(get_user)):
news = [doc async for doc in news_collection.find({}).limit(1000)]
return templates.TemplateResponse(
request=request, name="news.html", context={"user": user, "news": news}
)
@app.get("/pwa")
async def pwa(request: Request, user: Dict = Security(get_user)):
news = await get_latest_news()
return templates.TemplateResponse(
request=request, name="index.html", context={"user": user, "news": news, 'pwa':True}
)
@app.get("/pwa2")
async def pwa(request: Request, user: Dict = Security(get_user)):
news = await get_latest_news()
return templates.TemplateResponse(
request=request, name="index.html", context={"user": user, "news": news, 'pwa':True}
)
@app.get("/offline.html")
async def offline_html(request: Request, user: Dict = Security(get_user)):
news = await get_latest_news()
return templates.TemplateResponse(
request=request, name="index.html", context={"user": user, "news": news, 'pwa':True}
)
@app.get("/partner/galerist", response_class=HTMLResponse)
async def partner_galerist(
request: Request,
user: Dict = Security(get_user)
):
partner = [doc async for doc in users_collection.find({'roles': 'partner'},{'username': 1, 'description': 1})]
return templates.TemplateResponse(
request=request, name="partner/galerist.html", context={"id": 1, "user": user, "partner": partner}
)
@app.get("/partner", response_class=HTMLResponse)
async def partner(
request: Request,
user: Dict = Security(get_user)
):
return templates.TemplateResponse(
request=request, name="partner/index.html", context={"id": 1, "user": user}
)
@app.get("/artists", response_class=HTMLResponse)
async def artists(
request: Request,
user: Dict = Security(get_current_user, scopes=["users:read"])
):
users = [doc async for doc in users_collection.find({'roles':'user'}).limit(1000)]
return templates.TemplateResponse(
request=request, name="users/index.html", context={"id": 1, "user":user, "users": users}
)
@app.get("/my-paints", response_class=HTMLResponse)
async def My_paints(
request: Request,
user: Dict = Security(get_current_user, scopes=["read:self"])
):
user_id = user.get('_id')
if isinstance(user_id, int) or (is_str:=len(user_id) < 5):
doc = await users_collection.find_one({"id": int(user_id) if is_str else user_id}, {"_id": 1})
if not doc:
raise HTTPException(status_code=404, detail="Пользователь не найден")
return RedirectResponse(url=f"/artists/{doc.get('_id')}", status_code=301)
doc = users_collection.find_one({"_id": user_id})
if not doc:
raise HTTPException(status_code=404, detail="Пользователь не найден")
total_key = f"total_w_c_artists:{user_id}"
today_key = f"today_w_c_artists:{user_id}"
total_watch_count = await redis_client.incr(total_key)
today_date = datetime.now().strftime("%Y-%m-%d")
daily_key = f"{today_key}:{today_date}"
today_watch_count = await redis_client.incr(daily_key)
find = await users_collection.find_one({'_id':ObjectId(user_id)})
cards = [i async for i in posts_collection.find({'user._id': user_id}).sort('_id',-1).limit(1000)]
return templates.TemplateResponse(
request=request, name="users/user.html", context={"id": 1, "user":user, "find": find, "cards": cards,"total_watch_count": total_watch_count,"today_watch_count": today_watch_count,}
)
@app.get("/artists/{user_id}", response_class=HTMLResponse)
async def artists_id(
request: Request,
user_id: Union[int, str],
user: Dict = Security(get_current_user, scopes=["users:read"])
):
if isinstance(user_id, int) or (is_str:=len(user_id) < 5):
doc = await users_collection.find_one({"id": int(user_id) if is_str else user_id}, {"_id": 1})
if not doc:
raise HTTPException(status_code=404, detail="Пользователь не найден")
return RedirectResponse(url=f"/artists/{doc.get('_id')}", status_code=301)
doc = users_collection.find_one({"_id": user_id})
if not doc:
raise HTTPException(status_code=404, detail="Пользователь не найден")
total_key = f"total_w_c_artists:{user_id}"
today_key = f"today_w_c_artists:{user_id}"
total_watch_count = await redis_client.incr(total_key)
today_date = datetime.now().strftime("%Y-%m-%d")
daily_key = f"{today_key}:{today_date}"
today_watch_count = await redis_client.incr(daily_key)
find = await users_collection.find_one({'_id':ObjectId(user_id)})
cards = [i async for i in posts_collection.find({'user._id': user_id}).sort('_id',-1).limit(1000)]
return templates.TemplateResponse(
request=request, name="users/user.html", context={"id": 1, "user":user, "find": find, "cards": cards,"total_watch_count": total_watch_count,"today_watch_count": today_watch_count,}
)
@app.get("/form/competition", response_class=HTMLResponse)
async def form_(
request: Request,
user: Dict = Security(get_current_user, scopes=["create:super_items"])
):
return templates.TemplateResponse(
request=request, name="form/competition.html", context={"id": 1, "user": user}
)
@app.get("/form/competition/{competition_id}", response_class=HTMLResponse)
async def form_(
request: Request,
competition_id: str,
user: Dict = Security(get_current_user, scopes=["create:super_items"])
):
competition = await competitions_col.find_one({'_id': ObjectId(competition_id)})
data = json.dumps(competition, cls=MongoJSONEncoder, ensure_ascii=False)
return templates.TemplateResponse(
request=request, name="form/competition_edit.html", context={"competition": str(data), "competition_id": competition_id, "user": user}
)
@app.get('/news/{news_id}')
async def article_page(
news_id: Union[int, str],
requset: Request,
user: Dict = Security(get_user)
):
if isinstance(news_id, int) or (is_str:=len(news_id) < 5):
doc = await news_collection.find_one({"id": int(news_id) if is_str else news_id}, {"_id": 1})
if not doc:
raise HTTPException(status_code=404, detail="новость не найдена")
return RedirectResponse(url=f"/news/{doc.get('_id')}", status_code=301)
news = await news_collection.find_one({'_id':ObjectId(news_id)})
if not news:
raise HTTPException(status_code=404, detail="Новость не найдена")
return templates.TemplateResponse(
request=requset, name='news/index.html', context={"user":user, "news": news}
)
@app.get("/form/post", response_class=HTMLResponse)
async def form_post(
request: Request,
user: Dict = Security(get_current_user)
):
partner = [doc async for doc in users_collection.find({'roles': 'partner'},{'username': 1})]
service_social = [doc async for doc in social_collection.find({'service': True})]
user_social = [doc async for doc in social_collection.find({'active': True,'service': {'$exists': False},'user_id': str(user.get('_id'))})]
return templates.TemplateResponse(
request=request, name="posts/form.html", context={"id": 1, "user": user, 'user_social': user_social, 'partner': partner, 'service_social': service_social}
)
@app.get("/form/post/{post_id}", response_class=HTMLResponse)
async def form_post_edit(
request: Request,
post_id:str,
user: Dict = Security(get_current_user)
):
post = await posts_collection.find_one({'_id':ObjectId(post_id)})
data = json.dumps(post, cls=MongoJSONEncoder, ensure_ascii=False)
return templates.TemplateResponse(
request=request, name="form/post_edit.html", context={"post": data, "user": user}
)
@app.get("/form/news", response_class=HTMLResponse)
async def form_news(
request: Request,
user: Dict = Security(get_current_user)
):
return templates.TemplateResponse(
request=request, name="form/news.html", context={"id": 1, "user": user}
)
@app.get("/form/news/{news_id}", response_class=HTMLResponse)
async def form_news_edit(
request: Request,
news_id:str,
user: Dict = Security(get_current_user)
):
news = await news_collection.find_one({'_id':ObjectId(news_id)})
return templates.TemplateResponse(
request=request, name="form/news_edit.html", context={"news": news, "user": user}
)
@app.get("/form/exhibition", response_class=HTMLResponse)
async def form_exhibition(
request: Request,
user: Dict = Security(get_current_user)
):
return templates.TemplateResponse(
request=request, name="form/exhibition.html", context={"id": 1, "user": user}
)
@app.get("/form/exhibition/{exhibition_id}", response_class=HTMLResponse)
async def form_exhibition(
request: Request,
exhibition_id: str,
user: Dict = Security(get_current_user)
):
exhibition = await exhibition_collection.find_one({'_id': ObjectId(exhibition_id)},{'posts':0,'vk':0,'tg':0,'fb':0,'inst':0,'max':0})
data = json.dumps(exhibition, cls=MongoJSONEncoder, ensure_ascii=False)
return templates.TemplateResponse(
request=request, name="form/exhibition_edit.html", context={"id": 1, "user": user, "exhibition": str(data)}
)
@app.post("/api/avatar", response_model=dict)
async def avatar_change(
background_tasks: BackgroundTasks,
avatar: Optional[UploadFile] = File(...),
current_user: Dict = Security(get_current_user)
):
try:
avatar_id = await save_to_gridfs(avatar, 'users')
await users_collection.update_one({'_id': ObjectId(current_user.get('_id'))}, {'$set': {'img': avatar_id}})
current_user.update({'img': avatar_id})
background_tasks.add_task(logo_user_task, bson_clear(current_user))
return {'ok': True}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.put("/api/user/form", response_model=dict)
async def user_form(
request: Request,
current_user: Dict = Security(get_current_user)
):
try:
doc = await request.json()
if not isinstance(doc, dict):
raise HTTPException(status_code=400, detail="Invalid JSON body")
# Remove any protected fields from incoming update
update_doc = {k: v for k, v in doc.items() if k not in PROTECTED_FIELDS_USER}
if not update_doc:
return {"ok": True, "updated": False, "reason": "no allowed fields"}
result = users_collection.update_one(
{"_id": ObjectId(current_user["id"])},
{"$set": update_doc}
)
return {"ok": True, "updated": True}
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get('/paintings/{paint_id}')
async def paintings(
request: Request,
paint_id: Union[int, str],
user: Dict = Security(get_user)
):
if isinstance(paint_id, int) or (is_str:=len(paint_id) < 5):
doc = await posts_collection.find_one({"id": int(paint_id) if is_str else paint_id}, {"_id": 1})
if not doc:
raise HTTPException(status_code=404, detail="Картина не найдена")
return RedirectResponse(url=f"/paintings/{doc.get('_id')}", status_code=301)
paint = await posts_collection.find_one({'_id': ObjectId(paint_id)})
if not paint:
raise HTTPException(status_code=404, detail="Картина не найдена")
total_key = f"total_watch_count_paint:{paint_id}"
today_key = f"today_watch_count_paint:{paint_id}"
total_watch_count = await redis_client.incr(total_key)
today_date = datetime.now().strftime("%Y-%m-%d")
daily_key = f"{today_key}:{today_date}"
today_watch_count = await redis_client.incr(daily_key)
description = f"{paint['author']}\n"
description += f"Размер: {paint['size']}\n"
description += f"Материал: {paint['material']}\n"
description += f"Стоимость: {paint['price']}\n"
description += f"{paint['description']}"
return templates.TemplateResponse(
request=request, name="paintings/paint.html", context={"paint": paint, "user": user, "description": description,'today_watch_count':today_watch_count,'total_watch_count':total_watch_count}
)
@app.get('/post/{paint_id}')
async def old_post(
request: Request,
paint_id: int,
user: Dict = Security(get_user)
):
if isinstance(paint_id, int):
doc = await posts_collection.find_one({"id": paint_id}, {"_id": 1})
if not doc:
raise HTTPException(status_code=404, detail="Картина не найдена")
return RedirectResponse(url=f"/paintings/{doc.get('_id')}", status_code=301)
@app.delete("/api/post/{post_id}", response_model=dict)
async def post_delete(
post_id: Optional[str],
current_user: Dict = Security(get_current_user, scopes=["create:posts"])
):
await posts_collection.delete_one({'_id': ObjectId(post_id), 'user._id': current_user.get('_id')})
return {'ok': True}
@app.post("/api/post", response_model=dict)
async def post_painting(
background_tasks: BackgroundTasks,
author: str = Form(None),
grade: str = Form(None),
city: str = Form(None),
title: str = Form(...),
size: str = Form(None),
material: str = Form(None),
genre: str = Form(None),
price: str = Form(None),
description: str = Form(None),
post_medias: Optional[List[str]] = Form(...),
current_user: Dict = Security(get_current_user, scopes=["create:posts"])
):
try:
doc = {
"title": safe_strip(title),
"url": transliterate(f"{safe_strip(title)} {safe_strip(author, current_user['username'])} {safe_strip(city)}"),
"author": safe_strip(author, current_user['username']),
"grade": safe_strip(grade),
"city": safe_strip(city) or current_user.get('living_place'),
"size": safe_strip(size),
"status": "В каталоге",
"material": safe_strip(material),
"genre": safe_strip(genre),
"price": safe_strip(price),
"description": safe_strip(description),
"medias": post_medias or [],
"medias_original": post_medias or [],
"user": display_user(current_user),
**{i:[] for i in SOCIALS},
}
res = await posts_collection.insert_one(doc)
doc_out = {
"id": str(res.inserted_id),
"_id": str(res.inserted_id),
**{k: v for k, v in doc.items()}
}
doc_out = bson_clear(doc_out)
background_tasks.add_task(replace_post_medias, doc_out)
if PROD:
background_tasks.add_task(email_new_post, doc_out)
background_tasks.add_task(publish_social_post, doc_out)
if 'user' in current_user.get('roles',[]) and safe_strip(grade) != current_user.get('grade'):
await users_collection.update_one({'_id':ObjectId(current_user.get('_id'))},{"$set": {'grade': safe_strip(grade)}})
return doc_out
except Exception as e:
print(e)
raise HTTPException(status_code=500, detail=str(e))
@app.put("/api/post/{post_id}", response_model=dict)
async def put_painting(
background_tasks: BackgroundTasks,
post_id: str,
author: str = Form(None),
grade: str = Form(None),
city: str = Form(None),
title: str = Form(...),
size: str = Form(None),
material: str = Form(None),
genre: str = Form(None),
price: str = Form(None),
description: str = Form(None),
post_medias: Optional[List[str]] = Form(...),
current_user: Dict = Security(get_current_user, scopes=["create:posts"])
):
update_doc = {}
if title is not None:
update_doc["title"] = safe_strip(title)
update_doc["url"] = transliterate(f"{safe_strip(title)} {safe_strip(author)} {safe_strip(city)}")
if author is not None:
update_doc["author"] = safe_strip(author)
if grade is not None:
update_doc["grade"] = safe_strip(grade)
if city is not None:
update_doc["city"] = safe_strip(city)
if size is not None:
update_doc["size"] = safe_strip(size)
if material is not None:
update_doc["material"] = safe_strip(material)
if genre is not None:
update_doc["genre"] = safe_strip(genre)
if price is not None:
update_doc["price"] = safe_strip(price)
if description is not None:
update_doc["description"] = safe_strip(description)
if post_medias:
update_doc["medias_original"] = post_medias
res = await posts_collection.find_one_and_update(
{'_id': ObjectId(post_id)},
{'$set': update_doc},
return_document=False
)
links_media = list(zip(res.get('medias_original',[]), res.get('medias',[])))
if res is None:
raise HTTPException(status_code=404, detail="Картина не найдена")
doc_out = {
"_id":post_id,
"id":post_id,
**{k: v for k, v in update_doc.items()}
}
doc_out = bson_clear(doc_out)
background_tasks.add_task(update_social_post, doc_out)
background_tasks.add_task(replace_new_medias, bson_clear(doc_out), links_media)
return doc_out
@app.post("/api/posts/media/chunk")
async def upload_chunk(
chunk: UploadFile = File(...),
file_id: str = Form(...),
chunk_index: int = Form(...),
total_chunks: int = Form(...),
filename: str = Form(...),
content_type: str = Form(...),
current_user: Dict = Security(get_current_user, scopes=["create:posts"])
):
try:
data = await chunk.read()
b64_data = base64.b64encode(data).decode("ascii")
key = f"upload:{file_id}"
# Store meta only once
meta = await redis_client.hgetall(key)
if not meta:
await redis_client.hset(key, mapping={
"filename": filename,
"content_type": content_type,
"total_chunks": str(total_chunks),
"user_id": str(current_user.get("id", "")),
})
await redis_client.expire(key, CHUNK_TTL)
# Store the chunk
await redis_client.hset(key, f"chunk:{chunk_index}", b64_data)
await redis_client.expire(key, CHUNK_TTL)
return {"ok": True, "received": chunk_index}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/posts/media/finalize")
async def finalize_upload(
payload: dict,
current_user: Dict = Security(get_current_user, scopes=["create:posts"])
):
file_id = payload.get("file_id")
if not file_id:
raise HTTPException(status_code=400, detail="file_id required")
key = f"upload:{file_id}"
meta = await redis_client.hgetall(key)
if not meta:
raise HTTPException(status_code=400, detail="Unknown or expired file_id")
total_chunks = int(meta["total_chunks"])
filename = meta["filename"]
content_type = meta["content_type"]
# Collect all chunks
chunks = []
for i in range(total_chunks):
b64 = await redis_client.hget(key, f"chunk:{i}")
if b64 is None:
raise HTTPException(status_code=400, detail=f"Missing chunk {i}")
chunks.append(base64.b64decode(b64))
full_content = b"".join(chunks)
temp_file = StarletteUploadFile(
file=BytesIO(full_content),
filename=filename,
headers={"content-type": content_type}
)
try:
saved_name = await save_to_gridfs(temp_file, "posts")
return {"post_medias": [saved_name]}
finally:
# Cleanup
await redis_client.delete(key)
# Optional fallback (non-chunked)
@app.post("/api/posts/media", response_model=dict)
async def post_media(
medias: Optional[List[UploadFile]] = File(None),
current_user: Dict = Security(get_current_user, scopes=["create:posts"])
):
try:
post_medias = await asyncio.gather(
*[save_to_gridfs(media, "posts") for media in medias or []]
)
return {"post_medias": post_medias}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/news", response_model=dict)
async def post_news(
background_tasks: BackgroundTasks,
title: str = Form(...),
description: str = Form(None),
short_description: str = Form(None),
publish_on_platform: bool = Form(False),
post_medias: Optional[List[str]] = Form(...),
current_user: Dict = Security(get_current_user, scopes=["create:news"])
):
try:
doc = {
"title": safe_strip(title),
"short_description": safe_strip(short_description),
"publish_on_platform": publish_on_platform,
"description": safe_strip(description),
"media": post_medias,
"media_original": post_medias,
"user": display_user(current_user),
**{i:[] for i in SOCIALS},
}
res = await news_collection.insert_one(doc)
doc_out = {
"id": str(res.inserted_id),
**{k: v for k, v in doc.items()}
}
doc_out = bson_clear(doc_out)
# background_tasks.add_task(email_new_post, doc_out)
background_tasks.add_task(replace_news_medias, doc_out)
return doc_out
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.put("/api/news/{news_id}", response_model=dict)
async def put_news(
background_tasks: BackgroundTasks,
news_id: str,
title: str = Form(...),
description: str = Form(None),
short_description: str = Form(None),
publish_on_platform: bool = Form(False),
post_medias: Optional[List[str]] = Form(...),
current_user: Dict = Security(get_current_user, scopes=["create:news"])
):
update_doc = {}
if title is not None:
update_doc["title"] = safe_strip(title)
if description is not None:
update_doc["description"] = safe_strip(description)
if short_description is not None:
update_doc["short_description"] = safe_strip(short_description)
if publish_on_platform is not None:
update_doc["publish_on_platform"] = publish_on_platform
if post_medias:
update_doc["media"] = post_medias
update_doc["media_original"] = post_medias
res = await news_collection.update_one(
{'_id': ObjectId(news_id)},
{'$set': update_doc}
)
if res.modified_count == 0:
raise HTTPException(status_code=404, detail="Картина не найдена")
doc_out = {
"id":news_id,
**{k: v for k, v in update_doc.items()}
}
doc_out = bson_clear(doc_out)
return doc_out
@app.post("/api/news/media", response_model=dict)
async def news_media(
medias: Optional[List[UploadFile]] = File(None),
current_user: Dict = Security(get_current_user, scopes=["create:news"])
):
try:
media = await asyncio.gather(*[save_to_gridfs(media, 'news') for media in medias])
return {'media': media}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/api/read/notify", response_model=Dict)
async def read_notify(
user: Dict = Security(get_user)
):
await notify_collection.update_many({'user_id':user.get('_id')},{'$set':{'read':False}})
return {}
@app.post("/api/competition", response_model=Dict, tags=["Competition", "Partner"])
async def post_competition(
background_tasks: BackgroundTasks,
payload: CompetitionCreate = Depends(build_payload),
current_user: Dict = Security(get_current_user, scopes=["create:super_items"]),
):
try:
poster_id = await save_to_gridfs(payload.poster, "competition")
if payload.rules:
rules_id = await save_to_gridfs(payload.rules, "competition")
else:
rules_id = None
await users_collection.update_one(
{"_id": ObjectId(current_user["_id"])},
{"$addToSet": {"send_address": safe_strip(payload.address)}},
upsert=True,
)
doc = {
"url": transliterate(safe_strip(payload.title)),
"title": safe_strip(payload.title),
"topic": safe_strip(payload.topic) if payload.topic else None,
"type_": payload.type_,
"datetime_start": payload.datetime_start,
"date_finish": payload.date_finish,
"nominations": payload.nominations,
"address": safe_strip(payload.address),
"emailsNotify": payload.emailsNotify,
"serviceSocial": payload.serviceSocial,
"userSocial": payload.userSocial,
"short_desc": safe_strip(payload.short_desc) if payload.short_desc else None,
"poster_file_id": poster_id,
"poster_orig_id": poster_id,
"rules_file_id": rules_id,
"rules_link": payload.rules_link,
"status": 0,
"user": display_user(current_user),
"min_paintings": payload.min_paintings,
"max_paintings": payload.max_paintings,
**{i:[] for i in SOCIALS},
}
res = await competitions_col.insert_one(doc)
background_tasks.add_task(
poster_competition, bson_clear({**doc, "_id": res.inserted_id})
)
return {"ok": True, "competition_id": str(res.inserted_id)}
except Exception as exc:
logging.error(str(exc))
raise HTTPException(status_code=500, detail=str(exc))
@app.post("/api/competition/publish", response_model=Dict, tags=["Competition", "Partner"])
async def publish_competition(
background_tasks: BackgroundTasks,
payload: CompetitionPublishParams = Depends(build_publish_payload),
current_user: Dict = Security(get_current_user, scopes=["create:super_items"]),
):
try:
logging.info(
{
"current_user": current_user.get("_id"),
"competition_id": payload.competition_id,
"emailsNotify": payload.emailsNotify,
"serviceSocial": payload.serviceSocial,
"userSocial": payload.userSocial,
"usersSocial": payload.usersSocial,
}
)
competition = await competitions_col.find_one(
{
"user._id": current_user.get("_id"),
"_id": ObjectId(payload.competition_id),
}
)
if not competition:
raise HTTPException(status_code=404, detail="Конкурс не найден")
competition = bson_clear(competition)
if competition["datetime_start"] > datetime.now():
await competitions_col.update_one(
{"_id": ObjectId(payload.competition_id)},
{"$set": {"datetime_start": datetime.now()}},
)
if payload.emailsNotify:
background_tasks.add_task(invite_users_to_competitions, competition)
if any([payload.serviceSocial, payload.userSocial, payload.usersSocial]):
background_tasks.add_task(
service_social_competition,
competition,
payload.serviceSocial,
payload.userSocial,
payload.usersSocial,
)
return {"ok": True, "competition_id": payload.competition_id}
except Exception as exc:
logging.error(str(exc))
raise HTTPException(status_code=500, detail=str(exc))
@app.put("/api/competition/{competition_id}", response_model=Dict, tags=["Competition", "Partner"])
async def put_competition(
background_tasks: BackgroundTasks,
competition_id: str,
payload: CompetitionUpdate = Depends(build_update_payload),
current_user: Dict = Security(get_current_user, scopes=["create:super_items"]),
):
try:
update_doc = {
"user": display_user(current_user),
"rules_link": payload.rules_link,
"type_": payload.type_,
}
if payload.nominations is not None:
update_doc["nominations"] = payload.nominations
if payload.title is not None:
update_doc["title"] = safe_strip(payload.title)
if payload.topic is not None:
update_doc["topic"] = safe_strip(payload.topic)
if payload.datetime_start is not None:
update_doc["datetime_start"] = payload.datetime_start
if datetime.now(timezone.utc) + timedelta(hours=3) < payload.datetime_start:
update_doc["status"] = -1
if payload.date_finish is not None:
update_doc["date_finish"] = payload.date_finish
if payload.address is not None:
await users_collection.update_one(
{ "_id": ObjectId(current_user["_id"]) },
{ "$addToSet": { "send_address": safe_strip(payload.address) } },
upsert=True
)
update_doc["address"] = safe_strip(payload.address)
if payload.short_desc is not None:
update_doc["short_desc"] = safe_strip(payload.short_desc)
if payload.emailsNotify is not None:
update_doc["emailsNotify"] = payload.emailsNotify
if payload.serviceSocial is not None:
update_doc["serviceSocial"] = payload.serviceSocial
if payload.userSocial is not None:
update_doc["userSocial"] = payload.userSocial
if payload.min_paintings is not None:
update_doc["min_paintings"] = int(payload.min_paintings)
if payload.max_paintings is not None:
update_doc["max_paintings"] = int(payload.max_paintings)
if payload.poster:
poster_id = await save_to_gridfs(payload.poster, 'competition')
update_doc["poster_file_id"] = poster_id
background_tasks.add_task(poster_competition, bson_clear({**update_doc, '_id': competition_id}))
if payload.rules:
rules_id = await save_to_gridfs(payload.rules, 'competition')
update_doc["rules_file_id"] = rules_id
res = await competitions_col.update_one(
{'_id': ObjectId(competition_id)},
{'$set': update_doc}
)
return {'ok': True}
except Exception as exc:
logging.error(str(exc))
raise HTTPException(status_code=500, detail=str(exc))
@app.get("/api/exhibitions", response_model=List)
async def get_exhibitions():
exhibitions = []
current_time = datetime.now(timezone.utc) + timedelta(hours=3)
async for doc in exhibition_collection.find({
'$or': [
{'datetime_start': {'$lt': current_time}},
],
'prod': True
},{'vk': 0,'max': 0,'tg': 0,'fb': 0,'inst': 0}).sort('_id',-1).limit(1000):
exhibitions.append(bson_clear(doc))
return exhibitions
@app.post("/api/exhibition", response_model=Dict)
async def post_exhibition(
background_tasks: BackgroundTasks,
payload: ExhibitionCreate = Depends(build_exhibition_payload),
current_user: Dict = Security(get_current_user, scopes=["create:super_items"])
):
try:
poster_id = await save_to_gridfs(payload.poster, 'exhibition')
rules_id = await save_to_gridfs(payload.rules, 'exhibition') if payload.rules else None
tizer_id = await save_to_gridfs(payload.tizer, 'exhibition') if payload.tizer else None
await users_collection.update_one(
{"_id": ObjectId(current_user["_id"])},
{"$addToSet": {
"send_address": safe_strip(payload.address),
"places": safe_strip(payload.place)
}},
upsert=True
)
doc = {
"url": transliterate(safe_strip(payload.title)),
"title": safe_strip(payload.title),
"topic": safe_strip(payload.topic) if payload.topic else None,
"datetime_start": payload.datetime_start,
"date_finish": payload.date_finish,
"address": safe_strip(payload.address) if payload.address else None,
"place": safe_strip(payload.place) if payload.place else None,
"rules_file_id": rules_id,
"rules_link": payload.rules_link,
"tizer_file_id": tizer_id,
"tizer_link": payload.tizer_link,
"poster_orig_id": poster_id,
"poster_file_id": poster_id,
"status": 0,
"prod": PROD,
"user": display_user(current_user),
"posts": [],
**{i:[] for i in SOCIALS},
}
res = await exhibition_collection.insert_one(doc)
exhibition_id = res.inserted_id
background_tasks.add_task(
poster_exhibition,
bson_clear({**doc, '_id': exhibition_id})
)
return {'_id': str(exhibition_id)}
except Exception as e:
logging.error(str(e))
raise HTTPException(status_code=500, detail=str(e))
@app.put("/api/exhibition/{exhibition_id}", response_model=Dict)
async def put_exhibition(
background_tasks: BackgroundTasks,
exhibition_id: str,
payload: ExhibitionUpdate = Depends(build_update_exhibition_payload),
current_user: Dict = Security(get_current_user, scopes=["create:super_items"])
):
try:
update_doc = {
"user": display_user(current_user),
}
if payload.title is not None:
update_doc["title"] = safe_strip(payload.title)
if payload.topic is not None:
update_doc["topic"] = safe_strip(payload.topic)
if payload.place is not None:
await users_collection.update_one(
{"_id": ObjectId(current_user["_id"])},
{"$addToSet": {"places": safe_strip(payload.place)}},
upsert=True
)
update_doc["place"] = safe_strip(payload.place)
if payload.datetime_start is not None:
update_doc["datetime_start"] = payload.datetime_start
if payload.date_finish is not None:
update_doc["date_finish"] = payload.date_finish
if payload.address is not None:
await users_collection.update_one(
{"_id": ObjectId(current_user["_id"])},
{"$addToSet": {"send_address": safe_strip(payload.address)}},
upsert=True
)
update_doc["address"] = safe_strip(payload.address)
if payload.short_desc is not None:
update_doc["short_desc"] = safe_strip(payload.short_desc)
update_doc["rules_link"] = payload.rules_link
update_doc["tizer_link"] = payload.tizer_link
# Обработка файлов
if payload.poster:
poster_id = await save_to_gridfs(payload.poster, 'exhibition')
update_doc["poster_file_id"] = poster_id
update_doc["poster_orig_id"] = poster_id
background_tasks.add_task(
poster_exhibition,
bson_clear({**update_doc, '_id': exhibition_id})
)
if payload.tizer:
tizer_id = await save_to_gridfs(payload.tizer, 'exhibition')
update_doc["tizer_file_id"] = tizer_id
update_doc["tizer_link"] = None
if payload.rules:
rules_id = await save_to_gridfs(payload.rules, 'exhibition')
update_doc["rules_file_id"] = rules_id
update_doc["rules_link"] = None
# Если нужно добавить nominations / min_paintings и т.д. — расширь модель и этот блок
res = await exhibition_collection.update_one(
{'_id': ObjectId(exhibition_id)},
{'$set': update_doc}
)
return {'_id': exhibition_id}
except Exception as e:
logging.error(str(e))
raise HTTPException(status_code=500, detail=str(e))
async def stream_gridfs_file(file_id: str, bucket_name: str):
if not file_id:
return None
grid = gridfs_map.get(bucket_name)
if grid is None:
return None
try:
oid = ObjectId(file_id)
except Exception:
return None
stream = None
try:
stream = await grid.open_download_stream(oid)
filename = stream.filename or "file"
content_type = (
stream.content_type
or mimetypes.guess_type(filename)[0]
or "application/octet-stream"
)
file_length = getattr(stream, "length", None)
async def file_iterator(chunk_size: int = 64 * 1024):
try:
while True:
chunk = await stream.read(chunk_size)
if not chunk:
break
yield chunk
finally:
if stream is not None:
close = getattr(stream, "close", None)
if asyncio.iscoroutinefunction(close) or asyncio.iscoroutine(close):
await close()
elif callable(close):
close()
return filename, content_type, file_length, file_iterator
except Exception:
if stream is not None:
close = getattr(stream, "close", None)
if asyncio.iscoroutinefunction(close) or asyncio.iscoroutine(close):
await close()
elif callable(close):
close()
return None
def content_disposition_header(filename: str, inline: bool = True) -> str:
dispo_type = "inline" if inline else "attachment"
fallback = filename.encode("latin-1", errors="ignore").decode("latin-1")
fallback = fallback.replace('"', '\\"').replace("\\", "\\\\") or "file"
try:
filename.encode("latin-1")
return f'{dispo_type}; filename="{fallback}"'
except UnicodeEncodeError:
filename_quoted = quote(filename, safe="")
return f'{dispo_type}; filename="{fallback}"; filename*=UTF-8\'\'{filename_quoted}'
@app.get("/b/{bucket_name}/{file_id}")
async def get_file(bucket_name: str, file_id: str):
res = await stream_gridfs_file(file_id, bucket_name)
if res is None:
raise HTTPException(status_code=404, detail="File not found")
filename, content_type, file_length, iterator = res
headers = {
"Content-Disposition": content_disposition_header(filename, inline=True),
"Cache-Control": "public, max-age=3600",
"ETag": f'"{bucket_name}:{file_id}"',
"Vary": "Accept-Encoding",
}
if file_length is not None:
headers["Content-Length"] = str(file_length)
return StreamingResponse(iterator(), media_type=content_type, headers=headers)
@app.get("/a/{user_id}")
async def get_avatar(user_id: str):
user = await users_collection.find_one({'_id': ObjectId(user_id)},{'img':1})
file_id = user['img']
res = await stream_gridfs_file(file_id, 'users')
if res is None:
raise HTTPException(status_code=404, detail="File not found")
filename, content_type, file_length, iterator = res
headers = {
"Content-Disposition": content_disposition_header(filename, inline=True),
"Cache-Control": "public, max-age=3600",
"ETag": f'"users:{file_id}"',
"Vary": "Accept-Encoding",
}
if file_length is not None:
headers["Content-Length"] = str(file_length)
return StreamingResponse(iterator(), media_type=content_type, headers=headers)
async def unsubscribe_user(
email: str,
token: Optional[str] = None
) -> bool:
try:
if token:
stored_email = await redis_client.get(f"unsub:{token}")
if not stored_email or stored_email != email:
return False
await redis_client.delete(f"unsub:{token}")
result = await mongo_client['email']['tokens'].update_one(
{"token": token, "email": email, "used": False},
{"$set": {"used": True, "used_at": datetime.utcnow()}}
)
if result.modified_count == 0:
return False
await users_collection.update_one(
{"email": email},
{"$set": {
"email_subscribed": False,
"email_unsubscribed_at": datetime.utcnow(),
"email_unsubscribe_reason": "user_request"
}}
)
logging.info(f"✅ Пользователь отписался: {email}")
return True
except Exception as e:
logging.error(f"❌ Ошибка отписки {email}: {e}")
return False
@app.get("/api/v1/email/unsubscribe", response_class=HTMLResponse)
async def unsubscribe(
request: Request,
email: EmailStr,
token: Optional[str] = None,
):
logging.info(f"Отписка {token} {email}")
await unsubscribe_user(email)
return """
Отписка
✅ Вы успешно отписались
Вы больше не будете получать письма от нас.
Вернуться на сайт
"""
@app.get("/profile/admin/users", response_class=HTMLResponse)
async def profile_admin_statistics(
request: Request,
user: Dict = Security(get_current_user, scopes=["admin"])
):
users = [doc async for doc in users_collection.find().sort({'_id':-1})]
return templates.TemplateResponse(
request=request, name="profile/admin/users.html", context={"user": user,
'users': users
}
)
from collections import defaultdict
@app.get("/profile/admin/statistics", response_class=HTMLResponse)
async def profile_admin_statistics(
request: Request,
user: Dict = Security(get_current_user, scopes=["admin"])
):
pipeline1 = [
{"$match": {"roles": "user"}},
{
"$group": {
"_id": "$living_place",
"user_count": {"$sum": 1},
"user_ids": {"$push": "$_id"}
}
},
{"$sort": {"user_count": -1}}
]
living_stats = await users_collection.aggregate(pipeline1).to_list(length=None)
for stat in living_stats:
if stat["_id"]:
user_ids = stat["user_ids"] # list of user _id values (ObjectId or str)
# If posts store user._id as string, convert ObjectId -> str
user_id_strings = [str(u) for u in user_ids]
# post_count (use count_documents as you had)
stat["post_count"] = await posts_collection.count_documents(
{"user._id": {"$in": user_id_strings}}
)
# usernames
cursor = users_collection.find(
{"_id": {"$in": user_ids}},
{"username": 1, "_id": 1}
).limit(1000)
stat["usernames"] = await cursor.to_list(length=None)
# fetch posts for these users (limit/fields optional)
posts_cursor = posts_collection.find(
{"user._id": {"$in": user_id_strings}},
{"_id": 1, "title": 1, "created_at": 1, "user._id": 1}
).sort("created_at", -1)
stat["posts"] = await posts_cursor.to_list(length=None)
pipeline2 = [
{"$match": {"roles": "user"}},
{
"$project": {
"year_month": {
"$dateToString": {
"format": "%Y-%m",
"date": {"$toDate": "$_id"}
}
}
}
},
{
"$group": {
"_id": "$year_month",
"user_count": {"$sum": 1},
"user_ids": {"$push": "$_id"}
}
},
{"$sort": {"_id": -1}}
]
reg_stats = await users_collection.aggregate(pipeline2).to_list(length=None)
for stat in reg_stats:
user_ids = stat["user_ids"] # list of user _id values (ObjectId or str)
user_id_strings = [str(u) for u in user_ids] # use this if posts.user._id is a str
stat["post_count"] = await posts_collection.count_documents(
{"user._id": {"$in": user_id_strings}}
)
cursor = users_collection.find(
{"_id": {"$in": user_ids}},
{"username": 1, "_id": 0}
).limit(1000)
stat["usernames"] = await cursor.to_list(length=None)
# fetch posts (flat list)
posts_cursor = posts_collection.find(
{"user._id": {"$in": user_id_strings}},
{"_id": 1, "title": 1, "created_at": 1, "user._id": 1}
).sort("created_at", -1)
stat["posts"] = await posts_cursor.to_list(length=None)
# OR: group posts per user
posts = stat["posts"]
grouped = defaultdict(list)
for p in posts:
grouped[p["user"]["_id"]].append(p)
stat["posts_by_user"] = dict(grouped)
return templates.TemplateResponse(
request=request, name="profile/admin/statistics.html", context={"user": user, "living_stats": living_stats,
"reg_stats": reg_stats,
"all_users": await users_collection.count_documents({ 'roles': "user" }),
"all_city": len(await users_collection.distinct("living_place", filter={ 'roles': "user" }))
}
)
@app.post("/tg")
async def tg_result_save(data:dict, token: str = Security(get_token)):
item = data.get('item')
result = data.get('result', {})
if item:
update_spec = {"$push": {"tg": result}}
await mongo_client[item.get('db')][item.get('col')].update_one({'_id': ObjectId(item.get('_id'))}, update_spec)
@app.get("/profile/social", response_class=HTMLResponse)
async def profile_social(
request: Request,
user: Dict = Security(get_current_user)
):
socials = [doc async for doc in social_collection.find({'user_id': str(user.get('_id'))})]
return templates.TemplateResponse(
request=request, name="profile/social.html", context={"user": user, 'socials': socials}
)
@app.get("/profile/invitation/{invitation_id}", response_class=HTMLResponse)
async def profile_invitation(
request: Request,
invitation_id: str,
user: Dict = Security(get_current_user)
):
data = await invitations_collection.find_one({'_id': ObjectId(invitation_id)})
match_ = {
"$or": [
{"sender._id": user.get('_id')},
{"user._id": user.get('_id')},
{"receiver._id": user.get('_id')}
],
'prod': PROD,
}
if data.get('competition'):
match_.update({'competition_id': data.get('competition').get('_id')})
if data.get('exhibition_id'):
match_.update({'exhibition_id': data.get('exhibition_id')})
cursor = delivery_collection.aggregate([
{"$match": match_},
{"$sort": {"_id": -1}},
{"$group": {
"_id": "$post_id",
"doc": {"$first": "$$ROOT"}
}},
{"$replaceRoot": {"newRoot": "$doc"}},
{"$sort": {"_id": -1}},
])
posts = [doc async for doc in cursor]
return templates.TemplateResponse(
request=request, name="profile/invitation.html", context={ "user": user, "doc": data, "data": posts}
)
@app.get("/profile/invitation", response_class=HTMLResponse)
async def profile_invitation(
request: Request,
user: Dict = Security(get_current_user)
):
data = await invitations_collection.find({'prod': PROD,'user_id': user.get('_id')}).sort('_id', -1).to_list()
await notify_collection.update_many({'user_id':user.get('_id'),'type':'invitation'},{"$set":{'read':False}})
return templates.TemplateResponse(
request=request, name="profile/invitations.html", context={ "user": user, "data": data, "now": datetime.now()}
)
@app.post('/api/social/settings', response_model=dict)
async def social_setting(
id: Optional[str] = Body(None, embed=True),
key: Optional[str] = Body(None, embed=True),
check: Optional[bool] = Body(None, embed=True),
user: Dict = Security(get_current_user)
):
await social_collection.update_one({'_id': ObjectId(id), 'user_id': user.get('_id')},{'$set': {key:check}})
return {'ok': True}
async def get_delivery_post(post_id):
cursor = delivery_collection.aggregate([
{"$match": {
'post_id': post_id,
'prod': PROD,
}},
{"$sort": {"_id": -1}},
{"$group": {
"_id": "$post_id",
"doc": {"$first": "$$ROOT"}
}},
{"$replaceRoot": {"newRoot": "$doc"}},
{"$sort": {"_id": -1}},
])
data = [doc async for doc in cursor]
return data[0] if data else False
@app.post('/api/v1/views')
async def get_views_count(request: Request):
url = (await request.json())['url']
if 'artist' in url:
user_id = url.split('/')[-1]
total_key = f"total_w_c_artists:{user_id}"
today_key = f"today_w_c_artists:{user_id}"
total_watch_count = await redis_client.incr(total_key)
today_date = datetime.now().strftime("%Y-%m-%d")
daily_key = f"{today_key}:{today_date}"
today_watch_count = await redis_client.incr(daily_key)
return {"totalViews":total_watch_count,"todayViews":today_watch_count}
if 'post' in url:
paint_id = url.split('/')[-1]
total_key = f"total_watch_count_paint:{paint_id}"
today_key = f"today_watch_count_paint:{paint_id}"
total_watch_count = await redis_client.incr(total_key)
today_date = datetime.now().strftime("%Y-%m-%d")
daily_key = f"{today_key}:{today_date}"
today_watch_count = await redis_client.incr(daily_key)
return {"totalViews":total_watch_count,"todayViews":today_watch_count}
if 'news' in url:
news_id = url.split('/')[-1]
total_key = f"total_watch_count_news:{news_id}"
today_key = f"today_watch_count_news:{news_id}"
total_watch_count = await redis_client.incr(total_key)
today_date = datetime.now().strftime("%Y-%m-%d")
daily_key = f"{today_key}:{today_date}"
today_watch_count = await redis_client.incr(daily_key)
return {"totalViews":total_watch_count,"todayViews":today_watch_count}
return {"totalViews":0,"todayViews":0}
@app.get('/api/v1/generic/post', response_model=dict)
async def get_posts(
per_page: Optional[int] = Query(4),
page: Optional[int] = Query(1),
user_id: Optional[str] = Query(None),
filter: Optional[str] = Query(None, description="JSON filter string or omitted")
):
posts = []
if filter:
try:
filt = json.loads(filter)
mql = convert_to_mql(filt)
def mql_to_mongo(m):
if "and" in m:
return {"$and": [mql_to_mongo(x) for x in m["and"]]}
if "or" in m:
return {"$or": [mql_to_mongo(x) for x in m["or"]]}
field, op, value = m["field"], m["op"], m["value"]
if field == 'author_name':
field = 'author'
if field == 'user.gallery_name':
field = 'user.username'
if field == 'name':
field = 'title'
if op == "like":
return {field: {"$regex": safe_strip(value), "$options": "i"}}
return {field: value}
mongo_query = mql_to_mongo(mql)
for post in [doc async for doc in posts_collection.find(mongo_query).sort({"_id":-1}).limit(per_page * page + 8)][per_page * page - per_page:per_page * page]:
post['delivery'] = await get_delivery_post(str(post.get('_id')))
post['user']['avatar_url'] = f"/a/{post['user'].get('_id')}"
post['user']['name'] = post['user']['username']
photos = [{'url':f"/b/posts/{i}", 'type':'photo', 'sort_order':x+1} for x,i in enumerate(post['medias'])]
post['id'] = post.get('_id')
post['photos'] = photos
post['name'] = post['title']
post['author_name'] = post['author']
post['media'] = photos
posts.append(bson_clear(post))
post_count = await posts_collection.count_documents(mongo_query)
return {
'current_page': page,
'data':posts,
'total': post_count,
'to': 16,
'last_page': post_count / per_page,
'per_page': per_page
}
except json.JSONDecodeError:
pass
q = {}
if user_id:
try:
if len(str(user_id)) < 6:
user_id = int(user_id)
user = await users_collection.find_one({'id':user_id})
q['user._id'] = str(user["_id"])
else:
q['user._id'] = user_id
except:
q['user._id'] = user_id
for post in [doc async for doc in posts_collection.find(q).sort({"_id":-1}).limit(per_page * page + 8)][per_page * page - per_page:per_page * page]:
post['delivery'] = await get_delivery_post(str(post.get('_id')))
post['user']['avatar_url'] = f"/a/{post['user'].get('_id')}"
post['user']['name'] = post['user']['username']
post['author_name'] = post['author']
photos = [{'url':f"/b/posts/{i}", 'type':'photo', 'sort_order':x+1} for x,i in enumerate(post['medias'])]
post['id'] = post.get('_id')
post['photos'] = photos
post['media'] = photos
post['name'] = post['title']
posts.append(bson_clear(post))
post_count = await posts_collection.count_documents(q)
return {
'current_page': page,
'data':posts,
'total': post_count,
'to': 16,
'last_page': post_count / per_page,
'per_page': per_page
}
@app.get('/api/v2/statistics', response_model=dict)
async def statistics_v2(
user: Optional[str] = Query(None),
post: Optional[str] = Query(None),
):
if user:
return {"statistics":await users_collection.count_documents({ 'roles': "user" }),"additional_stats":len(await users_collection.distinct("living_place", filter={ 'roles': "user" }))}
else:
return {"statistics":await posts_collection.count_documents({}),"additional_stats":await users_collection.count_documents({ 'roles': "user" })}
@app.get('/api/v2/user/{user_id}', response_model=dict)
async def get_user_by_id(
user_id: Optional[str]
):
try:
user = await users_collection.find_one({'id': int(user_id)})
except:
user = await users_collection.find_one({'_id': ObjectId(user_id)})
timestamp = user.get('_id').generation_time
user['posts_count'] = await posts_collection.count_documents({'user._id':str(user.get('_id'))})
timestamp_str = timestamp.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"
user['created_at'] = timestamp_str
user['avatar_url'] = f"https://avtopost.kultmost.ru/a/{user.get('_id')}"
user['name'] = user['username']
user.pop('hashed_password')
user.pop('phone')
user.pop('email')
return {'data':bson_clear(user)}
@app.get('/api/v1/generic/post/{post_id}', response_model=dict)
async def get_post_by_id(
post_id: Optional[str]
):
try:
post = await posts_collection.find_one({'id': int(post_id)})
except:
post = await posts_collection.find_one({'_id': ObjectId(post_id)})
post['user']['avatar_url'] = f"/a/{post['user'].get('_id')}"
post['user']['name'] = post['user']['username']
photos = [{'url':f"/b/posts/{i}", 'type':'photo', 'sort_order':x+1} for x,i in enumerate(post['medias_original'])]
post['photos'] = photos
post['id'] = post.get('_id')
post['name'] = post['title']
post['media'] = photos
return bson_clear(post)
## Work
@app.get("/profile/invitation/competition/{competition_id}", response_class=HTMLResponse)
async def profile_invitation(
request: Request,
competition_id: str,
user: Dict = Security(get_current_user)
):
competition = await competitions_col.find_one({'_id': ObjectId(competition_id)})
req = await competition_request_col.find_one({'competition_id': competition_id, 'user._id': user.get('_id')})
return templates.TemplateResponse(
request=request, name="profile/invitation/competition.html", context={ "user": user, "competition": competition, "req": req}
)
@app.post('/api/invitation/accept/{invitations_id}')
async def invitation_accept(invitations_id: str, request: Request):
payload = await request.json()
posts_ids: List[str] = payload.get("posts", [])
# optional: валидация
if not isinstance(posts_ids, list):
posts_ids = []
await accept_invitations(invitations_id, posts_ids)
return {'ok': True}
@app.get('/profile/delivery/{post_id}')
async def profile_delivery_history_post(request: Request, post_id:str, user: Dict = Security(get_current_user)):
data = await delivery_collection.find({'post_id': post_id, "$or": [
{"sender._id": user.get('_id')},
{"user._id": user.get('_id')},
{"receiver._id": user.get('_id')}
]}).sort('_id', -1).to_list()
return templates.TemplateResponse(
request=request,
name="profile/delivery_history.html",
context={
"user": user,
"data": data,
}
)
# @app.get('/profile/delivery/{type_d}/{item_id}')
# async def profile_delivery_by_old(request: Request, type_d:str, item_id:str, user: Dict = Security(get_current_user)):
# cursor = delivery_collection.aggregate([
# {"$match": {
# "$or": [
# {"sender._id": user.get('_id')},
# {"user._id": user.get('_id')},
# {"receiver._id": user.get('_id')}
# ],
# f'{type_d}_id': ObjectId(item_id),
# 'prod': PROD,
# }},
# {"$sort": {"_id": -1}},
# {"$group": {
# "_id": "$post_id",
# "doc": {"$first": "$$ROOT"}
# }},
# {"$replaceRoot": {"newRoot": "$doc"}},
# {"$sort": {"_id": -1}},
# ])
# data = [doc async for doc in cursor]
# data.sort(key=lambda doc: doc["_id"].generation_time, reverse=True)
# return templates.TemplateResponse(
# request=request,
# name="profile/delivery_old.html",
# context={
# "user": user,
# "data": data,
# }
# )
@app.get('/profile/delivery/competition/{competition_id}')
async def profile_delivery_by_competition(request: Request, competition_id:str, user: Dict = Security(get_current_user)):
cursor = delivery_collection.aggregate([
{"$match": {
"$or": [
{"sender._id": user.get('_id')},
{"user._id": user.get('_id')},
{"receiver._id": user.get('_id')}
],
'prod': PROD,
}},
{"$sort": {"_id": -1}},
{"$group": {
"_id": "$post_id",
"doc": {"$first": "$$ROOT"}
}},
{"$replaceRoot": {"newRoot": "$doc"}},
{"$sort": {"_id": -1}},
])
data = [doc async for doc in cursor]
data.sort(key=lambda doc: doc["_id"].generation_time, reverse=True)
return templates.TemplateResponse(
request=request,
name="profile/delivery_old.html",
context={
"user": user,
"data": data,
}
)
@app.get('/profile/delivery/exhibition/{exhibition_id}')
async def profile_delivery_by_exhibition(request: Request, exhibition_id:str, user: Dict = Security(get_current_user)):
cursor = delivery_collection.aggregate([
{"$match": {
"$or": [
{"sender._id": user.get('_id')},
{"user._id": user.get('_id')},
{"receiver._id": user.get('_id')}
],
'exhibition_id': exhibition_id,
'prod': PROD,
}},
{"$sort": {"_id": -1}},
{"$group": {
"_id": "$post_id",
"doc": {"$first": "$$ROOT"}
}},
{"$replaceRoot": {"newRoot": "$doc"}},
{"$sort": {"_id": -1}},
])
data = [doc async for doc in cursor]
data.sort(key=lambda doc: doc["_id"].generation_time, reverse=True)
return templates.TemplateResponse(
request=request,
name="profile/delivery_old.html",
context={
"user": user,
"data": data,
}
)
@app.get(
"/profile/delivery",
response_class=HTMLResponse,
summary="Личный кабинет → Доставка Картин",
description="""
Отображает страницу с историей доставок пользователя.
Показывает **все** доставки, где пользователь является либо отправителем (`sender`),
либо получателем (`receiver`).
""",
tags=["Profile", "Delivery"],
)
async def profile_delivery(
request: Request,
user: Dict = Security(get_current_user)
):
"""
Рендерит HTML-страницу с доставками текущего пользователя.
Запрашивает все документы из коллекции `delivery`, где пользователь
участвует как отправитель или получатель.
Args:
request: FastAPI Request (для шаблонизатора)
user: Текущий авторизованный пользователь (из Security)
Returns:
HTMLResponse с шаблоном profile/delivery.html
"""
# Получаем все доставки пользователя (отправитель ИЛИ получатель)
cursor = delivery_collection.aggregate([
{"$match": {
"$or": [
{"sender._id": user.get('_id')},
{"user._id": user.get('_id')},
{"receiver._id": user.get('_id')}
],
'prod': PROD,
}},
{"$sort": {"_id": -1}}, # why not work
{"$group": {
"_id": "$post_id",
"doc": {"$first": "$$ROOT"}
}},
{"$replaceRoot": {"newRoot": "$doc"}},
{"$sort": {"_id": -1}},
])
data = [doc async for doc in cursor]
exhibitions = await exhibition_collection.find({'_id': {'$in': [ObjectId(i.get('exhibition_id')) for i in data if i.get('exhibition_id')]}}).to_list()
competitions = await competitions_col.find({'_id': {'$in': [i.get('competition_id') for i in data if i.get('competition_id')]}}).to_list()
return templates.TemplateResponse(
request=request,
name="profile/delivery.html",
context={
"user": user,
"exhibitions": exhibitions,
"competitions": competitions,
"data": data,
"now": datetime.now(),
"page_title": "Доставка",
"active_section": "delivery"
}
)
@app.post("/api/exhibition/send/invitations")
async def api_exhibition_send_invitations(
background_tasks: BackgroundTasks,
exhibition_id: Optional[str] = Form(...),
user: Dict = Security(get_current_user)
):
background_tasks.add_task(exhibition_send_invitations, exhibition_id)
logging.info(f"send exh inv {exhibition_id}")
return {"ok": True}
@app.post("/api/exhibition/post/select")
async def api_exhibition_post_select(
post_id: Optional[str] = Form(...),
exhibition_id: Optional[str] = Form(...),
user: Dict = Security(get_current_user)
):
await exhibition_collection.update_one(
{
"_id": ObjectId(exhibition_id),
"user._id": user.get("_id")
},
[
{
"$set": {
"posts": {
"$cond": [
{"$in": [post_id, {"$ifNull": ["$posts", []]}]},
{"$setDifference": ["$posts", [post_id]]},
{"$setUnion": ["$posts", [post_id]]}
]
}
}
}
]
)
return {"ok": True}
@app.get("/profile/partner/exhibition/{exhibition_id}/catalog", response_class=HTMLResponse)
async def profile_partner_exhibitions_catalog(
request: Request,
exhibition_id: str,
user: Dict = Security(get_current_user)
):
exhibition = await exhibition_collection.find_one({'_id': ObjectId(exhibition_id)})
cursor = posts_collection.aggregate([
{
"$match": {
"city": {"$nin": [None, ""]}
}
},
{ '$group': { '_id': "$city", 'count': { '$sum': 1 } } },
{ '$sort': { '_id': 1 } }
])
living_places = [doc async for doc in cursor]
cursor = posts_collection.aggregate([
{
"$match": {
"genre": {"$nin": [None, ""]}
}
},
{
"$project": {
"genres": {"$split": ["$genre", ","]}
}
},
{"$unwind": "$genres"},
{
"$match": {
"genres": {"$nin": ["", None]}
}
},
{
"$group": {
"_id": "$genres",
"count": {"$sum": 1}
}
},
{"$sort": {"_id": 1}}
])
genres = [{'_id': doc["_id"].strip(), 'count': doc["count"]} async for doc in cursor]
users = await users_collection.find(
{"roles": {"$in": ["partner", "galerist"]}},
{"username": 1, "_id": 0}
).to_list()
usernames = [u["username"] for u in users]
pipeline = [
{"$match": {"user.roles": {"$in": ["partner", "galerist"]}}},
{"$group": {"_id": "$user.username", "count": {"$sum": 1}}},
]
cursor = posts_collection.aggregate(pipeline)
post_counts = await cursor.to_list(length=None)
counts_by_username = {d["_id"]: d["count"] for d in post_counts}
galerists = [{"_id": uname, "count": counts_by_username.get(uname, '')} for uname in usernames]
posts = [doc async for doc in posts_collection.find().sort('_id', -1).limit(32)]
return templates.TemplateResponse(
request=request, name="profile/partner/exhibition catalog.html", context={"now": datetime.now(), "user": user, "galerists": galerists, "genres": genres, "living_places": living_places, "exhibition": exhibition, "posts": posts}
)
@app.get("/profile/partner/exhibition/{exhibition_id}", response_class=HTMLResponse)
async def profile_partner_exhibitions(
request: Request,
exhibition_id: str,
user: Dict = Security(get_current_user)
):
exhibition = await exhibition_collection.find_one({'_id': ObjectId(exhibition_id)})
obj_ids = [ObjectId(x) for x in exhibition.get('posts', [])]
posts = await posts_collection.find({"_id": {"$in": obj_ids}}).to_list() if obj_ids else []
d_posts = await delivery_collection.find({'post_id':{'$in':[str(i.get('_id')) for i in posts]}}).to_list()
for post in posts:
for doc in d_posts:
if str(post.get('_id')) == doc.get('post_id') and doc.get('receiver') and doc.get('receiver').get('_id') == str(user.get('_id')):
post.update({'status':doc.get('status'), 'receiver': doc.get('receiver'), 'd_id':doc.get('_id')})
return templates.TemplateResponse(
request=request, name="profile/partner/exhibition.html", context={"now": datetime.now(), "user": user, "exhibition": exhibition, "posts": posts}
)
@app.get("/profile/partner/exhibitions", response_class=HTMLResponse)
async def profile_exhibitions(
request: Request,
user: Dict = Security(get_current_user)
):
data = await exhibition_collection.find({'user._id': user.get('_id')}).sort('_id', -1).to_list()
return templates.TemplateResponse(
request=request, name="profile/partner/exhibitions.html", context={"now": datetime.now(), "user": user, "data": data}
)