Давно пишу ботов для телеграмм, использую 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 мне не удалось заставить писать рабочий код для телеграмм ботов. Намного больше ошибок.
Писать код легко, сложно и долго проверять что это работает. Еще сложнее не погружаясь в проект написать что-то более объемное по бизнес логике. То что здесь написано - уже на пределе нейронки, когда можно не используя свою голову писать приложение. Но чуть сложнее, и каждый новый кусок кода генерирует больше проблем чем пользы, связи нарушаются, прошлые правила забываются, общая идея теряется. И бездумная копипаста перестает работать совсем.