Давно пишу ботов для телеграмм, использую golang. Понадобился функционал - сканировать каналы по ссылке. Бот такое не может, это уже более сложное апи, порылся - нашел библиотеку на golang, попробовал - сложно. Нашел на питоне - проще. Но на питоне не хочется. Так родилась идея сделать простую обертку REST API для основного функционала: вступить в группу, прочитать сообщения, узнать информацию о группе, написать сообщение, и чтобы курлом все работало...
Кратко: https://github.com/pivolan/telegram_app_wrapper. Это стейтлесс на питоне простой, как три рубля. С такими методами:
POST /auth/send_code
POST /auth/verify_code
POST /auth/verify_password
DELETE /auth/logout
GET /chats
GET /messages/
GET /messages/media/{message_id}
POST /groups/join
POST /messages/send
POST /messages/send_with_file
DELETE /messages/delete
POST /messages/forward
POST /messages/edit
Сессия немного шифрованная и передается в заголовке на каждом запросе. От вас потребуется app_id, app_hash, номер телефона.
-
Отправляем номер телефона и API-креды (получаем строку сессии):
POST /auth/send_code
{
"phone": "+7XXXXXXXXXX",
"api_id": YOUR_API_ID,
"api_hash": "YOUR_API_HASH"
}-
Отправляем код из СМС/Telegram (используя полученную строку сессии в заголовке):
POST /auth/verify_code
X-Session-String: {session_string}
{ "code": "123456"}
-
Если включена двухфакторка, отправляем пароль:
POST /auth/verify_password
X-Session-String: {session_string}
{"password": "your_2fa_password"}
После успешной авторизации используем полученную строку сессии (X-Session-String) во всех последующих запросах. Сессия остаётся валидной до вызова logout или перезапуска сервера.
Немного опыта
изначально идея была такая: попросить у нейронки написать простой враппер для всех этих методов. А она прям много кода выдала, да еще и не рабочего. начал с простых шагов, как авторизоваться по этому апи, авторизация работала через консоль, т.е. запускаю скрипт на питоне, он в консоли меня спрашивает номер телефона, коды, потом ожидает ввода подтверждения, далее свой пароль. Потом это удалось вынести в рест апи, чтобы не через консоль. Первая мысли была сохранять сессии в базе, с привязкой к номеру. Это для личного использования просто. Возникла проблема, если сделать общедоступным, то нужна какая то авторизация, ключи, oauth - сложно, такое делать не хочется. В целом разобрался как работают сессии в телеграм, и сделал передачу сессии в заголовке на каждый апи запрос. Однако помимо сессии требуются api_id api_hash, каждый раз в открытую их передавать не хочется. По итогу сделал передачу этих данных только на этапе авторизации, далее уже отдаю шифрованный(простейшим способом) ключик, в нем содержится сессия и нужные ключи. В общем получилось stateless. Ну а дальше уже дело простое, каждый новый хендлер просто берет заголовок, восстанавливает сессию делает запрос отдает ответ.
Весь код написан с помощью claude.ai. Сам лишь разбил на файлы и ключи вставлял. Нейронки научились писать довольно объемный код и сразу рабочий, но вот связи между файлами по прежнему большая проблема. Утилита на коленке - супер, чуть сложнее, и приходится думать самому.
Про код
Использовал pydantic, fastapi, telethon. Весь код выложен на гитхаб. Так же запущен сервер где можно пощупать. Только свои аккаунты не пробуйте, создайте новый который не жалко.
Самое сложное было - авторизация, ее и покажу:
@app.post("/auth/send_code", response_model=AuthResponse)
async def send_code(credentials: ApiCredentials):
try:
# Create new client with provided credentials
client = TelegramClient(StringSession(), credentials.api_id, credentials.api_hash)
await client.connect()
# Send authentication code
await client.send_code_request(credentials.phone)
# Get session and combine with encrypted credentials
temp_session = client.session.save()
combined_session = encode_session_with_credentials(
temp_session,
credentials.api_id,
credentials.api_hash
)
# Store client
clients[combined_session] = client
return AuthResponse(
message="Verification code sent",
next_step="verify_code",
session_string=combined_session
)
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
@app.post("/auth/verify_code", response_model=AuthResponse)
async def verify_code(
verification_data: VerificationCode,
session_string: str = Header(..., alias="X-Session-String")
):
try:
if session_string not in clients:
raise HTTPException(status_code=401, detail="Invalid session")
client = clients[session_string]
session, api_id, api_hash = decode_session_with_credentials(session_string)
try:
# Try to sign in with the code
await client.sign_in(code=verification_data.code)
# Get new session and combine with credentials
new_session = client.session.save()
new_combined_session = encode_session_with_credentials(
new_session,
api_id,
api_hash
)
# Update clients dictionary
clients[new_combined_session] = client
del clients[session_string]
return AuthResponse(
message="Successfully authenticated",
next_step="completed",
session_string=new_combined_session
)
except Exception as e:
if "password" in str(e).lower():
return AuthResponse(
message="2FA password required",
next_step="verify_password",
session_string=session_string
)
raise e
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
@app.post("/auth/verify_password", response_model=AuthResponse)
async def verify_password(
password_data: Password,
session_string: str = Header(..., alias="X-Session-String")
):
try:
if session_string not in clients:
raise HTTPException(status_code=401, detail="Invalid session")
client = clients[session_string]
session, api_id, api_hash = decode_session_with_credentials(session_string)
# Sign in with password
await client.sign_in(password=password_data.password)
# Get new session and combine with credentials
new_session = client.session.save()
new_combined_session = encode_session_with_credentials(
new_session,
api_id,
api_hash
)
# Update clients dictionary
clients[new_combined_session] = client
del clients[session_string]
return AuthResponse(
message="Successfully authenticated with 2FA",
next_step="completed",
session_string=new_combined_session
)
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
Тут правда не совсем stateless, в переменной хранится, на диск не сохраняется.
И пример одного из методов, получить список чатов и каналов:
@app.get("/chats", response_model=ChatsResponse)
async def get_chats(
limit: int = 100,
session_string: str = Header(..., alias="X-Session-String")
):
try:
client = await get_client_from_session(session_string)
# Get dialogs
dialogs = await client.get_dialogs(limit=limit)
chats_list = []
for dialog in dialogs:
entity = dialog.entity
# Determine chat type
if isinstance(entity, Channel):
chat_type = "channel" if entity.broadcast else "supergroup"
elif isinstance(entity, Chat):
chat_type = "group"
elif isinstance(entity, User):
chat_type = "private"
else:
continue
chat_info = ChatInfo(
name=dialog.name,
id=dialog.id,
type=chat_type,
members_count=getattr(entity, 'participants_count', None),
is_private=not hasattr(entity, 'username') or entity.username is None,
username=getattr(entity, 'username', None)
)
chats_list.append(chat_info)
return ChatsResponse(
chats=chats_list,
total_count=len(chats_list)
)
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
Функционал
Работа с чатами и группами:
-
Получить список всех чатов:
GET /chats?limit=100 -
Подключиться к группам/каналам:
POST /groups/join { "group_identifier": "@username" // публичная группа по юзернейму "group_identifier": "https://t.me/groupname" // публичная группа по ссылке "group_identifier": "https://t.me/+InviteHash" // приватная группа по инвайт-ссылке }
Чтение сообщений:
-
Получить сообщения из чата (работает с любым типом идентификатора):
GET /messages/?chat_id={id}&limit=100 chat_id может быть: - числовой ID: 1234567890 - юзернейм: @username - приватный чат: user_id -
Дополнительные параметры для фильтрации:
GET /messages/?chat_id={id} &limit=50 &offset_id=0 // начать с определенного сообщения &search=keyword // поиск по тексту &from_date=2024-03-01T00:00:00Z // фильтр по дате &to_date=2024-03-14T23:59:59Z -
Скачать медиа из сообщения:
GET /messages/media/{message_id}?chat_id={chat_id}
Отправка сообщений:
-
Простое текстовое сообщение:
POST /messages/send { "chat_id": "@username", // работает с любым типом идентификатора "text": "сообщение", "reply_to_message_id": 123 // опционально для ответа } -
Сообщение с файлом:
POST /messages/send_with_file(multipart/form-data) -
Переслать сообщение:
POST /messages/forward { "from_chat_id": "@chat1", "to_chat_id": "@chat2", "message_id": 123 } -
Редактировать сообщение:
POST /messages/edit { "chat_id": "@chat", "message_id": "123", "new_text": "новый текст" } -
Удалить сообщения:
DELETE /messages/delete { "chat_id": "@chat", "message_ids": [123, 124, 125] }
Ограничения:
-
Работает как обычный пользовательский аккаунт, поэтому нет доступа к специальным возможностям ботов (кнопки, веб-хуки и т.д.)
-
Нет постоянного соединения/стриминга новых сообщений
-
Все операции выполняются в контексте одного аккаунта
-
Некоторые действия могут быть ограничены правами доступа в конкретном чате
Итог
Хотел сделать рекламного бота, который будет без человека премодерировать канал, прежде чем запускать рекламу, для этого нужна была возможность на этот канал зайти по ссылке и проверить что там есть. Чтобы купить рекламу своего канала в другом канале без участия человеческого менеджера.
Попробовал написать полностью с помощью нейронки весь код приложения, которое чуть сложнее чем пузырьковая сортировка. Кстати chatgpt мне не удалось заставить писать рабочий код для телеграмм ботов. Намного больше ошибок.
Писать код легко, сложно и долго проверять что это работает. Еще сложнее не погружаясь в проект написать что-то более объемное по бизнес логике. То что здесь написано - уже на пределе нейронки, когда можно не используя свою голову писать приложение. Но чуть сложнее, и каждый новый кусок кода генерирует больше проблем чем пользы, связи нарушаются, прошлые правила забываются, общая идея теряется. И бездумная копипаста перестает работать совсем.
Автор: pivolan
