В данном туториале мы научимся собирать данные и сообщения участников чатов и каналов Telegram, а также сохранять эту информацию в виде JSON-файлов, которые далее легко анализировать или экспортировать в базы данных.
Для указанных задач будет использоваться Python не ниже версии 3.5, а также высокоуровневая библиотека для работы с Telegram API – Telethon. Установить библиотеку можно с помощью менеджера пакетов pip:
pip3 install telethon
Регистрируем в Telegram новое приложение
Для подключения к Telegram API необходимы api_id и api_hash. Эти параметры выдаются при регистрации приложения в инструментах разработчика (при отсутствии доступа используйте VPN). Для авторизации указываем номер телефона, к которому привязан аккаунт Telegram.
Вводим пришедший в Telegram численно-буквенный код и попадаем на страницу регистрации нового приложения. Заполняем форму, достаточно первых двух граф:
В результате попадаем на страницу конфигурации приложения. Находим оба параметра, а также доступные MTProto-сервера и открытые (публичные) ключи.
Избегая проблем с безопасностью, сохраняем учетные данные в отдельном файле config.ini следующей структуры:
[Telegram]
api_id = Telegram-API-ID
api_hash = Telegram-API-Hash
username = Your-Telegram-Username
Поле username далее будет использоваться лишь для автоматического сохранения сессии под именем username.session. Одному клиенту соответствует одна сессия, учтите это в случае запуска нескольких клиентов.
Создаем клиент Telegram
Начнем с импорта библиотек.
import configparser
import json
from telethon.sync import TelegramClient
from telethon import connection
# для корректного переноса времени сообщений в json
from datetime import date, datetime
# классы для работы с каналами
from telethon.tl.functions.channels import GetParticipantsRequest
from telethon.tl.types import ChannelParticipantsSearch
# класс для работы с сообщениями
from telethon.tl.functions.messages import GetHistoryRequest
Встроенные модули configparser и json применяем соответственно для чтения параметров и вывода данных. Из библиотеки Telethon импортируем класс клиента Telegram и класс исключений. Внутренний модуль connection необходим при использовании прокси-сервера. Остальные элементы модуля telethon.tl используются для запросов необходимых нам списков (участников канала/чата и их сообщений).
Теперь считаем учетные данные из config.ini:
# Считываем учетные данные
config = configparser.ConfigParser()
config.read("config.ini")
# Присваиваем значения внутренним переменным
api_id = config['Telegram']['api_id']
api_hash = config['Telegram']['api_hash']
username = config['Telegram']['username']
Создадим объект клиента Telegram API:
client = TelegramClient(username, api_id, api_hash)
При необходимости прописываем прокси. При использовании протокола MTProxy прокси задается в виде кортежа (сервер, порт, ключ).
proxy = (proxy_server, proxy_port, proxy_key)
client = TelegramClient(username, api_id, api_hash,
connection=connection.ConnectionTcpMTProxyRandomizedIntermediate,
proxy=proxy)
Запускаем клиент:
client.start()
При первом запуске платформа запросит номер телефона, и вслед – код подтверждения. Так же, как если бы вы входили в учетную запись в приложении или браузере.
Для сбора, обработки и сохранения информации мы создадим две функции:
dump_all_participants(сhannel)заберет данные о пользователях администрируемого нами сообществаchannel;dump_all_messages(сhannel)соберет все сообщения. Для этой функции достаточно, чтобы у вас был доступ к сообществу (необязательно быть администратором).
Обе функции будут вызываться в теле функции main, в которой пользователь передаст ссылку на интересующий источник:
url = input("Введите ссылку на канал или чат: ")
channel = await client.get_entity(url)
Касательно написания вызова функций стоит оговориться, что Telethon является асинхронной библиотекой. Поэтому в коде используются операторы async и await. В связи с этим функция main полностью будет выглядеть так:
async def main():
url = input("Введите ссылку на канал или чат: ")
channel = await client.get_entity(url)
await dump_all_participants(channel)
await dump_all_messages(channel)
Заметим, что из-за асинхронности Telethon может некорректно работать в средах, использующих те же подходы (Anaconda, Spyder, Jupyter).
Рекомендуемым способом управления клиентом является менеджер контекстов with. Его мы запустим в конце скрипта после описания вложенных в main функций.
with client:
client.loop.run_until_complete(main())
Собираем данные об участниках
Telegram не выводит все запрашиваемые данные за один раз, а выдает их в пакетном режиме, по 100 записей за каждый запрос.
async def dump_all_participants(channel):
"""Записывает json-файл с информацией о всех участниках канала/чата"""
offset_user = 0 # номер участника, с которого начинается считывание
limit_user = 100 # максимальное число записей, передаваемых за один раз
all_participants = [] # список всех участников канала
filter_user = ChannelParticipantsSearch('')
while True:
participants = await client(GetParticipantsRequest(channel,
filter_user, offset_user, limit_user, hash=0))
if not participants.users:
break
all_participants.extend(participants.users)
offset_user += len(participants.users)
Устанавливаем ограничение в 100, начинаем со смещения 0, создаем список всех участников канала all_participants. Внутри бесконечного цикла передаем запрос GetParticipantsRequest.
Проверяем, есть ли у объекта participants свойство users. Если нет, выходим из цикла. В обратном случае добавляем новых членов в список all_participants, а длину полученного списка добавляем к смещению offset_user. Следующий запрос забирает пользователей, начиная с этого смещения. Цикл продолжается до тех пор, пока не соберет всех фолловеров канала.
Самый простой способ сохранить собранные данные в структурированном виде – воспользоваться форматом JSON. Базы данных, такие как MySQL, MongoDB и т. д., стоит рассматривать лишь для очень популярных каналов и большого количества сохраняемой информации. Либо если вы планируете такое расширение в будущем.
В JSON-файле можно хранить и всю информацию о каждом пользователе, но обычно достаточно лишь нескольких параметров. Покажем на примере, как ограничиться набором определенных данных:
all_users_details = [] # список словарей с интересующими параметрами участников канала
for participant in all_participants:
all_users_details.append({"id": participant.id,
"first_name": participant.first_name,
"last_name": participant.last_name,
"user": participant.username,
"phone": participant.phone,
"is_bot": participant.bot})
with open('channel_users.json', 'w', encoding='utf8') as outfile:
json.dump(all_users_details, outfile, ensure_ascii=False)
Итак, для каждого пользователя создается свой словарь данных и добавляется в общий список all_user_details, который записывается в JSON-файл.
Собираем сообщения
Ситуация со сбором сообщений идентична сбору сведений о пользователях. Отличия сводятся к трем пунктам:
- Вместо клиентского запроса
GetParticipantsRequestнеобходимо отправитьGetHistoryRequestсо своим набором параметров. Так же, как и в случае со списком участников запрос ограничен сотней записей за один раз. - Для списка сообщений важна их последовательность. Чтобы получать последние сообщения, нужно правильно задать смещение в
GetHistoryRequest(с конца). - Чтобы корректно сохранить данные о времени публикации сообщений в JSON-файле, нужно преобразовать формат времени.
Итоговый код:
import configparser
import json
from telethon.sync import TelegramClient
from telethon import connection
# для корректного переноса времени сообщений в json
from datetime import date, datetime
# классы для работы с каналами
from telethon.tl.functions.channels import GetParticipantsRequest
from telethon.tl.types import ChannelParticipantsSearch
# класс для работы с сообщениями
from telethon.tl.functions.messages import GetHistoryRequest
# Считываем учетные данные
config = configparser.ConfigParser()
config.read("config.ini")
# Присваиваем значения внутренним переменным
api_id = config['Telegram']['api_id']
api_hash = config['Telegram']['api_hash']
username = config['Telegram']['username']
proxy = (proxy_server, proxy_port, proxy_key)
client = TelegramClient(username, api_id, api_hash,
connection=connection.ConnectionTcpMTProxyRandomizedIntermediate,
proxy=proxy)
client.start()
async def dump_all_participants(channel):
"""Записывает json-файл с информацией о всех участниках канала/чата"""
offset_user = 0 # номер участника, с которого начинается считывание
limit_user = 100 # максимальное число записей, передаваемых за один раз
all_participants = [] # список всех участников канала
filter_user = ChannelParticipantsSearch('')
while True:
participants = await client(GetParticipantsRequest(channel,
filter_user, offset_user, limit_user, hash=0))
if not participants.users:
break
all_participants.extend(participants.users)
offset_user += len(participants.users)
all_users_details = [] # список словарей с интересующими параметрами участников канала
for participant in all_participants:
all_users_details.append({"id": participant.id,
"first_name": participant.first_name,
"last_name": participant.last_name,
"user": participant.username,
"phone": participant.phone,
"is_bot": participant.bot})
with open('channel_users.json', 'w', encoding='utf8') as outfile:
json.dump(all_users_details, outfile, ensure_ascii=False)
async def dump_all_messages(channel):
"""Записывает json-файл с информацией о всех сообщениях канала/чата"""
offset_msg = 0 # номер записи, с которой начинается считывание
limit_msg = 100 # максимальное число записей, передаваемых за один раз
all_messages = [] # список всех сообщений
total_messages = 0
total_count_limit = 0 # поменяйте это значение, если вам нужны не все сообщения
class DateTimeEncoder(json.JSONEncoder):
'''Класс для сериализации записи дат в JSON'''
def default(self, o):
if isinstance(o, datetime):
return o.isoformat()
if isinstance(o, bytes):
return list(o)
return json.JSONEncoder.default(self, o)
while True:
history = await client(GetHistoryRequest(
peer=channel,
offset_id=offset_msg,
offset_date=None, add_offset=0,
limit=limit_msg, max_id=0, min_id=0,
hash=0))
if not history.messages:
break
messages = history.messages
for message in messages:
all_messages.append(message.to_dict())
offset_msg = messages[len(messages) - 1].id
total_messages = len(all_messages)
if total_count_limit != 0 and total_messages >= total_count_limit:
break
with open('channel_messages.json', 'w', encoding='utf8') as outfile:
json.dump(all_messages, outfile, ensure_ascii=False, cls=DateTimeEncoder)
async def main():
url = input("Введите ссылку на канал или чат: ")
channel = await client.get_entity(url)
await dump_all_participants(channel)
await dump_all_messages(channel)
with client:
client.loop.run_until_complete(main())
Если для анализа сообщений потребуются не все записи, задайте их число в переменной total_count_limit. Если нужна только сборка сообщений канала, достаточно закомментировать вызов await dump_all_participants(channel).
Таким образом, с помощью Python и Telethon мы написали скрипт, собирающий и сохраняющий данные и реплики участников сообществ Telegram.
Комментарии