简介：面向希望用 Python 了解 Telegram 账户登录流程的开发者与学习者，这份程序型压缩包围绕“登录到电报”这一目标提供可参考的实现思路，适合具备基础 Python 语法认知、想研究第三方账号登录交互与客户端会话建立方式的人群。压缩包体积约 5.12MB，上游文件统计显示文件总数为 0，且未提供具体文件类型明细，因此更适合作为轻量级代码参考而非完整工程模板来使用。已有 5370 人浏览或学习，说明该主题在账号登录与自动化连接场景中具有一定关注度。读者可从中了解登录流程的组织方式、配置与运行思路以及常见异常排查方向，并据此梳理自己的 Python 小工具原型；若用于实际项目，仍需结合官方接口规范、账号安全策略与合规要求进行二次验证和改造。
1. 一个只有一行报错的登录脚本，把我卡了半个下午
前阵子帮人看一个自建的小工具，跑起来只吐一行AuthKeyUnregisteredError，日志里连是哪个账号、哪一步挂的都没有。翻回去看代码，整个登录逻辑塞在main.py的三十行里，验证码用input()硬等，登录成功后 session 文件写在临时目录，容器一重启就全没了。这份login-to-telegram源码包解决的正是这类问题：用 Python 把「手机号 → 验证码 → 两步验证 → 会话落盘」这条链路拆成可读、可复用、可排错的模块，而不是一坨跑一次就废的脚本。它适合想给自己账号做消息归档、通知转发、桌面端自动化的开发者，也适合刚开始接触 MTProto 客户端协议、想找一个能跑通的最小样例的人。下面按「先看结构、再看核心代码、最后看怎么改」的顺序拆。
2. 架构与目录：登录不是一次请求，而是一条状态机
2.1 三条登录路径与四个状态
很多人第一次写这类工具，会默认「登录 = 发验证码 + 填验证码」两步。真跑起来就会发现分支比想象的多：有开了两步验证的账号，有走二维码授权的登录方式，还有「上次已经登录过、这次只要把会话捞起来」的短路情况。源码包里把这四种情况收敛成一个状态机，状态只有四个，但覆盖了全部路径。
| 状态 | 触发条件 | 下一步动作 |
|---|---|---|
INIT | 进程启动，会话文件不存在或已失效 | 发起验证码请求 / 二维码登录 |
CODE_SENT | send_code_request返回成功 | 等待用户输入验证码 |
NEED_PASSWORD | 验证码校验通过但账号开启了两步验证 | 等待用户输入密码 |
AUTHORIZED | is_user_authorized()为真 | 直接返回当前用户信息 |
状态机的价值不在于「优雅」，而在于可恢复。用户在CODE_SENT阶段关掉终端去吃了个饭，回来再跑，程序应该能识别出「验证码已过期，需要重新发」，而不是傻等一个永远填不对的输入。常见做法是给每个状态配一个超时戳和一个持久化的中间态文件，状态字段写盘，重启时读回来判断是否过期。这一步不做，后面所有异常处理都是在补漏。
另外要区分「会话有效」和「账号可用」两件事。is_user_authorized()返回 True 只代表本地 auth key 还在，不代表这个 key 还能用——服务端可以随时吊销某个会话。所以状态机的AUTHORIZED出口必须再包一层网络调用（取一次自己的用户信息）做真实校验，失败就退回INIT。这个细节决定了你的定时任务在半夜会不会因为一个死会话直接崩掉。
2.2 目录结构与模块职责
源码包的目录不算深，但分层很清楚：core/放协议和登录逻辑，ui/只负责和用户交互，utils/是横切能力。这种拆法最大的好处是——想把它塞进 GUI 或者一个本地 HTTP 服务时，只要换掉ui/，core/一行不用动。
login-to-telegram/ ├── main.py # 入口，装配 settings + client + 登录流程 ├── config.py # 从环境变量读取配置，统一校验 ├── requirements.txt ├── .env.example # 配置模板，不含真实凭据 ├── core/ │ ├── client.py # 客户端构造：设备信息、超时、重试 │ ├── auth.py # 状态机主体：发码、验证、两步验证 │ ├── session_store.py # 会话的读写、加密、多账号路由 │ └── errors.py # 错误码到人话的映射表 ├── ui/ │ ├── cli.py # 命令行主循环 │ └── prompts.py # 输入校验：手机号格式、验证码位数 ├── utils/ │ ├── logger.py # 结构化日志 + 脱敏过滤器 │ └── mask.py # 手机号、会话串的掩码处理 └── sessions/ # 会话文件落盘目录，已在 .gitignore 里几个模块的职责边界值得单独说一句。config.py只做「读 + 校验 + 报错」，不做默认值兜底到能跑为止——缺api_id就直接抛异常退出，比跑到一半才发现要好。errors.py里维护的是一张纯字典，把协议层抛出来的异常类名映射成中文说明和「是否可重试」的布尔值，这个表是后面重试策略的唯一数据源，不要在业务代码里散写except。session_store.py是唯一允许直接碰sessions/目录的地方，其他模块一律通过它拿客户端实例。
2.3 环境准备与第一次跑通
依赖不多，核心就一个客户端库加一个配置管理。虚拟环境别省，这类工具经常要在不同机器上跑，系统 Python 里装一堆东西迟早出事。
# 建虚拟环境并安装依赖 python -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate pip install -r requirements.txt # 复制配置模板，填入自己的 api_id / api_hash / 手机号 cp .env.example .env.env里的字段不多，但每一项都会影响后续行为，尤其是设备信息那几项——它们会出现在账号的「活跃会话」列表里，填得太随意会让自己的设备列表看起来像一堆陌生人。
| 变量 | 示例 | 说明 |
|---|---|---|
API_ID | 1234567 | 应用标识，整数类型，缺了直接报错 |
API_HASH | abcdef... | 应用密钥，不要提交到仓库 |
PHONE | +8613800000000 | 带国际区号，程序里统一做一次格式化 |
SESSION_DIR | ./sessions | 会话文件目录，容器部署时挂卷 |
DEVICE_MODEL | Desktop-Tool | 显示在活跃会话里的设备名 |
APP_VERSION | 1.0.0 | 同上，建议固定，别每次随机 |
填完之后python main.py就能跑。第一次运行会依次问你手机号（如果没配）、验证码、可能还有两步验证密码；第二次运行如果会话还有效，应该直接打印出当前登录的账号名然后退出。判断「跑通了」的标准不是「没报错」，而是「第二次启动走了短路分支」。这一点在验收时很容易被忽略，很多人第一次登录成功就以为完事了，结果三天后发现每次都在重新登录。
提示：
sessions/目录务必加进.gitignore。会话文件等价于账号的登录凭据，落到公开仓库里等于把钥匙挂在门上。
3. 核心链路拆解：从 api_id 到第一个可用会话
3.1 客户端构造：参数决定了后面一半的坑
客户端实例是整个程序的根，构造时传进去的参数会一路影响到连接稳定性、活跃会话显示、以及异常类型。源码包把构造逻辑单独抽到core/client.py，就是为了让这些参数集中在一处、可被测试覆盖。
# core/client.py from telethon import TelegramClient from telethon.sessions import StringSession def build_client(settings, session=None) -> TelegramClient: """构造客户端并完成连接，失败时抛给上层统一处理""" client = TelegramClient( session or settings.session_dir / settings.session_name, settings.api_id, settings.api_hash, # 设备信息会出现在账号的活跃会话里，固定下来便于识别 device_model=settings.device_model, system_version=settings.system_version, app_version=settings.app_version, lang_code="zh", system_lang_code="zh-hans", # 连接层参数：默认值偏保守，长时间挂机需要显式调大 connection_retries=3, # 单次连接内部重试次数 retry_delay=2, # 每次重试间隔（秒） timeout=20, # 单请求超时，别设太小 auto_reconnect=True, # 断线自动重连，长驻进程必开 request_retries=3, ) return clientbuild_client刻意没有把client.connect()放在里面，而是让调用方显式 await。原因是连接是唯一一个「可能阻塞十几秒」的动作，放在构造函数里会让整个初始化过程变成一个黑匣子——出了问题你都不知道是参数错了还是网络问题。拆开之后，调用方可以给connect()单独套一层asyncio.wait_for，超时就明确报「连接超时」，而不是等一个含糊的堆栈。
device_model和app_version这两项，我一般会在配置里写死，不做随机化。随机设备名看起来"更安全"，实际后果是自己的活跃会话列表里全是看不懂的条目，某天想踢掉一个陌生会话时根本分不清哪个是自己的工具。
3.2 验证码登录与两步验证的实际写法
登录主体在core/auth.py，核心是一个带状态判断的异步函数。这里最容易写错的是两处：phone_code_hash的复用，以及两步验证的异常分支。
# core/auth.py from telethon.errors import ( SessionPasswordNeededError, PhoneCodeInvalidError, PhoneCodeExpiredError, ) async def login_with_code(client, phone: str, ask_code, ask_password): """ ask_code: (sent) -> str 拿验证码的回调，CLI 里就是 input ask_password: () -> str 拿两步验证密码的回调 返回当前登录用户对象 """ # 已有会话且真实有效，直接短路 if await client.is_user_authorized(): return await client.get_me() sent = await client.send_code_request(phone) # phone_code_hash 必须和本次请求绑定，不能缓存后重复使用 code = ask_code(sent) try: await client.sign_in( phone=phone, code=code, phone_code_hash=sent.phone_code_hash, ) except PhoneCodeInvalidError: # 验证码本身错了，可以原地重试，不消耗新的发送次数 raise except PhoneCodeExpiredError: # 过期只能重发，此时旧 hash 已失效 raise except SessionPasswordNeededError: # 走到这里说明账号开了两步验证，验证码这关已经过了 password = ask_password() await client.sign_in(password=password) return await client.get_me()这段代码的价值在于异常分工。PhoneCodeInvalidError和PhoneCodeExpiredError被分开处理，因为它们对用户的意义完全不同：前者是「你输错了，再输一次」，后者是「你磨蹭太久，得重新发一条」。如果把它们合并成一个except Exception然后统一提示「登录失败」，用户唯一能做的就是反复重试，而重试又可能触发频率限制，最后陷入更麻烦的死循环。
ask_code/ask_password做成回调而不是直接写input()，是为了让core/不依赖任何交互方式。命令行里传lambda s: input("验证码: ")，将来接图形界面就传一个读输入框的函数，接本地服务就传一个读请求体的函数，auth.py一个字都不用改。
还有一点：is_user_authorized()那行短路必须在发码之前。它在本地判断，不发网络请求，成本几乎为零，但能避免大量无谓的验证码发送。我见过不少实现把这行漏掉，结果每次启动都给账号发一条登录码，几天下来被限流了还不知道为什么。
3.3 会话落盘：文件会话与字符串会话的取舍
登录成功后最关键的产物不是用户信息，而是会话本身。源码包同时支持两种形态：SQLite 文件会话和字符串会话，由session_store.py统一封装。
# core/session_store.py from pathlib import Path from telethon.sessions import StringSession def load_session(path: Path): """优先读字符串会话（便于放到配置中心），回退到文件会话""" str_file = path.with_suffix(".session.str") if str_file.exists(): return StringSession(str_file.read_text().strip()) return str(path) # 交给库自己管理 SQLite 文件 def export_string(client, path: Path) -> None: """登录成功后导出一份字符串会话，便于迁移或备份""" data = StringSession.save(client.session) path.with_suffix(".session.str").write_text(data)| 形态 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| SQLite 文件 | 库原生支持，写入自动化 | 不支持多进程并发写 | 单机单进程的常驻脚本 |
| 字符串会话 | 可序列化，便于加密存储 | 需要自己管理导出时机 | 容器部署、多环境迁移 |
文件会话最省事，但有个硬约束：同一个文件不能被两个进程同时写，否则会遇到 SQLite 的锁冲突。字符串会话则相反，它把整个会话状态压成一段文本，你可以把它加密后丢进配置中心或者环境变量里，容器重建时直接读回来，不需要挂卷。
我一般会在登录成功后立刻导出一份字符串会话，作为「后悔药」。原因是文件会话一旦损坏（比如进程被 kill、磁盘写满），基本没法恢复；而字符串会话只要备份过一份，就还能把账号捞回来。导出时机选在login_with_code成功之后、进入业务逻辑之前，这时候会话状态是最干净的。
注意：字符串会话等同于账号凭据，凡是打印、写日志、上报的地方都必须经过掩码处理，只保留前 8 位和后 4 位。
4. 异常边界：把「能跑通」变成「能长期跑」
4.1 错误码映射与重试策略
协议层抛出的异常类型少说有二十来种，直接往业务代码里塞try/except是最快的写法，也是最难维护的写法。源码包的core/errors.py用一张表把它们收敛成三类：可重试、需用户介入、致命错误。
# core/errors.py from telethon.errors import ( FloodWaitError, AuthKeyUnregisteredError, SessionRevokedError, PhoneNumberBannedError, ApiIdInvalidError, ) # 需要等待后重试的异常：带上建议等待秒数 RETRYABLE = { "FloodWaitError": lambda e: e.seconds, "TimeoutError": lambda e: 5, "ConnectionError": lambda e: 5, "ServerError": lambda e: 10, } # 必须人工处理，重试没有任何意义 FATAL = { "AuthKeyUnregisteredError": "会话已失效，需要重新登录", "SessionRevokedError": "会话已在其他设备被终止，需要重新登录", "PhoneNumberBannedError": "该号码无法登录，请联系平台", "ApiIdInvalidError": "api_id / api_hash 不匹配，检查配置", } def classify(exc: Exception) -> tuple[str, int]: name = type(exc).__name__ if name in RETRYABLE: return ("retry", RETRYABLE[name](exc)) if name in FATAL: return ("fatal", 0) return ("unknown", 0)这张表的意义是把「要不要重试」这个决策从业务代码里抽出来，变成一个纯函数。好处是可以单独写测试覆盖：给一个FloodWaitError(seconds=30)，断言classify返回("retry", 30)，不用真去触发限流。
FloodWaitError是这类工具里最常遇到的一种。它带着一个seconds字段，明确告诉你还要等多久，照着等就行。真正要命的是「等完之后立刻再撞一次」——常见原因是程序里没有冷却状态，重试循环跑得比冷却时间还快。我的做法是把seconds写进一个本地冷却文件，进程启动时先读它，没到期就直接退出并提示剩余时间，这样即使被外部脚本反复拉起也不会加剧问题。
4.2 设备信息与登录通知
设备信息不只是「填着好看」。同账号的多设备登录会触发平台的登录通知，里面显示的就是device_model和app_version这几项。如果它们每次都变，账号的活跃会话列表会迅速膨胀成一堆无法辨认的条目，长期下来既难看也难清理。
写法上没什么技巧，就是别随机化。system_version用一个稳定的值（比如宿主机的系统版本），app_version用你自己的工具版本号，device_model用工具名。三样东西固定下来，每次登录在会话列表里都是同一条，想踢掉的时候一眼就能认出来。
另一个容易被忽略的点是「登录成功后主动取一次自己的信息」。这不只是为了打印欢迎语，而是给会话做一次真实校验。get_me()走的是完整请求链路，能验证 auth key 是否被服务端接受。如果这里失败，说明会话虽然本地判定有效，但服务端已经不认了，此时应该清掉本地会话并提示重新登录，而不是带着一个死会话继续跑业务逻辑。
4.3 日志脱敏与运行时可观测性
登录类工具最怕的不是报错，是「报错了但看不出是哪一步、哪个账号」。源码包里给日志加了一层过滤器，凡是被标记为敏感的字段在格式化阶段就被替换掉，避免下游有人直接把日志丢进公共的收集系统。
# utils/logger.py import logging, re MASK_RULES = [ (re.compile(r"(\+?\d{1,3})\d{4,8}(\d{4})"), r"\1****\2"), # 手机号 (re.compile(r"([A-Za-z0-9+/=]{8})[A-Za-z0-9+/=]{16,}([A-Za-z0-9+/=]{4})"), r"\1...\2"), # 会话串 ] class MaskFilter(logging.Filter): def filter(self, record: logging.LogRecord) -> bool: msg = record.getMessage() for pattern, repl in MASK_RULES: msg = pattern.sub(repl, msg) record.msg, record.args = msg, () return True这段过滤器的作用点很重要：它在getMessage()之后、真正输出之前动手，所以无论调用方用的是%s还是 f-string，最终落到磁盘上的都是脱敏后的文本。用正则而不是字段级控制，是因为日志里出现敏感信息的路径太多——异常堆栈、第三方库的调试输出、请求回显，逐处拦截不现实。
日志里我固定会打三样东西：当前状态机的状态、本次操作的耗时、以及是否命中了重试。有这三样，定位问题基本不用翻代码。比如看到「状态 CODE_SENT，耗时 0.3 秒，未重试」，紧接着一条PhoneCodeInvalidError，那就非常明确是输入错了，跟网络、跟会话都没关系。
5. 避坑记录：五个把我按在键盘前的问题
5.1 连接与发码阶段的三个坑
现象：第一次运行直接卡住，二十秒后抛TimeoutError，重跑一次又好了。
原因：客户端默认的连接参数偏保守，第一次握手要建多条连接，机器启动初期的 DNS 抖动或者本机负载高都会让其中一条慢下来，超过默认超时就被打断。
解决：把timeout显式设到 20 秒以上，connection_retries设 3，然后在connect()外层再套一层asyncio.wait_for做整体保护。两层超时要分开，内层管单次，外层管总时长，混在一起会出现「重试还在跑但外层已经放弃了」的尴尬局面。
现象：验证码明明填对了，还是报PhoneCodeExpiredError。
原因：验证码有效期只有几分钟，而input()会无限期阻塞。用户切出去处理别的事，回来再填，码早就作废了。另外一个高频原因是在重试逻辑里复用了上一次的phone_code_hash。
解决：ask_code回调外层加个软超时（比如 120 秒），到点直接放弃并提示重发；phone_code_hash只跟当次send_code_request的返回值绑定，绝不缓存到变量里跨次使用。
现象：连续调试几次之后，send_code_request开始抛FloodWaitError，提示要等几百秒。
原因：短时间内反复请求验证码，触发了频率限制。很多人遇到这种情况的第一反应是「换个号码试试」，实际上只会让问题扩大。
解决：捕获FloodWaitError读e.seconds，写进冷却文件，期间任何发码请求直接短路并打印剩余时间。冷却文件要放在会话目录旁边，别放/tmp——重启就丢的冷却等于没有冷却。
5.2 会话与运行阶段的两个坑
现象：本地调试一切正常，部署到容器后每天都要重新登录一次。
原因：会话文件写在了容器的可写层，容器重建就没了。或者用了字符串会话但没做持久化，每次启动都从空会话开始。
解决：文件会话挂载持久卷；字符串会话写进加密的配置存储。判断标准很简单——重启进程后看日志里有没有走「已授权短路」那条分支，走了才算真的持久化了。
现象：多开一个窗口跑同一份代码，报database is locked，或者其中一个进程的会话莫名失效。
原因：SQLite 文件会话不支持多进程并发写。两个进程同时持有同一个文件，写入顺序不确定，轻则锁冲突，重则会话状态被覆盖成中间态。
解决：给会话目录加一个文件锁，拿不到锁直接退出并提示；或者在设计上就规定单账号单进程，多账号按账号维度分目录。我倾向于后者，简单且没有额外依赖。真要做多进程共享，就把会话换成字符串形态，由主进程统一管理，子进程只读。
提示：这五条里有四条都能在本地复现，别等上线了再碰。尤其是冷却文件和多进程锁，本地多跑几次就能暴露出来。
6. 二次开发：把它改成你自己的登录中间件
代码能跑通只是起点，真正费力的是把它接进自己的系统。我从三个方向改过类似的项目，按投入产出排个序。
第一个改造点是回调注入。core/auth.py里的ask_code/ask_password已经是回调形式，你要做的是把它们换成自己的实现。接本地服务时，可以把登录拆成「发起」和「提交」两个接口：发起接口调用send_code_request并把phone_code_hash存进一个带 TTL 的内存字典，返回一个login_id；提交接口拿着login_id和验证码去sign_in。这样前端就不用维护长连接，用户刷新页面也不影响流程。改这个的时候要注意，内存字典必须带过期清理，否则跑久了会积累一堆永不释放的中间态。
第二个改造点是会话存储。默认实现落本地文件，稍微正经一点的部署都该换成加密存储。我的做法是用对称加密把字符串会话包一层，密钥走环境变量或系统密钥环，数据库里只存密文和一个账号标识。这样即使存储介质泄露，会话也不能直接被拿去用。代价是多了一个依赖和一个密钥管理问题，如果你的工具只在自己机器上跑，这一步可以先跳过。
| 改造点 | 涉及文件 | 工作量 | 风险 |
|---|---|---|---|
| 交互方式换成接口 | ui/+core/auth.py | 半天 | 中间态泄漏 |
| 会话加密存储 | core/session_store.py | 半天 | 密钥丢失不可恢复 |
| 多账号路由 | config.py+session_store.py | 一天 | 并发与目录隔离 |
| 结构化日志接入 | utils/logger.py | 两小时 | 脱敏规则漏配 |
第三个改造点是验证方式。登录逻辑最麻烦的地方在于它依赖真实账号，没法在 CI 里跑端到端。我的处理是把「决策」和「请求」分开测：classify()、状态迁移函数、手机号格式化这些纯函数用普通单测覆盖，构造各种异常对象断言返回值；真正的网络调用只留一个手工触发的冒烟脚本，跑之前先get_me()确认会话活着，再走一遍完整流程。冒烟脚本我会特意打印每一步的状态和耗时，因为登录失败时最有价值的信息往往不是异常本身，而是「卡在了哪一步」。
顺带说个具体的技巧：调试期把session_name加上时间戳后缀，每次跑都是一份全新的会话，能避免「旧会话掩盖了新代码 bug」这种玄学问题。等逻辑稳定了再把后缀去掉，切回固定名称。
血泪经验是——我曾经为了省事，把send_code_request和sign_in塞进同一个重试装饰器里，结果一次FloodWaitError之后，装饰器把发码也重试了，直接把冷却时间从 30 秒顶到了几小时。从那以后我每次给这类项目加重试，都强制走一遍「哪些异常可以重试、哪些操作幂等」的检查，发码和提交验证码永远分开处理。希望帮到你。
本文还有配套的精品资源，点击获取
