一个文件搞定:用 Python 标准库把网站表单推送到 Telegram(含防刷)
小网站最常见的一个需求:访客在页面上填一个表单,我要马上在手机上收到。邮件容易进垃圾箱,短信要花钱,自己搭后台又太重。Telegram 机器人刚好合适:免费、秒到、手机上就能直接回复访客。
我给自己的官网做的就是这个:一个 app.py,只用 Python 标准库,同时提供网页和表单接口。下面是做法和踩过的坑。
整体结构
kuaibuild/
├── app.py # 唯一的程序:提供网页 + 接收表单 + 推送 Telegram
├── .env # 配置:机器人 Token 和你的 chat id(不要上传到公开仓库)
├── index.html
└── assets/运行只有一句 python3 app.py,不需要 pip install 任何东西。前面放一层 Nginx 负责 HTTPS,把请求转给它。
第一步:创建机器人,拿到两个值
- 在 Telegram 里找
@BotFather,发/newbot,按提示起名字,得到 Token。 - 给你的新机器人随便发一句话,然后在浏览器打开
https://api.telegram.org/bot<你的Token>/getUpdates,在返回的 JSON 里找到chat→id,这就是 chat id。
两个值写进 .env:
BOT_TOKEN=填你的Token
CHAT_ID=填你的chat id
PORT=8787
ALLOWED_ORIGINS=https://你的域名Token 等于机器人的密码,不要发给任何人,也不要写进网页或提交到公开仓库。不小心泄露了,去
@BotFather发/revoke换一个。
第二步:推送本身只有十几行
Telegram 的 sendMessage 就是一个普通的 HTTPS 请求:
def tg_send(text):
req = urllib.request.Request(
"%s/bot%s/sendMessage" % (TELEGRAM_API, BOT_TOKEN),
data=json.dumps({"chat_id": CHAT_ID, "text": text, "parse_mode": "HTML",
"disable_web_page_preview": True}).encode(),
headers={"Content-Type": "application/json"}, method="POST",
)
try:
with urllib.request.urlopen(req, timeout=10) as r:
return r.status == 200
except Exception as e: # 异常信息里可能带 token,只记录类型,不记录内容
log("推送 Telegram 失败:%s" % type(e).__name__)
return False两个小细节:访客填的内容要先用 html.escape 转义再拼进消息(用了 HTML 模式),否则访客写个 < 就能让推送失败;异常信息里可能带着 Token,日志里只写异常类型。
第三步:防刷,这才是重点
表单一旦公开,很快就会被机器人盯上。我叠了几层,每一层都很简单:
| 手段 | 做法 | 挡掉什么 |
|---|---|---|
| 来源检查 | 只接受 Origin 是自己域名的请求 | 别的网站或脚本直接调接口 |
| 按 IP 限频 | 每分钟 5 次、每小时 12 次、每天 30 次 | 同一个人狂点或批量刷 |
| 蜜罐 | 页面里放一个人看不见的输入框,机器人才会填 | 自动填表的机器人 |
| 填写时间 | 页面打开不到 2 秒就提交,当作机器人 | 脚本直接提交 |
| 链接数量 | 描述里超过 3 个链接直接丢弃 | 广告和钓鱼 |
| 重复内容 | 10 分钟内相同内容只推送一次 | 手滑连点、重复灌水 |
| 全站总量 | 每小时最多推送 60 条 | 被分布式刷,保护你的 Telegram |
对蜜罐、填写时间和链接过多的请求,我的做法是返回成功,但不推送。不告诉对方“被识破了”,机器人就不会换招数。
多档限频的核心只有一个函数:
def too_many(ip, now, store=None, rules=None):
store = hits if store is None else store
rules = PER_IP_RULES if rules is None else rules # [(60, 5), (3600, 12), (86400, 30)]
longest = max(w for w, _ in rules)
q = store[ip]
while q and now - q[0] > longest: # 清掉过期的记录
q.popleft()
return any(sum(1 for t in q if now - t <= w) >= n for w, n in rules)每个 IP 记一个时间队列,三档规则各数一遍,任何一档超限就拒绝。不需要 Redis,进程里的 deque 就够用。
第四步:别让它崩,也别丢单
- 每个请求一个线程(
ThreadingHTTPServer),连接 10 秒没动静就断开,防止慢速连接占满。 - 统一兜底:用一个装饰器包住所有处理函数,出任何意外都记下完整错误、给访客返回 500,服务本身继续跑。
- 推送失败不丢单:Telegram 连不上时,把这条需求写进
failed_briefs.jsonl,之后可以手动查看。 - 只用 IPv4 连 Telegram:这是我踩的第一个坑,下面单独讲。
放在 Nginx 和 Cloudflare 后面,真实 IP 怎么拿
如果前面有 Nginx 和 Cloudflare,程序看到的连接地址永远是 127.0.0.1,按 IP 限频就等于把所有人当成一个人。要读 CF-Connecting-IP,但不能无条件相信这个请求头,否则任何人都能伪造它来绕过限频:
def client_ip(self):
# 只有本机(Nginx)转发过来的请求才信任这些头,直连的请求用连接地址
peer = self.client_address[0]
if peer in ("127.0.0.1", "::1"):
return (self.headers.get("CF-Connecting-IP") or self.headers.get("X-Real-IP") or peer).strip()
return peer程序只监听 127.0.0.1,外网只能通过 Nginx 进来,这个判断就是安全的。Nginx 这一段很短:
location / {
proxy_pass http://127.0.0.1:8787;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
client_max_body_size 32k; # 表单很小,挡掉异常大的请求
}踩过的坑:提交后要等好几秒才有反应
上线后我发现点“提交”会卡好几秒,访客以为没点上。原因是服务器的 IPv6 其实不通,Python 连 Telegram 时先等 IPv6 超时,才换 IPv4。
解决分两步:后端只对 telegram.org 用 IPv4;前端点击后立刻显示“正在提交…”并禁用按钮,20 秒超时。
_orig_getaddrinfo = socket.getaddrinfo
def _ipv4_for_telegram(host, *a, **kw):
res = _orig_getaddrinfo(host, *a, **kw)
if host and "telegram.org" in str(host):
return [r for r in res if r[0] == socket.AF_INET] or res
return res
socket.getaddrinfo = _ipv4_for_telegram凡是“用户点了按钮但页面没反应”的场景,都先给一个即时反馈,再去优化后台速度。
开机自启
写一个 systemd 服务,崩了自动重启:
[Unit]
After=network-online.target
[Service]
WorkingDirectory=/你的网站目录
ExecStart=/usr/bin/python3 /你的网站目录/app.py
Restart=always
RestartSec=3
[Install]
WantedBy=multi-user.targetsystemctl daemon-reload && systemctl enable --now kuaibuild小结
一个文件、零依赖、几层简单的防刷,就能让表单提交实时到手机,而且挺稳。如果你也需要这样的东西,又不想自己折腾,可以到官网说说你的需求,我来帮你做。
有想法或问题?去留言板留言 →