企业官网建设流程全解析

简介：这份开源资源面向Telegram机器人开发者与社群运营者，提供一套可直接部署的统计、记账与群发一体化解决方案，适合具备PHP与前端基础、希望快速搭建TG自动化工具的中高级开发者。压缩包共约2000个文件，整体16.65MB，以688个PHP后端逻辑、374个JavaScript交互脚本、187个HTML页面及141个TypeScript模块为主，辅以CSS样式、JSON配置、字体图标与多语言资源，前后端结构完整，便于二次开发与功能扩展。目前已有960人学习下载，说明该套源码在社群工具领域具备一定参考价值。读者可获得完整的账单统计与自动记账流程实现、群发消息调度逻辑、后台管理界面模板以及依赖配置与部署所需的环境文件，能够直接对照目录结构理解模块划分，快速替换配置并接入自有机器人，省去从零搭建的时间成本。

1. 从一份「统计机器人源码」说起：TG 记账群发机器人到底在解决什么问题

如果你在 Telegram 群里做过运营，大概率遇到过这种场景：几十个代理群、返佣群、任务群同时开着，每天有人报单、有人充值、有人提现，靠人工拿表格记，三天就乱。统计机器人源码这类东西，本质就是把「群内消息 → 结构化账目 → 定时汇总推送」这条链路自动化。标题里提到的 TG 记账群发机器人全开源版本，核心就三块能力：监听群消息并解析成账目、按用户或群维度做统计、把统计结果定时群发回指定会话。它适合谁？一是手里有多个 TG 群需要做流水分账的小团队，二是想拿一套能跑通的机器人骨架做二次开发的后端同学。这篇不吹概念，直接按「能复现」的标准拆开讲：环境怎么搭、消息怎么解析、账目怎么落库、群发怎么不翻车。

2. 把机器人跑起来：从 Bot Token 到第一条群消息落库

2.1 选型：为什么用长轮询而不是 Webhook

TG 机器人接消息有两条路：Webhook 和长轮询（getUpdates）。做记账群发这类场景，我一般优先长轮询，原因很实际——Webhook 要求你的服务有一个公网可达的 HTTPS 地址，本地调试、内网部署、临时换机都很麻烦；而记账机器人往往部署在一台普通云主机甚至家里的常开设备上，长轮询只要出网就能收消息，省掉证书和反代这一层。代价是实时性略低（取决于你设置的超时时间），但对记账这种秒级不敏感的业务完全够用。

选语言层面，Python 的生态最省事，python-telegram-bot或直接裸调 HTTP API 都行。为了让你看清底层，这里用裸 HTTP + SQLite 的最小组合，不引入重框架，方便你后面替换成自己的存储。

2.2 最小可运行骨架：拉取更新并写入 SQLite

先装依赖，只需要requests：

pip install requests

然后是主循环，核心逻辑是「拿 offset → 取更新 → 解析 → 落库 → 推进 offset」：

import requests, sqlite3, time TOKEN = "你的_BOT_TOKEN" API = f"https://api.telegram.org/bot{TOKEN}" # 初始化账目表：一条消息一行，保留原始文本便于回溯 conn = sqlite3.connect("ledger.db") conn.execute(""" CREATE TABLE IF NOT EXISTS raw_msg ( update_id INTEGER PRIMARY KEY, chat_id INTEGER, user_id INTEGER, username TEXT, text TEXT, ts INTEGER ) """) conn.commit() offset = 0 while True: # timeout=30 表示长轮询挂起 30 秒，减少空转请求 resp = requests.get(f"{API}/getUpdates", params={"offset": offset, "timeout": 30}, timeout=40).json() for upd in resp.get("result", []): offset = upd["update_id"] + 1 # 关键：推进 offset，否则重复消费 msg = upd.get("message") or {} text = msg.get("text") if not text: continue conn.execute( "INSERT OR IGNORE INTO raw_msg VALUES (?,?,?,?,?,?)", (upd["update_id"], msg["chat"]["id"], msg["from"]["id"], msg["from"].get("username", ""), text, msg["date"]) ) conn.commit() time.sleep(0.5)

逻辑说明：offset是 TG 的游标机制，你回传update_id + 1表示「这条我处理过了，下次从下一条开始」。如果不推进，机器人会反复收到同一条消息，账目就会重复记账——这是新手最常见的翻车点。参数上，timeout=30是服务端挂起时间，requests的timeout=40要略大于它，否则客户端先超时。INSERT OR IGNORE配合update_id主键，做了一层幂等兜底。

2.3 把原始消息解析成账目字段

原始文本落库只是第一步，真正要的是结构化账目。假设群内报单格式约定为+100 充值或-50 提现，解析函数这样写：

import re # 匹配形如 "+100 充值" / "-50 提现" / "+100.5 返佣" PATTERN = re.compile(r"^([+-])(\d+(?:\.\d{1,2})?)\s*(\S+)$") def parse_entry(text): m = PATTERN.match(text.strip()) if not m: return None sign, amount, kind = m.groups() value = float(amount) * (1 if sign == "+" else -1) return {"amount": value, "kind": kind} # 用法 print(parse_entry("+100 充值")) # {'amount': 100.0, 'kind': '充值'} print(parse_entry("-50 提现")) # {'amount': -50.0, 'kind': '提现'} print(parse_entry("随便聊聊")) # None

逻辑说明：正则用^...$锚定整行，避免「今天赚了+100 充值」这种闲聊被误判成账目。金额限制两位小数，是因为记账场景下超过两位基本是脏数据。kind用\S+抓非空白串，兼容「充值」「返佣」「提现」等自定义类型。解析失败返回None，调用方据此决定是忽略还是提示用户格式错误——这一步别偷懒直接抛异常，群里一条乱消息就能让机器人崩掉。

3. 统计与群发：账目聚合、定时任务和消息节流

3.1 按用户和群维度做聚合统计

账目落库后，统计就是 SQL 的活。下面按「用户 + 类型」汇总，并算出净额：

def summary_by_user(conn, chat_id): cur = conn.execute(""" SELECT username, SUM(CASE WHEN amount > 0 THEN amount ELSE 0 END) AS income, SUM(CASE WHEN amount < 0 THEN amount ELSE 0 END) AS outcome, SUM(amount) AS net FROM ledger WHERE chat_id = ? GROUP BY username ORDER BY net DESC """, (chat_id,)) return cur.fetchall()

逻辑说明：CASE WHEN把正负金额拆成收入和支出两列，net是净额。GROUP BY username前要确保username不为空，否则匿名用户会挤成一堆——实际项目里建议用user_id做分组键，username只做展示。ORDER BY net DESC让排行榜天然成型，直接能贴进群发消息。

3.2 定时群发：别让机器人被限流

群发最容易踩的坑是触发 TG 的频率限制。官方对同一 chat 的发送频率有约束，短时间狂发会被 429 甚至临时封禁。稳妥做法是加节流和重试：

import time, requests def send_message(chat_id, text): for attempt in range(3): r = requests.post(f"{API}/sendMessage", json={"chat_id": chat_id, "text": text}, timeout=10) if r.status_code == 200: return True if r.status_code == 429: # 读取服务端要求的等待秒数，默认退避 3 秒 wait = r.json().get("parameters", {}).get("retry_after", 3) time.sleep(wait) continue time.sleep(1) return False

逻辑说明：429 是限流信号，响应体里的retry_after是服务端明确告诉你要等几秒，照着等最稳，别自己拍脑袋。三次重试仍失败就放弃并记录日志，避免死循环拖垮整个定时任务。群发多条时，每条之间至少sleep(1)，这是血泪经验——曾经有项目一次性推 50 条汇总，直接被限流半小时。

3.3 用调度器把统计和群发串起来

定时用schedule库最轻量：

import schedule def daily_job(): rows = summary_by_user(conn, TARGET_CHAT_ID) lines = [f"{u}: 净额 {n:.2f}" for u, _, _, n in rows] send_message(TARGET_CHAT_ID, "今日汇总\n" + "\n".join(lines)) schedule.every().day.at("23:00").do(daily_job) while True: schedule.run_pending() time.sleep(1)

逻辑说明：daily_job把统计结果拼成多行文本一次发出，而不是每个用户发一条，这样既省请求又不容易触发限流。TARGET_CHAT_ID是汇总推送的目标会话，可以是群也可以是管理员私聊。注意schedule是单进程内存调度，重启会丢任务状态，生产环境建议换成带持久化的调度方案，或者干脆用系统 cron 触发一个一次性脚本。

4. 避坑与排查：记账群发机器人最常见的 5 个翻车点

4.1 消息重复记账

现象：同一笔充值在账目里出现两三次。原因：offset 没有正确推进，或者进程重启后从旧 offset 重新拉取。解决：每次处理完立即持久化 offset，别只放内存变量；入库用update_id做主键做幂等。

4.2 群消息收不到

现象：私聊机器人正常，群里 @ 它没反应。原因：TG 默认机器人只能收到「@机器人」或回复它的消息，普通群消息需要关闭隐私模式。解决：找 BotFather 用/setprivacy设为 Disable，或者把机器人设为群管理员。这一步不做，监听逻辑写得再对也收不到数据。

4.3 金额解析把闲聊当账目

现象：群里有人发「今天赚了+100 开心」被记成 100 元收入。原因：正则没锚定整行，或没做类型白名单。解决：^...$锚定 + 只接受预定义kind集合，其余一律忽略并可选提示。

4.4 群发被限流甚至封禁

现象：汇总消息发一半卡住，后续全部失败。原因：短时间高频调用 sendMessage。解决：加retry_after退避、条间sleep、合并消息减少条数。别用多线程并发群发，那是加速被封。

4.5 数据库并发写冲突

现象：多群同时来消息时报database is locked。原因：SQLite 默认写锁粒度粗，多线程写会冲突。解决：单线程串行写，或换 WAL 模式PRAGMA journal_mode=WAL，量大直接上 PostgreSQL。记账场景数据量不大，串行写通常就够。

5. 进阶：把开源版本改成你自己的记账规则引擎

拿到一份全开源版本，真正值钱的不是它现在能记什么，而是你能不能把「解析规则」和「统计口径」抽出来做成可配置的。我一般会做两件事。

第一，把正则和字段映射外置成配置。比如用一张rule表存pattern、kind、sign，启动时编译成多个正则，按顺序匹配。这样运营改报单格式不用动代码：

import re, json RULES = [ {"pattern": r"^\+(\d+(?:\.\d{1,2})?)\s*充值$", "kind": "充值", "sign": 1}, {"pattern": r"^-(\d+(?:\.\d{1,2})?)\s*提现$", "kind": "提现", "sign": -1}, ] COMPILED = [(re.compile(r["pattern"]), r["kind"], r["sign"]) for r in RULES] def parse_with_rules(text): for pat, kind, sign in COMPILED: m = pat.match(text.strip()) if m: return {"amount": float(m.group(1)) * sign, "kind": kind} return None

逻辑说明：规则按顺序匹配，第一条命中即返回，所以把更具体的规则放前面。sign决定正负，避免在正则里写死符号导致规则难维护。这套结构的好处是加一条新账目类型只是往RULES里加一行，不用碰解析函数。

第二，给统计加时间窗口参数。日报、周报、月报本质是同一条 SQL 换WHERE ts BETWEEN ? AND ?，把起止时间戳作为参数传进去，一个函数覆盖所有周期：

def summary_in_range(conn, chat_id, start_ts, end_ts): cur = conn.execute(""" SELECT username, SUM(amount) AS net FROM ledger WHERE chat_id = ? AND ts >= ? AND ts < ? GROUP BY username """, (chat_id, start_ts, end_ts)) return cur.fetchall()

验证方法很直接：拿一批已知金额的测试消息灌进去，手工算一遍净额，和 SQL 结果对拍。对不上就查时间戳边界——>= start和< end这种半开区间能避免跨周期重复统计，这是我在对账时踩过的最隐蔽的坑。

最后说个习惯：任何涉及金额的机器人，日志一定要记全「原始消息 + 解析结果 + 入库结果」三段，出问题时能一眼定位是解析错了还是统计错了。别嫌日志占地方，对账对不上的时候，它就是唯一的后悔药。希望帮到你。

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