简介这是一套面向Telegram社群运营者与开发者的全开源机器人源码聚焦群发、账单统计与自动记账三大场景适合需要搭建TG记账或统计机器人的中初级开发者直接部署与二次开发。压缩包共约2000个文件整体16.65MB以688个php后端逻辑、374个js交互脚本、187个html页面及141个ts模块为主辅以css、json配置、字体图标与多语言资源前后端结构完整便于按模块定位功能。目前已有960人学习下载说明该套源码在同类需求中具备一定参考价值。源码覆盖群发消息、账单汇总、自动记账等核心流程目录中保留了完整的依赖配置与静态资源读者可据此理解机器人消息调度、数据统计与账目记录的实现思路并在此基础上调整业务规则或接入自有服务快速搭建可用的TG记账统计工具。1. 从一份全开源的统计机器人源码说起TG记账群发到底怎么落地群里每天几百条交易消息刷屏手动记账对不上号月底盘账像开盲盒——这是不少做社群运营、小型电商或者资源对接的团队都遇到过的场景。这份「统计机器人源码TG记账群发机器源码人TG自动记账全开源版本」就是冲着这个痛点来的它把消息监听、关键词解析、账目统计、定时群发这几件事打包成一套可自部署的源码全部开源拿到手就能改。适合谁有一定 Python 基础、想自己掌控数据而不是把账目交给第三方机器人的开发者也适合想研究消息中间件和自动化调度怎么配合的运维同学。它解决的不是「有没有机器人」的问题而是「账目数据留在自己服务器上、规则自己定、群发节奏自己控」的问题。下面我按实际拆包和跑通的顺序把这份源码的结构、部署、参数和坑一次讲透。2. 拆开压缩包先看什么目录结构与运行依赖拿到源码别急着pip install先花十分钟把目录翻一遍能省掉后面一半的报错。这类 TG 记账群发项目通常由「监听层 解析层 存储层 调度层」四块组成目录命名各家不同但职责划分大同小异。2.1 典型目录布局与各模块职责我拆过的同类项目根目录一般长这样project/ ├── config/ │ ├── settings.yaml # 机器人 token、群组 ID、数据库连接 │ └── rules.yaml # 记账关键词、金额正则、群发模板 ├── core/ │ ├── listener.py # 消息监听与事件分发 │ ├── parser.py # 文本解析金额、币种、备注 │ ├── ledger.py # 账目写入与统计聚合 │ └── scheduler.py # 定时群发与报表推送 ├── storage/ │ ├── models.py # ORM 模型定义 │ └── migrations/ # 建表脚本 ├── utils/ │ ├── logger.py │ └── retry.py # 网络请求重试封装 ├── main.py # 入口 └── requirements.txtconfig/放的是所有可变参数改配置不用动代码这是判断一份源码是否「可维护」的第一眼标准。core/是业务核心listener.py负责接消息parser.py负责把「100 货款」这种文本拆成结构化数据ledger.py负责落库和汇总scheduler.py负责按 cron 或间隔把日报推到群里。storage/管数据持久化utils/是通用工具。如果你的包解压出来没有config/而是把 token 硬编码在main.py里那这份源码的成熟度就要打个问号得先做配置外置。2.2 依赖清单与版本约束requirements.txt是第二个必看文件。这类项目绕不开三个库消息客户端库、数据库驱动、调度库。常见组合如下依赖作用版本注意消息客户端库连接 TG、收发消息大版本升级常改 API锁小版本SQLAlchemyORM 与账目存储2.x 与 1.4 写法差异大APScheduler定时群发注意时区参数PyYAML读配置文件基本无坑redis去重与队列可选群发量大时才需要安装前先确认 Python 版本多数这类源码要求 3.9 以上。我一般会先建虚拟环境再装避免污染系统环境python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple用国内镜像源能明显加快安装尤其是依赖里有编译型包的时候。装完先别跑main.py用pip check看一眼依赖冲突有冲突先解决否则运行到一半才报错更费时间。2.3 配置文件里必须改的几项打开config/settings.yaml通常有这几项必须动bot: token: YOUR_BOT_TOKEN # 换成自己的机器人凭证 api_id: 0 # 应用 ID api_hash: # 应用哈希 group: listen_ids: [] # 要监听的群组 ID 列表 report_id: 0 # 报表推送目标群 database: url: sqlite:///ledger.db # 生产建议换 PostgreSQL/MySQL scheduler: timezone: Asia/Shanghai # 时区不对报表时间全乱listen_ids和report_id是最容易填错的地方群组 ID 通常是负数填成正数会一直收不到消息。database.url默认 SQLite 适合测试一旦群消息量大、并发写入多SQLite 的锁会拖垮整个流程生产环境务必换掉。时区这项看着不起眼但报表推送时间错八小时是新手最常见的翻车点。3. 消息监听与记账解析从原始文本到结构化账目配置改完核心就是让机器人「听懂」群里的消息。这一步的难点不在连接而在解析规则的鲁棒性——群里说话不会按你的格式来。3.1 监听层的事件分发逻辑监听层要做的事很单纯收到消息判断是不是需要处理的群、是不是需要记账的格式然后丢给解析层。常见写法是注册一个消息回调# core/listener.py from utils.logger import get_logger logger get_logger(__name__) class MessageListener: def __init__(self, client, parser, ledger, config): self.client client self.parser parser self.ledger ledger self.listen_ids set(config[group][listen_ids]) async def on_message(self, event): # 只处理配置里声明的群其他群直接丢弃减少无效解析 chat_id event.chat_id if chat_id not in self.listen_ids: return text event.message.message or # 解析层返回 None 表示这条消息不是账目跳过 record self.parser.parse(text) if record is None: return record.chat_id chat_id record.sender_id event.sender_id await self.ledger.save(record) logger.info(记账成功: %s, record)这段逻辑的关键在listen_ids的白名单过滤和parse返回None的短路设计。很多新手把解析失败也写进库导致账目表里一堆脏数据后面统计全乱。参数上event.chat_id和event.sender_id是消息自带的不用额外请求record是解析层产出的数据对象字段建议至少包含金额、币种、方向收入/支出、备注、时间戳。3.2 解析规则的正则设计与边界解析层是整份源码里最需要按自己业务改的地方。默认规则一般只认「100」「-50」这种最简格式实际群里可能是「收 100 货款」「支出50元」「100U」混着来。常见做法是用一组正则按优先级匹配# core/parser.py import re from dataclasses import dataclass dataclass class Record: amount: float currency: str direction: str # in / out note: str class Parser: # 按优先级排列先匹配到的先用 PATTERNS [ # 形如 100 或 -50.5带可选币种 (re.compile(r^\s*([-])\s*(\d(?:\.\d)?)\s*([A-Za-z]{0,5})), sign), # 形如 收100 / 支出50 (re.compile(r(收|收入|支出|付)\s*(\d(?:\.\d)?)\s*([A-Za-z]{0,5})), keyword), ] def parse(self, text): text text.strip() if not text: return None for pattern, kind in self.PATTERNS: m pattern.search(text) if not m: continue if kind sign: direction in if m.group(1) else out amount float(m.group(2)) currency m.group(3) or CNY else: direction in if m.group(1) in (收, 收入) else out amount float(m.group(2)) currency m.group(3) or CNY return Record(amountamount, currencycurrency, directiondirection, notetext) return None这里有几个参数要按业务调PATTERNS的顺序决定优先级把最严格的放前面能减少误判币种默认值CNY要按你的实际场景改做跨境对接的默认USDT更合理note直接存原文方便事后对账追溯。边界情况要特别注意金额带千分位逗号1,000默认正则匹配不到得在预处理里先去掉逗号负数金额和「-」符号要区分开否则支出会被记成负收入。3.3 账目落库与统计聚合解析出来的Record要落库统计层再基于库里的数据做聚合。表结构建议至少三张原始流水表、日汇总表、用户维度表。写入时用事务保证一致性# core/ledger.py from sqlalchemy import create_engine, Column, Integer, Float, String, DateTime from sqlalchemy.orm import declarative_base, sessionmaker from datetime import datetime Base declarative_base() class Flow(Base): __tablename__ flow id Column(Integer, primary_keyTrue) chat_id Column(Integer, indexTrue) sender_id Column(Integer, indexTrue) amount Column(Float) currency Column(String(8)) direction Column(String(4)) note Column(String(255)) created_at Column(DateTime, defaultdatetime.utcnow) class Ledger: def __init__(self, db_url): self.engine create_engine(db_url, pool_pre_pingTrue) Base.metadata.create_all(self.engine) self.Session sessionmaker(bindself.engine) async def save(self, record): with self.Session() as session: session.add(Flow( chat_idrecord.chat_id, sender_idrecord.sender_id, amountrecord.amount, currencyrecord.currency, directionrecord.direction, noterecord.note, )) session.commit()pool_pre_pingTrue是生产必备防止数据库连接空闲后被断开导致写入失败。indexTrue加在chat_id和sender_id上统计查询才快。日汇总可以用定时任务跑 SQL 聚合也可以写入时同步更新汇总表前者简单后者实时按数据量选。4. 定时群发与报表推送调度参数怎么设才不翻车记账只是前半程把统计结果按节奏推回群里才是闭环。调度这块的坑集中在时区、并发和消息频率上。4.1 调度器的初始化与任务注册用 APScheduler 做定时推送初始化时把时区显式写死# core/scheduler.py from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.cron import CronTrigger class ReportScheduler: def __init__(self, ledger, client, config): self.ledger ledger self.client client self.report_id config[group][report_id] # 显式指定时区避免用系统默认导致时间偏移 self.scheduler AsyncIOScheduler( timezoneconfig[scheduler][timezone] ) def start(self): # 每天 21:00 推日报 self.scheduler.add_job( self.push_daily, CronTrigger(hour21, minute0), iddaily_report, misfire_grace_time300, # 错过 5 分钟内仍执行 coalesceTrue, # 堆积的任务合并成一次 ) self.scheduler.start() async def push_daily(self): summary await self.ledger.daily_summary() text self.render(summary) await self.client.send_message(self.report_id, text)misfire_grace_time和coalesce是两个容易被忽略但很关键的参数。服务重启或短暂卡顿时任务可能错过触发点misfire_grace_time300让它在 5 分钟内补跑coalesceTrue防止补跑时把积压的多次任务全执行一遍导致群里被报表刷屏。4.2 群发消息的模板与频率控制群发模板建议放配置文件别硬编码在代码里改文案不用重新部署# config/rules.yaml templates: daily: | 【{date} 账目日报】 收入{income} {currency} 支出{expense} {currency} 净额{net} {currency} 笔数{count}渲染时用str.format或模板引擎都行。频率控制是重点TG 对机器人发消息有速率限制短时间内大量群发会触发限流甚至封禁。常见做法是加一个发送队列每条消息间隔 1 到 2 秒批量推送时用asyncio.sleep控制节奏。如果群发对象是多个群还要考虑失败重试utils/retry.py里通常封装了指数退避重试次数别超过 3 次否则一条消息卡住会拖垮整个队列。4.3 报表数据的聚合查询日报的数据来自聚合查询SQL 写不好会拖慢整个推送# core/ledger.py 中的聚合方法 async def daily_summary(self, dayNone): day day or datetime.utcnow().date() with self.Session() as session: rows session.query( Flow.direction, func.sum(Flow.amount).label(total), func.count(Flow.id).label(cnt), ).filter( func.date(Flow.created_at) day ).group_by(Flow.direction).all() income sum(r.total for r in rows if r.direction in) expense sum(r.total for r in rows if r.direction out) count sum(r.cnt for r in rows) return {income: income, expense: expense, net: income - expense, count: count}func.date(Flow.created_at)在数据量大时无法走索引生产环境建议额外存一个stat_date字段并加索引或者用范围查询created_at 当天零点 AND 次日零点。这是聚合查询最常见的性能坑测试数据少时看不出来上量后报表生成要等十几秒。5. 避坑与排查这几处报错我踩过不止一次源码跑不起来八成不是代码本身的问题而是环境和配置。下面几条是我实际部署时反复遇到的按「现象 → 原因 → 解决」列出来。现象一启动后一直收不到群消息日志无任何输出。原因listen_ids填的是群组用户名或正数 ID而实际事件里的chat_id是负数或者机器人根本没被拉进目标群。 解决先把机器人拉进群发一条测试消息在回调里打印event.chat_id拿到真实 ID再填回配置。别凭记忆填。现象二账目表里出现大量金额为 0 或重复的记录。原因解析正则过于宽松把「今天收入不错」这类闲聊也匹配成了账目或者消息编辑事件被重复触发。 解决收紧正则要求金额前后有明确边界在监听层对event.message.edit_date做判断编辑消息不重复记账或者用消息 ID 做去重。现象三日报推送时间比预期早或晚八小时。原因调度器用了系统默认时区而服务器时区是 UTC。 解决在AsyncIOScheduler初始化时显式传timezone并在配置里统一用Asia/Shanghai。别依赖系统环境变量。现象四群发到一半报连接超时后续消息全丢。原因没有做发送失败重试一条消息异常直接中断了整个循环。 解决把每条发送包在 try/except 里失败进重试队列用指数退避重试同时控制发送间隔别连续快速发送。现象五SQLite 数据库在高频写入时报database is locked。原因SQLite 同一时刻只允许一个写操作群消息密集时并发写入冲突。 解决测试阶段可以加timeout参数缓解生产环境直接换 PostgreSQL 或 MySQL连接池配好这个问题从根上消失。6. 进阶玩法把记账数据接出去做二次分析源码跑通只是起点真正有价值的是把账目数据用起来。我一般会在ledger.py里留一个导出接口把流水按天导成 CSV再接到自己的分析脚本里。比如按发送人维度统计贡献、按币种拆分收支、按小时看消息活跃度这些都能反过来指导群运营节奏。# 导出近 7 天流水为 CSV供外部分析 import csv from datetime import datetime, timedelta def export_recent(days7, pathexport.csv): since datetime.utcnow() - timedelta(daysdays) with Session() as session: rows session.query(Flow).filter(Flow.created_at since).all() with open(path, w, newline, encodingutf-8) as f: writer csv.writer(f) writer.writerow([时间, 群, 发送人, 方向, 金额, 币种, 备注]) for r in rows: writer.writerow([r.created_at, r.chat_id, r.sender_id, r.direction, r.amount, r.currency, r.note]) return len(rows)导出后可以用 pandas 做透视也可以直接丢进 BI 工具。验证数据对不对有个笨办法但很管用随机挑三天手动把群消息翻一遍和导出的流水逐条对对不上就回去查解析规则。我吃过一次亏正则把「退款 -50」记成了收入因为符号判断只看了数字前的字符没考虑「退款」这个前缀结果月度净额差了小两千。从那以后我每次改完解析规则都强制拿历史消息跑一遍回归确认没有翻转方向的记录才上线。希望这份拆解能帮你少走点弯路把这份源码真正用起来。本文还有配套的精品资源点击获取