基于Flask与Windows文件监控实现微信个人收款自动化处理

1. 项目缘起一个被低估的自动化场景几年前我在运营一个线上小社群主要靠微信群里的活动报名费来维持一些小额开销。每次有人转账我都得手动在Excel表格里记录核对金额、备注、时间再手动给对方的账户加上积分或开通权限。人少的时候还好一旦活动频繁起来这种重复劳动不仅枯燥还极易出错。漏记一笔、金额看错一位数后续的核对能让人崩溃。当时我就在想如果能有个东西在我收到微信转账后自动帮我完成记录和后续操作那该多省事。这就是“监听个人微信收款实现扫码充值”这个项目的核心诉求。它不是什么高深莫测的黑科技而是一个非常接地气的自动化需求。简单来说就是让你的电脑程序能“知道”你的微信个人收款码什么时候收到了钱收了多少钱以及付款方的备注信息然后自动触发后续的业务逻辑比如给你的网站用户充值余额、给游戏账号加钻石、或者像我一样更新社群成员的积分状态。很多人第一反应可能是去调用微信支付官方接口。但这里有个关键区别微信支付商户接口面向的是企业需要营业执照、对公账户、缴纳费率并且资金结算到对公户流程复杂不适合个人或小微项目。而我们这里讨论的个人微信收款资金直接进入个人零钱方便快捷零成本。官方并没有为这种场景提供标准的“收款通知”API。所以我们的核心思路变成了如何用一种“曲线救国”的方式让程序感知到收款行为。基于最新的技术动态和社区实践一个成熟且稳定的方案浮出水面利用微信PC版的本地数据存储机制 一个轻量的后端服务如Flask。这个方案完全在本地运行不涉及任何破解、封号风险只是读取微信客户端自己生成并存储的数据。下面我就结合热词中频繁出现的Flask、Windows等关键词把这个方案的原理、搭建步骤、避坑细节完整地拆解一遍。2. 核心原理微信的“账本”藏在哪里要让程序知道收款情况我们首先得找到微信记录交易数据的地方。微信PC版为了在本地实现快速的消息和交易记录同步、离线查看等功能会将大量数据加密后存储在本地电脑的硬盘上。我们的目标就是找到这个存储交易记录的“数据库”。经过多方验证和测试在Windows系统上这个关键文件通常位于C:\Users\[你的用户名]\Documents\WeChat Files\[你的微信ID]\Msg\Multi\MSG.db这个MSG.db是一个SQLite数据库文件。SQLite是一个轻量级的、文件式的数据库很多桌面应用都用它来存储本地数据。微信用它来存储各种消息记录其中就包括转账/收款消息。注意微信ID是一个由字母和数字组成的长串形如wxid_xxxxxxxxxxxxxx它对应你登录的这个微信号。每个登录的微信号都会在WeChat Files目录下有一个独立的文件夹。那么程序如何读取这个数据库呢直接去读这个MSG.db文件行不行理论上可以但存在几个大问题即时性程序需要不断地、主动地去轮询Polling这个数据库文件检查是否有新记录。这就像你每隔5秒去翻一下账本效率低且占用资源。并发与锁微信客户端本身也在频繁读写这个文件。如果我们的读取程序时机不对可能会遇到数据库被锁定的情况导致读取失败。可靠性直接文件操作错误处理比较麻烦。因此更优雅、更可靠的做法不是直接硬啃这个.db文件而是利用一个中间桥梁。社区里探索出一个巧妙的思路微信PC版在收到新消息包括收款消息时除了更新MSG.db还会在内存或临时目录中生成某种格式的缓存或提示文件。我们的程序可以通过监控这些文件的变化来触发读取数据库的操作。具体来说我们可以监控微信存储文件的目录WeChat Files\[微信ID]下的某些特定子目录或文件的时间戳、大小变化。一旦检测到变化就认为可能有新的交易消息产生此时再连接MSG.db数据库执行一次精准的查询获取最新的收款记录。这个“监控变化”的任务可以由一个常驻后台的Python脚本来完成。而当我们获取到收款信息金额、备注、时间后就需要一个“大脑”来处理这些信息比如判断该给哪个用户充值、调用哪个接口。这个“大脑”就是一个Web服务。这就是Flask框架登场的原因。3. 技术栈选型与环境搭建为什么是Flask从热词可以看出python flask、flask框架、flask下载安装是关联非常紧密的搜索词。Flask是一个用Python编写的轻量级Web应用框架它被称为“微框架”核心简单但扩展性强。对于我们这个项目来说它有几个无可替代的优势轻量快速没有一大堆强制性的依赖和复杂的目录结构几行代码就能拉起一个HTTP服务。易于集成我们的监控脚本和业务逻辑处理脚本通常也用Python写与Flask服务同属一个语言生态数据交换、函数调用非常方便。适合做APIFlask天生适合构建RESTful API。我们的服务核心就是提供一两个API接口一个用于接收监控脚本发来的收款信息另一个可能用于查询状态。整个项目的技术栈可以这样规划监控端Watcher一个Python脚本使用watchdog库监控微信文件目录的变化使用sqlite3库读取MSG.db。服务端Server一个Flask应用提供API接口处理收款逻辑并连接你自己的业务数据库如MySQL即热词中的pymysql。业务数据库用于存储用户信息、充值记录等。可以用MySQLpymysql如果更轻量也可以用SQLite。3.1 Windows环境下的准备工作既然热词明确指向了Windows我们就以Windows 10/11为例。你需要准备好以下环境安装Python前往Python官网下载最新稳定版的Windows安装包如Python 3.9。安装时务必勾选“Add Python to PATH”这样才能在命令行中直接使用python和pip命令。安装必要的Python包打开命令提示符CMD或 PowerShell执行以下命令pip install flask pip install watchdog pip install pymysql # 如果你用MySQL # 如果连接SQLitePython标准库自带sqlite3无需安装找到你的微信文件路径按照第2部分所述打开文件资源管理器导航到C:\Users\[你的用户名]\Documents\WeChat Files\找到以你微信ID命名的文件夹。记下这个完整路径例如C:\Users\John\Documents\WeChat Files\wxid_abcdefg123456。3.2 Flask服务端基础骨架搭建我们先从核心的Flask服务写起。创建一个名为app.py的文件。from flask import Flask, request, jsonify import pymysql import json import time app Flask(__name__) # 配置你的MySQL数据库连接如果使用SQLite配置方式不同 db_config { host: localhost, user: your_username, password: your_password, database: recharge_system, # 你创建的数据库名 charset: utf8mb4 } def get_db_connection(): 创建数据库连接 connection pymysql.connect(**db_config) return connection app.route(/api/notify, methods[POST]) def payment_notify(): 接收监控脚本发来的收款通知 预期接收的JSON数据格式 { transaction_id: 微信内部交易ID, amount: 100.00, # 单位元 payer_remark: 用户充值备注, receive_time: 2023-10-27 15:30:00 } data request.json if not data: return jsonify({code: 400, msg: Invalid JSON data}), 400 # 1. 基础数据校验 required_fields [amount, payer_remark] for field in required_fields: if field not in data: return jsonify({code: 400, msg: fMissing field: {field}}), 400 amount data[amount] payer_remark data[payer_remark].strip() # 去除前后空格 # 2. 业务逻辑根据备注识别用户并充值 # 这里假设用户充值时会备注自己的用户ID例如“充值用户123” # 你需要根据你的业务规则来解析 payer_remark user_id parse_user_id_from_remark(payer_remark) if not user_id: # 无法识别用户记录到日志或待处理表 log_unknown_payment(data) return jsonify({code: 200, msg: Payment received but user not identified, logged for manual review.}) # 3. 检查是否重复处理根据transaction_id transaction_id data.get(transaction_id) if transaction_id and is_duplicate_transaction(transaction_id): return jsonify({code: 200, msg: Duplicate transaction, ignored.}) # 4. 执行充值操作 try: connection get_db_connection() with connection.cursor() as cursor: # 开启事务 connection.begin() # 更新用户余额 sql_update UPDATE users SET balance balance %s WHERE id %s cursor.execute(sql_update, (amount, user_id)) # 插入充值记录 sql_insert INSERT INTO payment_records (user_id, amount, transaction_id, remark, status, create_time) VALUES (%s, %s, %s, %s, success, NOW()) cursor.execute(sql_insert, (user_id, amount, transaction_id, payer_remark)) # 提交事务 connection.commit() connection.close() # 5. 可选触发后续动作如发送站内信、邮件通知等 trigger_post_recharge_actions(user_id, amount) return jsonify({code: 200, msg: Recharge successful!}) except Exception as e: # 如果发生错误回滚事务并记录日志 if connection in locals(): connection.rollback() connection.close() print(fDatabase operation failed: {e}) return jsonify({code: 500, msg: Internal server error}), 500 def parse_user_id_from_remark(remark): 从备注信息中解析出用户ID这里需要你自定义规则 # 示例1备注就是纯数字用户ID if remark.isdigit(): return int(remark) # 示例2备注格式为“uid:123” if remark.startswith(uid:): try: return int(remark[4:]) except ValueError: return None # 示例3更复杂的解析比如从你的业务系统中查询 # ... 你的自定义逻辑 ... return None def is_duplicate_transaction(transaction_id): 检查交易ID是否已存在防止重复充值 try: connection get_db_connection() with connection.cursor() as cursor: sql SELECT id FROM payment_records WHERE transaction_id %s LIMIT 1 cursor.execute(sql, (transaction_id,)) result cursor.fetchone() connection.close() return result is not None except Exception as e: print(fError checking duplicate transaction: {e}) return False # 出错时保守处理不认为是重复让主流程去处理数据库唯一约束会最终拦截 def log_unknown_payment(data): 将无法识别的支付记录到日志或特定表 with open(unknown_payments.log, a, encodingutf-8) as f: f.write(f{time.time()}: {json.dumps(data, ensure_asciiFalse)}\n) # 或者写入数据库的 pending_review 表 def trigger_post_recharge_actions(user_id, amount): 充值成功后的后续动作 # 例如发送邮件、微信模板消息、更新缓存等 print(fUser {user_id} recharged {amount} yuan. Post-actions triggered.) # 这里可以调用你的其他服务函数 if __name__ __main__: # 运行Flask应用host0.0.0.0允许同一局域网内其他设备访问 # debugTrue 仅用于开发生产环境必须设为 False app.run(host0.0.0.0, port5000, debugTrue)这个app.py文件构建了一个最核心的HTTP服务。它监听/api/notify这个路径的POST请求。我们的监控脚本在检测到收款后就会把相关信息打包成JSON发送到这个接口。4. 监控脚本让程序“看见”微信收款Flask服务是“大脑”负责处理。监控脚本则是“眼睛”负责看。我们需要编写另一个Python脚本wechat_watcher.py。这个脚本的核心任务是监控微信存储目录特别是MSG子目录的文件变动。当检测到变动时等待一个短暂时间如1-2秒以确保微信客户端已完成写入。连接MSG.db数据库查询最新的收款记录。将查询到的、未处理过的记录发送给Flask服务的/api/notify接口。4.1 理解微信MSG.db的表结构关键这是整个项目最需要耐心和技巧的部分。微信的数据库表结构是未公开的我们需要通过工具如DB Browser for SQLite来探索。大致思路如下关闭微信客户端避免读写冲突。将MSG.db文件复制一份到其他位置作为备份和分析用。用SQLite工具打开备份文件查看有哪些表。通常与消息相关的表名可能包含Message、Chat等。通过分析字段名和内容来推测表用途。收款消息的type字段可能是一个特定的数字如49或某个子类型内容字段content可能是一个XML格式的字符串里面包含了转账金额、交易单号、备注等信息。以下是一个基于常见社区经验的模拟查询请注意实际表名和字段名可能需要你自行探索验证import sqlite3 import os import time import requests import json # 配置 WECHAT_DB_PATH rC:\Users\John\Documents\WeChat Files\wxid_abcdefg123456\Msg\Multi\MSG.db FLASK_SERVER_URL http://127.0.0.1:5000/api/notify # 记录上次处理过的消息ID避免重复发送 LAST_PROCESSED_MSG_ID 0 def query_new_payments(): 查询新的收款消息 global LAST_PROCESSED_MSG_ID if not os.path.exists(WECHAT_DB_PATH): print(fDatabase file not found: {WECHAT_DB_PATH}) return [] try: # 连接数据库注意使用只读模式避免干扰微信客户端 conn sqlite3.connect(ffile:{WECHAT_DB_PATH}?modero, uriTrue) conn.row_factory sqlite3.Row # 使返回结果为字典-like的Row对象 cursor conn.cursor() # 这是一个示例SQL你需要根据实际表结构调整 # 假设表名为 Message包含字段 MsgId, Type, Content, CreateTime # 假设收款消息的 Type 是 49并且在 Content 字段里包含特定标识 sql SELECT MsgId, Content, CreateTime FROM Message WHERE Type 49 AND MsgId ? AND Content LIKE %收款% -- 这是一个示例实际可能是别的关键词 ORDER BY MsgId ASC LIMIT 10 cursor.execute(sql, (LAST_PROCESSED_MSG_ID,)) rows cursor.fetchall() conn.close() payments [] for row in rows: msg_id row[MsgId] content row[Content] # 这里可能是XML create_time row[CreateTime] # 可能是时间戳 # 解析XML内容提取金额、交易单号、备注 # 这是一个解析示例实际情况非常复杂需要你仔细分析Content的结构 # 例如msg收款from.../fromamount100/amountremark用户123充值/remarktransid123456789/transid/收款/msg try: # 这里需要你写一个具体的解析函数 parse_payment_content(content) payment_info parse_payment_content(content) if payment_info: payment_info[msg_id] msg_id payment_info[receive_time] timestamp_to_str(create_time) payments.append(payment_info) # 更新最后处理的消息ID if msg_id LAST_PROCESSED_MSG_ID: LAST_PROCESSED_MSG_ID msg_id except Exception as e: print(fError parsing content for MsgId {msg_id}: {e}) continue return payments except sqlite3.Error as e: print(fDatabase error: {e}) return [] def parse_payment_content(content): 解析微信收款消息的XML内容。 这是最需要定制化的部分你需要根据自己微信收到的收款消息格式来编写。 一个笨但有效的方法是在微信上让朋友给你转一笔账然后立刻去数据库里查这条记录分析Content字段的完整内容。 # 简化示例假设我们能提取到以下信息 # 实际情况请使用 xml.etree.ElementTree 或正则表达式进行解析 import re amount_match re.search(ramount(\d(\.\d)?)/amount, content) remark_match re.search(rremark(.*?)/remark, content) transid_match re.search(rtransid(.*?)/transid, content) if amount_match: return { amount: float(amount_match.group(1)), payer_remark: remark_match.group(1) if remark_match else , transaction_id: transid_match.group(1) if transid_match else fmsg_{int(time.time())} } return None def timestamp_to_str(timestamp): 将微信可能使用的时间戳转换为可读字符串 # 微信的时间戳可能是秒或毫秒需要测试 try: # 先尝试秒 if timestamp 1000000000000: # 判断是否为毫秒级13位 timestamp timestamp / 1000 return time.strftime(%Y-%m-%d %H:%M:%S, time.localtime(timestamp)) except: return str(timestamp) def send_to_flask(payment_info): 将收款信息发送给Flask服务 try: headers {Content-Type: application/json} response requests.post(FLASK_SERVER_URL, datajson.dumps(payment_info), headersheaders, timeout5) if response.status_code 200: print(fSuccessfully notified server: {payment_info}) else: print(fServer returned error: {response.status_code}, {response.text}) except requests.exceptions.RequestException as e: print(fFailed to send notification: {e}) def main_watch_loop(): 主监控循环 print(fStarting WeChat payment watcher. Monitoring DB: {WECHAT_DB_PATH}) print(fFlask server: {FLASK_SERVER_URL}) print(Press CtrlC to stop.) # 初始化获取当前最新的消息ID避免处理历史记录 init_last_msg_id() while True: try: new_payments query_new_payments() for payment in new_payments: send_to_flask(payment) # 每3秒检查一次频率不宜过高 time.sleep(3) except KeyboardInterrupt: print(\nWatcher stopped by user.) break except Exception as e: print(fUnexpected error in main loop: {e}) time.sleep(10) # 出错后等待稍长时间 def init_last_msg_id(): 初始化 LAST_PROCESSED_MSG_ID 为当前最大的MsgId避免启动时处理旧消息 global LAST_PROCESSED_MSG_ID try: conn sqlite3.connect(ffile:{WECHAT_DB_PATH}?modero, uriTrue) cursor conn.cursor() cursor.execute(SELECT MAX(MsgId) FROM Message) # 假设表名和字段名 max_id cursor.fetchone()[0] if max_id: LAST_PROCESSED_MSG_ID max_id print(fInitialized last processed MsgId to: {LAST_PROCESSED_MSG_ID}) conn.close() except Exception as e: print(fCould not init last msg id: {e}) if __name__ __main__: main_watch_loop()4.2 使用Watchdog优化监控效率上面的循环查询 (time.sleep(3)) 是轮询不够高效。更优的方案是使用watchdog库监听文件系统事件。我们可以监控MSG.db文件本身或其父目录的修改事件。from watchdog.observers import Observer from watchdog.events import FileSystemEventHandler import threading class WeChatDBHandler(FileSystemEventHandler): 处理微信数据库文件变更事件 def on_modified(self, event): if not event.is_directory and event.src_path.endswith(MSG.db): print(fDetected change in: {event.src_path}) # 为了避免频繁触发可以在这里设置一个防抖延迟 # 例如启动一个线程等待2秒后再执行查询 threading.Thread(targetdelayed_query, args(2,)).start() def delayed_query(delay_seconds): 延迟查询避免微信正在写入时读取 time.sleep(delay_seconds) new_payments query_new_payments() for payment in new_payments: send_to_flask(payment) def main_watchdog(): 使用watchdog进行事件监听 path_to_watch os.path.dirname(WECHAT_DB_PATH) # 监控数据库所在目录 event_handler WeChatDBHandler() observer Observer() observer.schedule(event_handler, path_to_watch, recursiveFalse) observer.start() print(fStarted watchdog observer on: {path_to_watch}) try: while True: time.sleep(1) except KeyboardInterrupt: observer.stop() observer.join()将主函数main_watch_loop()替换为main_watchdog()监控脚本的效率会高很多只在文件实际发生变化时才工作。5. 部署、运行与实战避坑指南现在我们有了app.py(Flask服务) 和wechat_watcher.py(监控脚本)。如何让它们稳定地跑起来5.1 分步运行与测试启动Flask服务 打开一个命令行窗口导航到app.py所在目录运行python app.py你会看到类似* Running on http://0.0.0.0:5000的输出。保持这个窗口运行。启动监控脚本 打开另一个命令行窗口导航到wechat_watcher.py所在目录运行python wechat_watcher.py脚本开始运行打印出监控路径。模拟测试 这是最关键的一步。不要直接用真钱测试。你可以修改监控脚本在query_new_payments函数里暂时不查询数据库而是直接模拟返回一条支付数据用于测试Flask接口是否正常。# 在 query_new_payments 函数里临时添加测试代码 def query_new_payments(): # ... 原有代码 ... # 测试阶段直接返回模拟数据 test_payment { transaction_id: test_123, amount: 0.01, payer_remark: 测试用户666, receive_time: time.strftime(%Y-%m-%d %H:%M:%S) } return [test_payment] # 注释掉数据库查询部分用这个返回测试运行监控脚本观察Flask服务的控制台是否打印出处理日志以及是否返回“Recharge successful”。数据库联调 确保你的业务数据库如MySQL已经创建好users(用户) 表和payment_records(支付记录) 表并且app.py中的数据库配置正确。当模拟数据触发充值逻辑后检查数据库表中数据是否更新。5.2 生产环境部署要点开发测试没问题后要长期稳定运行需要考虑以下几点以Windows服务运行不能让两个命令行窗口一直开着。可以将python app.py和python wechat_watcher.py分别封装成Windows服务。推荐使用NSSM(the Non-Sucking Service Manager) 这个工具它可以将任何可执行程序包括Python脚本安装为系统服务并管理其启动、停止、重启和日志。为Flask服务创建服务nssm install MyFlaskService python c:\path\to\your\app.py为监控脚本创建服务nssm install WeChatWatcher python c:\path\to\your\wechat_watcher.py设置服务为自动启动并配置好工作目录和日志路径。处理微信多开如果你电脑上登录了多个微信每个微信ID都有独立的MSG.db。你的监控脚本需要能处理这种情况。可以修改脚本扫描WeChat Files目录下的所有子文件夹对每一个疑似微信ID的文件夹都启动一个监控线程或进程或者只监控你指定的主要账号。消息去重与幂等性这是保证系统可靠性的核心。必须确保同一笔收款不会导致重复充值。我们的方案采用了双重保障监控脚本去重通过记录LAST_PROCESSED_MSG_ID只处理比上次ID更大的消息。业务服务去重Flask接口中通过transaction_id在payment_records表中查询是否已存在。更严谨的做法是在数据库表中为transaction_id字段建立唯一索引这样即使逻辑层漏判数据库层也会抛出唯一约束异常从而阻止重复数据插入。错误处理与日志务必完善错误处理。网络请求可能失败数据库可能连接不上解析可能出错。所有关键步骤查询数据库、发送请求、更新余额都要有try...except包裹并将错误信息记录到文件日志中如Python的logging模块方便日后排查。app.py中已经包含了基本的错误处理和日志记录。安全考虑Flask服务运行在http://0.0.0.0:5000意味着同一局域网内的设备都能访问。在生产环境中这存在风险。建议添加简单的API密钥验证。在监控脚本发送请求时在Header中携带一个密钥Flask服务端验证这个密钥。或者使用反向代理如Nginx将Flask服务暴露到公网并配置HTTPS和防火墙规则只允许监控脚本所在服务器的IP访问。最安全的方式是Flask服务和监控脚本都部署在同一台电脑上Flask只监听127.0.0.1本地回环地址这样外部网络根本无法访问。5.3 我踩过的那些坑数据库被锁sqlite3.OperationalError: database is locked这是最常见的问题。微信客户端正在写入时我们的脚本去读就会冲突。解决方案使用只读模式连接sqlite3.connect(ffile:{db_path}?modero, uriTrue)。在检测到文件变更后等待1-2秒再查询给微信客户端留出写入完成的时间。如果还不行可以尝试复制MSG.db到一个临时位置再读取需确保微信未在写入时复制否则可能拿到不完整文件。表结构变更微信客户端更新后MSG.db的表结构或字段含义可能会变。解决方案不要将解析逻辑写死。将解析函数独立出来并做好日志。一旦发现解析失败或提取不到数据立即检查最新的数据库结构调整解析逻辑。监控脚本意外退出作为长期运行的服务稳定性第一。解决方案使用try...except捕获所有未知异常并在异常发生后等待一段时间后继续循环而不是直接崩溃。使用NSSM等工具将脚本安装为服务它们可以配置服务崩溃后自动重启。在脚本内添加“心跳”机制定期向一个日志文件或某个监控端点写入状态便于你知晓它是否还活着。备注信息识别率用户填写的备注五花八门parse_user_id_from_remark函数是业务逻辑的瓶颈。解决方案设计清晰的充值指引让用户按固定格式备注如“充值#用户ID”。对于无法识别的备注不要直接丢弃。像代码中那样记录到unknown_payments.log或一个待审核数据库表并提供一个人工处理的后台界面手动将支付记录与用户关联。这个项目从想法到实现涉及了文件监控、数据库解析、Web服务、业务逻辑整合等多个环节是一个非常好的全栈练手项目。它不要求你有多高深的算法知识但极其考验你的系统思维、问题分解和debug能力。当你看到第一笔测试收款成功触发自动充值时那种自动化带来的成就感是非常实在的。希望这份详细的拆解能帮你绕过我踩过的坑顺利搭建起属于自己的微信收款监听系统。