★ FEATURED ARTICLE 简介这是一套面向Telegram平台开发者的统计机器人源码合集聚焦群发、账单统计与自动记账三大核心功能适合有一定PHP与前端基础、希望快速搭建TG记账或群发工具的中高级开发者。压缩包共收录约2000个文件以688个PHP后端脚本、374个JavaScript交互逻辑、187个HTML页面及141个TypeScript模块为主另含79个CSS样式、55个JSON配置与31个GIF素材整体约16.65MB目录结构完整前后端资源齐备。目前已有960人学习下载说明该套源码在同类需求中具备一定参考价值。读者可从中获取完整的记账统计逻辑、群发消息调度实现、账单数据展示页面以及后台管理界面便于二次开发或直接部署快速理解Telegram机器人从消息接收、账目录入到统计输出的全链路实现思路。1. 从一份“统计机器人源码”说起TG记账群发机器人到底在解决什么问题如果你在群里做过记账、分账、统计流水这类事大概率经历过这样的场景群里几十号人每天有人报账、有人转账、有人要汇总靠人工翻聊天记录去核对一天下来眼睛都花了还容易漏。标题里这份“统计机器人源码TG记账群发机器源码人TG自动记账全开源版本”本质上就是把这套重复劳动交给一个跑在群里的机器人它监听消息、识别记账指令、把数据落到本地库、按需做群发提醒和统计汇总而且源码全开你能自己改逻辑、加字段、换存储。它适合三类人一是手里有社群、需要做轻量财务流水记录的小团队二是想学消息机器人从监听、解析到持久化完整链路的开发者三是已经有一版半成品、但统计和群发逻辑总是对不上账、想找一份可读源码对照排查的人。这篇不吹它能替代专业财务系统只讲清楚它怎么跑起来、参数怎么设、哪些地方最容易翻车。2. 消息监听与记账指令解析机器人怎么“听懂”一句话2.1 长轮询还是 Webhook先定通信方式机器人要拿到群消息常见做法有两种长轮询polling和 Webhook。长轮询是机器人主动去问服务端“有没有新消息”实现简单、对部署环境要求低本地开发机上就能跑Webhook 是服务端把消息推给你实时性好、省资源但需要一个能被外网访问的地址调试时多一层转发。我一般建议先用长轮询把整条链路跑通确认记账解析、入库、统计都对再考虑换 Webhook。原因是长轮询的失败模式很直观要么拿不到更新要么重复拿排查方向明确Webhook 一旦签名、路径、端口哪一环不对消息根本进不来新手容易卡在“机器人没反应”上找不到北。下面是一个长轮询拉取更新的最小骨架用 Python 写逻辑是“带偏移量循环拉取处理完再推进偏移量”import time import requests # 机器人令牌和API基地址实际使用时从配置读取不要硬编码 BOT_TOKEN YOUR_BOT_TOKEN API_BASE fhttps://api.telegram.org/bot{BOT_TOKEN} def get_updates(offsetNone, timeout30): 长轮询拉取更新offset 用于确认已处理到的位置 params {timeout: timeout} if offset is not None: params[offset] offset resp requests.get(f{API_BASE}/getUpdates, paramsparams, timeouttimeout 10) data resp.json() if not data.get(ok): # 失败时返回空列表由外层决定重试节奏 return [] return data.get(result, []) def main_loop(): offset None while True: updates get_updates(offset) for upd in updates: # 每处理一条就把 offset 推到它的 update_id 1 offset upd[update_id] 1 handle_update(upd) # 没有更新时短暂休眠避免空转打满CPU if not updates: time.sleep(1) def handle_update(upd): # 这里只做分发具体解析放到下一节 msg upd.get(message) or upd.get(edited_message) if not msg: return text msg.get(text, ) if text.startswith(/): parse_command(text, msg)逻辑说明getUpdates的offset参数是关键它告诉服务端“我已经处理到哪了”不传或传错会导致重复消费或丢消息。timeout是长轮询等待秒数设太短会频繁空请求设太长则消息延迟感明显30 秒是常见折中。handle_update里只做分发不掺业务方便后面单独测解析。参数上timeout建议 20 到 50 之间外层空转休眠别省我见过有人去掉sleep直接把接口打到限流。令牌一定走环境变量或配置文件源码里硬编码令牌是血泪教训一旦仓库外泄机器人就被人接管了。2.2 记账指令的解析把“50 午饭”变成一条结构化记录消息拿到了接下来是解析。记账指令通常长这样50 午饭、-30 打车、100 报销 备注张三。解析的核心是把“符号、金额、类目、备注”拆开并且要能容忍用户多打空格、用中文符号、金额带小数点。import re # 匹配形如 50、-30.5 开头后面跟类目和可选备注 RECORD_PATTERN re.compile( r^(?Psign[\-])\s*(?Pamount\d(?:\.\d{1,2})?)\s*(?Pcategory\S)(?:\s(?Pnote.))?$ ) def parse_record(text): 解析一条记账指令失败返回 None text text.strip().replace(, ).replace(, -) m RECORD_PATTERN.match(text) if not m: return None amount float(m.group(amount)) # 支出统一存负数收入存正数方便后续求和 if m.group(sign) -: amount -amount return { amount: amount, category: m.group(category), note: (m.group(note) or ).strip(), }逻辑说明正则里sign和amount分开捕获是为了先拿到绝对值再按符号定正负避免在正则里写两套分支。category用\S表示非空白连续字符能接住“午饭”“打车”这类词note用可选分组没有备注时为None。开头做了全角符号替换这是实际群里最常见的输入习惯不处理就会解析失败。参数上金额小数位限制两位符合日常记账精度如果你要支持“万”“k”这类单位得在amount后面再加一层单位换算别直接塞进正则否则可读性会崩。解析失败时不要静默丢弃应该回一条“格式不对示例50 午饭”否则用户以为记上了实际没入库对账时就是灾难。2.3 入库前的去重与幂等同一条消息别记两遍长轮询在网络抖动时可能重复投递同一条更新如果解析完直接插入账目就会翻倍。常见做法是用update_id或消息message_id做唯一键插入前先查。import sqlite3 def init_db(pathledger.db): conn sqlite3.connect(path) conn.execute( CREATE TABLE IF NOT EXISTS records ( id INTEGER PRIMARY KEY AUTOINCREMENT, msg_id INTEGER UNIQUE, -- 消息唯一ID用于幂等 user_id INTEGER, amount REAL, category TEXT, note TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ) conn.commit() return conn def save_record(conn, msg_id, user_id, record): 幂等写入重复 msg_id 直接忽略 try: conn.execute( INSERT INTO records (msg_id, user_id, amount, category, note) VALUES (?,?,?,?,?), (msg_id, user_id, record[amount], record[category], record[note]), ) conn.commit() return True except sqlite3.IntegrityError: # 唯一约束冲突说明这条已经记过 return False逻辑说明msg_id上加UNIQUE约束把幂等交给数据库比在应用层先查后插更稳因为查和插之间有并发窗口。捕获IntegrityError返回False调用方据此决定是否回复“已记录过”。created_at用数据库默认时间避免各机器时钟不一致。参数上user_id用来区分谁记的账统计时按人汇总就靠它。如果你要多群隔离再加一个chat_id字段并纳入唯一键组合。SQLite 适合单机小规模群多、写入频繁时换 PostgreSQL但表结构和幂等思路不变。3. 统计汇总与群发从流水到“这个月谁欠谁”3.1 按人、按类目、按周期做聚合记账数据进来后统计就是 SQL 的活。最常见的三个维度按人汇总净额、按类目看支出分布、按周期日/周/月出流水。下面这段把三个查询写在一起方便你按需取用。def summary_by_user(conn, start, end): 按用户汇总区间内的净额 cur conn.execute( SELECT user_id, SUM(amount) AS net, COUNT(*) AS cnt FROM records WHERE created_at BETWEEN ? AND ? GROUP BY user_id ORDER BY net ASC , (start, end)) return cur.fetchall() def summary_by_category(conn, start, end): 按类目汇总支出只看负数 cur conn.execute( SELECT category, SUM(amount) AS total FROM records WHERE created_at BETWEEN ? AND ? AND amount 0 GROUP BY category ORDER BY total ASC , (start, end)) return cur.fetchall() def summary_by_period(conn, start, end): 按天汇总净额用于画趋势 cur conn.execute( SELECT DATE(created_at) AS day, SUM(amount) AS net FROM records WHERE created_at BETWEEN ? AND ? GROUP BY day ORDER BY day ASC , (start, end)) return cur.fetchall()逻辑说明summary_by_user用SUM(amount)直接得净额因为收入正、支出负相加就是结余ORDER BY net ASC让欠得多的排前面方便催账。summary_by_category加amount 0只看支出避免收入类目混进来。summary_by_period用DATE()截断到天SQLite 里直接可用换其他库注意日期函数差异。参数上start和end传字符串时间格式要和created_at一致否则BETWEEN匹配不上。我一般把区间封装成“本月一号到此刻”别让用户手输时间减少格式错误。聚合结果为空时要有兜底文案别直接发一张空表。3.2 群发消息的节流与失败重试统计结果要发回群里群发最容易踩的坑是“发太快被限流”和“发失败没重试”。常见做法是加节流间隔并对可重试错误做退避。import time import requests def send_message(chat_id, text, retries3): 发送消息带指数退避重试 url f{API_BASE}/sendMessage for attempt in range(retries): try: resp requests.post(url, json{chat_id: chat_id, text: text}, timeout10) data resp.json() if data.get(ok): return True # 429 表示被限流按返回的 retry_after 等待 if resp.status_code 429: wait data.get(parameters, {}).get(retry_after, 2 ** attempt) time.sleep(wait) continue # 其他业务错误不重试直接返回 return False except requests.RequestException: # 网络异常按退避重试 time.sleep(2 ** attempt) return False def broadcast(chat_ids, text, interval1.0): 批量群发每条之间留间隔 results {} for cid in chat_ids: results[cid] send_message(cid, text) time.sleep(interval) return results逻辑说明send_message区分了三类情况——成功、限流429按服务端给的retry_after等、其他业务错误不重试重试也没用。网络异常单独捕获用2 ** attempt做指数退避。broadcast在每条之间sleep(interval)这是防限流最直接的手段。参数上interval设 1 秒对几十个群够用群特别多时适当加大retries别设太高3 次足够再多会拖慢整体。注意retry_after是服务端建议值优先用它而不是自己拍脑袋。群发内容里如果带用户昵称记得做转义特殊字符会导致发送失败。3.3 定时任务什么时候触发统计和群发统计和群发通常按周期触发常见做法是用调度库在固定时间跑。下面用schedule库演示每天固定时间出日报。import schedule import time def daily_job(): 每天定时执行的统计群发任务 conn init_db() start 2024-01-01 00:00:00 # 实际按当天零点计算 end 2024-01-01 23:59:59 rows summary_by_user(conn, start, end) if not rows: return lines [f用户 {uid}: 净额 {net:.2f}{cnt} 笔 for uid, net, cnt in rows] text 今日账目汇总\n \n.join(lines) broadcast([YOUR_CHAT_ID], text) # 每天 21:00 触发 schedule.every().day.at(21:00).do(daily_job) while True: schedule.run_pending() time.sleep(30)逻辑说明daily_job里先查再拼文案空结果直接返回避免发空消息。schedule.every().day.at()指定触发时间run_pending在循环里检查。时间区间实际应按运行当天动态计算示例里写死是为了展示结构。参数上触发时间选群里活跃时段太晚没人看太早数据不全。time.sleep(30)是调度轮询间隔30 秒精度对日报足够。注意调度进程要和机器人主进程分开或做好异常隔离否则统计任务崩了会把监听也带下去。4. 避坑与排查那些让账目对不上的细节4.1 现象账目金额翻倍。原因重复消费同一条更新。解决用msg_id唯一约束做幂等并确保offset在处理成功后才推进处理失败时不要推进让它下次重来。4.2 现象用户说记了账但查不到。原因解析失败被静默丢弃或全角符号、多余空格导致正则不匹配。解决解析失败必须回执提示正则前统一做strip和全角转半角金额和类目之间允许多个空格。4.3 现象群发到一半停了。原因触发限流后没有按retry_after等待或重试次数耗尽直接放弃。解决识别 429 并按服务端建议等待批量发送加固定间隔失败记录到日志便于补发。4.4 现象统计区间查不到数据。原因created_at存的是 UTC查询用的是本地时间两者差几个小时。解决统一时区要么入库就转本地要么查询时转换别一半一半。4.5 现象机器人突然不回消息。原因令牌泄露被他人占用或进程被限流封禁。解决令牌走配置不硬编码发现异常立即更换日志里记录每次请求状态便于定位是网络还是权限问题。5. 把源码改成你自己的字段扩展与验证习惯拿到一份全开源码真正有价值的是按自己场景改。最常见的扩展是加字段比如加chat_id做多群隔离加status标记已结算/未结算加tags做多标签分类。改表结构时SQLite 用ALTER TABLE ADD COLUMN加列历史数据该列默认为空统计时用COALESCE兜底。# 给已有表加字段历史行该列为 NULL conn.execute(ALTER TABLE records ADD COLUMN chat_id INTEGER) conn.execute(ALTER TABLE records ADD COLUMN status TEXT DEFAULT open) # 统计时用 COALESCE 处理 NULL避免整行被过滤 cur conn.execute( SELECT chat_id, SUM(amount) FROM records WHERE COALESCE(status, open) open GROUP BY chat_id )逻辑说明ALTER TABLE加列是轻量操作但要注意默认值只对新行生效历史行仍是 NULL所以查询里用COALESCE给默认值。status用来做结算流转统计只算未结算的结算后改状态账目就不会重复计入。验证习惯上我一般做三件事一是造几条边界数据金额为 0、超大金额、超长备注跑一遍解析和入库二是对同一区间用两种方式算总额比如 SQL 求和和 Python 遍历求和结果对不上就说明有隐藏过滤三是每次改完统计逻辑先在小群试发确认文案和数字都对再上大群。这套习惯帮我挡掉过好几次“看起来对、实际差几毛”的玄学问题。源码全开的好处就在这出问题能一路追到 SQL 和正则而不是对着黑匣子猜。希望帮到你。本文还有配套的精品资源点击获取
阅读完成 · 觉得有帮助？
