#!/usr/bin/env python3 """ Cookie应急处理脚本 当Cookie失效时，提供详细的恢复指南 """ import os import sys import json from datetime import datetime import subprocess class CookieEmergencyHandler: """Cookie应急处理器""" def __init__(self): self.cookie_file = os.path.expanduser("~/.xiaohongshu-cli/cookies.json") self.qqbot = None # 尝试导入QQBot try: from qqbot_integration import QQBotIntegration self.qqbot = QQBotIntegration() except: print("⚠️ QQBot集成不可用") def diagnose_cookie_issue(self): """诊断Cookie问题""" print("🔍 诊断Cookie问题...") issues = [] # 1. 检查文件是否存在 if not os.path.exists(self.cookie_file): issues.append({ 'type': 'FILE_MISSING', 'severity': 'CRITICAL', 'message': 'Cookie文件不存在', 'solution': '需要重新获取Cookie' }) return issues # 2. 检查文件权限 if not os.access(self.cookie_file, os.R_OK): issues.append({ 'type': 'PERMISSION_DENIED', 'severity': 'HIGH', 'message': 'Cookie文件不可读', 'solution': '检查文件权限: chmod 644 ~/.xiaohongshu-cli/cookies.json' }) # 3. 检查文件内容 try: with open(self.cookie_file, 'r', encoding='utf-8') as f: cookies = json.load(f) if not isinstance(cookies, dict): issues.append({ 'type': 'INVALID_FORMAT', 'severity': 'HIGH', 'message': 'Cookie文件格式错误', 'solution': 'Cookie文件应该是JSON字典格式' }) elif len(cookies) == 0: issues.append({ 'type': 'EMPTY_COOKIE', 'severity': 'HIGH', 'message': 'Cookie文件为空', 'solution': '需要重新获取Cookie' }) else: print(f"✅ Cookie文件格式正确，包含 {len(cookies)} 个Cookie") # 检查关键Cookie key_cookies = ['a1', 'webId', 'web_session', 'x-user-id-creator.xiaohongshu.com'] missing_keys = [] for key in key_cookies: if key not in cookies: missing_keys.append(key) if missing_keys: issues.append({ 'type': 'MISSING_KEY_COOKIES', 'severity': 'HIGH', 'message': f'缺少关键Cookie: {", ".join(missing_keys)}', 'solution': '这些Cookie对于登录至关重要，需要重新获取' }) except json.JSONDecodeError as e: issues.append({ 'type': 'JSON_PARSE_ERROR', 'severity': 'HIGH', 'message': f'JSON解析错误: {str(e)}', 'solution': 'Cookie文件损坏，需要重新获取' }) except Exception as e: issues.append({ 'type': 'UNKNOWN_ERROR', 'severity': 'HIGH', 'message': f'读取Cookie文件时出错: {str(e)}', 'solution': '检查文件完整性' }) # 4. 测试Cookie有效性 try: result = subprocess.run( ['xhs', 'status', '--json'], capture_output=True, text=True, timeout=10 ) if result.returncode != 0: issues.append({ 'type': 'XHS_COMMAND_FAILED', 'severity': 'HIGH', 'message': f'xhs命令执行失败: {result.stderr[:100]}', 'solution': '检查xhs-cli安装和配置' }) else: data = json.loads(result.stdout) if not data.get('ok') or not data.get('data', {}).get('authenticated'): issues.append({ 'type': 'COOKIE_INVALID', 'severity': 'CRITICAL', 'message': 'Cookie无效或已过期', 'solution': '需要重新登录小红书获取新Cookie' }) else: print(f"✅ Cookie验证通过，用户: {data['data']['user'].get('nickname')}") except subprocess.TimeoutExpired: issues.append({ 'type': 'CHECK_TIMEOUT', 'severity': 'MEDIUM', 'message': 'Cookie检查超时', 'solution': '网络问题或xhs-cli响应慢' }) except Exception as e: issues.append({ 'type': 'CHECK_ERROR', 'severity': 'HIGH', 'message': f'Cookie检查出错: {str(e)}', 'solution': '检查系统配置' }) return issues def generate_recovery_guide(self, issues): """生成恢复指南""" print("\n📋 生成恢复指南...") guide = "🔧 小红书Cookie恢复指南\n" guide += "=" * 50 + "\n\n" guide += f"⏰ 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n" # 问题摘要 critical_count = sum(1 for issue in issues if issue['severity'] == 'CRITICAL') high_count = sum(1 for issue in issues if issue['severity'] == 'HIGH') guide += f"📊 问题摘要:\n" guide += f" 严重问题: {critical_count} 个\n" guide += f" 高级问题: {high_count} 个\n" guide += f" 总问题数: {len(issues)} 个\n\n" # 详细问题 guide += "🔍 详细问题:\n" for i, issue in enumerate(issues, 1): emoji = "🔴" if issue['severity'] in ['CRITICAL', 'HIGH'] else "🟡" guide += f"{emoji} 问题 {i}: {issue['type']}\n" guide += f" 严重性: {issue['severity']}\n" guide += f" 描述: {issue['message']}\n" guide += f" 解决方案: {issue['solution']}\n\n" # 恢复步骤 guide += "🚀 恢复步骤:\n" guide += "1. 重新登录小红书\n" guide += " - 打开小红书网页版 (https://www.xiaohongshu.com)\n" guide += " - 使用账号密码登录\n" guide += " - 确保登录状态正常\n\n" guide += "2. 导出Cookie\n" guide += " - 安装EditThisCookie浏览器插件\n" guide += " - 在小红书页面点击插件图标\n" guide += " - 点击'导出'按钮，保存为cookies.txt\n\n" guide += "3. 导入新Cookie\n" guide += " cd ~/.nanobot/workspace/skills/xiaohongshu_integration\n" guide += " python3 cookie_manager.py import /path/to/cookies.txt --type editthis\n\n" guide += "4. 验证Cookie\n" guide += " python3 cookie_manager.py check\n" guide += " python3 xiaohongshu_publisher.py check\n\n" guide += "5. 测试发布\n" guide += " python3 xiaohongshu_publisher.py test\n\n" # 快速命令 guide += "⚡ 快速命令:\n" guide += " # 诊断问题\n" guide += " python3 cookie_emergency_handler.py diagnose\n\n" guide += " # 导入Cookie\n" guide += " python3 cookie_manager.py import cookies.txt\n\n" guide += " # 检查状态\n" guide += " python3 cookie_manager.py check\n\n" # 文件位置 guide += "📁 文件位置:\n" guide += " Cookie文件: ~/.xiaohongshu-cli/cookies.json\n" guide += " 备份目录: ~/.xiaohongshu-cli/backups/\n" guide += " 技能目录: ~/.nanobot/workspace/skills/xiaohongshu_integration/\n" return guide def send_emergency_alert(self, issues): """发送紧急告警""" if not self.qqbot: print("❌ QQBot不可用，无法发送紧急告警") return False print("🔔 发送紧急告警到QQBot...") # 生成告警消息 critical_issues = [i for i in issues if i['severity'] in ['CRITICAL', 'HIGH']] if not critical_issues: print("✅ 无严重问题，无需发送紧急告警") return True alert_message = f"发现 {len(critical_issues)} 个严重Cookie问题，需要立即处理！" # 构建详细信息 cookie_info = { '问题数量': f'{len(critical_issues)} 个严重问题', '最严重问题': critical_issues[0]['type'], '生成时间': datetime.now().strftime('%H:%M:%S'), '状态': '需要立即处理' } # 发送告警 return self.qqbot.send_cookie_alert('COOKIE_EXPIRED', alert_message, cookie_info) def create_backup(self): """创建Cookie备份""" print("💾 创建Cookie备份...") if not os.path.exists(self.cookie_file): print("❌ Cookie文件不存在，无法备份") return False try: # 使用cookie_manager备份 from cookie_manager import XiaohongshuCookieManager manager = XiaohongshuCookieManager() backup_file = manager.backup_current_cookies() if backup_file: print(f"✅ Cookie备份成功: {backup_file}") return backup_file else: print("❌ Cookie备份失败") return False except Exception as e: print(f"❌ 备份失败: {e}") # 手动备份 backup_dir = os.path.expanduser("~/.xiaohongshu-cli/emergency_backup") os.makedirs(backup_dir, exist_ok=True) timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_file = os.path.join(backup_dir, f"cookies_emergency_{timestamp}.json") import shutil shutil.copy2(self.cookie_file, backup_file) print(f"✅ 手动备份成功: {backup_file}") return backup_file def emergency_recovery(self): """应急恢复流程""" print("🚨 执行应急恢复流程...") # 1. 诊断问题 issues = self.diagnose_cookie_issue() # 2. 创建备份 backup_file = self.create_backup() # 3. 生成恢复指南 guide = self.generate_recovery_guide(issues) # 4. 保存指南到文件 guide_file = f"/tmp/xiaohongshu_recovery_guide_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt" with open(guide_file, 'w', encoding='utf-8') as f: f.write(guide) print(f"✅ 恢复指南已保存: {guide_file}") # 5. 发送紧急告警 if issues: self.send_emergency_alert(issues) # 6. 显示指南 print("\n" + "=" * 70) print("应急恢复指南") print("=" * 70) print(guide[:500] + "...") print(f"\n完整指南: {guide_file}") return guide_file def main(): """主函数""" import argparse parser = argparse.ArgumentParser(description='Cookie应急处理器') subparsers = parser.add_subparsers(dest='command', help='命令') # diagnose命令 subparsers.add_parser('diagnose', help='诊断Cookie问题') # recover命令 subparsers.add_parser('recover', help='执行应急恢复') # backup命令 subparsers.add_parser('backup', help='创建Cookie备份') # guide命令 subparsers.add_parser('guide', help='生成恢复指南') args = parser.parse_args() handler = CookieEmergencyHandler() if args.command == 'diagnose': issues = handler.diagnose_cookie_issue() if issues: print(f"🔍 发现 {len(issues)} 个问题:") for issue in issues: print(f" {issue['severity']}: {issue['type']} - {issue['message']}") else: print("✅ 未发现问题，Cookie状态正常") elif args.command == 'recover': handler.emergency_recovery() elif args.command == 'backup': handler.create_backup() elif args.command == 'guide': # 模拟一些问题 test_issues = [ { 'type': 'COOKIE_EXPIRED', 'severity': 'CRITICAL', 'message': 'Cookie已过期', 'solution': '重新登录获取新Cookie' }, { 'type': 'MISSING_KEY_COOKIES', 'severity': 'HIGH', 'message': '缺少关键Cookie: a1, webId', 'solution': '需要完整的Cookie集合' } ] guide = handler.generate_recovery_guide(test_issues) print(guide) else: parser.print_help() if __name__ == "__main__": main()