import re from datetime import datetime, time, timedelta import toml import os class InterventionBot: def __init__(self, config_path=None): # 加载配置 self.config = {} if config_path and os.path.exists(config_path): self.config = toml.load(config_path) # 从配置中获取关键词和阈值 keywords = self.config.get("Keywords", {}) time_window = self.config.get("TimeWindow", {}) reply_threshold = self.config.get("ReplyThreshold", {}) # 表情符号库 self.emojis = keywords.get("emojis", []) # 话题关键词 self.hot_topics = keywords.get("hot_topics", []) self.fish_keywords = keywords.get("fish_keywords", []) self.tech_keywords = keywords.get("tech_keywords", []) self.mechanism_keywords = keywords.get("mechanism_keywords", []) self.news_keywords = keywords.get("news_keywords",[]) # 早晨签到时间窗口 morning_start_hour = time_window.get("morning_start_hour", 8) morning_start_minute = time_window.get("morning_start_minute", 0) morning_end_hour = time_window.get("morning_end_hour", 8) morning_end_minute = time_window.get("morning_end_minute", 30) self.morning_window = ( time(morning_start_hour, morning_start_minute), time(morning_end_hour, morning_end_minute) ) # 回复阈值配置 self.messages_per_minute_threshold = reply_threshold.get("messages_per_minute_threshold", 3) self.analysis_window_minutes = reply_threshold.get("analysis_window_minutes", 5) # 冷却时间配置（秒） self.cooldown_seconds = 20 self.last_intervention_time = None # 最近话题记录 self.last_topic = None self.last_topic_time = None self.topic_cooldown = timedelta(seconds=60) def is_morning_window(self, timestamp): try: if isinstance(timestamp, float): message_datetime = datetime.fromtimestamp(timestamp) message_time = message_datetime.time() elif isinstance(timestamp, str): try: message_time = datetime.strptime(timestamp, "%Y-%m-%d %H:%M:%S").time() except ValueError: try: message_time = datetime.fromtimestamp(float(timestamp)).time() except: return False else: return False return self.morning_window[0] <= message_time <= self.morning_window[1] except Exception as e: print(f"[早晨窗口检测] 错误: {e}") return False def detect_topic(self, message): if not isinstance(message, str): return None message_lower = message.lower() if any(keyword in message_lower for keyword in self.fish_keywords): return "fish" if any(keyword in message_lower for keyword in self.tech_keywords): return "tech" if any(keyword in message_lower for keyword in self.mechanism_keywords): return "mechanism" if any(keyword in message_lower for keyword in self.news_keywords): return "news" if any(keyword in message_lower for keyword in self.hot_topics): return "hot_topic" return None def rule_morning_signin(self, timestamp, messages): return self.is_morning_window(timestamp) and any("签到" in msg or "早" in msg for msg in messages[-5:]) def rule_hot_topic(self, message, messages): return self.detect_topic(message) == "hot_topic" and len( [m for m in messages[-5:] if self.detect_topic(m) == "hot_topic"]) >= 3 def rule_tech_discussion(self, message, messages): return self.detect_topic(message) == "tech" def rule_fish_discussion(self, message, messages): return self.detect_topic(message) == "fish" def rule_mechanism_interaction(self, message, messages): return self.detect_topic(message) == "mechanism" def rule_humor_tease(self, message, messages): return any(emoji in message for emoji in self.emojis) or "哈哈" in message or len( [m for m in messages[-5:] if any(e in m for e in self.emojis)]) >= 2 def rule_news_reaction(self, message, messages): return self.detect_topic(message) == "news" def rule_high_reply_rate(self, timestamp, chat_log): try: if isinstance(timestamp, float): current_time = datetime.fromtimestamp(timestamp) elif isinstance(timestamp, str): try: current_time = datetime.strptime(timestamp, "%Y-%m-%d %H:%M:%S") except ValueError: try: current_time = datetime.fromtimestamp(float(timestamp)) except: current_time = datetime.now() else: current_time = datetime.now() window_start = current_time - timedelta(minutes=self.analysis_window_minutes) recent_messages = [] for msg in chat_log: try: msg_timestamp = msg.get("timestamp") if isinstance(msg_timestamp, float): msg_time = datetime.fromtimestamp(msg_timestamp) elif isinstance(msg_timestamp, str): try: msg_time = datetime.strptime(msg_timestamp, "%Y-%m-%d %H:%M:%S") except ValueError: try: msg_time = datetime.fromtimestamp(float(msg_timestamp)) except: continue else: continue if window_start <= msg_time <= current_time: recent_messages.append(msg) except (ValueError, KeyError, TypeError): continue if len(recent_messages) < self.messages_per_minute_threshold: return False messages_per_minute = len(recent_messages) / self.analysis_window_minutes if messages_per_minute >= self.messages_per_minute_threshold: print(f"[高频率检测] 当前消息频率: {messages_per_minute:.2f}/分钟，阈值: {self.messages_per_minute_threshold}/分钟") return messages_per_minute >= self.messages_per_minute_threshold except Exception as e: print(f"[高频率检测] 错误: {e}") return False def should_intervene(self, timestamp, message, messages, chat_log): rules = [ self.rule_morning_signin, self.rule_hot_topic, self.rule_tech_discussion, self.rule_fish_discussion, self.rule_mechanism_interaction, self.rule_humor_tease, self.rule_news_reaction, self.rule_high_reply_rate ] current_time = datetime.now() if not isinstance(timestamp, datetime) else timestamp if isinstance(timestamp, str): try: current_time = datetime.strptime(timestamp, "%Y-%m-%d %H:%M:%S") except: current_time = datetime.now() elif isinstance(timestamp, float): current_time = datetime.fromtimestamp(timestamp) if self.last_intervention_time: if (current_time - self.last_intervention_time).total_seconds() < self.cooldown_seconds: return False current_topic = self.detect_topic(message) if self.last_topic == current_topic and self.last_topic_time: if (current_time - self.last_topic_time) < self.topic_cooldown: return False for rule in rules: if rule == self.rule_morning_signin: if rule(timestamp, messages): self.last_intervention_time = current_time self.last_topic = current_topic self.last_topic_time = current_time return True elif rule == self.rule_high_reply_rate: if rule(timestamp, chat_log): self.last_intervention_time = current_time self.last_topic = current_topic self.last_topic_time = current_time return True elif rule(message, messages): self.last_intervention_time = current_time self.last_topic = current_topic self.last_topic_time = current_time return True return False def process_message(self, timestamp, message, messages, chat_log): return self.should_intervene(timestamp, message, messages, chat_log) def process_chat_log(self, chat_log): messages = [line["message"] for line in chat_log] results = [] for i, line in enumerate(chat_log): timestamp = line["timestamp"] message = line["message"] intervention = self.process_message(timestamp, message, messages[:i + 1], chat_log) results.append({ "timestamp": timestamp, "message": message, "intervention": intervention }) return results if __name__ == "__main__": sample_chat_log = [ {"timestamp": "2025-03-14 08:06:38", "user_id": "Jyunere", "message": "白嫖马斯克，每个月150刀的额度，应该能玩很久了。"}, {"timestamp": "2025-03-14 08:06:54", "user_id": "Jyunere", "message": "啥情况？卷了？"}, {"timestamp": "2025-03-14 08:07:20", "user_id": "wxid_qx4z0jq3rp3122", "message": "那你喝咖啡就好了"}, {"timestamp": "2025-03-14 09:12:28", "user_id": "Jyunere", "message": "我同事的鸿蒙确实流畅。"}, {"timestamp": "2025-03-14 09:35:21", "user_id": "Jyunere", "message": "垃圾MIUI"}, {"timestamp": "2025-05-21 14:31:57", "user_id": "wxid_4re8ddo26dxb52", "message": "年轻人随随便便就能深蹲200"}, {"timestamp": "2025-05-21 14:32:20", "user_id": "liu79830956", "message": "@水牛 过分了啊，报错还扣积分 赔我200"}, {"timestamp": "2025-05-21 14:32:39", "user_id": "Jyunere", "message": "哈哈，识别到指令了。"}, {"timestamp": "2025-05-21 14:32:42", "user_id": "wxid_z8uo70zywfpn12", "message": "检测到天 气了"}, {"timestamp": "2025-05-21 14:35:08", "user_id": "liu79830956", "message": "这螺蛳粉估计要明天也吃不上了[旺柴]"} ] bot = InterventionBot() results = bot.process_chat_log(sample_chat_log) for result in results: print(f"[{result['timestamp']}] Message: {result['message']}") print(f"Intervention: {result['intervention']}") print("-" * 50)