From 1a476c49c19d1ec2ab2cda1c2ffb5fd242288bb6 Mon Sep 17 00:00:00 2001 From: Fringg Date: Thu, 12 Feb 2026 19:13:51 +0300 Subject: [PATCH] feat: add cabinet admin API for pinned messages management - Full CRUD + broadcast/unpin/activate/deactivate endpoints - Admin auth required on all endpoints (get_current_admin_user) - Broadcast cooldown (60s) on all mass operation endpoints - Cached Bot singleton to prevent aiohttp session leaks - Guard against deleting active pinned messages (409 Conflict) - Route ordering: /active/* before /{message_id}/* to prevent path conflicts - Pydantic schemas with proper validation (file_id max_length=255) --- app/cabinet/routes/__init__.py | 2 + app/cabinet/routes/admin_pinned_messages.py | 397 ++++++++++++++++++++ app/cabinet/schemas/pinned_messages.py | 64 ++++ 3 files changed, 463 insertions(+) create mode 100644 app/cabinet/routes/admin_pinned_messages.py create mode 100644 app/cabinet/schemas/pinned_messages.py diff --git a/app/cabinet/routes/__init__.py b/app/cabinet/routes/__init__.py index a6bf5364..c879d128 100644 --- a/app/cabinet/routes/__init__.py +++ b/app/cabinet/routes/__init__.py @@ -9,6 +9,7 @@ from .admin_campaigns import router as admin_campaigns_router from .admin_email_templates import router as admin_email_templates_router from .admin_payment_methods import router as admin_payment_methods_router from .admin_payments import router as admin_payments_router +from .admin_pinned_messages import router as admin_pinned_messages_router from .admin_promo_offers import router as admin_promo_offers_router from .admin_promocodes import promo_groups_router as admin_promo_groups_router, router as admin_promocodes_router from .admin_remnawave import router as admin_remnawave_router @@ -89,6 +90,7 @@ router.include_router(admin_remnawave_router) router.include_router(admin_email_templates_router) router.include_router(admin_updates_router) router.include_router(admin_traffic_router) +router.include_router(admin_pinned_messages_router) # WebSocket route router.include_router(websocket_router) diff --git a/app/cabinet/routes/admin_pinned_messages.py b/app/cabinet/routes/admin_pinned_messages.py new file mode 100644 index 00000000..46776142 --- /dev/null +++ b/app/cabinet/routes/admin_pinned_messages.py @@ -0,0 +1,397 @@ +"""Admin routes for pinned messages in cabinet.""" + +import logging +import time +from datetime import datetime + +from aiogram import Bot +from aiogram.client.default import DefaultBotProperties +from aiogram.enums import ParseMode +from fastapi import APIRouter, Depends, HTTPException, Query, status +from sqlalchemy import func, select, update +from sqlalchemy.ext.asyncio import AsyncSession + +from app.config import settings +from app.database.models import PinnedMessage, User +from app.services.pinned_message_service import ( + broadcast_pinned_message, + deactivate_active_pinned_message, + get_active_pinned_message, + set_active_pinned_message, + unpin_active_pinned_message, +) +from app.utils.validators import sanitize_html, validate_html_tags + +from ..dependencies import get_cabinet_db, get_current_admin_user +from ..schemas.pinned_messages import ( + PinnedMessageBroadcastResponse, + PinnedMessageCreateRequest, + PinnedMessageListResponse, + PinnedMessageResponse, + PinnedMessageSettingsRequest, + PinnedMessageUnpinResponse, + PinnedMessageUpdateRequest, +) + + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix='/admin/pinned-messages', tags=['Cabinet Admin Pinned Messages']) + +# Broadcast cooldown: min 60 seconds between mass operations +_BROADCAST_COOLDOWN_SECONDS = 60 +_last_broadcast_time: float = 0.0 + + +def _check_broadcast_cooldown() -> None: + global _last_broadcast_time + now = time.monotonic() + elapsed = now - _last_broadcast_time + if _last_broadcast_time > 0 and elapsed < _BROADCAST_COOLDOWN_SECONDS: + remaining = int(_BROADCAST_COOLDOWN_SECONDS - elapsed) + raise HTTPException( + status.HTTP_429_TOO_MANY_REQUESTS, + f'Broadcast cooldown active. Try again in {remaining} seconds.', + ) + _last_broadcast_time = now + + +def _serialize_pinned_message(msg: PinnedMessage) -> PinnedMessageResponse: + return PinnedMessageResponse( + id=msg.id, + content=msg.content, + media_type=msg.media_type, + media_file_id=msg.media_file_id, + send_before_menu=msg.send_before_menu, + send_on_every_start=msg.send_on_every_start, + is_active=msg.is_active, + created_by=msg.created_by, + created_at=msg.created_at, + updated_at=msg.updated_at, + ) + + +_cached_bot: Bot | None = None + + +def _get_bot() -> Bot: + global _cached_bot + if _cached_bot is None: + _cached_bot = Bot( + token=settings.BOT_TOKEN, + default=DefaultBotProperties(parse_mode=ParseMode.HTML), + ) + return _cached_bot + + +# ============ List / Get Endpoints ============ + + +@router.get('', response_model=PinnedMessageListResponse) +async def list_pinned_messages( + admin: User = Depends(get_current_admin_user), + db: AsyncSession = Depends(get_cabinet_db), + limit: int = Query(20, ge=1, le=100), + offset: int = Query(0, ge=0), + active_only: bool = Query(False), +) -> PinnedMessageListResponse: + """Get list of pinned messages with pagination.""" + query = select(PinnedMessage).order_by(PinnedMessage.created_at.desc()) + count_query = select(func.count(PinnedMessage.id)) + + if active_only: + query = query.where(PinnedMessage.is_active.is_(True)) + count_query = count_query.where(PinnedMessage.is_active.is_(True)) + + total = await db.scalar(count_query) or 0 + result = await db.execute(query.offset(offset).limit(limit)) + items = result.scalars().all() + + return PinnedMessageListResponse( + items=[_serialize_pinned_message(msg) for msg in items], + total=int(total), + limit=limit, + offset=offset, + ) + + +@router.get('/active', response_model=PinnedMessageResponse | None) +async def get_active_message( + admin: User = Depends(get_current_admin_user), + db: AsyncSession = Depends(get_cabinet_db), +) -> PinnedMessageResponse | None: + """Get current active pinned message.""" + msg = await get_active_pinned_message(db) + if not msg: + return None + return _serialize_pinned_message(msg) + + +@router.get('/{message_id}', response_model=PinnedMessageResponse) +async def get_pinned_message( + message_id: int, + admin: User = Depends(get_current_admin_user), + db: AsyncSession = Depends(get_cabinet_db), +) -> PinnedMessageResponse: + """Get pinned message by ID.""" + result = await db.execute(select(PinnedMessage).where(PinnedMessage.id == message_id)) + msg = result.scalar_one_or_none() + if not msg: + raise HTTPException(status.HTTP_404_NOT_FOUND, 'Pinned message not found') + return _serialize_pinned_message(msg) + + +# ============ Create / Update Endpoints ============ + + +@router.post('', response_model=PinnedMessageBroadcastResponse, status_code=status.HTTP_201_CREATED) +async def create_pinned_message( + payload: PinnedMessageCreateRequest, + admin: User = Depends(get_current_admin_user), + db: AsyncSession = Depends(get_cabinet_db), +) -> PinnedMessageBroadcastResponse: + """ + Create a new pinned message. + + Automatically deactivates previous active message. + If broadcast=true, sends to all active users immediately. + """ + # Проверяем cooldown ДО мутации в БД + if payload.broadcast: + _check_broadcast_cooldown() + + content = payload.content.strip() + if not content and not payload.media: + raise HTTPException(status.HTTP_400_BAD_REQUEST, 'Either content or media must be provided') + + media_type = payload.media.type if payload.media else None + media_file_id = payload.media.file_id if payload.media else None + + try: + msg = await set_active_pinned_message( + db=db, + content=content, + created_by=admin.id, + media_type=media_type, + media_file_id=media_file_id, + send_before_menu=payload.send_before_menu, + send_on_every_start=payload.send_on_every_start, + ) + except ValueError as e: + raise HTTPException(status.HTTP_400_BAD_REQUEST, str(e)) + + sent_count = 0 + failed_count = 0 + + if payload.broadcast: + sent_count, failed_count = await broadcast_pinned_message(_get_bot(), db, msg) + + logger.info(f'Admin {admin.id} created pinned message #{msg.id} (broadcast={payload.broadcast})') + + return PinnedMessageBroadcastResponse( + message=_serialize_pinned_message(msg), + sent_count=sent_count, + failed_count=failed_count, + ) + + +@router.patch('/{message_id}', response_model=PinnedMessageResponse) +async def update_pinned_message( + message_id: int, + payload: PinnedMessageUpdateRequest, + admin: User = Depends(get_current_admin_user), + db: AsyncSession = Depends(get_cabinet_db), +) -> PinnedMessageResponse: + """Update a pinned message content, media, or settings.""" + result = await db.execute(select(PinnedMessage).where(PinnedMessage.id == message_id)) + msg = result.scalar_one_or_none() + if not msg: + raise HTTPException(status.HTTP_404_NOT_FOUND, 'Pinned message not found') + + if payload.content is not None: + sanitized = sanitize_html(payload.content) + is_valid, error = validate_html_tags(sanitized) + if not is_valid: + raise HTTPException(status.HTTP_400_BAD_REQUEST, error) + msg.content = sanitized + + if payload.media is not None: + msg.media_type = payload.media.type + msg.media_file_id = payload.media.file_id + + if payload.send_before_menu is not None: + msg.send_before_menu = payload.send_before_menu + + if payload.send_on_every_start is not None: + msg.send_on_every_start = payload.send_on_every_start + + msg.updated_at = datetime.utcnow() + await db.commit() + await db.refresh(msg) + + logger.info(f'Admin {admin.id} updated pinned message #{message_id}') + + return _serialize_pinned_message(msg) + + +@router.patch('/{message_id}/settings', response_model=PinnedMessageResponse) +async def update_pinned_message_settings( + message_id: int, + payload: PinnedMessageSettingsRequest, + admin: User = Depends(get_current_admin_user), + db: AsyncSession = Depends(get_cabinet_db), +) -> PinnedMessageResponse: + """Update only pinned message display settings.""" + result = await db.execute(select(PinnedMessage).where(PinnedMessage.id == message_id)) + msg = result.scalar_one_or_none() + if not msg: + raise HTTPException(status.HTTP_404_NOT_FOUND, 'Pinned message not found') + + if payload.send_before_menu is not None: + msg.send_before_menu = payload.send_before_menu + + if payload.send_on_every_start is not None: + msg.send_on_every_start = payload.send_on_every_start + + msg.updated_at = datetime.utcnow() + await db.commit() + await db.refresh(msg) + + return _serialize_pinned_message(msg) + + +# ============ Active Message Actions (before /{message_id} POST routes) ============ + + +@router.post('/active/deactivate', response_model=PinnedMessageResponse | None) +async def deactivate_active_message( + admin: User = Depends(get_current_admin_user), + db: AsyncSession = Depends(get_cabinet_db), +) -> PinnedMessageResponse | None: + """Deactivate the current active pinned message without unpinning from users.""" + msg = await deactivate_active_pinned_message(db) + if not msg: + return None + + logger.info(f'Admin {admin.id} deactivated pinned message #{msg.id}') + + return _serialize_pinned_message(msg) + + +@router.post('/active/unpin', response_model=PinnedMessageUnpinResponse) +async def unpin_active_message( + admin: User = Depends(get_current_admin_user), + db: AsyncSession = Depends(get_cabinet_db), +) -> PinnedMessageUnpinResponse: + """Unpin messages from all users and deactivate the active pinned message.""" + _check_broadcast_cooldown() + unpinned_count, failed_count, was_active = await unpin_active_pinned_message(_get_bot(), db) + + if was_active: + logger.info(f'Admin {admin.id} unpinned active message: unpinned={unpinned_count}, failed={failed_count}') + + return PinnedMessageUnpinResponse( + unpinned_count=unpinned_count, + failed_count=failed_count, + was_active=was_active, + ) + + +# ============ Per-Message Actions ============ + + +@router.post('/{message_id}/activate', response_model=PinnedMessageBroadcastResponse) +async def activate_pinned_message( + message_id: int, + broadcast: bool = Query(False), + admin: User = Depends(get_current_admin_user), + db: AsyncSession = Depends(get_cabinet_db), +) -> PinnedMessageBroadcastResponse: + """ + Activate a pinned message. + + Deactivates the current active message and activates the specified one. + If broadcast=true, sends to all active users immediately. + """ + # Проверяем cooldown ДО мутации в БД + if broadcast: + _check_broadcast_cooldown() + + result = await db.execute(select(PinnedMessage).where(PinnedMessage.id == message_id)) + msg = result.scalar_one_or_none() + if not msg: + raise HTTPException(status.HTTP_404_NOT_FOUND, 'Pinned message not found') + + await db.execute( + update(PinnedMessage) + .where(PinnedMessage.is_active.is_(True)) + .values(is_active=False, updated_at=datetime.utcnow()) + ) + + msg.is_active = True + msg.updated_at = datetime.utcnow() + await db.commit() + await db.refresh(msg) + + sent_count = 0 + failed_count = 0 + + if broadcast: + sent_count, failed_count = await broadcast_pinned_message(_get_bot(), db, msg) + + logger.info(f'Admin {admin.id} activated pinned message #{message_id} (broadcast={broadcast})') + + return PinnedMessageBroadcastResponse( + message=_serialize_pinned_message(msg), + sent_count=sent_count, + failed_count=failed_count, + ) + + +@router.post('/{message_id}/broadcast', response_model=PinnedMessageBroadcastResponse) +async def broadcast_message( + message_id: int, + admin: User = Depends(get_current_admin_user), + db: AsyncSession = Depends(get_cabinet_db), +) -> PinnedMessageBroadcastResponse: + """Broadcast a pinned message to all active users.""" + _check_broadcast_cooldown() + + result = await db.execute(select(PinnedMessage).where(PinnedMessage.id == message_id)) + msg = result.scalar_one_or_none() + if not msg: + raise HTTPException(status.HTTP_404_NOT_FOUND, 'Pinned message not found') + + sent_count, failed_count = await broadcast_pinned_message(_get_bot(), db, msg) + + logger.info(f'Admin {admin.id} broadcast pinned message #{message_id}: sent={sent_count}, failed={failed_count}') + + return PinnedMessageBroadcastResponse( + message=_serialize_pinned_message(msg), + sent_count=sent_count, + failed_count=failed_count, + ) + + +@router.delete('/{message_id}', status_code=status.HTTP_204_NO_CONTENT, response_model=None) +async def delete_pinned_message( + message_id: int, + admin: User = Depends(get_current_admin_user), + db: AsyncSession = Depends(get_cabinet_db), +) -> None: + """Delete a pinned message. Active messages must be deactivated first.""" + result = await db.execute(select(PinnedMessage).where(PinnedMessage.id == message_id)) + msg = result.scalar_one_or_none() + if not msg: + raise HTTPException(status.HTTP_404_NOT_FOUND, 'Pinned message not found') + + if msg.is_active: + raise HTTPException( + status.HTTP_409_CONFLICT, + 'Cannot delete active pinned message. Deactivate it first.', + ) + + await db.delete(msg) + await db.commit() + + logger.info(f'Admin {admin.id} deleted pinned message #{message_id}') diff --git a/app/cabinet/schemas/pinned_messages.py b/app/cabinet/schemas/pinned_messages.py new file mode 100644 index 00000000..a34f23ad --- /dev/null +++ b/app/cabinet/schemas/pinned_messages.py @@ -0,0 +1,64 @@ +"""Pydantic schemas for cabinet pinned messages.""" + +from __future__ import annotations + +from datetime import datetime + +from pydantic import BaseModel, Field + + +class PinnedMessageMedia(BaseModel): + type: str = Field(pattern=r'^(photo|video)$') + file_id: str = Field(..., min_length=1, max_length=255) + + +class PinnedMessageCreateRequest(BaseModel): + content: str = Field(..., min_length=1, max_length=4000) + media: PinnedMessageMedia | None = None + send_before_menu: bool = True + send_on_every_start: bool = True + broadcast: bool = False + + +class PinnedMessageUpdateRequest(BaseModel): + content: str | None = Field(None, max_length=4000) + send_before_menu: bool | None = None + send_on_every_start: bool | None = None + media: PinnedMessageMedia | None = None + + +class PinnedMessageSettingsRequest(BaseModel): + send_before_menu: bool | None = None + send_on_every_start: bool | None = None + + +class PinnedMessageResponse(BaseModel): + id: int + content: str | None + media_type: str | None = None + media_file_id: str | None = None + send_before_menu: bool + send_on_every_start: bool + is_active: bool + created_by: int | None = None + created_at: datetime + updated_at: datetime | None = None + + +class PinnedMessageBroadcastResponse(BaseModel): + message: PinnedMessageResponse + sent_count: int + failed_count: int + + +class PinnedMessageUnpinResponse(BaseModel): + unpinned_count: int + failed_count: int + was_active: bool + + +class PinnedMessageListResponse(BaseModel): + items: list[PinnedMessageResponse] + total: int + limit: int + offset: int