Привет! Сегодня разберём, как парсить Telegram каналы. Да, я знаю, что официальный Bot API не даёт доступ к каналам напрямую, но мы же не ищем лёгких путей, правда? 😎
Почему это сложнее, чем кажется?
Telegram — это не просто мессенджер, это целая экосистема с кучей защит. Но мы справимся! Есть два основных подхода: через Bot API (ограниченный) и через MTProto (более мощный, но сложнее).
Вариант 1: Bot API (для начинающих)
Если тебе нужно парсить только публичные каналы, где бот является администратором, то Bot API вполне сгодится:
import requests
import json
# Твой токен от @BotFather
BOT_TOKEN = "YOUR_BOT_TOKEN"
CHANNEL_USERNAME = "@your_channel"
def get_channel_messages(channel_username, limit=100):
"""Получаем сообщения из канала"""
url = f"https://api.telegram.org/bot{BOT_TOKEN}/getUpdates"
messages = []
offset = 0
while len(messages) < limit:
params = {
'offset': offset,
'timeout': 10
}
response = requests.get(url, params=params)
data = response.json()
if not data.get('ok'):
break
for update in data.get('result', []):
if 'channel_post' in update:
post = update['channel_post']
messages.append({
'text': post.get('text', ''),
'date': post.get('date', 0),
'message_id': post.get('message_id', 0)
})
if not data.get('result'):
break
offset = data['result'][-1]['update_id'] + 1
return messages
# Использование
messages = get_channel_messages("@example_channel", limit=50)
for msg in messages:
print(f"{msg['text'][:100]}...")
Вариант 2: Telethon (MTProto) — для профи
Если нужен полный доступ к каналам (даже приватным), используем Telethon — это Python библиотека для работы с MTProto API Telegram.
# Установка
# pip install telethon
from telethon import TelegramClient
from telethon.tl.functions.messages import GetHistoryRequest
from telethon.tl.types import PeerChannel
import asyncio
# Получаем эти данные на https://my.telegram.org
API_ID = 12345678 # Твой API ID
API_HASH = 'abcdefghijklmnopqrstuvwxyz123456' # Твой API Hash
async def parse_channel():
# Создаём клиент
client = TelegramClient('session_name', API_ID, API_HASH)
await client.start()
# Указываем канал (можно по username или ID)
channel = await client.get_entity('@example_channel')
# Получаем сообщения
messages = []
async for message in client.iter_messages(channel, limit=100):
messages.append({
'id': message.id,
'text': message.text,
'date': message.date,
'views': message.views,
'forwards': message.forwards
})
await client.disconnect()
return messages
# Запускаем
messages = asyncio.run(parse_channel())
for msg in messages:
print(f"ID: {msg['id']}, Views: {msg['views']}")
print(f"Text: {msg['text'][:100]}...\n")
Парсим медиафайлы
Часто нужно скачать не только текст, но и картинки, видео, документы. Вот как это сделать:
async def download_media_from_channel(channel_username):
client = TelegramClient('session', API_ID, API_HASH)
await client.start()
channel = await client.get_entity(channel_username)
async for message in client.iter_messages(channel, limit=50):
if message.media:
# Определяем тип медиа
if message.photo:
filename = f"photos/{message.id}.jpg"
elif message.video:
filename = f"videos/{message.id}.mp4"
elif message.document:
filename = f"docs/{message.document.id}.{message.document.mime_type.split('/')[-1]}"
else:
continue
# Скачиваем
await client.download_media(message, filename)
print(f"Скачано: {filename}")
await client.disconnect()
Мониторинг новых постов
Чтобы отслеживать новые посты в реальном времени, используем обработчик новых сообщений:
from telethon import events
@client.on(events.NewMessage(chats='@example_channel'))
async def handler_new_message(event):
message = event.message
print(f"Новый пост! ID: {message.id}")
print(f"Текст: {message.text}")
print(f"Дата: {message.date}")
# Можно сохранить в базу данных
# save_to_db(message)
# Запускаем клиент
client.start()
client.run_until_disconnected()
Извлечение участников канала
Если нужно получить список участников (для публичных каналов):
async def get_channel_participants(channel_username):
client = TelegramClient('session', API_ID, API_HASH)
await client.start()
channel = await client.get_entity(channel_username)
participants = []
async for user in client.iter_participants(channel):
participants.append({
'id': user.id,
'username': user.username,
'first_name': user.first_name,
'last_name': user.last_name
})
await client.disconnect()
return participants
# Использование
participants = asyncio.run(get_channel_participants('@example_channel'))
print(f"Всего участников: {len(participants)}")
Важные моменты
- Rate Limits: Telegram ограничивает количество запросов. Не делай слишком много запросов подряд — добавь задержки.
- Сессии: Telethon сохраняет сессию в файл. Не коммить этот файл в Git!
- Права доступа: Для приватных каналов нужно быть участником или администратором.
- API_ID и API_HASH: Получай на https://my.telegram.org — это безопасно и легально.
Обработка ошибок
Всегда обрабатывай ошибки — Telegram API может вернуть разные исключения:
from telethon.errors import FloodWaitError, ChannelPrivateError
async def safe_parse_channel(channel_username):
try:
client = TelegramClient('session', API_ID, API_HASH)
await client.start()
channel = await client.get_entity(channel_username)
messages = []
async for message in client.iter_messages(channel, limit=100):
messages.append(message.text)
except FloodWaitError as e:
print(f"Слишком много запросов! Жди {e.seconds} секунд")
await asyncio.sleep(e.seconds)
except ChannelPrivateError:
print("Канал приватный, нужен доступ")
except Exception as e:
print(f"Ошибка: {e}")
finally:
await client.disconnect()
Итоги
Парсинг Telegram каналов — это не так сложно, как кажется. Главное — выбрать правильный инструмент и не забывать про ограничения API. Telethon — мощный инструмент, который открывает много возможностей. Удачи в парсинге! 🚀