VPN/remnawave-bedolaga-telegram-bot
1
0
Fork 0
mirror of https://github.com/BEDOLAGA-DEV/remnawave-bedolaga-telegram-bot.git synced 2026-03-02 08:11:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Revert "Expose admin APIs for welcome texts and menu messages"

Browse Source
This commit is contained in:
Egor
2025-11-25 01:18:45 +03:00
committed by GitHub
parent 3881e7d1d5
commit 2d4b46afbd
8 changed files with 7 additions and 487 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
app/webapi/app.py
View File

@@ -20,8 +20,6 @@ from .routes import (
promocodes,
promo_groups,
promo_offers,
user_messages,
welcome_texts,
pages,
remnawave,
servers,
@@ -51,11 +49,7 @@ OPENAPI_TAGS = [
},
{
"name": "main-menu",
"description": "Управление кнопками и сообщениями главного меню Telegram-бота.",
},
{
"name": "welcome-texts",
"description": "Создание, редактирование и управление приветственными текстами.",
"description": "Управление кнопками главного меню Telegram-бота.",
},
{
"name": "users",
@@ -175,16 +169,6 @@ def create_web_api_app() -> FastAPI:
prefix="/main-menu/buttons",
tags=["main-menu"],
)
app.include_router(
user_messages.router,
prefix="/main-menu/messages",
tags=["main-menu"],
)
app.include_router(
welcome_texts.router,
prefix="/welcome-texts",
tags=["welcome-texts"],
)
app.include_router(pages.router, prefix="/pages", tags=["pages"])
app.include_router(promocodes.router, prefix="/promo-codes", tags=["promo-codes"])
app.include_router(broadcasts.router, prefix="/broadcasts", tags=["broadcasts"])

4
app/webapi/routes/__init__.py
View File

@@ -7,8 +7,6 @@ from . import (
partners,
polls,
promo_offers,
user_messages,
welcome_texts,
pages,
promo_groups,
servers,
@@ -32,8 +30,6 @@ __all__ = [
"partners",
"polls",
"promo_offers",
"user_messages",
"welcome_texts",
"pages",
"promo_groups",
"servers",

124
app/webapi/routes/user_messages.py
View File

@@ -1,124 +0,0 @@
from __future__ import annotations
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, Query, Response, Security, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.database.crud.user_message import (
create_user_message,
delete_user_message,
get_all_user_messages,
get_user_message_by_id,
get_user_messages_count,
toggle_user_message_status,
update_user_message,
)
from ..dependencies import get_db_session, require_api_token
from ..schemas.user_messages import (
UserMessageCreateRequest,
UserMessageListResponse,
UserMessageResponse,
UserMessageUpdateRequest,
)
router = APIRouter()
def _serialize(message) -> UserMessageResponse:
return UserMessageResponse(
id=message.id,
message_text=message.message_text,
is_active=message.is_active,
sort_order=message.sort_order,
created_by=message.created_by,
created_at=message.created_at,
updated_at=message.updated_at,
)
@router.get("", response_model=UserMessageListResponse)
async def list_user_messages(
_: Any = Security(require_api_token),
db: AsyncSession = Depends(get_db_session),
limit: int = Query(50, ge=1, le=200),
offset: int = Query(0, ge=0),
include_inactive: bool = Query(True, description="Включать неактивные сообщения"),
) -> UserMessageListResponse:
total = await get_user_messages_count(db, include_inactive=include_inactive)
messages = await get_all_user_messages(
db,
offset=offset,
limit=limit,
include_inactive=include_inactive,
)
return UserMessageListResponse(
items=[_serialize(message) for message in messages],
total=total,
limit=limit,
offset=offset,
)
@router.post("", response_model=UserMessageResponse, status_code=status.HTTP_201_CREATED)
async def create_user_message_endpoint(
payload: UserMessageCreateRequest,
token: Any = Security(require_api_token),
db: AsyncSession = Depends(get_db_session),
) -> UserMessageResponse:
created_by = getattr(token, "id", None)
message = await create_user_message(
db,
message_text=payload.message_text,
created_by=created_by,
is_active=payload.is_active,
sort_order=payload.sort_order,
)
return _serialize(message)
@router.patch("/{message_id}", response_model=UserMessageResponse)
async def update_user_message_endpoint(
message_id: int,
payload: UserMessageUpdateRequest,
_: Any = Security(require_api_token),
db: AsyncSession = Depends(get_db_session),
) -> UserMessageResponse:
update_payload = payload.dict(exclude_unset=True)
message = await update_user_message(db, message_id, **update_payload)
if not message:
raise HTTPException(status.HTTP_404_NOT_FOUND, "User message not found")
return _serialize(message)
@router.post("/{message_id}/toggle", response_model=UserMessageResponse)
async def toggle_user_message_endpoint(
message_id: int,
_: Any = Security(require_api_token),
db: AsyncSession = Depends(get_db_session),
) -> UserMessageResponse:
message = await toggle_user_message_status(db, message_id)
if not message:
raise HTTPException(status.HTTP_404_NOT_FOUND, "User message not found")
return _serialize(message)
@router.delete("/{message_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user_message_endpoint(
message_id: int,
_: Any = Security(require_api_token),
db: AsyncSession = Depends(get_db_session),
) -> Response:
message = await get_user_message_by_id(db, message_id)
if not message:
raise HTTPException(status.HTTP_404_NOT_FOUND, "User message not found")
await delete_user_message(db, message_id)
return Response(status_code=status.HTTP_204_NO_CONTENT)

122
app/webapi/routes/welcome_texts.py
View File

@@ -1,122 +0,0 @@
from __future__ import annotations
from typing import Any
from fastapi import APIRouter, Depends, HTTPException, Query, Response, Security, status
from sqlalchemy.ext.asyncio import AsyncSession
from app.database.crud.welcome_text import (
count_welcome_texts,
create_welcome_text,
delete_welcome_text,
get_welcome_text_by_id,
list_welcome_texts,
update_welcome_text,
)
from ..dependencies import get_db_session, require_api_token
from ..schemas.welcome_texts import (
WelcomeTextCreateRequest,
WelcomeTextListResponse,
WelcomeTextResponse,
WelcomeTextUpdateRequest,
)
router = APIRouter()
def _serialize(text) -> WelcomeTextResponse:
return WelcomeTextResponse(
id=text.id,
text=text.text_content,
is_active=text.is_active,
is_enabled=text.is_enabled,
created_by=text.created_by,
created_at=text.created_at,
updated_at=text.updated_at,
)
@router.get("", response_model=WelcomeTextListResponse)
async def list_welcome_texts_endpoint(
_: Any = Security(require_api_token),
db: AsyncSession = Depends(get_db_session),
limit: int = Query(50, ge=1, le=200),
offset: int = Query(0, ge=0),
include_inactive: bool = Query(True, description="Включать неактивные тексты"),
) -> WelcomeTextListResponse:
total = await count_welcome_texts(db, include_inactive=include_inactive)
records = await list_welcome_texts(
db,
limit=limit,
offset=offset,
include_inactive=include_inactive,
)
return WelcomeTextListResponse(
items=[_serialize(item) for item in records],
total=total,
limit=limit,
offset=offset,
)
@router.post("", response_model=WelcomeTextResponse, status_code=status.HTTP_201_CREATED)
async def create_welcome_text_endpoint(
payload: WelcomeTextCreateRequest,
token: Any = Security(require_api_token),
db: AsyncSession = Depends(get_db_session),
) -> WelcomeTextResponse:
created_by = getattr(token, "id", None)
record = await create_welcome_text(
db,
text_content=payload.text,
created_by=created_by,
is_enabled=payload.is_enabled,
is_active=payload.is_active,
)
return _serialize(record)
@router.get("/{welcome_text_id}", response_model=WelcomeTextResponse)
async def get_welcome_text_endpoint(
welcome_text_id: int,
_: Any = Security(require_api_token),
db: AsyncSession = Depends(get_db_session),
) -> WelcomeTextResponse:
record = await get_welcome_text_by_id(db, welcome_text_id)
if not record:
raise HTTPException(status.HTTP_404_NOT_FOUND, "Welcome text not found")
return _serialize(record)
@router.patch("/{welcome_text_id}", response_model=WelcomeTextResponse)
async def update_welcome_text_endpoint(
welcome_text_id: int,
payload: WelcomeTextUpdateRequest,
_: Any = Security(require_api_token),
db: AsyncSession = Depends(get_db_session),
) -> WelcomeTextResponse:
record = await get_welcome_text_by_id(db, welcome_text_id)
if not record:
raise HTTPException(status.HTTP_404_NOT_FOUND, "Welcome text not found")
update_payload = payload.dict(exclude_unset=True)
updated = await update_welcome_text(db, record, **update_payload)
return _serialize(updated)
@router.delete("/{welcome_text_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_welcome_text_endpoint(
welcome_text_id: int,
_: Any = Security(require_api_token),
db: AsyncSession = Depends(get_db_session),
) -> Response:
record = await get_welcome_text_by_id(db, welcome_text_id)
if not record:
raise HTTPException(status.HTTP_404_NOT_FOUND, "Welcome text not found")
await delete_welcome_text(db, record)
return Response(status_code=status.HTTP_204_NO_CONTENT)

50
app/webapi/schemas/user_messages.py
View File

@@ -1,50 +0,0 @@
from __future__ import annotations
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field, validator
def _normalize_text(value: str) -> str:
cleaned = (value or "").strip()
if not cleaned:
raise ValueError("Message text cannot be empty")
return cleaned
class UserMessageResponse(BaseModel):
id: int
message_text: str
is_active: bool
sort_order: int
created_by: Optional[int]
created_at: datetime
updated_at: datetime
class UserMessageCreateRequest(BaseModel):
message_text: str = Field(..., min_length=1, max_length=4000)
is_active: bool = True
sort_order: int = Field(0, ge=0)
_normalize_message_text = validator("message_text", allow_reuse=True)(_normalize_text)
class UserMessageUpdateRequest(BaseModel):
message_text: Optional[str] = Field(None, min_length=1, max_length=4000)
is_active: Optional[bool] = None
sort_order: Optional[int] = Field(None, ge=0)
@validator("message_text")
def validate_message_text(cls, value): # noqa: D401,B902
if value is None:
return value
return _normalize_text(value)
class UserMessageListResponse(BaseModel):
items: list[UserMessageResponse]
total: int
limit: int
offset: int

50
app/webapi/schemas/welcome_texts.py
View File

@@ -1,50 +0,0 @@
from __future__ import annotations
from datetime import datetime
from typing import Optional
from pydantic import BaseModel, Field, validator
def _normalize_text(value: str) -> str:
cleaned = (value or "").strip()
if not cleaned:
raise ValueError("Text cannot be empty")
return cleaned
class WelcomeTextResponse(BaseModel):
id: int
text: str
is_active: bool
is_enabled: bool
created_by: Optional[int]
created_at: datetime
updated_at: datetime
class WelcomeTextCreateRequest(BaseModel):
text: str = Field(..., min_length=1, max_length=4000)
is_enabled: bool = True
is_active: bool = True
_normalize_text = validator("text", allow_reuse=True)(_normalize_text)
class WelcomeTextUpdateRequest(BaseModel):
text: Optional[str] = Field(None, min_length=1, max_length=4000)
is_enabled: Optional[bool] = None
is_active: Optional[bool] = None
@validator("text")
def validate_text(cls, value): # noqa: D401,B902
if value is None:
return value
return _normalize_text(value)
class WelcomeTextListResponse(BaseModel):
items: list[WelcomeTextResponse]
total: int
limit: int
offset: int