From 328a20a7e31700b2013f283baf79086967dfdc8a Mon Sep 17 00:00:00 2001 From: IgorVolochay Date: Fri, 21 Aug 2026 12:11:03 +0300 Subject: [PATCH] Add moderation system --- app/main.py | 67 +++++++++++++++- app/mongo_worker.py | 21 +++++ app/rabbit_worker.py | 70 +++++++++++++++++ app/requirements.txt | 6 +- app/tg_bot.py | 178 +++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 340 insertions(+), 2 deletions(-) create mode 100644 app/rabbit_worker.py create mode 100644 app/tg_bot.py diff --git a/app/main.py b/app/main.py index cd19013..e6d1e30 100644 --- a/app/main.py +++ b/app/main.py @@ -1,17 +1,24 @@ import os +import secrets +import logging import uvicorn import asyncio from dotenv import load_dotenv -from fastapi import FastAPI, Depends, Response, status +from fastapi import FastAPI, Depends, Response, Header, HTTPException, status +from guard import SecurityMiddleware, SecurityConfig from schemas.api_schemas import BaseResponse, AddUserBody, AddCardBody, SelectChoice, ReactionCard, AddCommentBody from schemas.base_schemas import Card from mongo_worker import MongoWorker +from rabbit_worker import RabbitWorker from tools.base_moderation import moderate_text +logger = logging.getLogger(__name__) + + load_dotenv() disable_docs = os.getenv("DISABLE_DOCS", "true").lower() == "true" @@ -24,7 +31,34 @@ app: FastAPI = FastAPI( redoc_url=None if disable_docs else "/redoc", openapi_url=None if disable_docs else "/openapi.json", ) +config = SecurityConfig( + enable_rate_limiting=True, + rate_limit=120, + rate_limit_window=60, + enable_redis=False, + enable_ip_banning=True, + auto_ban_threshold=3, + auto_ban_duration=3600, + custom_log_file="security.log", +) + +app.add_middleware(SecurityMiddleware, config=config) mongo_worker = MongoWorker() +rabbit_worker = RabbitWorker() + +MODERATION_SECRET = os.getenv("MODERATION_SECRET", "change-me-in-production") + + +async def verify_moderation_secret( + x_moderation_secret: str = Header(..., alias="X-Moderation-Secret"), +) -> str: + """Проверяет секретный ключ модерации в заголовке запроса.""" + if not secrets.compare_digest(x_moderation_secret, MODERATION_SECRET): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Invalid moderation secret", + ) + return x_moderation_secret @app.on_event("startup") @@ -111,11 +145,42 @@ async def add_card( mongo: MongoWorker = Depends(lambda: mongo_worker),) -> BaseResponse: if moderate_text(new_card.choice_A) and moderate_text(new_card.choice_B): card = await mongo.add_card_by_api(new_card.choice_A, new_card.choice_B, new_card.author_id) + + # Отправляем карточку в RabbitMQ на ручную модерацию админом + try: + await rabbit_worker.send_to_moderation(card) + except Exception as exc: + logger.error("Failed to send card %s to moderation queue: %s", card.card_id, exc) + return BaseResponse(result=card) response.status_code = status.HTTP_400_BAD_REQUEST return BaseResponse(result="Card has not passed base moderation", error=True) +@app.patch("/card_accept", status_code=200, dependencies=[Depends(verify_moderation_secret)]) +async def card_accept( + card_id: int, + response: Response, + mongo: MongoWorker = Depends(lambda: mongo_worker),) -> BaseResponse: + """Принимает карточку — доступ только с секретным ключом.""" + result = await mongo.accept_card(card_id) + if result.error: + response.status_code = status.HTTP_404_NOT_FOUND + return result + + +@app.patch("/card_reject", status_code=200, dependencies=[Depends(verify_moderation_secret)]) +async def card_reject( + card_id: int, + response: Response, + mongo: MongoWorker = Depends(lambda: mongo_worker),) -> BaseResponse: + """Отклоняет карточку — доступ только с секретным ключом.""" + result = await mongo.reject_card(card_id) + if result.error: + response.status_code = status.HTTP_404_NOT_FOUND + return result + + @app.patch("/select_choice", status_code=200) async def select_choice( choice_data: SelectChoice, diff --git a/app/mongo_worker.py b/app/mongo_worker.py index 2e2ec7d..5a17d2a 100644 --- a/app/mongo_worker.py +++ b/app/mongo_worker.py @@ -141,6 +141,27 @@ class MongoWorker: logger.error("Failed to insert card: %s", exc) raise + async def accept_card(self, card_id: int) -> BaseResponse: + """Принимает карточку: ставит active_status=True и moderation_date=сейчас.""" + result = await self.game_data.find_one_and_update( + {"card_id": card_id}, + {"$set": { + "active_status": True, + "moderation_date": datetime.now().isoformat(), + }}, + return_document=ReturnDocument.AFTER, + ) + if not result: + return BaseResponse(result="Card doesn't exist", error=True) + return BaseResponse(result=Card.model_validate(result)) + + async def reject_card(self, card_id: int) -> BaseResponse: + """Отклоняет карточку: удаляет её из БД.""" + result = await self.game_data.delete_one({"card_id": card_id}) + if result.deleted_count == 0: + return BaseResponse(result="Card doesn't exist", error=True) + return BaseResponse(result=f"Card {card_id} rejected and deleted") + async def select_choice(self, card_id: int, choice: str) -> BaseResponse: if choice == "A": count_field = "count_choice_A" diff --git a/app/rabbit_worker.py b/app/rabbit_worker.py new file mode 100644 index 0000000..319069b --- /dev/null +++ b/app/rabbit_worker.py @@ -0,0 +1,70 @@ +import os +import json +import asyncio +import logging +from typing import Callable, Awaitable + +import aio_pika +from aio_pika.abc import AbstractIncomingMessage +from dotenv import load_dotenv + +from schemas.base_schemas import Card + + +logger = logging.getLogger(__name__) + + +class RabbitWorker: + def __init__(self): + load_dotenv() + self.url = ( + f"amqp://{os.getenv('RABBIT_USER')}:{os.getenv('RABBIT_PASS')}" + f"@{os.getenv('RABBIT_HOST')}:{os.getenv('RABBIT_PORT')}" + ) + + async def send_to_moderation(self, card: Card) -> None: + """Отправляет карточку в очередь модерации.""" + connection = await aio_pika.connect_robust(self.url) + async with connection: + channel = await connection.channel() + queue = await channel.declare_queue("moderation", durable=True) + await channel.default_exchange.publish( + aio_pika.Message( + body=card.model_dump_json().encode(), + delivery_mode=aio_pika.DeliveryMode.PERSISTENT, + ), + routing_key="moderation", + ) + logger.info("Card %s sent to moderation queue", card.card_id) + + async def consume_moderation( + self, + callback: Callable[[Card], Awaitable[None]], + ) -> None: + """Бесконечно слушает очередь модерации и вызывает callback для каждой карточки.""" + connection = await aio_pika.connect_robust(self.url) + async with connection: + channel = await connection.channel() + await channel.set_qos(prefetch_count=1) + queue = await channel.declare_queue("moderation", durable=True) + + logger.info("Started consuming moderation queue...") + + async def on_message(message: AbstractIncomingMessage) -> None: + async with message.process(): + try: + card_data = json.loads(message.body.decode()) + card = Card.model_validate(card_data) + await callback(card) + except Exception as exc: + logger.error("Error processing moderation message: %s", exc) + + await queue.consume(on_message) + + # Держим consumer живым, но позволяем отмену (Ctrl+C) + stop_event = asyncio.Event() + try: + await stop_event.wait() + except asyncio.CancelledError: + logger.info("Moderation consumer shutting down...") + raise \ No newline at end of file diff --git a/app/requirements.txt b/app/requirements.txt index 50718d4..a146643 100644 --- a/app/requirements.txt +++ b/app/requirements.txt @@ -1,4 +1,8 @@ fastapi==0.115.7 +fastapi_guard==7.6.0 motor==3.7.0 python-dotenv==1.0.1 -uvicorn==0.34.0 \ No newline at end of file +uvicorn==0.34.0 +aio-pika==10.0.1 +aiogram==3.18.0 +aiohttp==3.11.18 \ No newline at end of file diff --git a/app/tg_bot.py b/app/tg_bot.py new file mode 100644 index 0000000..9f1d37e --- /dev/null +++ b/app/tg_bot.py @@ -0,0 +1,178 @@ +""" +Telegram-бот модерации карточек. + +Слушает очередь RabbitMQ «moderation» и отправляет карточки +в чат администратору с inline-кнопками «Принять ✅» / «Отклонить ❌». + +При нажатии кнопки бот вызывает защищённые эндпоинты +/card_accept или /card_reject с секретным заголовком. +""" + +import os +import json +import asyncio +import logging +from datetime import datetime + +import aiohttp +from dotenv import load_dotenv +from aiogram import Bot, Dispatcher, F +from aiogram.types import CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup + +from schemas.base_schemas import Card +from rabbit_worker import RabbitWorker + + +load_dotenv() +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +# ── Конфигурация ────────────────────────────────────────────── +BOT_TOKEN = os.getenv("TG_BOT_TOKEN") +ADMIN_CHAT_ID = int(os.getenv("TG_ADMIN_CHAT_ID", "0")) +API_BASE_URL = os.getenv("API_BASE_URL", "http://localhost:5000") +MODERATION_SECRET = os.getenv("MODERATION_SECRET", "change-me-in-production") + +bot = Bot(token=BOT_TOKEN) +dp = Dispatcher() +rabbit = RabbitWorker() + + +# ── Отправка карточки администратору ────────────────────────── +async def _get_author_username(author_id: int) -> str: + """Запрашивает username автора через API.""" + try: + async with aiohttp.ClientSession() as session: + async with session.get( + f"{API_BASE_URL}/get_user", params={"user_id": author_id} + ) as resp: + if resp.status == 200: + data = await resp.json() + username = data.get("result", {}).get("username", "") + if username: + return f"@{username}" + except Exception as exc: + logger.warning("Failed to fetch username for %s: %s", author_id, exc) + return str(author_id) + + +def _format_date(iso_date: str) -> str: + """Преобразует ISO-дату в формат ДД.ММ.ГГГГ ЧЧ:ММ:СС.""" + try: + dt = datetime.fromisoformat(iso_date) + return dt.strftime("%d.%m.%Y %H:%M:%S") + except (ValueError, TypeError): + return iso_date + + +async def send_card_to_admin(card: Card) -> None: + """Формирует сообщение и inline-клавиатуру для карточки.""" + author_display = await _get_author_username(card.author_id) + date_display = _format_date(card.creation_date) + + text = ( + f"🆕 Новая карточка #{card.card_id}\n\n" + f"🅰️ {card.choice_A}\n" + f"🅱️ {card.choice_B}\n\n" + f"👤 Автор: {author_display}\n" + f"📅 Создана: {date_display}" + ) + keyboard = InlineKeyboardMarkup( + inline_keyboard=[ + [ + InlineKeyboardButton( + text="Принять ✅", + callback_data=f"accept:{card.card_id}", + ), + InlineKeyboardButton( + text="Отклонить ❌", + callback_data=f"reject:{card.card_id}", + ), + ] + ] + ) + await bot.send_message( + chat_id=ADMIN_CHAT_ID, + text=text, + reply_markup=keyboard, + parse_mode="HTML", + ) + logger.info("Sent card %s to admin chat", card.card_id) + + +# ── Вызов защищённых эндпоинтов API ────────────────────────── +async def call_moderation_api(action: str, card_id: int) -> dict: + """ + Вызывает /card_accept или /card_reject с секретным заголовком. + action: 'accept' | 'reject' + """ + endpoint = f"{API_BASE_URL}/card_{action}" + headers = {"X-Moderation-Secret": MODERATION_SECRET} + params = {"card_id": card_id} + + async with aiohttp.ClientSession() as session: + async with session.patch(endpoint, headers=headers, params=params) as resp: + data = await resp.json() + return data + + +# ── Обработчики callback-кнопок ─────────────────────────────── +@dp.callback_query(F.data.startswith("accept:")) +async def on_accept(callback: CallbackQuery) -> None: + card_id = int(callback.data.split(":")[1]) + result = await call_moderation_api("accept", card_id) + + if result.get("error"): + await callback.answer(f"Ошибка: {result['result']}", show_alert=True) + return + + await callback.message.edit_text( + callback.message.text + "\n\n✅ ПРИНЯТА", + parse_mode="HTML", + ) + await callback.answer("Карточка принята!") + logger.info("Card %s accepted by admin", card_id) + + +@dp.callback_query(F.data.startswith("reject:")) +async def on_reject(callback: CallbackQuery) -> None: + card_id = int(callback.data.split(":")[1]) + result = await call_moderation_api("reject", card_id) + + if result.get("error"): + await callback.answer(f"Ошибка: {result['result']}", show_alert=True) + return + + await callback.message.edit_text( + callback.message.text + "\n\n❌ ОТКЛОНЕНА", + parse_mode="HTML", + ) + await callback.answer("Карточка отклонена!") + logger.info("Card %s rejected by admin", card_id) + +# ── Lifecycle-хуки aiogram ──────────────────────────────────── +_rabbit_task: asyncio.Task | None = None + + +@dp.startup() +async def on_startup() -> None: + global _rabbit_task + _rabbit_task = asyncio.create_task( + rabbit.consume_moderation(send_card_to_admin) + ) + logger.info("Moderation bot started, RabbitMQ consumer running") + + +@dp.shutdown() +async def on_shutdown() -> None: + if _rabbit_task: + _rabbit_task.cancel() + try: + await _rabbit_task + except asyncio.CancelledError: + pass + logger.info("Moderation bot stopped") + + +if __name__ == "__main__": + dp.run_polling(bot)