diff --git a/browser_login.py b/browser_login.py index 49c3144..734d09d 100644 --- a/browser_login.py +++ b/browser_login.py @@ -129,6 +129,99 @@ class BrowserLogin: logger.error(f"检查登录状态失败: {e}") return False + def import_all_cookies(self, cookies: List[Dict]) -> bool: + """ + 一次性导入所有 cookies 到浏览器 + + Args: + cookies: cookie 列表 + + Returns: + 是否成功 + """ + if not self.browser or not self.tab: + self._create_browser() + + success_count = 0 + fail_count = 0 + + logger.info(f"开始导入 {len(cookies)} 个 cookie") + for cookie in cookies: + try: + cookie_dict = { + 'name': cookie.get('name', ''), + 'value': cookie.get('value', ''), + 'domain': cookie.get('domain', ''), + 'path': cookie.get('path', '/'), + } + if cookie.get('secure'): + cookie_dict['secure'] = True + if cookie.get('httpOnly'): + cookie_dict['httpOnly'] = True + if cookie.get('sameSite'): + cookie_dict['sameSite'] = cookie.get('sameSite') + + self.tab.set.cookies(cookie_dict) + success_count += 1 + except Exception as e: + fail_count += 1 + + logger.info(f"Cookie 导入完成: 成功 {success_count}, 失败 {fail_count}") + return fail_count == 0 + + def verify_login(self, url: str, check_selector: str = "", success_text: str = "") -> bool: + """ + 验证登录状态（cookies已预先导入） + + Args: + url: 网站地址 + check_selector: 登录检测选择器（可选） + success_text: 登录成功时显示的文本（可选） + + Returns: + 是否已登录 + """ + if not self.browser or not self.tab: + self._create_browser() + + try: + self.tab.get(url) + time.sleep(5) + + # 如果没有指定选择器，只检查页面是否成功加载 + if not check_selector: + logger.info(" 未指定登录检测选择器，跳过登录验证") + return True + + # 尝试查找元素 + try: + element = self.tab.ele(check_selector, timeout=10) + + if element: + element_text = element.text or "" + logger.info(f" 找到元素，文本内容: {element_text[:50]}...") + + if not success_text: + logger.info(" 未指定成功文本，找到元素即认为登录成功") + return True + + if success_text in element_text: + logger.info(f" 检测到成功文本: {success_text}") + return True + else: + logger.warning(f" 未检测到成功文本 '{success_text}'，元素文本: {element_text[:100]}") + return False + else: + logger.warning(f" 未找到元素: {check_selector}") + return False + except Exception as ele_error: + logger.warning(f" 查找元素失败: {ele_error}") + return False + + except Exception as e: + logger.error(f"验证登录状态失败: {e}") + return False + def login_with_cookies( self, url: str, @@ -152,20 +245,38 @@ class BrowserLogin: self._create_browser() try: - # 访问网站 + from urllib.parse import urlparse + parsed = urlparse(url) + base_url = f"{parsed.scheme}://{parsed.netloc}" + + self.tab.get(base_url) + time.sleep(1) + + logger.info(f"开始设置 {len(cookies)} 个 cookie") + for cookie in cookies: + try: + cookie_dict = { + 'name': cookie.get('name', ''), + 'value': cookie.get('value', ''), + 'domain': cookie.get('domain', parsed.netloc), + 'path': cookie.get('path', '/'), + } + if cookie.get('secure'): + cookie_dict['secure'] = True + if cookie.get('httpOnly'): + cookie_dict['httpOnly'] = True + if cookie.get('sameSite'): + cookie_dict['sameSite'] = cookie.get('sameSite') + + self.tab.set.cookies(cookie_dict) + logger.info(f"设置 cookie: {cookie_dict['name']}") + except Exception as e: + logger.error(f"设置 cookie 失败 {cookie.get('name')}: {e}") + self.tab.get(url) - # 添加 cookies - for cookie in cookies: - self.tab.set.cookies(cookie) - - # 刷新页面 - self.tab.refresh() - - # 等待页面加载 time.sleep(3) - # 检查登录状态 element = self.tab.ele(check_selector, timeout=10) if element and element.text and success_text in element.text: diff --git a/config.json b/config.json index 93e73c5..737c680 100644 --- a/config.json +++ b/config.json @@ -1 +1,52 @@ -{"users": [{"name": "用户A", "cookie_cloud": {"uuid": "hu6n2vcUmzpu7mqUN2rVCg", "password": "jtDM5dV9AyqXkZdQVeA9f6", "api_url": "https://movie-pilot.org/cookiecloud"}, "notification": {"iyuu_token": "IYUU37629Tc1d371c7ce99a49ff9778e196286b7e4592be191", "max_fail_count": 3}, "browser": {"type": "edge", "headless": false}, "websites": [{"name": "网站A", "url": "https://lmkbi.95155.com/bi-system/#/carUnpetrifiedComAll", "login_check_selector": "#user-info", "success_text": "彭峰"}]}]} \ No newline at end of file +{ + "users": [ + { + "name": "用户A", + "cookie_cloud": { + "uuid": "hu6n2vcUmzpu7mqUN2rVCg", + "password": "jtDM5dV9AyqXkZdQVeA9f6", + "api_url": "https://movie-pilot.org/cookiecloud" + }, + "notification": { + "iyuu_token": "IYUU37629Tc1d371c7ce99a49ff9778e196286b7e4592be191", + "max_fail_count": 3 + }, + "browser": { + "type": "edge", + "headless": false + }, + "websites": [ + { + "name": "网站A", + "url": "https://lmkbi.95155.com/bi-system/#/carUnpetrifiedComAll", + "login_check_selector": "", + "success_text": "" + } + ] + }, + { + "name": "用户B", + "cookie_cloud": { + "uuid": "qyZpTrgiP5mVwiRZwfBhjz", + "password": "iGn1B4FnWftp3oj4Ko3jxA", + "api_url": "http://192.168.1.100:3000/cookiecloud" + }, + "notification": { + "iyuu_token": "IYUU37629Tc1d371c7ce99a49ff9778e196286b7e4592be191", + "max_fail_count": 3 + }, + "browser": { + "type": "edge", + "headless": false + }, + "websites": [ + { + "name": "网站A", + "url": "https://lmkbi.95155.com/bi-system/#/carUnpetrifiedComAll", + "login_check_selector": "", + "success_text": "" + } + ] + } + ] +} \ No newline at end of file diff --git a/cookie_cloud.py b/cookie_cloud.py index 97f733a..21b15d1 100644 --- a/cookie_cloud.py +++ b/cookie_cloud.py @@ -1,11 +1,11 @@ -"""Cookie Cloud 模块 - 从 Cookie Cloud 服务器获取并解密 cookie 数据""" -import hashlib +"""Cookie Cloud 模块 - 从 Cookie Cloud 服务器获取 cookie 数据""" import json +import hashlib import base64 -from typing import Dict, List, Optional -import requests +from typing import Dict, List, Optional, Tuple +import urllib.request +import urllib.error from Crypto.Cipher import AES -from Crypto.Util.Padding import unpad from applogger import get_logger logger = get_logger(__name__) @@ -27,77 +27,82 @@ class CookieCloud: self.uuid = uuid self.password = password - def decrypt(self, encrypted_data: str) -> Dict: - """ - 解密 Cookie Cloud 数据 + def _get_crypt_key(self) -> bytes: + """生成加密密钥""" + combined_string = f"{self.uuid}-{self.password}" + return hashlib.md5(combined_string.encode('utf-8')).hexdigest()[:16].encode("utf-8") - Args: - encrypted_data: 加密的数据字符串 + def _bytes_to_key(self, data: bytes, salt: bytes, output: int = 48) -> bytes: + """OpenSSL EVP_BytesToKey 密钥派生算法""" + assert len(salt) == 8, len(salt) + data += salt + key = hashlib.md5(data).digest() + final_key = key + while len(final_key) < output: + key = hashlib.md5(key + data).digest() + final_key += key + return final_key[:output] - Returns: - 解密后的数据字典，包含 cookie_data 和 local_storage_data - """ - # 生成解密密钥: md5(uuid+password) 取前 16 个字符 - key_str = f"{self.uuid}{self.password}" - key = hashlib.md5(key_str.encode()).hexdigest()[:16].encode() - print(f"生成的密钥: {key}") - print(f"密钥长度: {len(key)}") - - # 解码 Base64 - encrypted = base64.b64decode(encrypted_data) - print(f"Base64 解码后长度: {len(encrypted)}") - print(f"解码后前 20 字节: {encrypted[:20]}") - - # 检查是否有盐值（OpenSSL 格式） - if encrypted.startswith(b'Salted__'): - print("检测到 OpenSSL 格式数据") - # 对于 OpenSSL 格式，我们需要跳过盐值部分 - # 但根据用户要求，我们直接使用16字符密钥，不考虑盐值 - ciphertext = encrypted[16:] # 跳过 "Salted__" 和盐值 - print(f"密文长度: {len(ciphertext)}") - else: - print("未检测到 OpenSSL 格式，使用整个数据作为密文") - ciphertext = encrypted - - # 使用固定的 IV (16 个 0 字节) - iv = b'\x00' * 16 - print(f"使用的 IV: {iv}") - - # AES-128-CBC 解密（16字节密钥） - cipher = AES.new(key, AES.MODE_CBC, iv) - decrypted = cipher.decrypt(ciphertext) - print(f"解密后长度: {len(decrypted)}") - print(f"解密后前 100 字节: {decrypted[:100]}") - + def _decrypt(self, encrypted: str, passphrase: bytes) -> bytes: + """解密数据""" + encrypted_bytes = base64.b64decode(encrypted) + assert encrypted_bytes.startswith(b"Salted__"), "Invalid encrypted data format" + salt = encrypted_bytes[8:16] + key_iv = self._bytes_to_key(passphrase, salt, 32 + 16) + key = key_iv[:32] + iv = key_iv[32:] + aes = AES.new(key, AES.MODE_CBC, iv) + decrypted_padded = aes.decrypt(encrypted_bytes[16:]) + padding_length = decrypted_padded[-1] + if isinstance(padding_length, str): + padding_length = ord(padding_length) + return decrypted_padded[:-padding_length] + + def _download_data(self) -> Tuple[Optional[Dict], str]: + """下载并解密所有数据""" try: - # 移除 PKCS#7 填充 - decrypted = unpad(decrypted, AES.block_size) - print(f"移除填充后长度: {len(decrypted)}") - print(f"移除填充后前 100 字节: {decrypted[:100]}") + url = f"{self.api_url}/get/{self.uuid}" + request = urllib.request.Request( + url, + headers={ + 'Content-Type': 'application/json', + 'User-Agent': 'CookieCloud-Client/1.0' + }, + method='GET' + ) - # 尝试解析 JSON - result = json.loads(decrypted.decode('utf-8')) - print("JSON 解析成功") - return result - except Exception as e: - print(f"解密失败: {e}") - # 尝试直接截取可能的有效数据 + response = urllib.request.urlopen(request, timeout=30) + + if response.status != 200: + return None, f"服务器返回错误状态码: {response.status}" + + result = json.loads(response.read().decode('utf-8')) + + if not result: + return None, "服务器返回数据为空" + + encrypted = result.get("encrypted") + if not encrypted: + return None, "未获取到cookie密文" + + crypt_key = self._get_crypt_key() try: - # 尝试找到 JSON 开始的位置 - for i in range(len(decrypted)): - try: - test_data = decrypted[i:] - test_str = test_data.decode('utf-8', errors='ignore') - if test_str.strip().startswith('{'): - print(f"找到可能的 JSON 开始位置: {i}") - result = json.loads(test_str) - print("JSON 解析成功（直接截取）") - return result - except: - continue - except: - pass - raise + decrypted_data = self._decrypt(encrypted, crypt_key) + result = json.loads(decrypted_data.decode("utf-8")) + except Exception as e: + return None, f"cookie解密失败: {str(e)}" + + if not result: + return None, "cookie解密为空" + + return result, "" + + except urllib.error.HTTPError as e: + return None, f"HTTP错误: {e.code} {e.reason}" + except urllib.error.URLError as e: + return None, f"网络连接失败: {e.reason}" + except Exception as e: + return None, f"下载失败: {str(e)}" def get_cookies(self) -> Dict[str, List[Dict]]: """ @@ -107,49 +112,34 @@ class CookieCloud: 按域名分组的 cookie 数据字典 格式: {"domain.com": [{"name": "cookie1", "value": "val1", ...}, ...], ...} """ - url = f"{self.api_url}/get/{self.uuid}" + data, error = self._download_data() - try: - response = requests.get(url, timeout=30) - response.raise_for_status() - data = response.json() - - logger.info(f"Cookie Cloud 响应: {json.dumps(data, indent=2)}") - - if not data or 'encrypted' not in data: - logger.error("响应中没有 encrypted 字段") - return {} - - # 打印完整的加密数据长度和前200个字符 - logger.info(f"加密数据长度: {len(data['encrypted'])}") - logger.info(f"加密数据前 200 个字符: {data['encrypted'][:200]}...") - - try: - decrypted_data = self.decrypt(data['encrypted']) - logger.info(f"解密后的数据: {json.dumps(decrypted_data, indent=2)}") - except Exception as e: - logger.error(f"解密失败: {e}") - return {} - - # 解析 cookie_data - cookie_dict = {} - if 'cookie_data' in decrypted_data: - for domain, cookies in decrypted_data['cookie_data'].items(): - cookie_dict[domain] = [] - for cookie in cookies: - # 处理 sameSite 字段 - if cookie.get('sameSite') == 'unspecified': - cookie['sameSite'] = 'Lax' - cookie_dict[domain].append(cookie) - - return cookie_dict - - except requests.RequestException as e: - logger.error(f"请求 Cookie Cloud 失败: {e}") - return {} - except Exception as e: - logger.error(f"处理 Cookie Cloud 数据失败: {e}") + if error: + logger.error(error) return {} + + # 处理数据结构 + cookie_data = {} + + # 兼容直接按域名分组的格式 + if isinstance(data, dict) and not data.get('cookie_data'): + cookie_data = data + # 兼容包含 cookie_data 的格式 + elif isinstance(data, dict) and data.get('cookie_data'): + cookie_data = data.get('cookie_data', {}) + + # 处理 sameSite 字段 + processed_cookies = {} + for domain, cookies in cookie_data.items(): + if not cookies: + continue + processed_cookies[domain] = [] + for cookie in cookies: + if cookie.get('sameSite') == 'unspecified': + cookie['sameSite'] = 'Lax' + processed_cookies[domain].append(cookie) + + return processed_cookies def get_cookies_for_domain(self, domain: str) -> List[Dict]: """ @@ -163,16 +153,16 @@ class CookieCloud: """ all_cookies = self.get_cookies() - # 查找匹配的域名 + matched_cookies = [] for cookie_domain, cookies in all_cookies.items(): - if domain in cookie_domain or cookie_domain in domain: - return cookies + clean_domain = cookie_domain.lstrip('.') + if domain in clean_domain or clean_domain in domain: + matched_cookies.extend(cookies) - return [] + return matched_cookies if __name__ == "__main__": - # 测试代码 cc = CookieCloud( api_url="https://movie-pilot.org/cookiecloud", uuid="hu6n2vcUmzpu7mqUN2rVCg", @@ -180,4 +170,7 @@ if __name__ == "__main__": ) cookies = cc.get_cookies() - logger.info(json.dumps(cookies, indent=2, ensure_ascii=False)) \ No newline at end of file + logger.info(f"获取到 {len(cookies)} 个域名的 cookies") + + lmkbi_cookies = cc.get_cookies_for_domain('lmkbi.95155.com') + logger.info(f"获取到 {len(lmkbi_cookies)} 个 lmkbi.95155.com 的 cookies") diff --git a/monitor.py b/monitor.py index 2591464..968252d 100644 --- a/monitor.py +++ b/monitor.py @@ -1,7 +1,5 @@ """Cookie 监控主程序""" import json -import os -import logging from datetime import datetime from cookie_cloud import CookieCloud from notifier import IYUUNotifier, FailureTracker @@ -15,237 +13,200 @@ 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}") + logger.error(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 + return urlparse(url).netloc def process_user(self, user_config: dict): """ - 处理单个用户的所有网站 - - Args: - user_config: 用户配置 + 处理单个用户：一次性导入所有cookies，然后验证各网站登录状态 """ user_name = user_config.get('name', '未知用户') - self._log(f"开始处理用户: {user_name}") + logger.info(f"{'='*20} 开始处理用户: {user_name} {'='*20}") - # 获取用户配置 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', []) - # 初始化爱语飞飞通知 + if not websites: + logger.info(f"用户 {user_name} 没有配置网站，跳过") + return + 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', '') ) - # 初始化浏览器 + logger.info(f"步骤1: 从 Cookie Cloud 获取所有 cookies") + all_cookies = cookie_cloud.get_cookies() + total_domains = len(all_cookies) + total_cookies = sum(len(c) for c in all_cookies.values()) + logger.info(f"获取到 {total_domains} 个域名，共 {total_cookies} 个 cookie") + browser = BrowserLogin( browser_type=browser_config.get('type', 'edge'), headless=browser_config.get('headless', False) ) + results = [] + try: - # 处理每个网站 + logger.info(f"步骤2: 一次性导入所有 cookies 到浏览器") + all_cookies_list = [] + for cookies in all_cookies.values(): + all_cookies_list.extend(cookies) + + import_success = browser.import_all_cookies(all_cookies_list) + if import_success: + logger.info(f"成功导入 {len(all_cookies_list)} 个 cookie") + else: + logger.warning(f"部分 cookie 导入失败") + + logger.info(f"步骤3: 依次验证各网站登录状态") for website in websites: - self.process_website( + result = 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() + results.append(result) - self._log(f"用户 {user_name} 处理完成") + if not result['success']: + self.failure_tracker.check_and_notify( + user_name=user_name, + website_name=result['name'], + max_fail_count=max_fail_count, + notifier=notifier, + error_msg=result['error'] + ) + else: + self.failure_tracker.reset_fail_count(user_name, result['name']) + + self._print_summary(user_name, results) + + finally: + browser.close() def process_website( self, user_name: str, website_config: dict, - cookie_cloud: CookieCloud, - notifier: IYUUNotifier, - max_fail_count: int, browser: BrowserLogin - ): + ) -> dict: """ - 处理单个网站 - - Args: - user_name: 用户名 - website_config: 网站配置 - cookie_cloud: Cookie Cloud 客户端 - notifier: 爱语飞飞通知客户端 - max_fail_count: 最大失败次数 - browser: 浏览器客户端 + 验证网站登录状态（cookies已预先导入） + + Returns: + dict: {'name': 网站名, 'url': URL, 'success': 是否成功, 'error': 错误信息} """ 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}") + logger.info(f"\n验证网站: {website_name}") + logger.info(f" URL: {url}") - # 获取域名 - domain = self._get_domain_from_url(url) + result = { + 'name': website_name, + 'url': url, + 'success': False, + 'error': '' + } + + if not url: + result['error'] = 'URL 未配置' + logger.error(f" 失败: {result['error']}") + return result 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 + login_success = browser.verify_login( + url=url, + check_selector=check_selector, + success_text=success_text ) - if should_stop: - self._log(f"已达到最大失败次数，已发送通知并重置计数: {website_name}") + if login_success: + result['success'] = True + logger.info(f" ✓ 登录验证成功") + + new_cookies = browser.refresh_and_save_cookies(url) + if new_cookies: + self.cookie_manager.save_cookies(user_name, website_name, new_cookies) + logger.info(f" 已保存 {len(new_cookies)} 个新 cookie") else: - fail_count = self.failure_tracker.get_fail_count(user_name, website_name) - self._log(f"当前失败次数: {fail_count}/{max_fail_count}") + result['error'] = '登录状态验证失败（cookies可能已过期或选择器配置错误）' + logger.warning(f" ✗ {result['error']}") 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 - ) + result['error'] = f'处理异常: {str(e)}' + logger.error(f" 异常: {e}") + + return result + + def _print_summary(self, user_name: str, results: list): + """打印处理结果汇总""" + success_count = sum(1 for r in results if r['success']) + fail_count = len(results) - success_count + + logger.info(f"\n{'='*20} 用户 {user_name} 处理结果汇总 {'='*20}") + logger.info(f"总网站数: {len(results)}") + logger.info(f"成功: {success_count}, 失败: {fail_count}") + + if fail_count > 0: + logger.info("\n失败详情:") + for r in results: + if not r['success']: + logger.info(f" - {r['name']}: {r['error']}") def run(self): """运行监控""" - self._log("===== Cookie 监控开始 =====") + logger.info("="*50) + logger.info("Cookie 监控开始") + logger.info(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") + logger.info("="*50) users = self.config.get('users', []) if not users: - self._log("未找到用户配置") + logger.error("未找到用户配置") 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 + logger.error(f"处理用户失败: {e}") - self._log(f"===== Cookie 监控结束 =====") - self._log(f"成功处理: {success_count} 个用户，失败: {fail_count} 个用户") + logger.info("\n" + "="*50) + logger.info("Cookie 监控结束") + logger.info("="*50) def main(): - """主函数""" - # 配置日志 setup_logging() - monitor = CookieMonitor() monitor.run() if __name__ == "__main__": - main() \ No newline at end of file + main()