1. 项目概述什么是Cleaner Bot如果你也像我一样每天被各种群聊里的垃圾信息、过期文件、无效通知搞得心烦意乱那“Cleaner Bot”这个项目你一定会感兴趣。简单来说Cleaner Bot就是一个能帮你自动清理数字空间的智能机器人。它不是一个具体的软件而是一套自动化清理的解决方案你可以把它理解为一个24小时在线的“数字保洁员”。这个项目的核心价值在于它解放了我们的双手和注意力。无论是微信、QQ、钉钉等社交/办公群组里堆积如山的广告、过期的活动通知还是云盘、本地文件夹里那些早已忘记的临时文件、重复照片甚至是邮箱里塞满的订阅邮件Cleaner Bot都能根据你设定的规则自动、精准地进行识别和清理。它解决的不仅仅是“空间”问题更是“信息过载”带来的效率低下和焦虑感。对于需要管理大量社群的组织者、追求极简数字生活的个人或是单纯想保持工作环境整洁的开发者来说这都是一件能极大提升幸福感的工具。2. 核心设计思路如何让机器人“聪明”地打扫一个合格的保洁员不能只会无差别地乱扔东西。Cleaner Bot的设计核心就是如何让它具备“判断力”知道什么该留什么该丢。这背后是一套从“感知”到“决策”再到“执行”的完整逻辑链条。2.1 规则引擎定义清理的“宪法”一切自动化的前提是明确的规则。Cleaner Bot的规则引擎是其大脑。我们需要定义多种维度的规则时间规则最基础的规则。例如“清理7天前的所有图片缓存”、“删除超过30天未读的促销邮件”。内容规则基于关键词、正则表达式或文件类型的匹配。例如在群聊中自动删除包含“扫码领红包”、“加我好友”等关键词的消息在文件夹中删除所有.tmp或.log临时文件。来源规则指定清理范围。例如只清理“摸鱼交流群”而放过“项目攻坚组”只清理“下载”文件夹而不动“工作文档”。条件组合规则高级规则允许“与或非”逻辑组合。例如“删除来自‘XX营销号’且内容包含‘限时折扣’并且发送时间在凌晨2点到5点之间的消息”。这能精准打击特定模式的垃圾信息。规则的设计需要平衡“彻底”和“安全”。过于宽松则效果不佳过于严苛则可能误伤重要信息。我的经验是初期规则宜松不宜紧可以先从“时间规则”这种绝对安全的维度入手运行观察一段时间再逐步添加更精细的内容规则。2.2 对接平台打通清理的“任督二脉”Cleaner Bot需要与目标平台进行交互。根据平台开放程度主要有两种方式官方API对接对于提供了开放接口的平台如企业微信、钉钉、部分邮箱服务、云存储服务这是最稳定、最合规的方式。通过APIBot可以合法地读取消息列表、文件列表并执行删除操作。这种方式权限清晰行为可控。自动化脚本模拟对于未开放API或API功能受限的平台如个人微信、QQ则可能需要借助一些自动化工具如基于pyautogui、selenium的脚本来模拟用户操作。这种方式存在较大风险包括但不限于账号被封禁、操作不稳定、违反平台用户协议等必须慎之又慎。在个人学习和小范围测试中务必使用小号并加入大量随机延迟和容错处理模拟真人操作。注意任何自动化操作都必须遵守平台规则。优先寻找和使用官方支持的自动化方案如邮箱的过滤器、云盘的自动清理功能在不得不使用非官方方式时务必以学习研究为目的并严格控制频率和范围避免对平台和其他用户造成影响。2.3 执行与日志留下清晰的“工作记录”执行环节不仅要“能做”还要“可查”、“可逆”。Cleaner Bot的每一次清理操作都必须记录详尽的日志至少包括操作时间、目标平台/位置、清理内容摘要如消息ID、文件名、触发的规则。这不仅是审计的需要更是出问题后能及时恢复的关键。对于文件删除一个最佳实践是“先移动后删除”。即先将被规则匹配的文件移动到某个“回收站”或“待审查”目录保留一段时间如7天再由另一个定时任务进行最终物理删除。这为误操作提供了宝贵的缓冲期。对于消息虽然通常无法“移动”但完整的日志可以让你知道“谁”被“何时”以“何理由”删除方便追溯。3. 关键技术点与工具选型解析实现一个健壮的Cleaner Bot涉及多个技术栈的选择。这里我基于常见实践分享一套高可行性的方案。3.1 后端核心Python 异步框架Python因其丰富的库和简洁的语法是构建此类自动化脚本的首选。核心框架选择FastAPI / Sanic如果你希望Bot提供一个Web控制面板来管理规则、查看日志那么选择一个异步Web框架是必要的。FastAPI现代、性能好适合快速构建API。APScheduler这是定时任务调度的不二之选。它可以非常方便地配置Cron式的定时任务例如“每天凌晨3点执行一次全量扫描清理”。SQLAlchemy SQLite用于存储规则配置、执行日志。SQLite足够轻量无需单独部署数据库服务适合个人或小团队使用。SQLAlchemy作为ORM让数据操作更优雅。3.2 平台对接库根据目标平台选择相应的SDK或库企业微信/钉钉使用官方提供的Python SDK。这些SDK通常封装完善认证、消息收发、文件管理等功能一应俱全。邮箱处理使用imaplib(收取) 和smtplib(发送) 标准库或更高级的aioimaplib进行异步操作。配合email库解析邮件头和信息。文件系统操作Python内置的os、shutil、pathlib模块就足够强大用于本地文件的遍历、移动、删除。云存储如对接阿里云OSS、腾讯云COS都有官方Python SDK可直接操作云端文件。3.3 消息内容分析与匹配这是体现Bot“智能”的关键关键词匹配简单的字符串in操作或正则表达式re模块足以应对大部分场景。相似度匹配对于变着花样的广告如“加V12345”和“薇❤abcde”可以使用文本相似度算法。jieba分词后结合gensim或sentence-transformers计算余弦相似度与预设的垃圾信息样本库进行比对。图片/文件识别对于清理重复图片或特定表情包可以使用imagehash库计算图片的感知哈希pHash快速比对相似图片。对于文件可以通过计算MD5/SHA1来精确识别重复文件。3.4 部署与运行本地运行最简单的方式在个人电脑上以脚本或服务形式常驻运行。适合清理本地文件和个人账号。服务器部署对于需要7x24小时运行或处理公共/团队资源的Bot建议部署在云服务器上。使用systemd或supervisor来管理进程确保其稳定运行并在崩溃后自动重启。容器化使用Docker将Bot及其依赖环境打包可以极大简化部署并保证环境一致性。这对于需要对接多种复杂依赖的项目尤其有用。4. 实操构建一个微信群聊Cleaner Bot的完整实现下面我将以一个清理微信非官方API方式仅供技术研究群聊广告的Bot为例拆解核心实现步骤。再次强调此方法风险极高务必在测试环境或小号中进行并严格遵守平台规则。4.1 环境准备与依赖安装首先我们需要一个能控制微信客户端的库。这里以itchat的一个分支wxauto假设实际需寻找当前可用的库为例它基于Windows UI自动化。实际上由于微信客户端频繁更新这类库极不稳定这里主要展示思路。# 假设的依赖实际请根据可用库调整 pip install wxauto schedule sqlalchemy我们使用SQLite记录日志。# config.py - 配置和数据库模型 from sqlalchemy import create_engine, Column, String, Integer, DateTime, Text from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from datetime import datetime engine create_engine(sqlite:///cleaner_bot.db) Base declarative_base() class CleanupLog(Base): __tablename__ cleanup_logs id Column(Integer, primary_keyTrue) timestamp Column(DateTime, defaultdatetime.now) chatroom_name Column(String) message_sender Column(String) message_content Column(Text) matched_rule Column(String) action Column(String) # deleted, flagged Base.metadata.create_all(engine) Session sessionmaker(bindengine)4.2 规则定义与匹配逻辑我们定义几条简单的规则并编写匹配函数。# rules.py import re from datetime import datetime, timedelta class CleanupRule: def __init__(self, name, pattern, rule_typekeyword, case_sensitiveFalse, expire_hoursNone): self.name name self.pattern pattern # 可以是关键词字符串、正则表达式字符串或文件扩展名 self.rule_type rule_type # keyword, regex, file_type self.case_sensitive case_sensitive self.expire_hours expire_hours # 消息存在多久后才可被清理 def match(self, content, send_timeNone): 匹配消息内容或文件类型 if self.expire_hours and send_time: if datetime.now() - send_time timedelta(hoursself.expire_hours): return False # 未过期不清理 content_to_check content if self.case_sensitive else content.lower() pattern_to_check self.pattern if self.case_sensitive else self.pattern.lower() if self.rule_type keyword: return pattern_to_check in content_to_check elif self.rule_type regex: try: return re.search(pattern_to_check, content_to_check) is not None except re.error: return False elif self.rule_type file_type: # 假设content是文件名 return content.endswith(self.pattern) return False # 定义规则列表 RULES [ CleanupRule(广告-扫码, r扫码.*(加群|领红包|咨询|关注), rule_typeregex), CleanupRule(广告-微信号, r[VvWw][:]?\s*[a-zA-Z0-9_\-]{5,}, rule_typeregex), CleanupRule(临时文件, .tmp, rule_typefile_type), # 一条时间规则仅清理12小时前的广告 CleanupRule(过期广告, r扫码.*(加群|领红包), rule_typeregex, expire_hours12), ]4.3 主循环与清理执行这里是核心的调度和执行逻辑。由于直接操作微信UI不稳定代码中加入大量异常处理和延迟。# bot_core.py import time import schedule from datetime import datetime from sqlalchemy.orm import Session from config import Session as DBSession, CleanupLog from rules import RULES # 假设的微信控制库导入 # from wxauto import WeChat class WeChatCleanerBot: def __init__(self): # self.wx WeChat() # 实际初始化 self.db_session DBSession() self.target_chatrooms [测试群, 另一个群] # 指定要清理的群 def check_and_clean_messages(self): 检查并清理消息的主函数 print(f[{datetime.now()}] 开始扫描群聊消息...) for chatroom in self.target_chatrooms: try: # 模拟获取最近N条消息实际中需要根据库的API来获取 # messages self.wx.get_chatroom_messages(chatroom, count100) messages self._mock_get_messages(chatroom) # 模拟数据 for msg in messages: if self._should_clean(msg): self._perform_cleanup(chatroom, msg) time.sleep(1) # 重要每次操作后随机延迟模拟真人 except Exception as e: print(f处理群聊【{chatroom}】时出错: {e}) self._log_to_db(chatroom, None, None, fERROR: {e}, error) def _should_clean(self, msg): 判断单条消息是否应该被清理 for rule in RULES: if rule.match(msg[content], msg.get(send_time)): return True, rule.name return False, None def _perform_cleanup(self, chatroom, msg): 执行清理操作并记录日志 should_clean, rule_name self._should_clean(msg) if not should_clean: return # 实际执行删除操作这里用打印模拟 # success self.wx.delete_message(msg[id]) success True # 模拟成功 action deleted if success else failed print(f - 在【{chatroom}】中{action}一条消息。发送者{msg[sender]} 内容{msg[content][:50]}... 规则{rule_name}) # 记录日志到数据库 self._log_to_db(chatroom, msg[sender], msg[content], rule_name, action) def _log_to_db(self, chatroom, sender, content, rule, action): log CleanupLog( chatroom_namechatroom, message_sendersender, message_contentcontent, matched_rulerule, actionaction ) self.db_session.add(log) self.db_session.commit() def _mock_get_messages(self, chatroom): 模拟获取消息实际项目中替换为真实API调用 import random mock_msgs [ {sender:用户A, content:大家好扫码加群领福利, send_time: datetime.now()}, {sender:用户B, content:今晚开会。, send_time: datetime.now()}, {sender:广告哥, content:VX123456 低价出货, send_time: datetime.now()}, ] return mock_msgs def run(self): 启动定时任务 schedule.every(30).minutes.do(self.check_and_clean_messages) # 每30分钟运行一次 print(Cleaner Bot 已启动定时任务每30分钟执行一次。) while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次任务 if __name__ __main__: bot WeChatCleanerBot() bot.run()4.4 增加安全性与误杀恢复为了防止误删重要消息我们可以引入“白名单”和“确认机制”。白名单用户群主、管理员或特定重要成员的发言永远不清理。关键词白名单包含特定关键词如“会议”、“紧急”、“报销”的消息即使匹配广告规则也跳过。二次确认机制对于高置信度的广告如匹配多条规则直接删除对于低置信度的可以标记出来发送到另一个“待审核”频道或给自己发一条提醒由人工最终确认。恢复则依赖于之前提到的完整日志。如果发现误删可以第一时间查看日志找到被删除消息的准确内容、发送者和时间然后在群里进行解释或请求发送者重发。对于文件清理因为采用了“先移动后删除”的策略恢复就是简单的文件移动操作。5. 避坑指南与实战经验在开发和运行Cleaner Bot的过程中我踩过不少坑这里总结几条血泪教训频率与延迟是生命线无论是调用API还是模拟操作频率过高是导致账号被风控或封禁的最主要原因。务必为所有自动化操作添加随机延迟time.sleep(random.uniform(2, 5))并且将扫描/清理频率设置在平台能接受的合理范围如每小时一次而不是每分钟一次。规则测试务必充分上线任何新规则前一定要在测试环境或小范围进行至少一周的“观察模式”运行。即Bot只记录匹配到的消息和将要执行的操作而不实际删除。通过日志分析规则的准确率和误杀率反复调整优化。做好异常处理与状态监控网络会波动平台API会变更客户端会升级。你的Bot必须有完善的异常捕获和重试机制。同时最好实现一个简单的心跳或状态上报功能比如每天定时向你的邮箱发送一份运行报告让你知道它还在正常工作。法律与隐私红线清理他人消息或文件时必须明确权限。在社群中最好事先公告征得成员同意。清理公司内部资料必须有明确的授权和合规流程。绝对不要用Bot去做爬取隐私、恶意刷屏等违法违规的事情。代码与配置分离将群聊名单、关键词规则、延迟参数等所有可配置项都放在配置文件如config.yaml或数据库中。这样在调整时无需修改代码也方便进行版本管理。单一职责与模块化将规则匹配、平台操作、日志记录、任务调度等模块清晰地分开。这样当需要从清理微信切换到清理钉钉时你只需要替换“平台操作”这个模块其他部分可以复用。构建一个Cleaner Bot的过程更像是在制定一套精细的数字空间管理策略。技术实现只是手段真正的核心在于你对信息价值的判断和取舍。当你看到那些无关紧要的冗余信息被自动清走重要的内容得以凸显时那种数字世界焕然一新的清爽感就是对这个项目最好的回报。从最简单的定时清理临时文件开始逐步扩展它的能力你会发现自己对自动化工具的理解和应用水平也在不断提升。 构建智能Cleaner Bot：自动化清理数字空间的技术实现与实战
1. 项目概述什么是Cleaner Bot如果你也像我一样每天被各种群聊里的垃圾信息、过期文件、无效通知搞得心烦意乱那“Cleaner Bot”这个项目你一定会感兴趣。简单来说Cleaner Bot就是一个能帮你自动清理数字空间的智能机器人。它不是一个具体的软件而是一套自动化清理的解决方案你可以把它理解为一个24小时在线的“数字保洁员”。这个项目的核心价值在于它解放了我们的双手和注意力。无论是微信、QQ、钉钉等社交/办公群组里堆积如山的广告、过期的活动通知还是云盘、本地文件夹里那些早已忘记的临时文件、重复照片甚至是邮箱里塞满的订阅邮件Cleaner Bot都能根据你设定的规则自动、精准地进行识别和清理。它解决的不仅仅是“空间”问题更是“信息过载”带来的效率低下和焦虑感。对于需要管理大量社群的组织者、追求极简数字生活的个人或是单纯想保持工作环境整洁的开发者来说这都是一件能极大提升幸福感的工具。2. 核心设计思路如何让机器人“聪明”地打扫一个合格的保洁员不能只会无差别地乱扔东西。Cleaner Bot的设计核心就是如何让它具备“判断力”知道什么该留什么该丢。这背后是一套从“感知”到“决策”再到“执行”的完整逻辑链条。2.1 规则引擎定义清理的“宪法”一切自动化的前提是明确的规则。Cleaner Bot的规则引擎是其大脑。我们需要定义多种维度的规则时间规则最基础的规则。例如“清理7天前的所有图片缓存”、“删除超过30天未读的促销邮件”。内容规则基于关键词、正则表达式或文件类型的匹配。例如在群聊中自动删除包含“扫码领红包”、“加我好友”等关键词的消息在文件夹中删除所有.tmp或.log临时文件。来源规则指定清理范围。例如只清理“摸鱼交流群”而放过“项目攻坚组”只清理“下载”文件夹而不动“工作文档”。条件组合规则高级规则允许“与或非”逻辑组合。例如“删除来自‘XX营销号’且内容包含‘限时折扣’并且发送时间在凌晨2点到5点之间的消息”。这能精准打击特定模式的垃圾信息。规则的设计需要平衡“彻底”和“安全”。过于宽松则效果不佳过于严苛则可能误伤重要信息。我的经验是初期规则宜松不宜紧可以先从“时间规则”这种绝对安全的维度入手运行观察一段时间再逐步添加更精细的内容规则。2.2 对接平台打通清理的“任督二脉”Cleaner Bot需要与目标平台进行交互。根据平台开放程度主要有两种方式官方API对接对于提供了开放接口的平台如企业微信、钉钉、部分邮箱服务、云存储服务这是最稳定、最合规的方式。通过APIBot可以合法地读取消息列表、文件列表并执行删除操作。这种方式权限清晰行为可控。自动化脚本模拟对于未开放API或API功能受限的平台如个人微信、QQ则可能需要借助一些自动化工具如基于pyautogui、selenium的脚本来模拟用户操作。这种方式存在较大风险包括但不限于账号被封禁、操作不稳定、违反平台用户协议等必须慎之又慎。在个人学习和小范围测试中务必使用小号并加入大量随机延迟和容错处理模拟真人操作。注意任何自动化操作都必须遵守平台规则。优先寻找和使用官方支持的自动化方案如邮箱的过滤器、云盘的自动清理功能在不得不使用非官方方式时务必以学习研究为目的并严格控制频率和范围避免对平台和其他用户造成影响。2.3 执行与日志留下清晰的“工作记录”执行环节不仅要“能做”还要“可查”、“可逆”。Cleaner Bot的每一次清理操作都必须记录详尽的日志至少包括操作时间、目标平台/位置、清理内容摘要如消息ID、文件名、触发的规则。这不仅是审计的需要更是出问题后能及时恢复的关键。对于文件删除一个最佳实践是“先移动后删除”。即先将被规则匹配的文件移动到某个“回收站”或“待审查”目录保留一段时间如7天再由另一个定时任务进行最终物理删除。这为误操作提供了宝贵的缓冲期。对于消息虽然通常无法“移动”但完整的日志可以让你知道“谁”被“何时”以“何理由”删除方便追溯。3. 关键技术点与工具选型解析实现一个健壮的Cleaner Bot涉及多个技术栈的选择。这里我基于常见实践分享一套高可行性的方案。3.1 后端核心Python 异步框架Python因其丰富的库和简洁的语法是构建此类自动化脚本的首选。核心框架选择FastAPI / Sanic如果你希望Bot提供一个Web控制面板来管理规则、查看日志那么选择一个异步Web框架是必要的。FastAPI现代、性能好适合快速构建API。APScheduler这是定时任务调度的不二之选。它可以非常方便地配置Cron式的定时任务例如“每天凌晨3点执行一次全量扫描清理”。SQLAlchemy SQLite用于存储规则配置、执行日志。SQLite足够轻量无需单独部署数据库服务适合个人或小团队使用。SQLAlchemy作为ORM让数据操作更优雅。3.2 平台对接库根据目标平台选择相应的SDK或库企业微信/钉钉使用官方提供的Python SDK。这些SDK通常封装完善认证、消息收发、文件管理等功能一应俱全。邮箱处理使用imaplib(收取) 和smtplib(发送) 标准库或更高级的aioimaplib进行异步操作。配合email库解析邮件头和信息。文件系统操作Python内置的os、shutil、pathlib模块就足够强大用于本地文件的遍历、移动、删除。云存储如对接阿里云OSS、腾讯云COS都有官方Python SDK可直接操作云端文件。3.3 消息内容分析与匹配这是体现Bot“智能”的关键关键词匹配简单的字符串in操作或正则表达式re模块足以应对大部分场景。相似度匹配对于变着花样的广告如“加V12345”和“薇❤abcde”可以使用文本相似度算法。jieba分词后结合gensim或sentence-transformers计算余弦相似度与预设的垃圾信息样本库进行比对。图片/文件识别对于清理重复图片或特定表情包可以使用imagehash库计算图片的感知哈希pHash快速比对相似图片。对于文件可以通过计算MD5/SHA1来精确识别重复文件。3.4 部署与运行本地运行最简单的方式在个人电脑上以脚本或服务形式常驻运行。适合清理本地文件和个人账号。服务器部署对于需要7x24小时运行或处理公共/团队资源的Bot建议部署在云服务器上。使用systemd或supervisor来管理进程确保其稳定运行并在崩溃后自动重启。容器化使用Docker将Bot及其依赖环境打包可以极大简化部署并保证环境一致性。这对于需要对接多种复杂依赖的项目尤其有用。4. 实操构建一个微信群聊Cleaner Bot的完整实现下面我将以一个清理微信非官方API方式仅供技术研究群聊广告的Bot为例拆解核心实现步骤。再次强调此方法风险极高务必在测试环境或小号中进行并严格遵守平台规则。4.1 环境准备与依赖安装首先我们需要一个能控制微信客户端的库。这里以itchat的一个分支wxauto假设实际需寻找当前可用的库为例它基于Windows UI自动化。实际上由于微信客户端频繁更新这类库极不稳定这里主要展示思路。# 假设的依赖实际请根据可用库调整 pip install wxauto schedule sqlalchemy我们使用SQLite记录日志。# config.py - 配置和数据库模型 from sqlalchemy import create_engine, Column, String, Integer, DateTime, Text from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from datetime import datetime engine create_engine(sqlite:///cleaner_bot.db) Base declarative_base() class CleanupLog(Base): __tablename__ cleanup_logs id Column(Integer, primary_keyTrue) timestamp Column(DateTime, defaultdatetime.now) chatroom_name Column(String) message_sender Column(String) message_content Column(Text) matched_rule Column(String) action Column(String) # deleted, flagged Base.metadata.create_all(engine) Session sessionmaker(bindengine)4.2 规则定义与匹配逻辑我们定义几条简单的规则并编写匹配函数。# rules.py import re from datetime import datetime, timedelta class CleanupRule: def __init__(self, name, pattern, rule_typekeyword, case_sensitiveFalse, expire_hoursNone): self.name name self.pattern pattern # 可以是关键词字符串、正则表达式字符串或文件扩展名 self.rule_type rule_type # keyword, regex, file_type self.case_sensitive case_sensitive self.expire_hours expire_hours # 消息存在多久后才可被清理 def match(self, content, send_timeNone): 匹配消息内容或文件类型 if self.expire_hours and send_time: if datetime.now() - send_time timedelta(hoursself.expire_hours): return False # 未过期不清理 content_to_check content if self.case_sensitive else content.lower() pattern_to_check self.pattern if self.case_sensitive else self.pattern.lower() if self.rule_type keyword: return pattern_to_check in content_to_check elif self.rule_type regex: try: return re.search(pattern_to_check, content_to_check) is not None except re.error: return False elif self.rule_type file_type: # 假设content是文件名 return content.endswith(self.pattern) return False # 定义规则列表 RULES [ CleanupRule(广告-扫码, r扫码.*(加群|领红包|咨询|关注), rule_typeregex), CleanupRule(广告-微信号, r[VvWw][:]?\s*[a-zA-Z0-9_\-]{5,}, rule_typeregex), CleanupRule(临时文件, .tmp, rule_typefile_type), # 一条时间规则仅清理12小时前的广告 CleanupRule(过期广告, r扫码.*(加群|领红包), rule_typeregex, expire_hours12), ]4.3 主循环与清理执行这里是核心的调度和执行逻辑。由于直接操作微信UI不稳定代码中加入大量异常处理和延迟。# bot_core.py import time import schedule from datetime import datetime from sqlalchemy.orm import Session from config import Session as DBSession, CleanupLog from rules import RULES # 假设的微信控制库导入 # from wxauto import WeChat class WeChatCleanerBot: def __init__(self): # self.wx WeChat() # 实际初始化 self.db_session DBSession() self.target_chatrooms [测试群, 另一个群] # 指定要清理的群 def check_and_clean_messages(self): 检查并清理消息的主函数 print(f[{datetime.now()}] 开始扫描群聊消息...) for chatroom in self.target_chatrooms: try: # 模拟获取最近N条消息实际中需要根据库的API来获取 # messages self.wx.get_chatroom_messages(chatroom, count100) messages self._mock_get_messages(chatroom) # 模拟数据 for msg in messages: if self._should_clean(msg): self._perform_cleanup(chatroom, msg) time.sleep(1) # 重要每次操作后随机延迟模拟真人 except Exception as e: print(f处理群聊【{chatroom}】时出错: {e}) self._log_to_db(chatroom, None, None, fERROR: {e}, error) def _should_clean(self, msg): 判断单条消息是否应该被清理 for rule in RULES: if rule.match(msg[content], msg.get(send_time)): return True, rule.name return False, None def _perform_cleanup(self, chatroom, msg): 执行清理操作并记录日志 should_clean, rule_name self._should_clean(msg) if not should_clean: return # 实际执行删除操作这里用打印模拟 # success self.wx.delete_message(msg[id]) success True # 模拟成功 action deleted if success else failed print(f - 在【{chatroom}】中{action}一条消息。发送者{msg[sender]} 内容{msg[content][:50]}... 规则{rule_name}) # 记录日志到数据库 self._log_to_db(chatroom, msg[sender], msg[content], rule_name, action) def _log_to_db(self, chatroom, sender, content, rule, action): log CleanupLog( chatroom_namechatroom, message_sendersender, message_contentcontent, matched_rulerule, actionaction ) self.db_session.add(log) self.db_session.commit() def _mock_get_messages(self, chatroom): 模拟获取消息实际项目中替换为真实API调用 import random mock_msgs [ {sender:用户A, content:大家好扫码加群领福利, send_time: datetime.now()}, {sender:用户B, content:今晚开会。, send_time: datetime.now()}, {sender:广告哥, content:VX123456 低价出货, send_time: datetime.now()}, ] return mock_msgs def run(self): 启动定时任务 schedule.every(30).minutes.do(self.check_and_clean_messages) # 每30分钟运行一次 print(Cleaner Bot 已启动定时任务每30分钟执行一次。) while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次任务 if __name__ __main__: bot WeChatCleanerBot() bot.run()4.4 增加安全性与误杀恢复为了防止误删重要消息我们可以引入“白名单”和“确认机制”。白名单用户群主、管理员或特定重要成员的发言永远不清理。关键词白名单包含特定关键词如“会议”、“紧急”、“报销”的消息即使匹配广告规则也跳过。二次确认机制对于高置信度的广告如匹配多条规则直接删除对于低置信度的可以标记出来发送到另一个“待审核”频道或给自己发一条提醒由人工最终确认。恢复则依赖于之前提到的完整日志。如果发现误删可以第一时间查看日志找到被删除消息的准确内容、发送者和时间然后在群里进行解释或请求发送者重发。对于文件清理因为采用了“先移动后删除”的策略恢复就是简单的文件移动操作。5. 避坑指南与实战经验在开发和运行Cleaner Bot的过程中我踩过不少坑这里总结几条血泪教训频率与延迟是生命线无论是调用API还是模拟操作频率过高是导致账号被风控或封禁的最主要原因。务必为所有自动化操作添加随机延迟time.sleep(random.uniform(2, 5))并且将扫描/清理频率设置在平台能接受的合理范围如每小时一次而不是每分钟一次。规则测试务必充分上线任何新规则前一定要在测试环境或小范围进行至少一周的“观察模式”运行。即Bot只记录匹配到的消息和将要执行的操作而不实际删除。通过日志分析规则的准确率和误杀率反复调整优化。做好异常处理与状态监控网络会波动平台API会变更客户端会升级。你的Bot必须有完善的异常捕获和重试机制。同时最好实现一个简单的心跳或状态上报功能比如每天定时向你的邮箱发送一份运行报告让你知道它还在正常工作。法律与隐私红线清理他人消息或文件时必须明确权限。在社群中最好事先公告征得成员同意。清理公司内部资料必须有明确的授权和合规流程。绝对不要用Bot去做爬取隐私、恶意刷屏等违法违规的事情。代码与配置分离将群聊名单、关键词规则、延迟参数等所有可配置项都放在配置文件如config.yaml或数据库中。这样在调整时无需修改代码也方便进行版本管理。单一职责与模块化将规则匹配、平台操作、日志记录、任务调度等模块清晰地分开。这样当需要从清理微信切换到清理钉钉时你只需要替换“平台操作”这个模块其他部分可以复用。构建一个Cleaner Bot的过程更像是在制定一套精细的数字空间管理策略。技术实现只是手段真正的核心在于你对信息价值的判断和取舍。当你看到那些无关紧要的冗余信息被自动清走重要的内容得以凸显时那种数字世界焕然一新的清爽感就是对这个项目最好的回报。从最简单的定时清理临时文件开始逐步扩展它的能力你会发现自己对自动化工具的理解和应用水平也在不断提升。