Skip to content
This repository was archived by the owner on Feb 11, 2026. It is now read-only.

Refactoring telegram bridge #43

Merged
ink-developer merged 2 commits into from
Feb 2, 2026
Merged

Refactoring telegram bridge #43

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
17fd084
Refactoring
Rise0x00 Jan 25, 2026
e9c72bd
Fix KeyError
Rise0x00 Jan 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
252 changes: 113 additions & 139 deletions examples/telegram_bridge.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,194 +1,168 @@
# Всякая всячина
import asyncio

# Библиотека для работы с файлами
from io import BytesIO

import aiohttp

# Импорты библиотеки aiogram для TG-бота
from aiogram import Bot, Dispatcher, types
# УСТАНОВИТЬ ЗАВИСИМОСТИ - pip install maxapi-python aiogram==3.22.0
import asyncio # Для асинхронности
import aiohttp # Для асинхронных реквестов
from io import BytesIO # Для хранения ответов файлом в RAM
from aiogram import Bot, Dispatcher, types # Для ТГ

# Импорты библиотеки PyMax
from pymax import MaxClient, Message, Photo
from pymax import MaxClient, Message
from pymax.types import FileAttach, PhotoAttach, VideoAttach

# УСТАНОВИТЬ ЗАВИСИМОСТИ - pip install maxapi-python aiogram==3.22.0


# Настройки ботов
PHONE = "+79998887766" # Номер телефона Max
telegram_bot_TOKEN = "token" # Токен TG-бота

chats = { # В формате айди чата в Max: айди чата в Telegram (айди чата Max можно узнать из ссылки на чат в веб версии web.max.ru)
# Формат: id чата в Max: id чата в Tg
# (Id чата в Max можно узнать из ссылки на чат в веб версии web.max.ru)
chats = {
-68690734055662: -1003177746657,
}


# Создаём зеркальный массив для отправки из Telegram в Max
# Создаём зеркальный словарь для отправки из Telegram в Max
chats_telegram = {value: key for key, value in chats.items()}

max_client = MaxClient(phone=PHONE, work_dir="cache", reconnect=True) # Инициализация клиента Max

# Инициализация клиента MAX
client = MaxClient(phone=PHONE, work_dir="cache", reconnect=True)


# Инициализация TG-бота
telegram_bot = Bot(token=telegram_bot_TOKEN)
telegram_bot = Bot(token=telegram_bot_TOKEN) # Инициализация TG-бота
dp = Dispatcher()

async def download_file_bytes(url: str) -> BytesIO:
"""Загружает файл по URL и возвращает его в виде BytesIO."""
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
response.raise_for_status() # Кидаем exception в случае ошибки HTTP
file_bytes = BytesIO(await response.read()) # Читаем ответ в файлоподобный объект
file_bytes.name = response.headers.get("X-File-Name") # Ставим "файлу" имя из заголовков ответа
return file_bytes
Comment on lines +28 to +35
Copy link
Contributor

@coderabbitai coderabbitai bot Jan 25, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Provide a fallback filename when header is missing.

response.headers.get("X-File-Name") returns None if the header is absent. This propagates to Telegram sends where filename=video_bytes.name could be None, potentially causing files to be sent without a recognizable name.

🐛 Suggested fix 
             file_bytes = BytesIO(await response.read()) # Читаем ответ в файлоподобный объект
-            file_bytes.name = response.headers.get("X-File-Name") # Ставим "файлу" имя из заголовков ответа
+            file_bytes.name = response.headers.get("X-File-Name") or "file" # Ставим "файлу" имя из заголовков ответа
             return file_bytes
🧰 Tools
🪛 Ruff (0.14.13)

29-29: Docstring contains ambiguous е (CYRILLIC SMALL LETTER IE). Did you mean e (LATIN SMALL LETTER E)?

(RUF002)

29-29: Docstring contains ambiguous г (CYRILLIC SMALL LETTER GHE). Did you mean r (LATIN SMALL LETTER R)?

(RUF002)

29-29: Docstring contains ambiguous о (CYRILLIC SMALL LETTER O). Did you mean o (LATIN SMALL LETTER O)?

(RUF002)

🤖 Prompt for AI Agents 
In `@examples/telegram_bridge.py` around lines 28 - 35, The download_file_bytes
function sets file_bytes.name from response.headers.get("X-File-Name") which can
be None; update download_file_bytes to provide a sensible fallback filename when
the header is missing (e.g., derive from URL path or use a constant like
"file.bin" or preserve original extension if present) so that file_bytes.name is
never None before returning; ensure you still set file_bytes.name using the
header value when present and only use the fallback when that header is falsy.


# Обработчик входящих сообщений MAX
@client.on_message()
@max_client.on_message()
async def handle_message(message: Message) -> None:
try:
tg_id = chats[message.chat_id]
tg_id = chats[message.chat_id] # pyright: ignore[reportArgumentType]
except KeyError:
return

sender = await client.get_user(user_id=message.sender)

if message.attaches:
for attach in message.attaches:
# Проверка на видео
if isinstance(attach, VideoAttach):
async with aiohttp.ClientSession() as session:
try:
# Получаем видео по айди
video = await client.get_video_by_id(
chat_id=message.chat_id,
message_id=message.id,
video_id=attach.video_id,
)

# Загружаем видео по URL
async with session.get(video.url) as response:
response.raise_for_status() # Проверка на ошибки HTTP
video_bytes = BytesIO(await response.read())
video_bytes.name = response.headers.get("X-File-Name")

# Отправляем видео через телеграм бота
await telegram_bot.send_video(
chat_id=tg_id,
caption=f"{sender.names[0].name}: {message.text}",
video=types.BufferedInputFile(
video_bytes.getvalue(), filename=video_bytes.name
),
)

# Очищаем память
video_bytes.close()

except aiohttp.ClientError as e:
print(f"Ошибка при загрузке видео: {e}")
except Exception as e:
print(f"Ошибка при отправке видео: {e}")

# Проверка на изображение
elif isinstance(attach, PhotoAttach):
async with aiohttp.ClientSession() as session:
try:
# Загружаем изображение по URL
async with session.get(attach.base_url) as response:
response.raise_for_status() # Проверка на ошибки HTTP
photo_bytes = BytesIO(await response.read())
photo_bytes.name = response.headers.get("X-File-Name")

# Отправляем фото через телеграм бота
await telegram_bot.send_photo(
chat_id=tg_id,
caption=f"{sender.names[0].name}: {message.text}",
photo=types.BufferedInputFile(
photo_bytes.getvalue(), filename=photo_bytes.name
),
)

# Очищаем память
photo_bytes.close()

except aiohttp.ClientError as e:
print(f"Ошибка при загрузке изображения: {e}")
except Exception as e:
print(f"Ошибка при отправке фото: {e}")

# Проверка на файл
elif isinstance(attach, FileAttach):
async with aiohttp.ClientSession() as session:
try:
# Получаем файл по айди
file = await client.get_file_by_id(
chat_id=message.chat_id,
message_id=message.id,
file_id=attach.file_id,
)

# Загружаем файл по URL
async with session.get(file.url) as response:
response.raise_for_status() # Проверка на ошибки HTTP
file_bytes = BytesIO(await response.read())
file_bytes.name = response.headers.get("X-File-Name")

# Отправляем файл через телеграм бота
await telegram_bot.send_document(
chat_id=tg_id,
caption=f"{sender.names[0].name}: {message.text}",
document=types.BufferedInputFile(
file_bytes.getvalue(), filename=file_bytes.name
),
)

# Очищаем память
file_bytes.close()

except aiohttp.ClientError as e:
print(f"Ошибка при загрузке файла: {e}")
except Exception as e:
print(f"Ошибка при отправке файла: {e}")
sender = await max_client.get_user(user_id=message.sender) # pyright: ignore[reportArgumentType]

Comment on lines +45 to +46
Copy link
Contributor

@coderabbitai coderabbitai bot Jan 25, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Handle None sender to prevent AttributeError.

get_user() can return None (per API docs in user.py). The sender variable is later accessed at lines 64, 83, 109, and 122 via sender.names[0].name, which will crash if sender is None.

🐛 Suggested fix 
     sender = await max_client.get_user(user_id=message.sender) # pyright: ignore[reportArgumentType]
+    sender_name = sender.names[0].name if sender and sender.names else "Unknown"

Then replace all sender.names[0].name usages with sender_name.

🤖 Prompt for AI Agents 
In `@examples/telegram_bridge.py` around lines 45 - 46, get_user() can return None
so the code must guard against a missing sender: after calling
max_client.get_user(...) assign a safe sender_name (e.g., use
sender.names[0].name if sender is not None else a fallback like f"User
{message.sender}" or "Unknown") and then replace all direct accesses to
sender.names[0].name with this sender_name; update the code around the sender
assignment (the sender variable from max_client.get_user) and usages at the
later points so they use sender_name and avoid AttributeError when sender is
None.

if message.attaches: # Проверка на наличие вложений
for attach in message.attaches: # Перебор всех вложений
if isinstance(attach, VideoAttach): # Проверка на видео
try:
# Получаем видео из max по айди
video = await max_client.get_video_by_id(
chat_id=message.chat_id, # pyright: ignore[reportArgumentType]
message_id=message.id,
video_id=attach.video_id
)

# Загружаем видео по URL
video_bytes = await download_file_bytes(video.url) # pyright: ignore[reportOptionalMemberAccess]

# Отправляем видео через тг
await telegram_bot.send_video(
chat_id=tg_id,
caption=f"{sender.names[0].name}: {message.text}", # pyright: ignore[reportOptionalMemberAccess]
video=types.BufferedInputFile(video_bytes.getvalue(), filename=video_bytes.name)
)

video_bytes.close() # Удаляем видео из памяти

except aiohttp.ClientError as e:
print(f"Ошибка при загрузке видео: {e}")
except Exception as e:
print(f"Ошибка при отправке видео: {e}")
Comment on lines +49 to +73
Copy link
Contributor

@coderabbitai coderabbitai bot Jan 25, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Check for None video before accessing .url.

get_video_by_id() returns VideoRequest | None. Line 59 accesses video.url with only a pyright suppression — if video is None, this raises AttributeError at runtime.

🐛 Suggested fix 
                 video = await max_client.get_video_by_id(
                     chat_id=message.chat_id, # pyright: ignore[reportArgumentType]
                     message_id=message.id,
                     video_id=attach.video_id
                 )
+                if not video or not video.url:
+                    print(f"Не удалось получить видео {attach.video_id}")
+                    continue
                     
                 # Загружаем видео по URL
-                video_bytes = await download_file_bytes(video.url) # pyright: ignore[reportOptionalMemberAccess]
+                video_bytes = await download_file_bytes(video.url)
🧰 Tools
🪛 Ruff (0.14.13)

72-72: Do not catch blind exception: Exception

(BLE001)

🤖 Prompt for AI Agents 
In `@examples/telegram_bridge.py` around lines 49 - 73, get_video_by_id can return
None but the code immediately uses video.url, risking AttributeError; update the
VideoAttach handling block (around get_video_by_id, download_file_bytes, and
telegram_bot.send_video) to check whether video is None after calling
max_client.get_video_by_id and, if so, log or print a clear message and skip
sending (continue/return) before accessing video.url; only call
download_file_bytes and operate on video_bytes when video is not None, and
ensure video_bytes.close() is executed only when it was opened.


elif isinstance(attach, PhotoAttach): # Проверка на фото
try:
# Загружаем изображение по URL
photo_bytes = await download_file_bytes(attach.base_url) # pyright: ignore[reportOptionalMemberAccess]

# Отправляем фото через тг бота
await telegram_bot.send_photo(
chat_id=tg_id,
caption=f"{sender.names[0].name}: {message.text}", # pyright: ignore[reportOptionalMemberAccess]
photo=types.BufferedInputFile(photo_bytes.getvalue(), filename=photo_bytes.name)
)

photo_bytes.close() # Удаляем фото из памяти

except aiohttp.ClientError as e:
print(f"Ошибка при загрузке изображения: {e}")
except Exception as e:
print(f"Ошибка при отправке фото: {e}")

elif isinstance(attach, FileAttach): # Проверка на файл
try:
# Получаем файл по айди
file = await max_client.get_file_by_id(
chat_id=message.chat_id, # pyright: ignore[reportArgumentType]
message_id=message.id,
file_id=attach.file_id
)

# Загружаем файл по URL
file_bytes = await download_file_bytes(file.url) # pyright: ignore[reportOptionalMemberAccess]

# Отправляем файл через тг бота
await telegram_bot.send_document(
chat_id=tg_id,
caption=f"{sender.names[0].name}: {message.text}", # pyright: ignore[reportOptionalMemberAccess]
document=types.BufferedInputFile(file_bytes.getvalue(), filename=file_bytes.name)
)

file_bytes.close() # Удаляем файл из памяти

except aiohttp.ClientError as e:
print(f"Ошибка при загрузке файла: {e}")
except Exception as e:
print(f"Ошибка при отправке файла: {e}")
Comment on lines +75 to +118
Copy link
Contributor

@coderabbitai coderabbitai bot Jan 25, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Same None safety issues apply to photo and file handling.

  • Line 78: attach.base_url may be None for PhotoAttach.
  • Line 104: file.url requires a None check after get_file_by_id().

Apply the same guard pattern as suggested for video:

🐛 Suggested fixes

For photo (line 78):

             elif isinstance(attach, PhotoAttach): # Проверка на фото
                 try:
+                    if not attach.base_url:
+                        print(f"Нет URL для фото")
+                        continue
                     # Загружаем изображение по URL
-                    photo_bytes = await download_file_bytes(attach.base_url) # pyright: ignore[reportOptionalMemberAccess]
+                    photo_bytes = await download_file_bytes(attach.base_url)

For file (after line 101):

                     file = await max_client.get_file_by_id(
                         chat_id=message.chat_id,
                         message_id=message.id,
                         file_id=attach.file_id
                     )
+                    if not file or not file.url:
+                        print(f"Не удалось получить файл {attach.file_id}")
+                        continue

                     # Загружаем файл по URL
-                    file_bytes = await download_file_bytes(file.url) # pyright: ignore[reportOptionalMemberAccess]
+                    file_bytes = await download_file_bytes(file.url)
🧰 Tools
🪛 Ruff (0.14.13)

91-91: Do not catch blind exception: Exception

(BLE001)

117-117: Do not catch blind exception: Exception

(BLE001)

🤖 Prompt for AI Agents 
In `@examples/telegram_bridge.py` around lines 75 - 118, The photo and file
branches need the same None-safety guards as the video path: before calling
download_file_bytes, verify that attach.base_url (for PhotoAttach) is not None
and that the result of max_client.get_file_by_id (for FileAttach) is not None
and has a non-None file.url; if any are None, log/skip gracefully instead of
calling download_file_bytes. Update the PhotoAttach branch (inspect
attach.base_url) and the FileAttach branch (check the returned file and file.url
from get_file_by_id) to return/continue on None, and only call
download_file_bytes, send_photo/send_document, and close the bytes when the URL
and byte-object exist.

else:
await telegram_bot.send_message(
chat_id=tg_id, text=f"{sender.names[0].name}: {message.text}"
chat_id=tg_id,
text=f"{sender.names[0].name}: {message.text}" # pyright: ignore[reportOptionalMemberAccess]
)


# Обработчик запуска клиента, функция выводит все сообщения из чата "Избранное"
@client.on_start
@max_client.on_start
async def handle_start() -> None:
print("Клиент запущен")

# Получение истории сообщений
history = await client.fetch_history(chat_id=0)
history = await max_client.fetch_history(chat_id=0)
if history:
for message in history:
user = await client.get_user(message.sender)
user = await max_client.get_user(message.sender) # pyright: ignore[reportArgumentType]
if user:
print(f"{user.names[0].name}: {message.text}")


# Обработчик сообщений Telegram
# Обработчик сообщений из Telegram
@dp.message()
async def handle_tg_message(message: types.Message, bot: Bot) -> None:
max_id = chats_telegram[message.chat.id]
await client.send_message(
chat_id=max_id,
text=f"{message.from_user.first_name}: {message.text}",
notify=True,
)

try:
max_id = chats_telegram[message.chat.id]
await max_client.send_message(
chat_id=max_id,
text=f"{message.from_user.first_name}: {message.text}", # pyright: ignore[reportOptionalMemberAccess]
)
except KeyError:
return

# Раннер ботов
async def main() -> None:
# TG-бот в фоне
telegram_bot_task = asyncio.create_task(dp.start_polling(telegram_bot))

try:
await client.start()
await max_client.start()
finally:
await client.close()
await max_client.close()
telegram_bot_task.cancel()


if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("Программа остановлена пользователем.")
print("Программа остановлена пользователем.")