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

perf: cache logo file_id to avoid re-uploading on every message

Browse Source 
After first logo upload, Telegram returns a file_id that can be reused
for all subsequent sends. This eliminates 3-4 second delay per message
caused by re-uploading the same file from disk every time.
This commit is contained in:
Fringg
2026-02-09 18:14:54 +03:00
parent 49871f82f3
commit 142ff14a50
4 changed files with 57 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
app/handlers/start.py
View File

@@ -1880,9 +1880,7 @@ async def required_sub_channel_check(
menu_text = await get_main_menu_text(user, texts, db)
from aiogram.types import FSInputFile
from app.utils.message_patch import LOGO_PATH
from app.utils.message_patch import _cache_logo_file_id, get_logo_media
is_admin = settings.is_admin(user.telegram_id)
is_moderator = (not is_admin) and SupportSettingsService.is_moderator(user.telegram_id)
@@ -1909,13 +1907,14 @@ async def required_sub_channel_check(
)
if settings.ENABLE_LOGO_MODE:
await bot.send_photo(
_result = await bot.send_photo(
chat_id=query.from_user.id,
photo=FSInputFile(LOGO_PATH),
photo=get_logo_media(),
caption=menu_text,
reply_markup=keyboard,
parse_mode='HTML',
)
_cache_logo_file_id(_result)
else:
await bot.send_message(
chat_id=query.from_user.id,
@@ -1975,9 +1974,7 @@ async def required_sub_channel_check(
menu_text = await get_main_menu_text(user, texts, db)
from aiogram.types import FSInputFile
from app.utils.message_patch import LOGO_PATH
from app.utils.message_patch import _cache_logo_file_id, get_logo_media
is_admin = settings.is_admin(user.telegram_id)
is_moderator = (not is_admin) and SupportSettingsService.is_moderator(user.telegram_id)
@@ -2004,13 +2001,14 @@ async def required_sub_channel_check(
)
if settings.ENABLE_LOGO_MODE:
await bot.send_photo(
_result = await bot.send_photo(
chat_id=query.from_user.id,
photo=FSInputFile(LOGO_PATH),
photo=get_logo_media(),
caption=menu_text,
reply_markup=keyboard,
parse_mode='HTML',
)
_cache_logo_file_id(_result)
else:
await bot.send_message(
chat_id=query.from_user.id,
@@ -2030,19 +2028,18 @@ async def required_sub_channel_check(
)
await state.set_state(RegistrationStates.waiting_for_referral_code)
else:
from aiogram.types import FSInputFile
from app.utils.message_patch import LOGO_PATH
from app.utils.message_patch import _cache_logo_file_id, get_logo_media
rules_text = await get_rules(language)
if settings.ENABLE_LOGO_MODE:
await bot.send_photo(
_result = await bot.send_photo(
chat_id=query.from_user.id,
photo=FSInputFile(LOGO_PATH),
photo=get_logo_media(),
caption=rules_text,
reply_markup=get_rules_keyboard(language),
)
_cache_logo_file_id(_result)
else:
await bot.send_message(
chat_id=query.from_user.id,

9
app/services/monitoring_service.py
View File

@@ -6,7 +6,6 @@ from typing import Any
from aiogram.enums import ChatMemberStatus
from aiogram.exceptions import TelegramBadRequest, TelegramForbiddenError
from aiogram.types import FSInputFile
from sqlalchemy import and_, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
@@ -108,13 +107,17 @@ class MonitoringService:
if settings.ENABLE_LOGO_MODE and LOGO_PATH.exists() and (text is None or len(text) <= 1000):
try:
return await self.bot.send_photo(
from app.utils.message_patch import _cache_logo_file_id, get_logo_media
result = await self.bot.send_photo(
chat_id=chat_id,
photo=FSInputFile(LOGO_PATH),
photo=get_logo_media(),
caption=text,
reply_markup=reply_markup,
parse_mode=parse_mode,
)
_cache_logo_file_id(result)
return result
except TelegramBadRequest as exc:
logger.warning(
'Не удалось отправить сообщение с логотипом пользователю %s: %s. Отправляем текстовое сообщение.',

35
app/utils/message_patch.py
View File

@@ -10,6 +10,28 @@ from app.localization.texts import get_texts
LOGO_PATH = Path(settings.LOGO_FILE)
_PRIVACY_RESTRICTED_CODE = 'BUTTON_USER_PRIVACY_RESTRICTED'
# Кеш file_id логотипа: после первой загрузки Telegram возвращает file_id,
# который можно переиспользовать без повторной загрузки файла (экономит 3-4 сек)
_logo_file_id: str | None = None
def get_logo_media():
"""Возвращает кешированный file_id или FSInputFile для логотипа."""
if _logo_file_id:
return _logo_file_id
return FSInputFile(LOGO_PATH)
def _cache_logo_file_id(result: Message | None) -> None:
"""Извлекает и кеширует file_id логотипа из ответа Telegram."""
global _logo_file_id
if _logo_file_id or result is None:
return
if hasattr(result, 'photo') and result.photo:
_logo_file_id = result.photo[-1].file_id
_TOPIC_REQUIRED_ERRORS = (
'topic must be specified',
'TOPIC_CLOSED',
@@ -110,8 +132,9 @@ async def _answer_with_photo(self: Message, text: str = None, **kwargs):
if LOGO_PATH.exists():
try:
# Отправляем caption как есть; при ошибке парсинга ниже сработает фоллбек
return await self.answer_photo(FSInputFile(LOGO_PATH), caption=text, **kwargs)
result = await self.answer_photo(get_logo_media(), caption=text, **kwargs)
_cache_logo_file_id(result)
return result
except TelegramBadRequest as error:
if is_topic_required_error(error):
# Канал с топиками — просто игнорируем, нельзя ответить без message_thread_id
@@ -163,12 +186,8 @@ async def _edit_with_photo(self: Message, text: str, **kwargs):
return await _original_answer(self, text, **kwargs)
except Exception:
pass
# Всегда используем логотип если включен режим логотипа,
# кроме специальных случаев (QR сообщения)
if (settings.ENABLE_LOGO_MODE and LOGO_PATH.exists() and not is_qr_message(self)) or (
is_qr_message(self) and LOGO_PATH.exists()
):
media = FSInputFile(LOGO_PATH)
if LOGO_PATH.exists():
media = get_logo_media()
else:
media = self.photo[-1].file_id
media_kwargs = {'media': media, 'caption': text}

24
app/utils/photo_message.py
View File

@@ -3,13 +3,15 @@ import logging
from aiogram import types
from aiogram.exceptions import TelegramBadRequest, TelegramNetworkError
from aiogram.types import FSInputFile, InaccessibleMessage, InputMediaPhoto
from aiogram.types import InaccessibleMessage, InputMediaPhoto
from app.config import settings
from .message_patch import (
LOGO_PATH,
_cache_logo_file_id,
append_privacy_hint,
get_logo_media,
is_privacy_restricted_error,
is_qr_message,
prepare_privacy_safe_kwargs,
@@ -23,17 +25,13 @@ RETRY_DELAY = 0.5
def _resolve_media(message: types.Message):
# Если сообщение недоступно, возвращаем логотип по умолчанию
if isinstance(message, InaccessibleMessage):
return FSInputFile(LOGO_PATH)
# Всегда используем логотип если включен режим логотипа,
# кроме специальных случаев (QR сообщения)
return get_logo_media()
if settings.ENABLE_LOGO_MODE and not is_qr_message(message):
return FSInputFile(LOGO_PATH)
# Только если режим логотипа выключен, используем фото из сообщения
return get_logo_media()
if message.photo:
return message.photo[-1].file_id
return FSInputFile(LOGO_PATH)
return get_logo_media()
def _get_language(callback: types.CallbackQuery) -> str | None:
@@ -91,12 +89,13 @@ async def edit_or_answer_photo(
if isinstance(callback.message, InaccessibleMessage):
try:
if settings.ENABLE_LOGO_MODE and LOGO_PATH.exists():
await callback.message.answer_photo(
photo=FSInputFile(LOGO_PATH),
result = await callback.message.answer_photo(
photo=get_logo_media(),
caption=caption,
reply_markup=keyboard,
parse_mode=resolved_parse_mode,
)
_cache_logo_file_id(result)
else:
await callback.message.answer(
caption,
@@ -183,12 +182,13 @@ async def edit_or_answer_photo(
pass
try:
# Отправим как фото с логотипом
await callback.message.answer_photo(
photo=media if isinstance(media, FSInputFile) else FSInputFile(LOGO_PATH),
result = await callback.message.answer_photo(
photo=get_logo_media(),
caption=caption,
reply_markup=keyboard,
parse_mode=resolved_parse_mode,
)
_cache_logo_file_id(result)
except TelegramBadRequest as photo_error:
await _answer_text(callback, caption, keyboard, resolved_parse_mode, photo_error)
except Exception: