"""Cookie 监控主程序""" import json import os import logging from datetime import datetime from cookie_cloud import CookieCloud from notifier import IYUUNotifier, FailureTracker from browser_login import BrowserLogin, CookieManager from applogger import setup_logging, get_logger logger = get_logger(__name__) class CookieMonitor: """Cookie 监控器""" def __init__(self, config_file: str = "config.json"): """ 初始化 Cookie 监控器 Args: config_file: 配置文件路径 """ self.config_file = config_file self.config = self._load_config() self.cookie_manager = CookieManager("cookies.json") self.failure_tracker = FailureTracker("state.json") def _load_config(self) -> dict: """加载配置文件""" try: with open(self.config_file, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: print(f"加载配置文件失败: {e}") return {} def _log(self, message: str): """记录日志""" logger.info(message) def _get_domain_from_url(self, url: str) -> str: """从 URL 中提取域名""" from urllib.parse import urlparse parsed = urlparse(url) return parsed.netloc def process_user(self, user_config: dict): """ 处理单个用户的所有网站 Args: user_config: 用户配置 """ user_name = user_config.get('name', '未知用户') self._log(f"开始处理用户: {user_name}") # 获取用户配置 cookie_cloud_config = user_config.get('cookie_cloud', {}) notification_config = user_config.get('notification', {}) browser_config = user_config.get('browser', {}) websites = user_config.get('websites', []) # 初始化爱语飞飞通知 iyuu_token = notification_config.get('iyuu_token', '') max_fail_count = notification_config.get('max_fail_count', 3) notifier = IYUUNotifier(iyuu_token) if iyuu_token else None # 初始化 Cookie Cloud cookie_cloud = CookieCloud( api_url=cookie_cloud_config.get('api_url', ''), uuid=cookie_cloud_config.get('uuid', ''), password=cookie_cloud_config.get('password', '') ) # 初始化浏览器 browser = BrowserLogin( browser_type=browser_config.get('type', 'edge'), headless=browser_config.get('headless', False) ) try: # 处理每个网站 for website in websites: self.process_website( user_name=user_name, website_config=website, cookie_cloud=cookie_cloud, notifier=notifier, max_fail_count=max_fail_count, browser=browser ) finally: # 关闭浏览器 browser.close() self._log(f"用户 {user_name} 处理完成") def process_website( self, user_name: str, website_config: dict, cookie_cloud: CookieCloud, notifier: IYUUNotifier, max_fail_count: int, browser: BrowserLogin ): """ 处理单个网站 Args: user_name: 用户名 website_config: 网站配置 cookie_cloud: Cookie Cloud 客户端 notifier: 爱语飞飞通知客户端 max_fail_count: 最大失败次数 browser: 浏览器客户端 """ website_name = website_config.get('name', '未知网站') url = website_config.get('url', '') check_selector = website_config.get('login_check_selector', '') success_text = website_config.get('success_text', '') self._log(f"处理网站: {user_name} - {website_name}") # 获取域名 domain = self._get_domain_from_url(url) try: # 第一步：使用本地 cookie 登录 self._log(f"尝试使用本地 cookie 登录: {website_name}") local_cookies = self.cookie_manager.get_cookies(user_name, website_name) if local_cookies: login_success = browser.login_with_cookies( url=url, cookies=local_cookies, check_selector=check_selector, success_text=success_text ) if login_success: self._log(f"本地 cookie 登录成功: {website_name}") # 刷新页面并保存新的 cookie new_cookies = browser.refresh_and_save_cookies(url) if new_cookies: self.cookie_manager.save_cookies(user_name, website_name, new_cookies) self._log(f"已更新 cookie: {website_name}") # 重置失败计数 self.failure_tracker.reset_fail_count(user_name, website_name) return self._log(f"本地 cookie 登录失败: {website_name}") # 第二步：从 Cookie Cloud 获取 cookie 并重试 self._log(f"从 Cookie Cloud 获取 cookie: {website_name}") cloud_cookies = cookie_cloud.get_cookies_for_domain(domain) if cloud_cookies: login_success = browser.login_with_cookies( url=url, cookies=cloud_cookies, check_selector=check_selector, success_text=success_text ) if login_success: self._log(f"Cookie Cloud 登录成功: {website_name}") # 刷新页面并保存新的 cookie new_cookies = browser.refresh_and_save_cookies(url) if new_cookies: self.cookie_manager.save_cookies(user_name, website_name, new_cookies) self._log(f"已更新 cookie: {website_name}") # 重置失败计数 self.failure_tracker.reset_fail_count(user_name, website_name) return self._log(f"Cookie Cloud 登录失败: {website_name}") else: self._log(f"未从 Cookie Cloud 获取到 cookie: {website_name}") # 第三步：登录失败，增加失败计数并检查是否需要通知 self._log(f"登录失败: {website_name}") error_msg = f"{website_name} 登录失败，请检查 Cookie Cloud 是否有最新的 cookie" should_stop = self.failure_tracker.check_and_notify( user_name=user_name, website_name=website_name, max_fail_count=max_fail_count, notifier=notifier, error_msg=error_msg ) if should_stop: self._log(f"已达到最大失败次数，已发送通知并重置计数: {website_name}") else: fail_count = self.failure_tracker.get_fail_count(user_name, website_name) self._log(f"当前失败次数: {fail_count}/{max_fail_count}") except Exception as e: self._log(f"处理网站 {website_name} 时发生异常: {e}") error_msg = f"{website_name} 处理异常: {str(e)}" should_stop = self.failure_tracker.check_and_notify( user_name=user_name, website_name=website_name, max_fail_count=max_fail_count, notifier=notifier, error_msg=error_msg ) def run(self): """运行监控""" self._log("===== Cookie 监控开始 =====") users = self.config.get('users', []) if not users: self._log("未找到用户配置") return success_count = 0 fail_count = 0 for user_config in users: try: self.process_user(user_config) success_count += 1 except Exception as e: self._log(f"处理用户失败: {e}") fail_count += 1 self._log(f"===== Cookie 监控结束 =====") self._log(f"成功处理: {success_count} 个用户，失败: {fail_count} 个用户") def main(): """主函数""" # 配置日志 setup_logging() monitor = CookieMonitor() monitor.run() if __name__ == "__main__": main()