#!/usr/bin/env python3 """ 小红书Cookie监控器 监控Cookie有效期，通过QQBot发送告警 """ import os import sys import json import time from datetime import datetime, timedelta import subprocess import sqlite3 class XiaohongshuCookieMonitor: """小红书Cookie监控器""" def __init__(self): self.cookie_file = os.path.expanduser("~/.xiaohongshu-cli/cookies.json") self.monitor_db = "/home/ubuntu/.nanobot/workspace/xiaohongshu_monitor.db" self.qqbot_channel = "qq" self.qqbot_chat_id = "FAB5DE2C5FD0EE22CC0350B40094C646" # 初始化数据库 self._init_database() def _init_database(self): """初始化监控数据库""" conn = sqlite3.connect(self.monitor_db) cursor = conn.cursor() # 创建监控记录表 cursor.execute(''' CREATE TABLE IF NOT EXISTS cookie_monitor ( id INTEGER PRIMARY KEY AUTOINCREMENT, check_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, cookie_valid BOOLEAN, user_id TEXT, username TEXT, nickname TEXT, cookie_count INTEGER, error_message TEXT, notified BOOLEAN DEFAULT 0, notification_time TIMESTAMP ) ''') # 创建告警记录表 cursor.execute(''' CREATE TABLE IF NOT EXISTS alerts ( id INTEGER PRIMARY KEY AUTOINCREMENT, alert_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP, alert_type TEXT, alert_message TEXT, cookie_status TEXT, sent_to_qqbot BOOLEAN DEFAULT 0, qqbot_response TEXT ) ''') conn.commit() conn.close() def check_cookie_validity(self): """检查Cookie有效性""" print(f"🔍 检查Cookie有效性: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") if not os.path.exists(self.cookie_file): self._record_check_result(False, error_message="Cookie文件不存在") return False try: # 使用xhs命令检查登录状态 result = subprocess.run( ['xhs', 'status', '--json'], capture_output=True, text=True, timeout=10 ) if result.returncode == 0: data = json.loads(result.stdout) if data.get('ok') and data.get('data', {}).get('authenticated'): user = data['data']['user'] # 读取Cookie文件获取详细信息 with open(self.cookie_file, 'r', encoding='utf-8') as f: cookies = json.load(f) self._record_check_result( True, user_id=user.get('id'), username=user.get('username'), nickname=user.get('nickname'), cookie_count=len(cookies) ) print(f"✅ Cookie有效") print(f" 用户ID: {user.get('id')}") print(f" 用户名: {user.get('username')}") print(f" 昵称: {user.get('nickname')}") print(f" Cookie数量: {len(cookies)}") return True else: error_msg = f"Cookie无效: {data}" self._record_check_result(False, error_message=error_msg) print(f"❌ {error_msg}") return False else: error_msg = f"检查失败: {result.stderr[:100]}" self._record_check_result(False, error_message=error_msg) print(f"❌ {error_msg}") return False except subprocess.TimeoutExpired: error_msg = "检查超时" self._record_check_result(False, error_message=error_msg) print(f"⏰ {error_msg}") return False except Exception as e: error_msg = f"检查异常: {str(e)}" self._record_check_result(False, error_message=error_msg) print(f"❌ {error_msg}") return False def _record_check_result(self, valid, **kwargs): """记录检查结果到数据库""" conn = sqlite3.connect(self.monitor_db) cursor = conn.cursor() cursor.execute(''' INSERT INTO cookie_monitor (check_time, cookie_valid, user_id, username, nickname, cookie_count, error_message) VALUES (?, ?, ?, ?, ?, ?, ?) ''', ( datetime.now().isoformat(), valid, kwargs.get('user_id'), kwargs.get('username'), kwargs.get('nickname'), kwargs.get('cookie_count'), kwargs.get('error_message') )) conn.commit() conn.close() def analyze_cookie_health(self): """分析Cookie健康状况""" print("\n📊 分析Cookie健康状况...") conn = sqlite3.connect(self.monitor_db) cursor = conn.cursor() # 获取最近24小时的检查记录 cursor.execute(''' SELECT COUNT(*) as total_checks, SUM(CASE WHEN cookie_valid THEN 1 ELSE 0 END) as valid_checks, MIN(check_time) as first_check, MAX(check_time) as last_check FROM cookie_monitor WHERE check_time >= datetime('now', '-24 hours') ''') stats = cursor.fetchone() if stats and stats[0] > 0: total_checks, valid_checks, first_check, last_check = stats success_rate = (valid_checks / total_checks) * 100 if total_checks > 0 else 0 print(f" 最近24小时检查: {total_checks} 次") print(f" 有效检查: {valid_checks} 次") print(f" 成功率: {success_rate:.1f}%") print(f" 首次检查: {first_check}") print(f" 最后检查: {last_check}") # 检查是否需要告警 if success_rate < 50: alert_msg = f"Cookie健康状况不佳！最近24小时成功率仅{success_rate:.1f}%" self.send_alert("COOKIE_HEALTH_POOR", alert_msg) return "POOR" elif success_rate < 80: alert_msg = f"Cookie健康状况一般，成功率{success_rate:.1f}%" self.send_alert("COOKIE_HEALTH_WARNING", alert_msg) return "WARNING" else: return "GOOD" else: print(" 无最近24小时检查记录") return "UNKNOWN" conn.close() def check_cookie_expiry_warning(self): """检查Cookie过期预警""" print("\n⚠️ 检查Cookie过期预警...") if not os.path.exists(self.cookie_file): return "NO_COOKIE_FILE" try: # 检查Cookie文件修改时间 mtime = os.path.getmtime(self.cookie_file) cookie_age_days = (time.time() - mtime) / (24 * 3600) print(f" Cookie文件年龄: {cookie_age_days:.1f} 天") # 预警阈值 if cookie_age_days > 180: # 6个月 alert_msg = f"Cookie文件已存在{cookie_age_days:.0f}天，可能即将过期！" self.send_alert("COOKIE_EXPIRY_WARNING", alert_msg) return "EXPIRY_WARNING" elif cookie_age_days > 300: # 10个月 alert_msg = f"Cookie文件已存在{cookie_age_days:.0f}天，很可能已过期！" self.send_alert("COOKIE_EXPIRY_CRITICAL", alert_msg) return "EXPIRY_CRITICAL" else: return "OK" except Exception as e: print(f"❌ 检查Cookie年龄失败: {e}") return "ERROR" def send_alert(self, alert_type, message): """发送告警到QQBot""" print(f"🔔 发送告警: {alert_type}") print(f" 消息: {message}") # 记录告警到数据库 conn = sqlite3.connect(self.monitor_db) cursor = conn.cursor() cursor.execute(''' INSERT INTO alerts (alert_time, alert_type, alert_message, cookie_status) VALUES (?, ?, ?, ?) ''', ( datetime.now().isoformat(), alert_type, message, "需要检查" )) alert_id = cursor.lastrowid # 发送到QQBot try: # 构建QQBot消息 qqbot_message = self._format_qqbot_message(alert_type, message) # 这里需要调用QQBot的API # 由于QQBot集成方式未知，我们先记录到数据库 # 实际发送需要根据QQBot的API实现 cursor.execute(''' UPDATE alerts SET sent_to_qqbot = 1, qqbot_response = '已记录，需要手动发送' WHERE id = ? ''', (alert_id,)) conn.commit() print(f"✅ 告警已记录到数据库 (ID: {alert_id})") print(f" QQBot消息: {qqbot_message}") # 返回消息供手动发送 return qqbot_message except Exception as e: print(f"❌ 发送告警失败: {e}") cursor.execute(''' UPDATE alerts SET qqbot_response = ? WHERE id = ? ''', (f"发送失败: {str(e)}", alert_id)) conn.commit() return None finally: conn.close() def _format_qqbot_message(self, alert_type, message): """格式化QQBot消息""" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") if alert_type == "COOKIE_EXPIRY_CRITICAL": emoji = "🔴" title = "【紧急】小红书Cookie已过期" elif alert_type == "COOKIE_EXPIRY_WARNING": emoji = "🟡" title = "【警告】小红书Cookie即将过期" elif alert_type == "COOKIE_HEALTH_POOR": emoji = "🔴" title = "【异常】小红书Cookie健康状况不佳" elif alert_type == "COOKIE_HEALTH_WARNING": emoji = "🟡" title = "【注意】小红书Cookie健康状况一般" else: emoji = "⚪" title = "【通知】小红书Cookie状态更新" qqbot_message = f"{emoji} {title}\n" qqbot_message += f"⏰ 时间: {timestamp}\n" qqbot_message += f"📝 详情: {message}\n\n" qqbot_message += "🔧 需要操作:\n" qqbot_message += "1. 重新登录小红书获取新Cookie\n" qqbot_message += "2. 使用EditThisCookie导出Cookie\n" qqbot_message += "3. 运行: python3 cookie_manager.py import cookies.txt\n" qqbot_message += "4. 测试: python3 cookie_manager.py check\n\n" qqbot_message += "📁 文件位置: ~/.xiaohongshu-cli/cookies.json" return qqbot_message def get_recent_alerts(self, hours=24): """获取最近告警""" conn = sqlite3.connect(self.monitor_db) cursor = conn.cursor() cursor.execute(''' SELECT alert_time, alert_type, alert_message, sent_to_qqbot FROM alerts WHERE alert_time >= datetime('now', ?) ORDER BY alert_time DESC ''', (f'-{hours} hours',)) alerts = cursor.fetchall() conn.close() return alerts def generate_daily_report(self): """生成每日报告""" print("\n📋 生成每日报告...") conn = sqlite3.connect(self.monitor_db) cursor = conn.cursor() # 今日统计 today = datetime.now().strftime("%Y-%m-%d") cursor.execute(''' SELECT COUNT(*) as total_checks, SUM(CASE WHEN cookie_valid THEN 1 ELSE 0 END) as valid_checks FROM cookie_monitor WHERE date(check_time) = ? ''', (today,)) today_stats = cursor.fetchone() # 告警统计 cursor.execute(''' SELECT COUNT(*) as total_alerts FROM alerts WHERE date(alert_time) = ? ''', (today,)) today_alerts = cursor.fetchone() # 生成报告 report = f"📊 小红书Cookie监控日报\n" report += f"📅 日期: {today}\n" report += f"⏰ 生成时间: {datetime.now().strftime('%H:%M:%S')}\n\n" if today_stats and today_stats[0] > 0: total_checks, valid_checks = today_stats success_rate = (valid_checks / total_checks) * 100 report += f"✅ 今日检查: {total_checks} 次\n" report += f"✅ 有效检查: {valid_checks} 次\n" report += f"📈 成功率: {success_rate:.1f}%\n" else: report += "ℹ️ 今日无检查记录\n" if today_alerts and today_alerts[0] > 0: report += f"⚠️ 今日告警: {today_alerts[0]} 条\n" else: report += "✅ 今日无告警\n" # Cookie文件信息 if os.path.exists(self.cookie_file): mtime = os.path.getmtime(self.cookie_file) cookie_age_days = (time.time() - mtime) / (24 * 3600) report += f"\n📁 Cookie文件信息:\n" report += f" 位置: {self.cookie_file}\n" report += f" 年龄: {cookie_age_days:.1f} 天\n" with open(self.cookie_file, 'r', encoding='utf-8') as f: cookies = json.load(f) report += f" Cookie数量: {len(cookies)} 个\n" else: report += "\n❌ Cookie文件不存在\n" report += "\n🔧 建议:\n" if today_stats and today_stats[0] > 0 and (valid_checks / total_checks) < 0.8: report += " ⚠️ Cookie成功率较低，建议检查Cookie有效性\n" else: report += " ✅ Cookie状态良好，无需操作\n" conn.close() print(report) return report def monitor_and_alert(self): """监控并发送告警""" print("=" * 70) print("小红书Cookie监控系统") print("=" * 70) # 1. 检查Cookie有效性 is_valid = self.check_cookie_validity() # 2. 分析健康状况 health_status = self.analyze_cookie_health() # 3. 检查过期预警 expiry_status = self.check_cookie_expiry_warning() # 4. 生成报告 report = self.generate_daily_report() # 5. 如果需要，发送告警 if not is_valid or health_status in ["POOR", "WARNING"] or expiry_status in ["EXPIRY_WARNING", "EXPIRY_CRITICAL"]: alert_message = f"Cookie状态异常！有效性: {is_valid}, 健康: {health_status}, 过期: {expiry_status}" qqbot_message = self.send_alert("COOKIE_STATUS_ALERT", alert_message) print("\n🔔 已生成QQBot告警消息:") print("-" * 50) print(qqbot_message) print("-" * 50) print("\n💡 请手动发送此消息到QQBot") print("\n" + "=" * 70) print("监控完成") print("=" * 70) return is_valid def main(): """主函数""" import argparse parser = argparse.ArgumentParser(description='小红书Cookie监控器') subparsers = parser.add_subparsers(dest='command', help='命令') # monitor命令 subparsers.add_parser('monitor', help='执行监控并发送告警') # check命令 subparsers.add_parser('check', help='检查Cookie有效性') # report命令 subparsers.add_parser('report', help='生成每日报告') # alerts命令 alerts_parser = subparsers.add_parser('alerts', help='查看告警') alerts_parser.add_argument('--hours', type=int, default=24, help='查看最近多少小时的告警') # test-alert命令 subparsers.add_parser('test-alert', help='测试告警功能') args = parser.parse_args() monitor = XiaohongshuCookieMonitor() if args.command == 'monitor': monitor.monitor_and_alert() elif args.command == 'check': monitor.check_cookie_validity() elif args.command == 'report': monitor.generate_daily_report() elif args.command == 'alerts': alerts = monitor.get_recent_alerts(args.hours) if alerts: print(f"📋 最近{args.hours}小时告警:") for alert in alerts: alert_time, alert_type, alert_message, sent = alert sent_status = "✅" if sent else "❌" print(f" {alert_time} [{alert_type}] {sent_status}") print(f" {alert_message[:100]}...") else: print(f"✅ 最近{args.hours}小时无告警") elif args.command == 'test-alert': test_message = "测试告警: Cookie监控系统工作正常" qqbot_message = monitor.send_alert("TEST_ALERT", test_message) print("测试告警已发送:") print(qqbot_message) else: parser.print_help() if __name__ == "__main__": main()