☰
Telegram机器人源码落地:Webhook接入、后台管理与群发避坑指南
2026/10/28 23:56:56 网站建设 项目流程

简介:全功能Telegram机器人多功能后台源码内含可直接上手的完整项目,面向需要快速搭建TG机器人服务的开发者与社群运营者。内置TRX兑换、闪兑、关键字监控、群管、USDT监控、余额查询、推广分享等常见功能,并配有可视化后台管理界面,可对各项功能进行配置与开关,适合中高级PHP开发者二次开发或直接部署使用,也对有PHP基础的初级开发者具有模块设计参考价值。资源共36个文件,以28个PHP脚本为核心逻辑,辅以PNG图标界面、TXT使用说明、env环境配置示例及HTML页面等,整包约28.63MB,目录结构紧凑,便于对照部署与快速排错。目前已有2468人学习下载。通过源码可掌握机器人功能模块的代码组织方式、后台与机器人交互的实现思路,并参考环境变量配置与部署要点,对搭建生产级TG机器人具有直接参考价值。

1. 全功能 tgbot 源码第一晚:先分清“能跑”和“敢用”

拿到一份标注“全功能”的 tgbot 源码,大多数人第一件事就是解压、找 token、直接跑。结果往往是机器人能收发消息,但那个“多功能有后台版”里的后台登录页要么转圈,要么白屏。问题不在源码,而在这类 Telegram 机器人工程根本不是一套程序,而是三套东西捆在一起:负责收发消息的 Bot 服务、存放用户和任务数据的数据库、给运营用的 Web 管理后台。三者靠同一个 token 或连接串才能串起来,任何一个点断掉,整个包就只剩“挂机”这个功能。

这篇文章以带后台的多功能 Telegram 机器人为对象,从 token 获取、webhook 接入、模块拆分、后台表结构到群发任务队列,按落地顺序讲清楚。读完你会知道怎么把一个源码包变成自己能改、能交付、能交给运营用的服务,也知道哪些坑根本不该再踩一遍。

2. 后端先通:从 BotFather 取 token 到 webhook 最小闭环

2.1 BotFather 创建机器人与 token 的权限边界

Telegram 机器人没有“注册码”,只有 BotFather 这一个入口。找一个账号对 BotFather 发/newbot,按提示填显示名和用户名,最后会给出一行123456789:AAF...格式的 token。这个 token 就是机器人的全部控制权:拿到它的人能通过 HTTP API 任意发消息、删消息、改头像、踢人。所以源码包落地前的第一件事不是看代码,而是确认你手里的 token 值,然后决定是复用还是换掉。

很多人把 BotFather 里的“权限”和后台的“权限”搞混。BotFather 能设置的是/setjoingroups(是否允许被拉进群)、/setprivacy(群聊里是否只接收被提到的消息)、/setcommands(命令菜单)这些平台级开关;而后台管理系统里的“角色”“管理员名单”,是应用层自己写的逻辑,Telegram 根本不知道。也就是说,你把某人设为后台超管,只是数据库里一行记录,和 token 权限互不影响。这个边界想不清楚,后面做权限模块时很容易写出“以为限制了、其实谁都能调”的漏洞。

一个建议:凡是源码包里带着的 token 或.env,落地前统一用 BotFather 的/mybots → API Token → Revoke token换掉。Revoke 是一次性操作,旧 token 立刻失效,新 token 立即生效,相当于给源码包吃了颗后悔药,比去翻 git 历史删提交省事得多。换完的新 token 只写进服务器环境变量,不落进任何文件。

2.2 polling 还是 webhook:一张对比表帮你选

Bot 服务代码写得再好,接收消息的方式选错,后台体验都会变怪。Telegram Bot API 只提供两种收消息的姿势:getUpdates 轮询(polling)和 setWebhook 回调(webhook)。多数源码包默认用 polling,因为本地开发最省事;但带后台的多功能机器人一旦上服务器,我一般会切到 webhook。

两边的差别集中在延迟、端口和出站请求上。

对比维度polling(getUpdates)webhook
启动成本低,本机直接跑需要一个公网 HTTPS 域名
消息延迟取决于轮询间隔,通常 1~2 秒毫秒级,Telegram 主动推过来
服务器出站请求每 1~2 秒一次 getUpdates,空闲也占连接有更新才请求,空闲零占用
多实例部署两个实例抢更新,必须做锁或切分天然只推给一个地址,扩展时再分发
本地联调方便,适合写代码阶段需要临时暴露一个公网回调地址
后台联动事件流在 bot 进程内部,后台要再去查库回调入口就是 HTTP 接口,后台能复用同一套鉴权

对带后台的机器人来说,webhook 还有一个隐蔽好处:Telegram 会在回调请求头里带X-Telegram-Bot-Api-Secret-Token,你可以用它验证来源,等于给 bot 入口加了一道前置守卫。polling 模式下没有这层机制,任何能拿到 token 的人都可以模拟 Bot API 向你的服务灌数据。

团队开发阶段我会用 polling 开着热重载写代码;进入交付阶段必切 webhook。两种方式共用同一套业务代码,切换只是配置环境变量TG_BOT_MODE=polling/webhook,不用动 handler。

2.3 FastAPI 挂接收端点:最小可用的 tgbot 代码

切 webhook 时,最简单可靠的做法是用 FastAPI 自己写接收端点,而不是套大而全的框架。下面是一份可以直接抄的最小工程,核心文件叫main.py,包含设置 webhook 用到的全部逻辑。

# main.py import os from fastapi import FastAPI, Request, Response from aiogram import Bot, Dispatcher from aiogram.types import Update TG_BOT_TOKEN = os.environ["TG_BOT_TOKEN"] WEBHOOK_SECRET = os.environ["TG_WEBHOOK_SECRET"] # 自定义,不告诉别人 app = FastAPI() bot = Bot(token=TG_BOT_TOKEN) dp = Dispatcher() # 业务 handler 由其它模块通过 dp.include_router(...) 注册进来 @app.post("/webhook") async def telegram_webhook(request: Request): # 1. 先验证来源,不是 Telegram 的直接拒绝 secret = request.headers.get("X-Telegram-Bot-Api-Secret-Token", "") if secret != WEBHOOK_SECRET: return Response(status_code=403) # 2. 把 JSON 更新转成 aiogram 的 Update 对象 update = Update.model_validate(await request.json()) # 3. 交给 dispatcher 按已注册的规则路由 await dp.feed_update(bot, update) return Response(status_code=200) @app.get("/healthz") async def healthz(): return {"ok": True}

逻辑说明:第一个if做来源校验,避免别人拿假更新刷你的接口;第二步会 pydantic 校验 Telegram 传过来的 JSON,字段不合法直接抛异常;第三步feed_update把更新对象喂给 dispatcher,由命令过滤器、消息类型过滤器决定去向。/healthz是健康检查端点,后面第 6 章会用到。

代码跑起来后,还要在服务器上执行一次 setWebhook,把 Telegram 的推流地址指到你的域名:

curl -X POST "https://api.telegram.org/bot${TG_BOT_TOKEN}/setWebhook" \ -F "url=https://bot.example.com/webhook" \ -F "secret_token=${TG_WEBHOOK_SECRET}"

参数说明:url必须是公网可达的 HTTPS 地址,并且和代码里的路由完全一致;secret_token会原样出现在每次回调的 header 里,代码端校验用的就是这个值。设置完可以用getWebhookInfo确认状态:

curl "https://api.telegram.org/bot${TG_BOT_TOKEN}/getWebhookInfo"

返回里的url应是你填的地址,pending_update_count一般应为 0。如果发现last_error_message有内容,说明回调地址没通,先去查 HTTPS 证书和域名解析,而不是急着改代码。

3. 多功能从哪来:命令分发、按钮回调和 FSM 的三层设计

3.1 分发器:让命令、文本、按钮回调查询各回各家

“全功能”的源码往往把几百个功能塞进同一个文件,看起来功能很多,实际上是把自己的维护时间搭进去了。功能多的机器人,核心不是“功能枚举”,而是分发器的路由能力。aiogram 3 的Router机制就是为了解决这个问题:每个业务模块注册自己的 Router,主程序只负责把多个 Router 合并到 Dispatcher。

# routers/start.py from aiogram import Router from aiogram.filters import CommandStart, Command from aiogram.types import Message router = Router() # 这个模块只处理 /start、/help @router.message(CommandStart()) async def start_cmd(message: Message): await message.answer( "功能菜单:\n" "/search 搜索\n" "/subscribe 订阅\n" "/report 日报" ) @router.message(Command("help")) async def help_cmd(message: Message): await message.answer("把 / 开头的命令发给我就行,其它文本走搜索。")
# main.py 中合并 from routers import start, search, subscribe, admin for r in [start.router, search.router, subscribe.router, admin.router]: dp.include_router(r)

逻辑说明:@router.message(CommandStart())表示这个函数只接/start命令;Command("help")单独接/help。不在这些命令里的普通文本,由别的 Router 的文本过滤器去接,互不干扰。每个业务模块有自己的 Router 文件,加新功能时新建一个文件、在 main.py 里多 include 一行,老功能完全不受影响。这个模式是“多功能”源码能不能长期维护的分水岭:所有 handler 写在一个文件里的包,改一个功能等于重试整个机器人。

3.2 权限分级:管理员身份放数据库而不是写死在代码里

很多源码包权限做得粗糙,直接在文件顶部写ADMIN_IDS = [123456789]。这种做法有两个问题:一是改管理员要重启进程,二是所有看到源码的人都知道谁有权限,后台形同虚设。带后台版的机器人,权限应该是一张数据库表,后台管理页改的是这张表,bot 服务的鉴权也读这张表。

# utils/permissions.py from aiogram import types import db async def is_admin(user_id: int) -> bool: row = await db.fetch_one( "SELECT role FROM admins WHERE tg_id=%s AND is_active=1", (user_id,) ) return row is not None async def is_super_admin(user_id: int) -> bool: row = await db.fetch_one( "SELECT role FROM admins WHERE tg_id=%s AND is_active=1", (user_id,) ) return bool(row and row["role"] == "super_admin")

反例:

# 不要这样写 ADMIN_IDS = {123456789} # 改权限还要改代码重启 @router.message(Command("broadcast")) async def broadcast(message: types.Message): if message.from_user.id not in ADMIN_IDS: await message.answer("无权限") return

参数说明:admins表里tg_id是管理员的 Telegram 用户编号,role区分 operator 和 super_admin 两种级别,is_active用于临时停用账号而不删记录。鉴权函数每次查询都走主键索引,对消息处理性能没有实质影响;真要压性能,可以套一层 30 秒的缓存。用数据库而不是硬编码的最大好处是:后台改完管理员名单,立即生效,不用重启 bot。

3.3 多步会话:用 FSM 收集用户输入不丢上下文

多功能机器人难免有“多步交互”场景:用户点按钮下单,机器人要先问商品、再问数量、最后确认。如果不用状态机,每一步都当成独立消息处理,用户答非所问时就会串台。aiogram 3 内置的 FSM(有限状态机)专门处理这个,它允许你给每个用户标记“当前在哪个流程的第几步”。

from aiogram.fsm.context import FSMContext from aiogram.fsm.state import State, StatesGroup class OrderFlow(StatesGroup): waiting_product = State() waiting_qty = State() @router.message(Command("order")) async def start_order(message: types.Message, state: FSMContext): await state.set_state(OrderFlow.waiting_product) await message.answer("请输入商品编号") @router.message(OrderFlow.waiting_product) async def get_product(message: types.Message, state: FSMContext): await state.update_data(product=message.text) await state.set_state(OrderFlow.waiting_qty) await message.answer("数量填数字,例如 2") @router.message(OrderFlow.waiting_qty) async def get_qty(message: types.Message, state: FSMContext): await state.update_data(qty=message.text) data = await state.get_data() await message.answer(f"订单确认: {data['product']} x {data['qty']}") await state.clear() # 流程结束,清掉状态

逻辑说明:set_state把用户推进到下一步,update_data把当步输入暂存在状态里,get_data汇总整个流程的数据,clear()退出状态机。如果用户中途发了一个和流程无关的命令,需要额外写一个进入新命令时state.clear()的兜底,否则用户会被卡在上一个流程里出不来。

FSM 默认把状态存内存,单进程没问题;一旦你有两个 bot 实例或发布到多机上,必须把存储切到 Redis,否则用户请求被两个实例分别处理,状态会互相覆盖。这是“本地跑得好好的,上服务器就玄学”的高发根源之一。

4. 有后台版的管理面板:表结构、登录鉴权与群发任务队列

4.1 后台在管什么:用户、群组、群发任务三类业务表

所谓的“有后台版”,本质是给 Telegram 机器人外挂了一个管理界面。后台管理的对象,逃不开三类:用户、群组、广播任务。数据库设计得好不好,决定后台页面能不能撑住,以及群发任务炸了之后你还能不能定位问题。下面是这套机器人最常用的一张建表脚本。

CREATE TABLE users ( id BIGINT PRIMARY KEY, username VARCHAR(64) NULL, first_name VARCHAR(128) NULL, is_blocked TINYINT(1) NOT NULL DEFAULT 0, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, last_seen_at DATETIME NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; CREATE TABLE admins ( id INT AUTO_INCREMENT PRIMARY KEY, tg_id BIGINT NULL, panel_username VARCHAR(64) NOT NULL UNIQUE, panel_password_hash VARCHAR(128) NOT NULL, role ENUM('operator', 'super_admin') NOT NULL DEFAULT 'operator', is_active TINYINT(1) NOT NULL DEFAULT 1, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; CREATE TABLE broadcast_tasks ( id INT AUTO_INCREMENT PRIMARY KEY, target_type ENUM('all', 'users', 'groups') NOT NULL, content TEXT NOT NULL, status ENUM('pending', 'running', 'done', 'failed') NOT NULL DEFAULT 'pending', done_count INT NOT NULL DEFAULT 0, fail_count INT NOT NULL DEFAULT 0, created_by INT NOT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, finished_at DATETIME NULL ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

表结构说明:users.id直接存 Telegram user_id,BIGINT 是因为 id 已经超过 32 位整数上限;is_blocked不删用户,只做拉黑标记。admins里的panel_password_hash只存哈希,不存明文,登录校验用check_password_hash比对。broadcast_tasks.status用枚举记录任务生命周期,done_count/fail_count每发完一批就回写,后台的进度条读这两个字段。

4.2 Web 后台登录:用独立后台账号而不是共享数据库密码

后台登录是最容易被源码包坑到的地方。有些包把后台的登录密码直接写死在配置文件里,或者更离谱,让后台登录时去比对数据库账号密码。前者的后果是换服务器就得改代码,后者的后果是运营人员一旦拿到数据库密码,整个服务对他就没有保密性了。

我用 Flask 写后台时,登录逻辑非常简单:前端表单把用户名和密码 POST 到/login,后端从admins表查出哈希比对,通过后把admin_id写进 session。

# admin_server.py from flask import Flask, request, session, redirect, url_for from werkzeug.security import check_password_hash import db app = Flask(__name__) app.secret_key = "从环境变量读取,不写死在源码里" @app.post("/login") def login(): username = request.form.get("username", "") password = request.form.get("password", "") row = db.fetch_one( "SELECT id, panel_password_hash, role, is_active " "FROM admins WHERE panel_username=%s", (username,) ) if not row or not row["is_active"]: return ("账号不存在或已停用", 403) if not check_password_hash(row["panel_password_hash"], password): return ("密码错误", 403) session["admin_id"] = row["id"] session["admin_role"] = row["role"] session.permanent = True # 使用 Flask 的 permanent_session_lifetime 控制时长 return redirect(url_for("dashboard")) # 其它路由统一加这个装饰器 def login_required(handler): def wrapper(*args, **kwargs): if not session.get("admin_id"): return redirect(url_for("login")) return handler(*args, **kwargs) wrapper.__name__ = handler.__name__ return wrapper

参数说明:permanent_session_lifetime建议设 8 小时,后台会话过期后强制重新登录。这里的关键点是:后台登录走的是admins表,和数据库系统的账号体系完全隔离。运营人员登录后台用的是面板账号,DBA 管数据库用数据库账号,两者不该混为一谈。session.permanent = True会启用 Flask 的 cookie 持久化,注意生产环境上 HTTPS,否则 session cookie 明文传输等于把钥匙送人。

4.3 后台发起群发任务:队列 + 频率控制,避免 bot 假死

群发是“多功能后台版”里最常被要求的功能,也是最容易把机器人打挂的功能。直接 for 循环给每个用户发消息,会让 bot 进程卡在发消息上,期间新消息无法处理,Telegram 侧又会因为响应超时不断重试,最后整个进程被拖死。正确做法是:后台只负责往broadcast_tasks插一行任务,bot 进程里跑一个独立 worker 去消费任务。

# worker.py import asyncio from aiogram.exceptions import TelegramRetryAfter import db async def consume_broadcast(bot): while True: task = await db.fetch_one( "SELECT id, target_type, content FROM broadcast_tasks " "WHERE status='pending' ORDER BY id LIMIT 1 FOR UPDATE SKIP LOCKED" ) if task is None: await asyncio.sleep(2) continue await db.execute("UPDATE broadcast_tasks SET status='running' WHERE id=%s", task["id"]) done, failed = 0, 0 async for user_id in iter_target_users(task["target_type"]): try: await bot.send_message(user_id, task["content"]) done += 1 except TelegramRetryAfter as e: await asyncio.sleep(e.retry_after) # 尊重限流,等它说的时间 failed += 1 except Exception: failed += 1 if (done + failed) % 50 == 0: await db.execute( "UPDATE broadcast_tasks SET done_count=%s, fail_count=%s " "WHERE id=%s", (done, failed, task["id"]) ) await asyncio.sleep(1 / 20) # 每秒最多 20 条 await db.execute( "UPDATE broadcast_tasks SET status='done', done_count=%s, " "fail_count=%s, finished_at=NOW() WHERE id=%s", (done, failed, task["id"]) )

参数说明:FOR UPDATE SKIP LOCKED保证多个 worker 同时拉取时不重复消费同一条任务,这是数据库层面的锁机制,比单纯标记 status 可靠。asyncio.sleep(1 / 20)把发送速率控制在每秒 20 条以内,Telegram 对 Bot API 的速率限制是每秒约 30 条消息,留 30% 余量能有效避开 429 惩罚。每 50 条回写一次进度,后台页面看到的进度是按批次刷新的,不是一次跳到底。

群发 worker 这个思路的意义在于:它让 bot 处理群发任务和用户主动消息互不阻塞。后台页面即使同时发起几十个任务,也只是往库里插了几行记录,真正的压力发生在 worker 内部,而 worker 的速率已经被 sleep 控制住了。

5. tgbot 落地避坑:token 泄漏、409 冲突、乱码与群发翻车的 5 个排查记录

5.1 webhook 和 polling 抢同一 token:409 Conflict 与重试风暴

现象:配置了 webhook 之后,本地调试时手滑执行了dp.run_polling(),日志里持续刷409 Conflict;或者反过来,在 setWebhook 已存在的状态下去掉 webhook 重跑 polling,更新就像抽风一样时来时不来。

原因:一个 bot token 同一时刻只允许一种接收方式。webhook 存在时 polling 会收到 409;而 webhook 已经被设置时,旧地址仍会继续收到 Telegram 的重试请求,直到pending_update_count归零或last_error_message不断累积。

解决:先用 deleteWebhook 一键清掉再切模式。

curl -X POST "https://api.telegram.org/bot${TG_BOT_TOKEN}/deleteWebhook?drop_pending_updates=true"

确认返回{"ok":true}后,再执行想要的方式。记住顺序永远是一删一设,别直接覆盖。生产环境建议在启动脚本里先探getWebhookInfo,如果 url 不是当前域名的地址就报警,而不是默默重置。

5.2 token 和后台密码被 git 提交后又推到了远端

现象:源码包 clone 到新服务器时,发现.env、config.php、settings.json里的 token 和数据库密码已经在 git commit 历史里躺着,推到远端后谁都看得到。更难受的是,项目已经跑了一个月,换 token 意味着所有线上服务要重新依赖新凭据。

原因:交付源码时习惯把真实配置一并打包,解压的人随手git init又把整个目录提交了。这属于把“本地环境变量”和“源码”混在一个版本库里。

解决:第一时间在 BotFather 上执行 Revoke token,新 token 只放在服务器的环境变量或.env且.gitignore掉这个文件。密码型凭据同样处理:先在后台系统里改掉密码,再清理 git 历史不迟——但重点是改了密码之后不要再把新密码提交回去。这一个动作同时解决眼前泄漏和未来重置两个问题,是我处理所有源码包的第一优先级。

5.3 后台用户名和消息内容全变问号:字符集不是 utf8mb4

现象:Telegram 消息里的 emoji、中文昵称后台能显示,但一保存就变成一堆问号或�,重开页面后内容损坏。

原因:建库时用了CHARACTER SET utf8。MySQL 的utf8最多只支持 3 字节,而 emoji 是 4 字节编码,存不进去就变?。Telegram 用户昵称里的特殊字符也一样。很多老源码包建库脚本写的是utf8,因为当年 MySQL 版本里utf8mb4还不够普及。

解决:确认库、表、连接串三层都统一成utf8mb4。建库时写CREATE DATABASE bot DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci,已存在的表执行ALTER TABLE users CONVERT TO CHARACTER SET utf8mb4。连接串里也要带charset=utf8mb4,Python 这侧如果用 SQLAlchemy,就是create_engine(url + "?charset=utf8mb4")。三处都对了,emoji 才算真正落库。

5.4 群发 1 万条消息时 bot 无响应:Flood Control 与同步阻塞

现象:后台发起群发后,前两分钟机器人还能回消息,之后点它任何命令都像失联,过十几分钟又自动恢复。

原因:两件事叠一起。第一,没用上面第 4.3 节的 worker 设计,群发直接在主事件循环里 for 循环发送,循环期间新消息完全进不了 miain loop;第二,没做频率控制,触发 Telegram 的 Flood Control(429),被短暂封禁后开始退避,机器人看起来就是“卡死”。

解决:群发逻辑必须独立到 worker 进程或独立协程,main loop 只负责单个用户消息的收发;发送频率写入asyncio.sleep(1 / 20)让发送和重试都可控;捕获TelegramRetryAfter异常,等待retry_after秒数而不是粗暴 sleep 固定时间。改完之后群发期间用户消息依然能秒回,这才算把后台功能加稳了。

5.5 后台任务进度一直 0%:任务状态没有真正落库

现象:后台点击“开始群发”,任务列表里 status 一直是 pending,进度条纹丝不动,过几分钟刷新变成 done,但过程和结果完全是黑的。

原因:任务开始、回写进度、标记完成分别散落在不同函数里,而且没及时 commit。MySQL 默认自动提交是开的,但如果用的是同一个连接池、某个异常把事务回滚了,后面的更新就被吞掉。后端返回成功不等于数据库真的写进去了。

解决:任务状态机的三个关键点都要写库:开始前把 pending 改成 running,每 50 条回写 done_count/fail_count,结束 write finished_at 和 done。每个更新点用独立事务,别把一次任务的上百次进度更新包在一个大事务里。另外后台页面读进度时不要直接读整个任务列表,只按status IN ('running')过滤出活跃任务再展示,避免全表扫描拖慢后台响应。

这五条是我处理建议是从真实运维视角提取的通用经验。你把它们当作检查清单,逐条过源码,基本能把“能跑的 bot”升级成“敢交给客户的 bot”。

6. 进阶用法:健康检查、自动重置 webhook 与备份前的恢复演练

后台和 bot 都跑顺之后,还有一件值得做的事:给整套服务加上自愈能力。每天晚上用getWebhookInfo检查 webhook 是否还指向正确地址,比第二天发现 bot 没回消息再排查强得多。

先把这个检查写成一行命令,设成 cron 每晚执行。

#!/bin/bash # health_check.sh INFO=$(curl -s "https://api.telegram.org/bot${TG_BOT_TOKEN}/getWebhookInfo") URL=$(echo "$INFO" | grep -o '"url":"[^"]*"' | cut -d'"' -f4) if [ "$URL" != "https://bot.example.com/webhook" ]; then curl -s -X POST "https://api.telegram.org/bot${TG_BOT_TOKEN}/setWebhook" \ -F "url=https://bot.example.com/webhook" \ -F "secret_token=${TG_WEBHOOK_SECRET}" echo "$(date) webhook reset" >> /var/log/tgbot_health.log fi

脚本逻辑不复杂,但行为是重点:只有 URL 不匹配才重置,不会每次 cron 都打断 Telegram 推流。同时检查/healthz端点,如果后台进程挂了而 cron 在另一台机器上,就用 supervisorctl 或 systemd 拉起进程。这两者合起来,能覆盖掉 bot 进程挂掉和 webhook 被误改两种最常见故障。

另一个舍得花时间的习惯是恢复演练。群发任务跑多了,库里会积累大量真实用户数据,备份一定要做,但更关键的是备份之后要真的恢复一次。我一般每两周从备份里起一个临时库,跑一遍后台的未完成任务列表,确认时间戳、计数都正确,才安心继续积累生产库。备份文件本身有没有可恢复性,不试永远不知道,这是我最看重的一条纪律。

每次接手这种源码包,我的流程也已经固定下来:先 Revoke token,再查 webhook 状态,再建库跑迁移,最后开 worker。步骤从不变更,因为每一道都挡过线上事故。希望帮到你。


以上。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询