Telethon Python MTProto Telegram 客户端库技术分析

一、概述

1. 简介

A. 是什么

Telethon 是一个基于 asyncio 的纯 Python 3 MTProto 库，用于与 Telegram API 进行交互。它支持用户账户和机器人账户两种模式，是 Bot API 的替代方案。

B. 为什么值得研究

  • 纯 Python 实现，无需依赖 C 扩展库
  • 完整实现 MTProto 协议，直接访问 Telegram API
  • 支持 asyncio 异步编程模型，性能优异
  • 活跃的社区维护，11.2k+ GitHub Stars
  • 被超过 69,900 个项目使用

C. 学完能做什么

  • 开发 Telegram 机器人和用户自动化工具
  • 实现 Telegram 消息收发、文件传输功能
  • 构建自定义的 Telegram 客户端应用
  • 理解 MTProto 协议的工作原理

2. 前置知识

A. 必备技能

  • Python 3 基础编程能力
  • 异步编程概念（async/await）
  • Telegram 基本使用经验

B. 推荐知识

  • asyncio 库的使用
  • MTProto 协议基础
  • API 开发经验

二、环境准备

1. 系统要求

  • Python 3.6 或更高版本
  • asyncio 支持（Python 3.7+ 内置）
  • 网络连接

2. 安装步骤

基础安装

pip3 install telethon

完整安装（包含加密库依赖）

pip3 install telethon[cryptg]

3. 获取 API 凭证

在使用 Telethon 之前，必须从 Telegram 获取 API 凭证：

  1. 访问 https://my.telegram.org
  2. 登录你的 Telegram 账户

  3. 进入 API Development 部分获取：

    • api_id：数字标识符
    • api_hash：密钥字符串

4. 验证安装

from telethon import TelegramClient
print(Telethon.__version__)

三、核心概念

1. 基本术语

MTProto

MTProto 是 Telegram 开发的专有网络协议，用于客户端与服务器之间的通信。它提供了：

  • 高效的数据传输
  • 端到端加密
  • 数据完整性验证

Session（会话）

Session 是客户端与 Telegram 服务器之间的持久化连接状态，包含：

  • 认证信息
  • 服务器配置
  • 缓存数据

Events（事件）

Telethon 使用事件驱动模型处理消息和通知：

  • NewMessage：新消息事件
  • CallbackQuery：回调查询
  • Raw：原始事件

2. 工作原理

graph TB
    A[Python 应用] -->|API 调用| B[Telethon 客户端]
    B -->|MTProto 协议| C[Telegram 服务器]
    C -->|响应| B
    B -->|异步事件| D[事件处理器]
    D -->|回调| A
    B -->|持久化| E[Session 文件]

Telethon 工作原理图

3. 架构设计

Telethon 采用分层架构：

graph LR
    A[应用层] -->|API| B[Telethon Client]
    B -->|请求| C[TL 层]
    C -->|序列化| D[MTProto 层]
    D -->|网络| E[Transport 层]
    E -->|TCP| F[TG Server]

Telethon 架构图

组件说明：

  • 应用层：用户代码，调用 Telethon API
  • Telethon Client：提供高级 API 接口
  • TL 层：Type Language，定义 API 数据结构
  • MTProto 层：协议实现，处理加密和传输
  • Transport 层：网络传输抽象

四、快速上手

1. Hello World 示例

创建客户端

from telethon import TelegramClient

# 使用从 my.telegram.org 获取的凭据
api_id = 12345
api_hash = '0123456789abcdef0123456789abcdef'

# 创建客户端，session_name 会保存登录状态
client = TelegramClient('session_name', api_id, api_hash)

# 启动客户端（首次运行会要求输入手机号和验证码）
client.start()

基本操作

# 获取当前账户信息
print(client.get_me().stringify())

# 发送消息
client.send_message('username', 'Hello! Talking to you from Telethon')

# 发送文件
client.send_file('username', '/path/to/file.jpg')

# 下载个人资料照片
client.download_profile_photo('me')

# 获取消息历史
messages = client.get_messages('username')
messages[0].download_media()

2. 事件处理

消息监听器

from telethon import events

@client.on(events.NewMessage(pattern='(?i)hi|hello'))
async def handler(event):
    await event.respond('Hey!')

多种事件类型

# 新消息
@client.on(events.NewMessage)
async def new_message_handler(event):
    print(event.message.text)

# 被提及
@client.on(events.NewMessage(incoming=True, forwards=False))
async def mention_handler(event):
    if event.message.mentioned:
        await event.reply('你提到我了！')

# 私聊消息
@client.on(events.NewMessage(chats='username'))
async def private_handler(event):
    print(f'私聊消息：{event.message.text}')

3. 代码讲解

异步编程模型

Telethon 全面采用 asyncio，所有 I/O 操作都是异步的：

async def main():
    # 异步获取对话列表
    dialogs = await client.get_dialogs()
    async for dialog in dialogs:
        print(dialog.name)

# 运行异步函数
client.loop.run_until_complete(main())

Session 管理

Session 文件保存了登录状态，避免每次都输入验证码：

  • 默认位置：当前目录下的 .session 文件
  • 可自定义路径：TelegramClient('/path/to/session', ...)
  • 包含加密的认证密钥

五、核心功能

1. 消息操作

发送消息

# 文本消息
await client.send_message('username', 'Hello')

# 格式化消息
await client.send_message('username', '**粗体** 和 _斜体_', parse_mode='markdown')

# 发送多个实体
await client.send_message('username', '消息1')
await client.send_message('username', '消息2')

获取消息

# 获取最新消息
messages = await client.get_messages('username', limit=10)

# 通过 ID 获取
message = await client.get_messages('username', ids=12345)

# 搜索消息
messages = await client.get_messages('username', search='关键词')

编辑和删除

# 编辑消息
await client.edit_message('username', message_id, '新内容')

# 删除消息
await client.delete_messages('username', message_ids, revoke=True)

2. 文件处理

上传文件

# 发送图片
await client.send_file('username', 'photo.jpg')

# 发送文档
await client.send_file('username', 'document.pdf')

# 发送多个文件
await client.send_file('username', ['file1.jpg', 'file2.jpg'])

# 带进度上传
def progress(current, total):
    print(f'{current / total * 100:.1f}%')

await client.send_file('username', 'large_file.mp4', progress_callback=progress)

下载文件

# 下载媒体文件
await message.download_media('/path/to/save/')

# 下载个人资料照片
await client.download_profile_photo('username', 'photo.jpg')

# 下载聊天媒体
async for message in client.iter_messages('username'):
    if message.media:
        await message.download_media()

3. 聊天管理

获取对话列表

# 获取所有对话
dialogs = await client.get_dialogs()

# 遍历对话
async for dialog in client.iter_dialogs():
    print(f'{dialog.name}: {dialog.id}')

群组操作

# 创建群组
from telethon.tl.functions.messages import CreateChatRequest
result = await client(CreateChatRequest(users=['user1', 'user2'], title='新群组'))

# 邀请成员
await client(InviteToChannelRequest(channel=group, users=['username']))

# 获取成员
members = await client.get_participants(group)

4. 用户和频道

获取用户信息

# 获取完整用户信息
user = await client.get_entity('username')

# 获取多个用户
users = await client.get_entity(['user1', 'user2', 'user3'])

频道操作

# 加入频道
await client(JoinChannelRequest(channel))

# 离开频道
await client(LeaveChannelRequest(channel))

# 获取频道消息
messages = await client.get_messages(channel, limit=100)

六、高级功能

1. Bot API 替代

Telethon 可以作为 Bot API 的替代方案，提供更强大的功能：

# 创建 Bot 客户端
bot = TelegramClient('bot_session', api_id, api_hash).start(bot_token='你的bot_token')

# 处理命令
@bot.on(events.NewMessage(pattern='/start'))
async def start(event):
    await event.reply('欢迎！')

# 处理回调查询
@bot.on(events.CallbackQuery)
async def callback(event):
    await event.answer('你点击了按钮！')

相比 Bot API 的优势

  • 直接访问 Telegram API，功能更完整
  • 支持 Bot API 不提供的功能
  • 更低的延迟
  • 可以使用用户账户模式

2. 自定义过滤器

from telethon.tl.custom import Message

def my_filter(event):
    message = event.message
    return message.out and '关键字' in message.text

@client.on(events.NewMessage(func=my_filter))
async def custom_handler(event):
    print('匹配的消息：', event.message.text)

3. 并发处理

import asyncio

async def process_chat(chat):
    async for message in client.iter_messages(chat):
        # 处理消息
        pass

async def main():
    chats = ['chat1', 'chat2', 'chat3']
    tasks = [process_chat(chat) for chat in chats]
    await asyncio.gather(*tasks)

client.loop.run_until_complete(main())

七、最佳实践

1. 错误处理

from telethon.errors import SessionPasswordNeededError
from telethon.tl.types import InputPeerUser

try:
    await client.send_message('username', 'Hello')
except SessionPasswordNeededError:
    print('需要两步验证密码')
    await client.sign_in(password='你的密码')
except Exception as e:
    print(f'错误：{e}')

2. 性能优化

使用连接池

# 配置连接池
client = TelegramClient(
    'session',
    api_id,
    api_hash,
    connection_retries=5,
    retry_delay=1,
    timeout=10
)

批量操作

# 批量获取信息
from telethon.tl.functions.messages import GetHistoryRequest

result = await client(GetHistoryRequest(
    peer='username',
    limit=100,
    offset_date=None,
    offset_id=0,
    max_id=0,
    min_id=0,
    add_offset=0,
    hash=0
))

3. 安全建议

保护 API 凭证

# 不要硬编码凭据
import os
from dotenv import load_dotenv

load_dotenv()
api_id = int(os.getenv('TELEGRAM_API_ID'))
api_hash = os.getenv('TELEGRAM_API_HASH')

Session 文件安全

  • Session 文件包含认证密钥，需要妥善保管
  • 不要将 Session 文件提交到版本控制
  • 生产环境使用加密存储

4. 遵守 Telegram ToS

使用 Telethon 时必须遵守 Telegram 服务条款：

  • 不要发送垃圾消息
  • 不要频繁调用 API
  • 不要尝试破解或滥用 Telegram 服务
  • 违规可能导致账户被封禁

八、实战案例

1. 自动回复机器人

from telethon import TelegramClient, events

client = TelegramClient('bot_session', api_id, api_hash).start(bot_token='你的token')

@client.on(events.NewMessage(pattern='/help'))
async def help_handler(event):
    help_text = '''
    可用命令：
    /start - 开始使用
    /help - 显示帮助
    /info - 获取信息
    '''
    await event.reply(help_text)

@client.on(events.NewMessage(pattern='/info'))
async def info_handler(event):
    sender = await event.get_sender()
    await event.reply(f'你的 ID：{sender.id}')

print('Bot 运行中...')
client.run_until_disconnected()

2. 消息转发器

from telethon import TelegramClient, events

client = TelegramClient('forwarder_session', api_id, api_hash)

SOURCE_CHAT = 'source_chat'
TARGET_CHAT = 'target_chat'

@client.on(events.NewMessage(chats=SOURCE_CHAT))
async def forward_handler(event):
    # 转发消息
    await client.forward_messages(TARGET_CHAT, event.message)

print('转发器运行中...')
client.start()
client.run_until_disconnected()

3. 文件备份工具

from telethon import TelegramClient
import asyncio

client = TelegramClient('backup_session', api_id, api_hash)

async def backup_chat(chat_username):
    """备份聊天中的所有媒体文件"""
    os.makedirs(f'backup/{chat_username}', exist_ok=True)

    async for message in client.iter_messages(chat_username):
        if message.media:
            file_path = await message.download_media(f'backup/{chat_username}/')
            print(f'已下载：{file_path}')

async def main():
    chats = ['chat1', 'chat2', 'chat3']
    for chat in chats:
        await backup_chat(chat)

client.start()
client.loop.run_until_complete(main())

4. 消息分析工具

from telethon import TelegramClient
from collections import Counter
import datetime

client = TelegramClient('analysis_session', api_id, api_hash)

async def analyze_chat(chat_username):
    """分析聊天统计"""
    messages = await client.get_messages(chat_username, limit=1000)

    # 消息统计
    total = len(messages)
    text_only = sum(1 for m in messages if m.text and not m.media)
    media_only = sum(1 for m in messages if m.media)

    # 最活跃用户
    senders = Counter()
    for m in messages:
        if m.sender:
            senders[m.sender.first_name or m.sender.username] += 1

    print(f'''
    聊天分析：{chat_username}
    总消息数：{total}
    纯文本：{text_only}
    媒体文件：{media_only}
    最活跃用户：{senders.most_common(5)}
    ''')

client.start()
client.loop.run_until_complete(analyze_chat('目标用户名'))

九、故障排查

1. 常见问题

认证失败

# 问题：FloodWaitError
# 原因：请求过于频繁
# 解决：等待指定时间后重试

from telethon.errors import FloodWaitError

try:
    await client.send_message('username', 'Hello')
except FloodWaitError as e:
    print(f'需要等待 {e.seconds} 秒')
    await asyncio.sleep(e.seconds)

连接超时

# 配置超时和重试
client = TelegramClient(
    'session',
    api_id,
    api_hash,
    timeout=30,
    connection_retries=5
)

Session 损坏

# 删除损坏的 session 文件重新登录
import os
if os.path.exists('session_name.session'):
    os.remove('session_name.session')

2. 调试技巧

启用日志

import logging
logging.basicConfig(level=logging.DEBUG)

# 或者只显示 Telethon 日志
logging.getLogger('telethon').setLevel(logging.DEBUG)

测试连接

# 测试 API 连接
async def test_connection():
    me = await client.get_me()
    print(f'连接成功，用户：{me.first_name}')

client.loop.run_until_complete(test_connection())

十、扩展生态

1. 相关项目

Telethon 生态系统

  • telethon_generator：TL 定义生成器
  • telethon_examples：官方示例集合
  • docs.telethon.dev：完整文档

替代方案

  • Pyrogram：另一个 MTProto Python 库
  • python-telegram-bot：Bot API 封装

2. 版本迁移

从 1.0 之前版本迁移

Telethon 1.0 版本进行了重大更新，迁移时需要注意：

  • 大量 API 变更
  • 异步模型调整
  • 事件系统重构

参考官方文档的"Compatibility and Convenience"章节。

十一、项目数据

1. GitHub 统计

  • Stars：11,2k
  • Forks：1.5k
  • Contributors：184
  • Used by：69,900+ 项目
  • 语言：Python 97.5%

2. 开发状态

  • 最新版本：v1.x
  • 许可证：MIT License
  • 文档：docs.telethon.dev
  • 活跃维护：持续更新

参考资料

  1. Telethon GitHub 仓库
  2. Telethon 官方文档
  3. Telegram API 开发者平台
  4. MTProto 协议文档
最后修改：2026 年 01 月 15 日
© 允许规范转载
如果觉得我的文章对你有用，请随意赞赏