#!/usr/bin/env python3 """ QQBot集成模块 用于发送Cookie过期告警到QQ """ import os import sys import json import time from datetime import datetime import subprocess class QQBotIntegration: """QQBot集成""" def __init__(self): self.channel = "qq" self.chat_id = "FAB5DE2C5FD0EE22CC0350B40094C646" self.nanobot_path = "/home/ubuntu/.local/bin/nanobot" def send_message_via_nanobot(self, message): """通过nanobot发送消息到QQ""" print(f"📤 通过nanobot发送消息到QQ...") print(f" 频道: {self.channel}") print(f" 聊天ID: {self.chat_id}") print(f" 消息长度: {len(message)} 字符") try: # 使用nanobot的message工具发送消息 # 由于nanobot的API限制，我们通过exec调用 cmd = [ self.nanobot_path, '--channel', self.channel, '--chat-id', self.chat_id, 'message', message ] print(f" 执行命令: {' '.join(cmd[:5])}...") result = subprocess.run( cmd, capture_output=True, text=True, timeout=30 ) print(f" 返回码: {result.returncode}") if result.returncode == 0: print("✅ 消息发送成功") return True else: print(f"❌ 消息发送失败") if result.stdout: print(f" 输出: {result.stdout[:200]}") if result.stderr: print(f" 错误: {result.stderr[:200]}") return False except subprocess.TimeoutExpired: print("⏰ 发送消息超时") return False except Exception as e: print(f"❌ 发送消息异常: {e}") return False def send_cookie_alert(self, alert_type, alert_message, cookie_info=None): """发送Cookie告警""" print(f"🔔 发送Cookie告警: {alert_type}") # 构建告警消息 message = self._format_cookie_alert_message(alert_type, alert_message, cookie_info) # 发送消息 success = self.send_message_via_nanobot(message) # 记录发送结果 self._log_alert_sent(alert_type, success, message) return success def _format_cookie_alert_message(self, alert_type, alert_message, cookie_info=None): """格式化Cookie告警消息""" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") # 根据告警类型设置emoji和标题 alert_config = { "COOKIE_EXPIRED": { "emoji": "🔴", "title": "【紧急】小红书Cookie已过期", "priority": "高" }, "COOKIE_EXPIRY_WARNING": { "emoji": "🟡", "title": "【警告】小红书Cookie即将过期", "priority": "中" }, "COOKIE_INVALID": { "emoji": "🔴", "title": "【错误】小红书Cookie无效", "priority": "高" }, "PUBLISH_FAILED": { "emoji": "🔴", "title": "【失败】小红书发布失败", "priority": "高" }, "COOKIE_HEALTH_POOR": { "emoji": "🟡", "title": "【注意】Cookie健康状况不佳", "priority": "中" }, "DAILY_REPORT": { "emoji": "📊", "title": "【日报】小红书Cookie监控报告", "priority": "低" } } config = alert_config.get(alert_type, { "emoji": "⚪", "title": "【通知】小红书系统消息", "priority": "低" }) # 构建消息 message = f"{config['emoji']} {config['title']}\n" message += f"⏰ 时间: {timestamp}\n" message += f"📊 优先级: {config['priority']}\n\n" message += f"📝 详情: {alert_message}\n\n" # 添加Cookie信息 if cookie_info: message += "🔑 Cookie信息:\n" for key, value in cookie_info.items(): if value: message += f" {key}: {value}\n" message += "\n" # 添加操作指南 message += "🔧 需要操作:\n" if alert_type in ["COOKIE_EXPIRED", "COOKIE_INVALID", "COOKIE_EXPIRY_WARNING"]: message += "1. 重新登录小红书获取新Cookie\n" message += "2. 使用EditThisCookie插件导出Cookie\n" message += "3. 运行命令更新Cookie:\n" message += " cd ~/.nanobot/workspace/skills/xiaohongshu_integration\n" message += " python3 cookie_manager.py import /path/to/cookies.txt\n" message += "4. 验证Cookie:\n" message += " python3 cookie_manager.py check\n" elif alert_type == "PUBLISH_FAILED": message += "1. 检查Cookie有效性:\n" message += " python3 cookie_manager.py check\n" message += "2. 如果Cookie无效，按上述步骤更新\n" message += "3. 手动测试发布:\n" message += " python3 xiaohongshu_publisher.py check\n" elif alert_type == "DAILY_REPORT": message += "1. 查看详细报告:\n" message += " python3 cookie_monitor.py report\n" message += "2. 查看最近告警:\n" message += " python3 cookie_monitor.py alerts\n" message += "\n📁 相关文件:\n" message += " Cookie文件: ~/.xiaohongshu-cli/cookies.json\n" message += " 监控数据库: ~/.nanobot/workspace/xiaohongshu_monitor.db\n" message += " 技能目录: ~/.nanobot/workspace/skills/xiaohongshu_integration/\n" # 添加快速命令 message += "\n⚡ 快速命令:\n" message += " 检查Cookie: python3 cookie_manager.py check\n" message += " 监控状态: python3 cookie_monitor.py monitor\n" message += " 测试发布: python3 xiaohongshu_publisher.py test\n" return message def _log_alert_sent(self, alert_type, success, message): """记录告警发送""" log_dir = "/home/ubuntu/.nanobot/workspace/qqbot_logs" os.makedirs(log_dir, exist_ok=True) log_file = os.path.join(log_dir, f"alerts_{datetime.now().strftime('%Y%m%d')}.log") timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") status = "✅" if success else "❌" log_entry = f"[{timestamp}] {status} {alert_type}\n" log_entry += f" 消息: {message[:100]}...\n" log_entry += "-" * 50 + "\n" with open(log_file, 'a', encoding='utf-8') as f: f.write(log_entry) print(f"📝 告警已记录到: {log_file}") def send_daily_report(self, report_content): """发送每日报告""" print("📊 发送每日报告到QQ...") message = self._format_cookie_alert_message("DAILY_REPORT", report_content) # 如果报告太长，分割发送 if len(message) > 1000: print("⚠️ 报告过长，进行分割...") parts = self._split_long_message(message, max_length=800) success_count = 0 for i, part in enumerate(parts, 1): print(f" 发送第 {i}/{len(parts)} 部分...") success = self.send_message_via_nanobot(part) if success: success_count += 1 time.sleep(1) # 避免发送过快 return success_count == len(parts) else: return self.send_message_via_nanobot(message) def _split_long_message(self, message, max_length=800): """分割长消息""" lines = message.split('\n') parts = [] current_part = [] current_length = 0 for line in lines: line_length = len(line) + 1 # +1 for newline if current_length + line_length > max_length and current_part: parts.append('\n'.join(current_part)) current_part = [line] current_length = line_length else: current_part.append(line) current_length += line_length if current_part: parts.append('\n'.join(current_part)) # 添加页码 for i in range(len(parts)): parts[i] = f"📄 第 {i+1}/{len(parts)} 部分\n\n" + parts[i] return parts def test_connection(self): """测试QQBot连接""" print("🧪 测试QQBot连接...") test_message = "🔧 测试消息: 小红书Cookie监控系统连接测试\n" test_message += f"⏰ 时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n" test_message += "✅ 如果收到此消息，说明QQBot连接正常" success = self.send_message_via_nanobot(test_message) if success: print("✅ QQBot连接测试成功") else: print("❌ QQBot连接测试失败") return success def main(): """主函数""" import argparse parser = argparse.ArgumentParser(description='QQBot集成工具') subparsers = parser.add_subparsers(dest='command', help='命令') # test命令 subparsers.add_parser('test', help='测试QQBot连接') # alert命令 alert_parser = subparsers.add_parser('alert', help='发送告警') alert_parser.add_argument('type', help='告警类型') alert_parser.add_argument('message', help='告警消息') # report命令 report_parser = subparsers.add_parser('report', help='发送报告') report_parser.add_argument('report_file', help='报告文件路径') args = parser.parse_args() qqbot = QQBotIntegration() if args.command == 'test': qqbot.test_connection() elif args.command == 'alert': qqbot.send_cookie_alert(args.type, args.message) elif args.command == 'report': if os.path.exists(args.report_file): with open(args.report_file, 'r', encoding='utf-8') as f: report_content = f.read() qqbot.send_daily_report(report_content) else: print(f"❌ 报告文件不存在: {args.report_file}") else: parser.print_help() if __name__ == "__main__": main()