用 Telethon 编写简单的 TeleGram Python 机器人
Telethon 是一个面向 Python 的 Telegram 客户端库,使用 MTProto 协议处理登录、消息和媒体事件。它把网络请求、账号会话和事件分发封装在异步接口中,适合编写消息监听、关键词回复和自动化处理脚本。常见 Bot API 方案使用机器人令牌,而 Telethon 通常登录一个用户账号,因此脚本只能访问该账号本来可见的会话。
下面的示例会在私聊中处理 /start 与 /echo:前者发送问候,后者把命令后的文字回显给发送者。准备 Python 3.10 以上版本、Telegram 账号和开发者凭据即可开始;测试时使用测试账号,部署前遵守平台规则并控制私聊数据的保存范围。
运行环境与开发者凭据
先准备 Python 3.10 或更高版本,并在独立虚拟环境中安装依赖。这样不会污染系统环境,也方便以后升级或删除项目。
python -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install telethon python-dotenv
Windows 系统可使用 .\.venv\Scripts\activate 激活环境。若还要在桌面端或手机端核对客户端,可在客户端下载说明中按 Windows、macOS、Linux、iOS、Android 或网页端选择版本;客户端只负责界面和账号登录,脚本仍需要单独配置。
接着在 Telegram 开发者门户创建应用,取得数字形式的 API ID 和字符串形式的 API_HASH。它们是 Telethon 连接账号所需的应用凭据,不要与 Bot API 的机器人令牌混用。将凭据放进项目目录的 .env 文件,代码中只读取环境变量:
API_ID=12345678
API_HASH=替换为真实值
编写首个消息响应机器人
新建 echo_bot.py,让程序启动后保持连接,并在收到私聊消息时执行对应命令。下面的代码只处理私聊,避免机器人意外回应群组或频道内容。
import asyncio
import os
from dotenv import load_dotenv
from telethon import TelegramClient, events
load_dotenv()
client = TelegramClient(
"echo_session",
int(os.environ["API_ID"]),
os.environ["API_HASH"],
)
@client.on(events.NewMessage(incoming=True))
async def echo_handler(event):
if not event.is_private:
return
text = event.message.text or ""
if text.startswith("/start"):
await event.respond("你好,我是简单的 Python 机器人。发送 /echo 内容 试试。")
elif text.startswith("/echo "):
content = text[6:].strip()
if content:
await event.respond(f"收到:{content}")
else:
await event.respond("请在命令后写入内容。")
async def main():
await client.start()
await client.run_until_disconnected()
if __name__ == "__main__":
asyncio.run(main())
events.NewMessage(incoming=True) 只接收新收到的消息,event.is_private 将处理范围限制在私聊。await event.respond 会把回复发回原聊天;TelegramClient 则负责维护登录状态和异步连接。若要增加功能,可以在同一个事件处理器里继续判断其他命令。
本地运行与首次登录
在项目目录执行 python echo_bot.py。第一次运行时,终端会询问手机号和 Telegram 验证码;如果账号开启了两步验证,还需要输入对应的密码。验证成功后,程序会生成 echo_session.session 会话文件,后续启动通常可以直接恢复连接。
测试时,用另一个 Telegram 账号给运行脚本的账号发送私聊消息。先发送 /start 查看问候,再发送 /echo 你好 检查文字回显。由于事件过滤器只接收传入消息,脚本账号自己发出的内容不会触发处理器。
如果出现 ApiIdInvalidError,先检查 .env 是否位于运行目录,以及 API_ID、API_HASH 是否填错;验证码过期时重新运行即可。遇到 FloodWaitError 时不要连续重试,应按照服务器返回的等待时间暂停请求。
网络连接与稳定运行
如果运行电脑的无线信号不稳定,消息监听可能频繁断开。可以参考无线中继设置调整 Edimax 路由器,通过以太网连接主路由并扩大无线覆盖。网络连通后,再检查系统时间、代理设置和 Telegram 客户端的连接状态。
Telethon 的事件处理器运行在异步环境中,不要在其中调用阻塞式的 time.sleep 或耗时的同步网络请求,否则其他消息也会被卡住。需要等待时使用 await asyncio.sleep;触发 FloodWaitError 时按照返回的秒数等待,并在重试前确认操作不会造成重复回复。
若脚本需要长时间运行,可以使用 Linux 的 systemd、macOS 的 launchd 或 Windows 任务计划程序托管进程,并保存基础日志。同一会话文件不要同时交给多个进程使用,否则可能产生数据库锁定或重复处理消息的情况。
安全边界与上线检查清单
API_HASH 和 .session 文件都应视为敏感凭据。前者代表应用身份,后者保存账号会话,泄露后可能让他人取得不必要的访问能力。上线前可以按两组清单核对:
凭据保护
- 将
.env、echo_session.session和运行日志加入.gitignore,不要提交到代码仓库。 - 不要在截图、问题反馈页面或日志中公开
API_HASH、验证码和会话文件;一旦泄露,应立即更换相关凭据。 - 为登录账号开启两步验证,并限制保存会话文件的目录权限。
访问范围与数据处理
- 默认只处理私聊消息;若要接收群组内容,先确认账号拥有相应权限,并告知参与者。
- 为批量回复设置频率上限,遇到
FloodWaitError时按服务器给出的秒数等待。 - 日志只记录命令、时间等元数据,避免保存私聊正文、联系人信息和媒体文件。
- 遵守 Telegram 的服务规则、群组规则以及敏感内容限制,不把脚本用于骚扰或未授权转发。
完成检查后,再根据实际使用场景调整命令和日志字段。任何凭据都不应写进公开仓库,测试账号也不要与重要私人账号混用。
立即运行并扩展你的机器人
这个回声脚本可以作为消息自动化的起点,例如增加帮助命令、关键词回复、定时提醒或文件处理。使用数据库保存状态时,要让异步处理器正确管理连接;部署到服务器前,则应先确认登录流程、异常重启和日志轮换都能正常工作。若目标是面向公众的正式服务,应优先评估官方 Bot API,避免把个人账号会话当作公共机器人使用。
现在进入项目目录,执行 python echo_bot.py,用另一个账号发送 /start 和 /echo 你好,确认收到回复后即可完成第一次部署。