diff --git a/browser_login.py b/browser_login.py deleted file mode 100644 index 8af7074..0000000 --- a/browser_login.py +++ /dev/null @@ -1,548 +0,0 @@ -"""浏览器登录模块 - 使用 DrissionPage 控制浏览器进行登录检测和 cookie 管理""" -from typing import Dict, List, Optional -import json -import time -from DrissionPage import Chromium, ChromiumOptions -from applogger import get_logger - -logger = get_logger(__name__) - - -class BrowserLogin: - """浏览器登录客户端""" - - def __init__(self, browser_type: str = "edge", headless: bool = False): - """ - 初始化浏览器登录客户端 - - Args: - browser_type: 浏览器类型 (chrome 或 edge) - headless: 是否无头模式 - """ - self.browser_type = browser_type.lower() - self.headless = headless - self.browser = None - self.tab = None - - def _create_browser(self): - """创建浏览器实例""" - co = ChromiumOptions() - - # 设置浏览器路径 - if self.browser_type == "edge": - co.set_browser_path("C:\\Program Files (x86)\\Microsoft\\Edge\\Application\\msedge.exe") - elif self.browser_type == "chrome": - co.set_browser_path("chrome") - - # 设置无头模式 - if self.headless: - co.headless = True - - # 设置其他选项 - co.set_argument('--no-sandbox') - co.set_argument('--disable-dev-shm-usage') - - # 创建浏览器 - self.browser = Chromium(co) - self.tab = self.browser.new_tab() - - def _close_browser(self): - """关闭浏览器""" - if self.browser: - try: - self.browser.quit() - except Exception as e: - logger.error(f"关闭浏览器失败: {e}") - finally: - self.browser = None - self.tab = None - - def _get_domain_from_url(self, url: str) -> str: - """ - 从 URL 中提取域名 - - Args: - url: 网站地址 - - Returns: - 域名 - """ - from urllib.parse import urlparse - parsed = urlparse(url) - return parsed.netloc - - def get_all_cookies(self) -> List[Dict]: - """ - 获取当前浏览器实例的所有 cookie - - Returns: - 所有 cookie 的列表 - """ - # 先获取 cookies 对象，然后调用 as_dict 方法 - cookies_obj = self.tab.cookies - cookies = cookies_obj() if callable(cookies_obj) else cookies_obj - cookies = cookies.as_dict() if hasattr(cookies, 'as_dict') else cookies - - # 确保返回完整的 cookie 列表，包含所有属性 - formatted_cookies = [] - for cookie in cookies: - formatted_cookie = { - 'name': cookie.get('name', ''), - 'value': cookie.get('value', ''), - 'domain': cookie.get('domain', ''), - 'path': cookie.get('path', '/'), - 'expires': cookie.get('expirationDate', None), - 'secure': cookie.get('secure', False), - 'httpOnly': cookie.get('httpOnly', False), - 'sameSite': cookie.get('sameSite', 'Lax') - } - formatted_cookies.append(formatted_cookie) - - return formatted_cookies - - def check_login(self, url: str, check_selector: str, success_text: str) -> bool: - """ - 检查是否已登录 - - 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(2) - - # 检查登录状态 - element = self.tab.ele(check_selector, timeout=10) - - if element and element.text and success_text in element.text: - return True - - return False - - except Exception as e: - 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") - - try: - # 准备所有 cookie 数据 - formatted_cookies = [] - for cookie in cookies: - 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') - formatted_cookies.append(cookie_dict) - - # 根据 DrissionPage 文档，set.cookies() 方法可以直接接收列表格式的多个 cookie - logger.info("使用一次性注入方式导入 cookie") - self.tab.set.cookies(formatted_cookies) - success_count = len(formatted_cookies) - - except Exception as e: - logger.error(f"导入 cookie 失败: {e}") - # 回退到逐个注入 - logger.info("回退到逐个注入方式导入 cookie") - for cookie_dict in formatted_cookies: - try: - 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, - cookies: List[Dict], - check_selector: str, - success_text: str - ) -> bool: - """ - 使用 cookie 登录 - - Args: - url: 网站地址 - cookies: cookie 列表 - check_selector: 登录检测选择器 - success_text: 登录成功时显示的文本 - - Returns: - 是否登录成功 - """ - if not self.browser or not self.tab: - 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) - - time.sleep(3) - - element = self.tab.ele(check_selector, timeout=10) - - if element and element.text and success_text in element.text: - return True - - return False - - except Exception as e: - logger.error(f"使用 cookie 登录失败: {e}") - return False - - def refresh_and_save_cookies(self, url: str) -> List[Dict]: - """ - 刷新页面并保存新的 cookie - - Args: - url: 网站地址 - - Returns: - cookie 列表 - """ - if not self.browser or not self.tab: - self._create_browser() - - try: - # 刷新页面 - self.tab.refresh() - - # 等待页面加载 - time.sleep(3) - - # 获取 cookies - cookies_obj = self.browser.cookies - cookies = cookies_obj() if callable(cookies_obj) else cookies_obj - cookies = cookies.as_dict() if hasattr(cookies, 'as_dict') else cookies - - # 确保返回的是列表 - if isinstance(cookies, list): - return cookies - elif isinstance(cookies, dict): - # 如果是字典，转换为列表 - return [cookies] - else: - # 如果是其他类型，返回空列表 - return [] - - except Exception as e: - logger.error(f"刷新页面并保存 cookie 失败: {e}") - return [] - - def close(self): - """关闭浏览器""" - self._close_browser() - - -class CookieManager: - """Cookie 管理器 - 实现浏览器实例级别的 Cookie 共享""" - - def __init__(self, cookie_file: str): - """ - 初始化 Cookie 管理器 - - Args: - cookie_file: cookie 文件路径 - """ - self.cookie_file = cookie_file - self.cookies = self._load_cookies() - - def _load_cookies(self) -> Dict[str, List[Dict]]: - """加载 cookie 文件""" - import os - - if os.path.exists(self.cookie_file): - try: - with open(self.cookie_file, 'r', encoding='utf-8') as f: - data = json.load(f) - # 兼容旧格式 - if isinstance(data, dict) and any(isinstance(v, dict) for v in data.values()): - # 转换旧格式为新格式 - new_data = {} - for user, websites in data.items(): - all_cookies = [] - for site_cookies in websites.values(): - all_cookies.extend(site_cookies) - new_data[user] = all_cookies - return new_data - return data - except Exception as e: - logger.error(f"加载 cookie 文件失败: {e}") - - return {} - - def _save_cookies(self): - """保存 cookie 文件""" - try: - with open(self.cookie_file, 'w', encoding='utf-8') as f: - json.dump(self.cookies, f, indent=2, ensure_ascii=False) - except Exception as e: - logger.error(f"保存 cookie 文件失败: {e}") - - def get_all_cookies(self, user_name: str) -> List[Dict]: - """ - 获取指定用户的所有 cookie - - Args: - user_name: 用户名 - - Returns: - cookie 列表 - """ - return self.cookies.get(user_name, []) - - def save_all_cookies(self, user_name: str, cookies: List[Dict]): - """ - 保存指定用户的所有 cookie - - Args: - user_name: 用户名 - cookies: cookie 列表 - """ - self.cookies[user_name] = cookies - self._save_cookies() - - def save_cookies(self, user_name: str, website_name: str, cookies: List[Dict]): - """ - 保存指定用户和网站的 cookie（兼容旧接口） - - Args: - user_name: 用户名 - website_name: 网站名（未使用，保持兼容） - cookies: cookie 列表 - """ - # 保持向后兼容，将网站的cookie添加到用户的所有cookie中 - if user_name not in self.cookies: - self.cookies[user_name] = [] - - # 合并新cookie到现有cookie中 - existing_cookies = self.cookies[user_name] - - # 创建现有cookie的索引，用于快速查找 - cookie_index = {} - for i, cookie in enumerate(existing_cookies): - if isinstance(cookie, dict): - key = (cookie.get('name'), cookie.get('domain'), cookie.get('path')) - cookie_index[key] = i - - # 添加或更新cookie - if isinstance(cookies, list): - valid_cookies = [] - for cookie in cookies: - if isinstance(cookie, dict): - valid_cookies.append(cookie) - - for cookie in valid_cookies: - key = (cookie.get('name'), cookie.get('domain'), cookie.get('path')) - if key in cookie_index: - # 更新现有cookie - existing_cookies[cookie_index[key]] = cookie - else: - # 添加新cookie - existing_cookies.append(cookie) - - if valid_cookies: - logger.info(f" 已保存 {len(valid_cookies)} 个新 cookie") - - self._save_cookies() - - def add_cookie(self, user_name: str, cookie: Dict): - """ - 为指定用户添加单个 cookie - - Args: - user_name: 用户名 - cookie: cookie 字典 - """ - if user_name not in self.cookies: - self.cookies[user_name] = [] - - # 检查是否已存在相同的 cookie - existing_index = -1 - for i, existing_cookie in enumerate(self.cookies[user_name]): - if (existing_cookie.get('name') == cookie.get('name') and - existing_cookie.get('domain') == cookie.get('domain') and - existing_cookie.get('path') == cookie.get('path')): - existing_index = i - break - - if existing_index >= 0: - # 更新现有 cookie - self.cookies[user_name][existing_index] = cookie - else: - # 添加新 cookie - self.cookies[user_name].append(cookie) - - self._save_cookies() - - def remove_cookie(self, user_name: str, cookie_name: str, domain: str): - """ - 从指定用户中移除指定的 cookie - - Args: - user_name: 用户名 - cookie_name: cookie 名称 - domain: cookie 域名 - """ - if user_name in self.cookies: - self.cookies[user_name] = [ - cookie for cookie in self.cookies[user_name] - if not (cookie.get('name') == cookie_name and cookie.get('domain') == domain) - ] - self._save_cookies() - - def clear_cookies(self, user_name: str): - """ - 清空指定用户的所有 cookie - - Args: - user_name: 用户名 - """ - if user_name in self.cookies: - self.cookies[user_name] = [] - self._save_cookies() - - def get_cookies_for_domain(self, user_name: str, domain: str) -> List[Dict]: - """ - 获取指定用户的指定域名的 cookie - - Args: - user_name: 用户名 - domain: 域名 - - Returns: - cookie 列表 - """ - user_cookies = self.get_all_cookies(user_name) - filtered_cookies = [] - - for cookie in user_cookies: - cookie_domain = cookie.get('domain', '') - # 统一域名格式 - if cookie_domain.startswith('.'): - cookie_domain = cookie_domain[1:] - - if domain in cookie_domain or cookie_domain in domain: - filtered_cookies.append(cookie) - - return filtered_cookies - - diff --git a/browser_tool.py b/browser_tool.py deleted file mode 100644 index b84242c..0000000 --- a/browser_tool.py +++ /dev/null @@ -1,179 +0,0 @@ -import tkinter as tk -from tkinter import ttk, messagebox -from DrissionPage import ChromiumPage, ChromiumOptions -import json -import threading -import time - -class BrowserTool: - def __init__(self, root): - self.root = root - self.root.title('浏览器Cookie管理工具') - self.root.geometry('500x200') - self.root.resizable(False, False) - - self.page = None - self.cookie_file = 'd:\\test\\openUrl\\cookies.json' - - self.setup_ui() - - def setup_ui(self): - frame = ttk.Frame(self.root, padding=20) - frame.pack(fill=tk.BOTH, expand=True) - - ttk.Label(frame, text='网址:').grid(row=0, column=0, sticky=tk.W, pady=5) - self.url_entry = ttk.Entry(frame, width=60) - self.url_entry.grid(row=0, column=1, columnspan=3, pady=5, sticky=tk.W) - self.url_entry.insert(0, 'https://yunlu.sinopec.com/AddressBook/AddressBookCheck') - - btn_frame = ttk.Frame(frame) - btn_frame.grid(row=1, column=0, columnspan=4, pady=20) - - self.open_btn = ttk.Button(btn_frame, text='打开', width=15, command=self.open_browser) - self.open_btn.pack(side=tk.LEFT, padx=10) - - self.save_btn = ttk.Button(btn_frame, text='保存', width=15, command=self.save_cookies) - self.save_btn.pack(side=tk.LEFT, padx=10) - - self.import_btn = ttk.Button(btn_frame, text='导入', width=15, command=self.import_cookies) - self.import_btn.pack(side=tk.LEFT, padx=10) - - self.status_var = tk.StringVar(value='就绪') - ttk.Label(frame, textvariable=self.status_var).grid(row=2, column=0, columnspan=4, pady=10) - - def log(self, msg): - self.status_var.set(msg) - self.root.update() - - def init_browser(self): - co = ChromiumOptions() - co.set_browser_path(r'C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe') - self.page = ChromiumPage(addr_or_opts=co) - - def wait_page_loaded(self, timeout=30): - if self.page: - self.page.wait.doc_loaded(timeout=timeout) - self.page.ele('tag:body', timeout=timeout) - - def open_browser(self): - def run(): - try: - self.log('正在启动浏览器...') - self.init_browser() - - url = self.url_entry.get().strip() - if not url: - messagebox.showerror('错误', '请输入网址') - return - - self.log(f'正在打开: {url}') - self.page.get(url) - self.wait_page_loaded() - self.log('页面加载完成') - - except Exception as e: - self.log(f'错误: {str(e)}') - messagebox.showerror('错误', str(e)) - - threading.Thread(target=run, daemon=True).start() - - def save_cookies(self): - def run(): - try: - if not self.page: - messagebox.showwarning('提示', '请先打开浏览器') - return - - self.log('正在保存Cookies...') - cookies = self.page.cookies() - - with open(self.cookie_file, 'w', encoding='utf-8') as f: - json.dump(cookies, f, ensure_ascii=False, indent=2) - - self.log(f'Cookies已保存到 {self.cookie_file}') - - self.log('正在关闭浏览器...') - self.page.quit() - self.page = None - self.log('浏览器已关闭') - - messagebox.showinfo('成功', 'Cookies保存成功，浏览器已关闭') - - except Exception as e: - self.log(f'错误: {str(e)}') - messagebox.showerror('错误', str(e)) - - threading.Thread(target=run, daemon=True).start() - - def import_cookies(self): - def run(): - try: - import os - if not os.path.exists(self.cookie_file): - messagebox.showwarning('提示', 'Cookie文件不存在，请先保存Cookies') - return - - self.log('正在启动浏览器...') - self.init_browser() - - self.log('正在加载Cookies...') - with open(self.cookie_file, 'r', encoding='utf-8') as f: - cookies = json.load(f) - - url = self.url_entry.get().strip() - if not url: - messagebox.showerror('错误', '请输入网址') - return - - from urllib.parse import urlparse - parsed = urlparse(url) - domain = parsed.netloc - - self.page.get('about:blank') - - for cookie in cookies: - cookie_dict = { - 'name': cookie.get('name'), - 'value': cookie.get('value'), - 'domain': cookie.get('domain', domain), - 'path': cookie.get('path', '/') - } - try: - self.page.set.cookies(cookie_dict) - except Exception as ce: - self.log(f'设置Cookie失败: {cookie.get("name")} - {str(ce)}') - - self.log(f'已导入 {len(cookies)} 个Cookie') - - self.log(f'正在打开: {url}') - self.page.get(url) - self.wait_page_loaded() - - self.log('刷新页面验证登录状态...') - for i in range(3): - self.page.refresh() - self.wait_page_loaded() - time.sleep(1) - self.log(f'刷新第 {i+1} 次...') - - self.log('页面加载完成') - - except Exception as e: - self.log(f'错误: {str(e)}') - messagebox.showerror('错误', str(e)) - - threading.Thread(target=run, daemon=True).start() - - def on_closing(self): - if self.page: - try: - self.page.quit() - except: - pass - self.root.destroy() - -if __name__ == '__main__': - root = tk.Tk() - app = BrowserTool(root) - root.protocol('WM_DELETE_WINDOW', app.on_closing) - root.mainloop() diff --git a/config.json b/config.json index 64f9a7b..79113e5 100644 --- a/config.json +++ b/config.json @@ -1,7 +1,7 @@ { "users": [ { - "name": "用户A", + "name": "小余", "cookie_cloud": { "uuid": "hu6n2vcUmzpu7mqUN2rVCg", "password": "jtDM5dV9AyqXkZdQVeA9f6", @@ -18,7 +18,7 @@ }, "websites": [ { - "name": "网站A", + "name": "柴油联名卡", "url": "https://lmkbi.95155.com/bi-system/#/carUnpetrifiedComAll", "login_check_selector": "", "success_text": "彭峰" @@ -26,7 +26,7 @@ ] }, { - "name": "用户B", + "name": "彭峰(个人)", "cookie_cloud": { "uuid": "qyZpTrgiP5mVwiRZwfBhjz", "password": "iGn1B4FnWftp3oj4Ko3jxA", @@ -42,12 +42,6 @@ "headless": false }, "websites": [ - { - "name": "网站A", - "url": "https://lmkbi.95155.com/bi-system/#/carUnpetrifiedComAll", - "login_check_selector": "", - "success_text": "彭峰" - }, { "name": "网站B", "url": "https://192.168.1.200:8006/", @@ -55,6 +49,49 @@ "success_text": "彭峰" } ] + }, + { + "name": "彭峰(公司)", + "cookie_cloud": { + "uuid": "aCgAWN5DsYhP88FnTmrr3H", + "password": "bK8dfh1a6ZZ79jWeeMedjN", + "api_url": "https://movie-pilot.org/cookiecloud" + }, + "notification": { + "iyuu_token": "IYUU37629Tc1d371c7ce99a49ff9778e196286b7e4592be191", + "max_fail_count": 3, + "notify_on_success": true + }, + "browser": { + "type": "edge", + "headless": false + }, + "websites": [ + { + "name": "柴油联名卡", + "url": "https://lmkbi.95155.com/bi-system/#/carUnpetrifiedComAll", + "login_check_selector": "", + "success_text": "彭峰" + }, + { + "name": "加油卡交易", + "url": "https://cardweb.salecard.sinopec.com/card/index.jsp", + "login_check_selector": "", + "success_text": "彭峰" + }, + { + "name": "数字销售门户", + "url": "https://ewcc.sinopec.com/?authConfigs=37d5aed075525d4fa0fe635231cba447", + "login_check_selector": "", + "success_text": "彭峰" + }, + { + "name": "飞牛存储", + "url": "http://10.190.20.156:5666/", + "login_check_selector": "", + "success_text": "彭峰" + } + ] } ] -} +} \ No newline at end of file diff --git a/config_manager.py b/config_manager.py new file mode 100644 index 0000000..fa71447 --- /dev/null +++ b/config_manager.py @@ -0,0 +1,292 @@ +"""配置管理模块""" +import json +import os +from logger import get_logger + +logger = get_logger(__name__) + +DEFAULT_CONFIG_FILE = "config.json" +DEFAULT_COOKIE_DATA_FILE = "cookie_data.json" + + +class ConfigManager: + """配置管理器""" + + def __init__(self, config_file: str = DEFAULT_CONFIG_FILE): + """ + 初始化配置管理器 + + Args: + config_file: 配置文件路径 + """ + self.config_file = config_file + self.config = self._load_config() + + def _load_config(self) -> dict: + """加载配置文件""" + try: + if os.path.exists(self.config_file): + with open(self.config_file, 'r', encoding='utf-8') as f: + config = json.load(f) + logger.info(f"配置文件加载成功: {self.config_file}") + return config + else: + logger.warning(f"配置文件不存在: {self.config_file}") + return {} + except Exception as e: + logger.error(f"加载配置文件失败: {e}") + return {} + + def save_config(self, config: dict) -> bool: + """ + 保存配置文件 + + Args: + config: 配置字典 + + Returns: + 是否保存成功 + """ + try: + with open(self.config_file, 'w', encoding='utf-8') as f: + json.dump(config, f, indent=2, ensure_ascii=False) + logger.info(f"配置文件保存成功: {self.config_file}") + return True + except Exception as e: + logger.error(f"保存配置文件失败: {e}") + return False + + def get(self, key: str, default=None): + """ + 获取配置项 + + Args: + key: 配置键 + default: 默认值 + + Returns: + 配置值 + """ + return self.config.get(key, default) + + def set(self, key: str, value): + """ + 设置配置项 + + Args: + key: 配置键 + value: 配置值 + """ + self.config[key] = value + self.save_config(self.config) + + def get_users(self) -> list: + """ + 获取所有用户配置 + + Returns: + 用户配置列表 + """ + return self.config.get('users', []) + + def get_user_config(self, user_name: str) -> dict: + """ + 获取指定用户的配置 + + Args: + user_name: 用户名 + + Returns: + 用户配置字典 + """ + users = self.get_users() + for user in users: + if user.get('name') == user_name: + return user + return {} + + def get_cookie_cloud_config(self, user_name: str) -> dict: + """ + 获取用户的Cookie Cloud配置 + + Args: + user_name: 用户名 + + Returns: + Cookie Cloud配置字典 + """ + user_config = self.get_user_config(user_name) + return user_config.get('cookie_cloud', {}) + + def get_notification_config(self, user_name: str) -> dict: + """ + 获取用户的通知配置 + + Args: + user_name: 用户名 + + Returns: + 通知配置字典 + """ + user_config = self.get_user_config(user_name) + return user_config.get('notification', {}) + + def get_browser_config(self, user_name: str) -> dict: + """ + 获取用户的浏览器配置 + + Args: + user_name: 用户名 + + Returns: + 浏览器配置字典 + """ + user_config = self.get_user_config(user_name) + return user_config.get('browser', {}) + + def get_websites(self, user_name: str) -> list: + """ + 获取用户的网站配置 + + Args: + user_name: 用户名 + + Returns: + 网站配置列表 + """ + user_config = self.get_user_config(user_name) + return user_config.get('websites', []) + + +class CookieDataManager: + """Cookie数据管理器""" + + def __init__(self, data_file: str = DEFAULT_COOKIE_DATA_FILE): + """ + 初始化Cookie数据管理器 + + Args: + data_file: Cookie数据文件路径 + """ + self.data_file = data_file + self.data = self._load_data() + + def _load_data(self) -> dict: + """加载Cookie数据文件""" + try: + if os.path.exists(self.data_file): + with open(self.data_file, 'r', encoding='utf-8') as f: + data = json.load(f) + logger.info(f"Cookie数据文件加载成功: {self.data_file}") + return data + else: + logger.warning(f"Cookie数据文件不存在: {self.data_file}") + return {} + except Exception as e: + logger.error(f"加载Cookie数据文件失败: {e}") + return {} + + def save_data(self, data: dict) -> bool: + """ + 保存Cookie数据文件 + + Args: + data: Cookie数据字典 + + Returns: + 是否保存成功 + """ + try: + with open(self.data_file, 'w', encoding='utf-8') as f: + json.dump(data, f, indent=2, ensure_ascii=False) + logger.info(f"Cookie数据文件保存成功: {self.data_file}") + return True + except Exception as e: + logger.error(f"保存Cookie数据文件失败: {e}") + return False + + def get_user_cookies(self, user_name: str) -> dict: + """ + 获取用户的Cookie数据 + + Args: + user_name: 用户名 + + Returns: + 用户Cookie数据字典 + """ + return self.data.get(user_name, {}) + + def save_user_cookies(self, user_name: str, cookies: list, url: str, website_name: str = ""): + """ + 保存用户的Cookie数据 + + Args: + user_name: 用户名 + cookies: Cookie列表 + url: 网站URL + website_name: 网站名称 + """ + if user_name not in self.data: + self.data[user_name] = {} + + from urllib.parse import urlparse + parsed = urlparse(url) + domain = parsed.netloc + + cookie_entry = { + "cookies": cookies, + "url": url, + "website_name": website_name + } + + self.data[user_name][domain] = cookie_entry + self.save_data(self.data) + logger.info(f"已保存 {len(cookies)} 个 cookie ({user_name} -> {domain})") + + def get_domain_cookies(self, user_name: str, domain: str) -> dict: + """ + 获取指定域名的Cookie数据 + + Args: + user_name: 用户名 + domain: 域名 + + Returns: + 域名Cookie数据字典 + """ + user_cookies = self.get_user_cookies(user_name) + + for cookie_domain, cookie_data in user_cookies.items(): + clean_domain = cookie_domain.lstrip('.') + if domain in clean_domain or clean_domain in domain: + return cookie_data + + return {} + + def save_simple_cookies(self, cookies: list, url: str): + """ + 保存简单的Cookie数据（用于Cookie提取器） + + Args: + cookies: Cookie列表 + url: 网站URL + """ + data = { + "cookies": cookies, + "url": url + } + self.save_data(data) + logger.info(f"已保存 {len(cookies)} 个 cookie") + + def load_simple_cookies(self) -> dict: + """ + 加载简单的Cookie数据（用于Cookie提取器） + + Returns: + Cookie数据字典 + """ + return { + "cookies": self.data.get("cookies", []), + "url": self.data.get("url", "") + } diff --git a/cookie_cloud.py b/cookie_cloud.py index 9208bf0..0941306 100644 --- a/cookie_cloud.py +++ b/cookie_cloud.py @@ -7,7 +7,7 @@ import urllib.request import urllib.error from Crypto.Cipher import AES from dataclasses import dataclass -from applogger import get_logger +from logger import get_logger logger = get_logger(__name__) @@ -253,6 +253,3 @@ class CookieCloud: logger.info(f"获取到所有数据: {len(result['cookies'])} 个域名的 cookies, {len(result['local_storage'])} 个域名的 local storage") return result - - - diff --git a/cookie_data.json b/cookie_data.json index c026136..3851720 100644 --- a/cookie_data.json +++ b/cookie_data.json @@ -1,21 +1,38 @@ { - "用户A": { + "小余": { "lmkbi.95155.com": { "cookies": [], "url": "https://lmkbi.95155.com/bi-system/#/carUnpetrifiedComAll", - "website_name": "网站A" + "website_name": "柴油联名卡" } }, - "用户B": { - "lmkbi.95155.com": { - "cookies": [], - "url": "https://lmkbi.95155.com/bi-system/#/carUnpetrifiedComAll", - "website_name": "网站A" - }, + "彭峰(个人)": { "192.168.1.200:8006": { "cookies": [], "url": "https://192.168.1.200:8006/", "website_name": "网站B" } + }, + "彭峰(公司)": { + "lmkbi.95155.com": { + "cookies": [], + "url": "https://lmkbi.95155.com/bi-system/#/carUnpetrifiedComAll", + "website_name": "柴油联名卡" + }, + "cardweb.salecard.sinopec.com": { + "cookies": [], + "url": "https://cardweb.salecard.sinopec.com/card/index.jsp", + "website_name": "加油卡交易" + }, + "ewcc.sinopec.com": { + "cookies": [], + "url": "https://ewcc.sinopec.com/?authConfigs=37d5aed075525d4fa0fe635231cba447", + "website_name": "数字销售门户" + }, + "10.190.20.156:5666": { + "cookies": [], + "url": "http://10.190.20.156:5666/", + "website_name": "飞牛存储" + } } } \ No newline at end of file diff --git a/cookie_extractor.py b/cookie_extractor.py index e26619b..474acff 100644 --- a/cookie_extractor.py +++ b/cookie_extractor.py @@ -1,35 +1,35 @@ """Cookie提取器 - 使用DrissionPage库""" import tkinter as tk from tkinter import ttk, messagebox -import json -import os -import logging -from DrissionPage import ChromiumPage, ChromiumOptions +from browser_manager import BrowserManager +from config_manager import CookieDataManager +from logger import get_logger + +logger = get_logger(__name__) -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s - %(levelname)s - %(message)s', - handlers=[ - logging.FileHandler('cookie_extractor.log'), - logging.StreamHandler() - ] -) -logger = logging.getLogger(__name__) class CookieExtractorApp: """Cookie提取器应用""" def __init__(self, root): + """ + 初始化应用 + + Args: + root: 根窗口 + """ self.root = root self.root.title("Cookie提取器") self.root.geometry("600x250") self.root.resizable(True, True) - self.page = None + self.browser_manager = BrowserManager() + self.cookie_data_manager = CookieDataManager() self._create_widgets() def _create_widgets(self): + """创建界面控件""" main_frame = ttk.Frame(self.root, padding="20") main_frame.pack(fill=tk.BOTH, expand=True) @@ -63,18 +63,6 @@ class CookieExtractorApp: ttk.Label(status_frame, text="状态:").pack(side=tk.LEFT, padx=5) ttk.Label(status_frame, textvariable=self.status_var, foreground="blue").pack(side=tk.LEFT, padx=5) - def _init_browser(self): - """初始化浏览器""" - co = ChromiumOptions() - co.set_browser_path(r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe") - self.page = ChromiumPage(addr_or_opts=co) - - def _wait_page_loaded(self, timeout=30): - """等待页面加载完成""" - if self.page: - self.page.wait.doc_loaded(timeout=timeout) - self.page.ele('tag:body', timeout=timeout) - def open_website(self): """打开指定网站""" url = self.url_var.get().strip() @@ -93,13 +81,13 @@ class CookieExtractorApp: self.status_var.set("正在打开浏览器...") self.root.update() - self._init_browser() + self.browser_manager.init_browser() self.status_var.set(f"正在访问: {url}") self.root.update() - self.page.get(url) - self._wait_page_loaded() + self.browser_manager.page.get(url) + self.browser_manager.wait_page_loaded() self.status_var.set(f"已打开: {url}") self.save_button.config(state=tk.NORMAL) @@ -113,30 +101,25 @@ class CookieExtractorApp: def save_cookies(self): """保存当前的cookie信息，并强制关闭浏览器""" - if not self.page: + if not self.browser_manager.page: messagebox.showerror("错误", "请先打开网站") return try: - self.status_var.set("正在保存Cookies...") + self.status_var.set("正在提取cookies...") self.root.update() - cookies = self.page.cookies() + cookies = self.browser_manager.get_cookies() + logger.info(f"成功提取 {len(cookies)} 个 cookies") - data = { - "cookies": cookies, - "url": self.url_var.get() - } + self.cookie_data_manager.save_simple_cookies(cookies, self.url_var.get()) - output_file = "cookie_data.json" - with open(output_file, 'w', encoding='utf-8') as f: - json.dump(data, f, indent=2, ensure_ascii=False) + self.status_var.set("正在关闭浏览器...") + self.root.update() - self.page.quit() - self.page = None - self.save_button.config(state=tk.DISABLED) + self._cleanup_browser() - self.status_var.set(f"Cookies已保存到 {output_file}") + self.status_var.set(f"Cookies已保存到 cookie_data.json") messagebox.showinfo('成功', 'Cookies保存成功，浏览器已关闭') except Exception as e: @@ -146,23 +129,13 @@ class CookieExtractorApp: self._cleanup_browser() def import_cookies(self): - """打开浏览器，导入cookie和localStorage信息，然后打开指定网站""" + """打开浏览器，导入cookie信息，然后打开指定网站""" import time try: - output_file = "cookie_data.json" - if not os.path.exists(output_file): - messagebox.showerror("错误", f"文件不存在: {output_file}") - return - - self.status_var.set("正在读取cookies...") - self.root.update() - - with open(output_file, 'r', encoding='utf-8') as f: - data = json.load(f) + data = self.cookie_data_manager.load_simple_cookies() url = data.get("url", "") cookies = data.get("cookies", []) - local_storage = data.get("local_storage", {}) if not url: url = self.url_var.get().strip() @@ -179,11 +152,11 @@ class CookieExtractorApp: self.status_var.set("正在打开浏览器...") self.root.update() - self._init_browser() + self.browser_manager.init_browser() self.status_var.set("等待浏览器就绪...") self.root.update() - self.page.get("about:blank") + self.browser_manager.page.get("about:blank") time.sleep(1) from urllib.parse import urlparse @@ -193,55 +166,19 @@ class CookieExtractorApp: if cookies: self.status_var.set("正在导入cookies...") self.root.update() - - success_count = 0 - for cookie in cookies: - try: - if not cookie.get('name') or not cookie.get('value'): - continue - - cookie_dict = { - 'name': cookie.get('name'), - 'value': cookie.get('value'), - 'domain': cookie.get('domain', domain), - 'path': cookie.get('path', '/') - } - - self.page.set.cookies(cookie_dict) - success_count += 1 - except Exception as e: - logger.error(f"导入cookie失败: {cookie.get('name')} - {e}") - + success_count = self.browser_manager.import_cookies(cookies, domain) logger.info(f"成功导入 {success_count} 个 cookies") - if local_storage: - self.status_var.set("正在导入localStorage...") - self.root.update() - - ls_count = 0 - for ls_domain, items in local_storage.items(): - try: - for key, value in items.items(): - try: - self.page.set.local_storage(key, value) - ls_count += 1 - except Exception as e: - logger.error(f"导入localStorage失败: {key} - {e}") - except Exception as e: - logger.error(f"导入localStorage域失败: {ls_domain} - {e}") - - logger.info(f"成功导入 {ls_count} 个 localStorage") - self.status_var.set(f"正在访问: {url}") self.root.update() - self.page.get(url) - self._wait_page_loaded() + self.browser_manager.page.get(url) + self.browser_manager.wait_page_loaded() self.status_var.set("刷新页面验证登录状态...") self.root.update() for i in range(3): - self.page.refresh() - self._wait_page_loaded() + self.browser_manager.page.refresh() + self.browser_manager.wait_page_loaded() time.sleep(1) self.status_var.set(f"刷新第 {i+1} 次...") self.root.update() @@ -258,21 +195,19 @@ class CookieExtractorApp: self._cleanup_browser() def close_app(self): + """关闭应用""" self._cleanup_browser() self.root.destroy() def _cleanup_browser(self): - if self.page: - try: - self.page.quit() - except Exception as e: - logger.error(f"关闭浏览器失败: {e}") - finally: - self.page = None - self.save_button.config(state=tk.DISABLED) - self.status_var.set("就绪") + """清理浏览器资源""" + self.browser_manager.close() + self.save_button.config(state=tk.DISABLED) + self.status_var.set("就绪") + def main(): + """主函数""" root = tk.Tk() app = CookieExtractorApp(root) @@ -283,5 +218,6 @@ def main(): root.protocol("WM_DELETE_WINDOW", on_closing) root.mainloop() + if __name__ == "__main__": main() diff --git a/applogger.py b/logger.py similarity index 93% rename from applogger.py rename to logger.py index 9686b5b..9395894 100644 --- a/applogger.py +++ b/logger.py @@ -1,4 +1,4 @@ -"""日志配置模块""" +"""日志管理模块""" import os import logging from logging.handlers import TimedRotatingFileHandler @@ -15,7 +15,7 @@ def setup_logging( ) -> None: """ 配置日志系统 - + Args: log_dir: 日志目录 log_file: 日志文件名 @@ -28,7 +28,7 @@ def setup_logging( # 确保日志目录存在 if not os.path.exists(log_dir): os.makedirs(log_dir) - + # 配置日志处理器 handlers = [ TimedRotatingFileHandler( @@ -40,7 +40,7 @@ def setup_logging( ), logging.StreamHandler() ] - + # 配置日志 logging.basicConfig( level=level, @@ -52,11 +52,11 @@ def setup_logging( def get_logger(name: str) -> logging.Logger: """ 获取日志记录器 - + Args: name: 日志记录器名称，通常使用 __name__ - + Returns: Logger 实例 """ - return logging.getLogger(name) \ No newline at end of file + return logging.getLogger(name) diff --git a/monitor.py b/monitor.py index 70708dd..5a0bc1b 100644 --- a/monitor.py +++ b/monitor.py @@ -1,13 +1,12 @@ """Cookie 监控主程序""" -import json import time -import os -from datetime import datetime, time as dt_time +from datetime import datetime from urllib.parse import urlparse from cookie_cloud import CookieCloud -from notifier import IYUUNotifier -from applogger import setup_logging, get_logger -from DrissionPage import ChromiumPage, ChromiumOptions +from notification import create_notification_manager +from config_manager import ConfigManager, CookieDataManager +from browser_manager import BrowserManager +from logger import get_logger logger = get_logger(__name__) @@ -16,6 +15,9 @@ STATE_FILE = "notify_state.json" def load_notify_state() -> dict: + """加载通知状态""" + import json + import os if os.path.exists(STATE_FILE): try: with open(STATE_FILE, 'r', encoding='utf-8') as f: @@ -26,6 +28,8 @@ def load_notify_state() -> dict: def save_notify_state(state: dict): + """保存通知状态""" + import json try: with open(STATE_FILE, 'w', encoding='utf-8') as f: json.dump(state, f, indent=2, ensure_ascii=False) @@ -34,164 +38,55 @@ def save_notify_state(state: dict): def should_send_notification(user_name: str) -> bool: + """判断是否应该发送通知""" return False def mark_notification_sent(user_name: str): + """标记通知已发送""" now = datetime.now() state = load_notify_state() - + if user_name not in state: state[user_name] = {} - + state[user_name]['date'] = now.strftime('%Y-%m-%d') state[user_name]['hour'] = now.hour - + save_notify_state(state) -class BrowserManager: - """浏览器管理器""" - - def __init__(self, browser_type: str = 'edge', headless: bool = False): - self.browser_type = browser_type - self.headless = headless - self.page = None - - def init_browser(self): - """初始化浏览器""" - co = ChromiumOptions() - if self.browser_type == 'edge': - co.set_browser_path(r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe") - co.set_argument('--no-sandbox') - co.set_argument('--disable-dev-shm-usage') - if self.headless: - co.set_argument('--headless') - self.page = ChromiumPage(addr_or_opts=co) - - def wait_page_loaded(self, timeout=30): - """等待页面加载完成""" - if self.page: - self.page.wait.doc_loaded(timeout=timeout) - self.page.ele('tag:body', timeout=timeout) - - def ensure_connection(self): - """确保浏览器连接正常""" - try: - if self.page: - self.page.get("about:blank") - return True - except: - pass - - try: - self.init_browser() - self.page.get("about:blank") - return True - except Exception as e: - logger.error(f"无法建立浏览器连接: {e}") - return False - - def import_cookies(self, cookies: list, domain: str): - """导入cookies""" - if not self.ensure_connection(): - return 0 - - success_count = 0 - for cookie in cookies: - try: - if not cookie.get('name') or not cookie.get('value'): - continue - - cookie_dict = { - 'name': cookie.get('name'), - 'value': cookie.get('value'), - 'domain': cookie.get('domain', domain), - 'path': cookie.get('path', '/') - } - - self.page.set.cookies(cookie_dict) - success_count += 1 - except Exception as e: - pass - - return success_count - - def verify_login(self, url: str, check_selector: str = "", success_text: str = "") -> bool: - """验证登录状态""" - if not self.ensure_connection(): - return False - - try: - self.page.get(url) - self.wait_page_loaded() - - for i in range(3): - time.sleep(1) - self.page.refresh() - self.wait_page_loaded() - - if not check_selector: - return True - - try: - element = self.page.ele(check_selector, timeout=10) - if element: - element_text = element.text or "" - if not success_text: - return True - if success_text in element_text: - return True - except: - pass - - return False - - except Exception as e: - logger.error(f"验证登录状态失败: {e}") - return False - - def close(self): - """关闭浏览器""" - if self.page: - try: - self.page.quit() - except: - pass - finally: - self.page = None - - class CookieMonitor: """Cookie 监控器""" def __init__(self, config_file: str = "config.json"): - self.config_file = config_file - self.config = self._load_config() + """ + 初始化监控器 + + Args: + config_file: 配置文件路径 + """ + self.config_manager = ConfigManager(config_file) + self.cookie_data_manager = CookieDataManager() self.all_results = [] - 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: - logger.error(f"加载配置文件失败: {e}") - return {} - def process_user(self, user_config: dict): - """处理单个用户：依次验证各网站登录状态""" + """ + 处理单个用户：依次验证各网站登录状态 + + Args: + user_config: 用户配置字典 + """ user_name = user_config.get('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 + browser_config = user_config.get('browser', {}) browser = BrowserManager( browser_type=browser_config.get('type', 'edge'), headless=browser_config.get('headless', False) @@ -199,7 +94,7 @@ class CookieMonitor: user_results = { 'user_name': user_name, - 'notification_config': notification_config, + 'notification_config': user_config.get('notification', {}), 'websites': [], 'success_count': 0, 'fail_count': 0 @@ -207,9 +102,10 @@ class CookieMonitor: try: logger.info(f"步骤1: 从本地 cookie_data.json 读取用户 cookies") - local_cookies = self._load_local_cookies(user_name) - + local_cookies = self.cookie_data_manager.get_user_cookies(user_name) + logger.info(f"步骤1.5: 从 Cookie Cloud 获取所有 cookies") + cookie_cloud_config = user_config.get('cookie_cloud', {}) cookie_cloud = CookieCloud( api_url=cookie_cloud_config.get('api_url', ''), uuid=cookie_cloud_config.get('uuid', ''), @@ -219,7 +115,7 @@ class CookieMonitor: total_domains = len(all_cookies) total_cookies = sum(len(c) for c in all_cookies.values()) logger.info(f"获取到 {total_domains} 个域名，共 {total_cookies} 个 cookie") - + logger.info(f"步骤2: 初始化浏览器") browser.init_browser() @@ -233,7 +129,7 @@ class CookieMonitor: local_cookies=local_cookies ) user_results['websites'].append(result) - + if result['success']: user_results['success_count'] += 1 else: @@ -241,7 +137,7 @@ class CookieMonitor: self.all_results.append(user_results) self._print_summary(user_name, user_results['websites']) - + logger.info(f"\n检查用户 {user_name} 是否需要发送通知") if should_send_notification(user_name): self._send_user_notification(user_results) @@ -249,19 +145,6 @@ class CookieMonitor: finally: browser.close() - def _load_local_cookies(self, user_name: str) -> dict: - """从本地 cookie_data.json 读取用户的 cookies""" - filename = "cookie_data.json" - if not os.path.exists(filename): - return {} - - try: - with open(filename, 'r', encoding='utf-8') as f: - cookie_data = json.load(f) - return cookie_data.get(user_name, {}) - except: - return {} - def process_website( self, user_name: str, @@ -270,12 +153,24 @@ class CookieMonitor: all_cookies: dict, local_cookies: dict ) -> dict: - """验证网站登录状态""" + """ + 验证网站登录状态 + + Args: + user_name: 用户名 + website_config: 网站配置 + browser: 浏览器管理器 + all_cookies: 所有cookies + local_cookies: 本地cookies + + Returns: + 验证结果字典 + """ 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', '') - + logger.info(f"\n验证网站: {website_name}") logger.info(f" URL: {url}") @@ -294,20 +189,26 @@ class CookieMonitor: try: parsed = urlparse(url) domain = parsed.netloc - + domain_cookies = [] - for cookie_domain, cookies in local_cookies.items(): + for cookie_domain, cookie_data in local_cookies.items(): clean_domain = cookie_domain.lstrip('.') if domain in clean_domain or clean_domain in domain: + if isinstance(cookie_data, dict): + cookies = cookie_data.get('cookies', []) + elif isinstance(cookie_data, list): + cookies = cookie_data + else: + cookies = [] domain_cookies.extend(cookies) - + if domain_cookies: logger.info(f" 找到 {len(domain_cookies)} 个本地 cookie") success_count = browser.import_cookies(domain_cookies, domain) logger.info(f" 成功导入 {success_count} 个 cookie") else: logger.warning(f" 未找到本地 cookie") - + login_success = browser.verify_login( url=url, check_selector=check_selector, @@ -317,35 +218,59 @@ class CookieMonitor: if login_success: result['success'] = True logger.info(f" ✓ 使用本地 cookie 登录验证成功") - - self._save_cookies_to_file(domain_cookies, url, website_name, user_name) + + if domain_cookies: + logger.info(f" 保存 {len(domain_cookies)} 个 cookie 到本地文件") + self.cookie_data_manager.save_user_cookies( + user_name=user_name, + cookies=domain_cookies, + url=url, + website_name=website_name + ) + else: + logger.warning(f" 登录验证成功但 cookies 为空，跳过保存") else: logger.warning(f" ✗ 本地 cookie 登录失败，尝试使用服务器 cookie") - + domain_cookies = [] - for cookie_domain, cookies in all_cookies.items(): + for cookie_domain, cookie_data in all_cookies.items(): clean_domain = cookie_domain.lstrip('.') if domain in clean_domain or clean_domain in domain: + if isinstance(cookie_data, dict): + cookies = cookie_data.get('cookies', []) + elif isinstance(cookie_data, list): + cookies = cookie_data + else: + cookies = [] domain_cookies.extend(cookies) - + if domain_cookies: logger.info(f" 找到 {len(domain_cookies)} 个服务器 cookie") success_count = browser.import_cookies(domain_cookies, domain) logger.info(f" 成功导入 {success_count} 个 cookie") else: logger.warning(f" 未找到服务器 cookie") - + login_success = browser.verify_login( url=url, check_selector=check_selector, success_text=success_text ) - + if login_success: result['success'] = True logger.info(f" ✓ 使用服务器 cookie 登录验证成功") - - self._save_cookies_to_file(domain_cookies, url, website_name, user_name) + + if domain_cookies: + logger.info(f" 保存 {len(domain_cookies)} 个 cookie 到本地文件") + self.cookie_data_manager.save_user_cookies( + user_name=user_name, + cookies=domain_cookies, + url=url, + website_name=website_name + ) + else: + logger.warning(f" 登录验证成功但 cookies 为空，跳过保存") else: result['error'] = '登录状态验证失败（本地和服务器 cookie 都无效）' logger.warning(f" ✗ {result['error']}") @@ -356,49 +281,6 @@ class CookieMonitor: return result - def _save_cookies_to_file(self, cookies: list, url: str, website_name: str, user_name: str): - """保存 cookies 到文件（按用户和域名区分）""" - try: - parsed = urlparse(url) - domain = parsed.netloc - - cookie_entry = { - "cookies": [], - "url": url, - "website_name": website_name - } - - for cookie in cookies: - if cookie.get('name') and cookie.get('value'): - cookie_entry["cookies"].append({ - "name": cookie.get('name'), - "value": cookie.get('value'), - "domain": cookie.get('domain', domain) - }) - - filename = "cookie_data.json" - - cookie_data = {} - if os.path.exists(filename): - with open(filename, 'r', encoding='utf-8') as f: - try: - cookie_data = json.load(f) - except: - cookie_data = {} - - if user_name not in cookie_data: - cookie_data[user_name] = {} - - cookie_data[user_name][domain] = cookie_entry - - with open(filename, 'w', encoding='utf-8') as f: - json.dump(cookie_data, f, indent=2, ensure_ascii=False) - - logger.info(f" 已保存 {len(cookie_entry['cookies'])} 个 cookie 到 {filename} ({user_name} -> {domain})") - - except Exception as e: - logger.error(f" 保存 cookie 失败: {e}") - def _print_summary(self, user_name: str, results: list): """打印处理结果汇总""" success_count = sum(1 for r in results if r['success']) @@ -418,18 +300,18 @@ class CookieMonitor: """发送单个用户的通知""" user_name = user_result['user_name'] notification_config = user_result.get('notification_config', {}) - iyuu_token = notification_config.get('iyuu_token', '') - - if not iyuu_token: + + notifier = create_notification_manager(notification_config) + + if not notifier: + logger.warning(f"用户 {user_name} 未配置通知") return - - notifier = IYUUNotifier(iyuu_token) - + success_count = user_result['success_count'] fail_count = user_result['fail_count'] - + title = f"【Cookie监控】{user_name} - 成功{success_count} 失败{fail_count}" - + content_lines = [ f"**{user_name} Cookie监控报告**", f"", @@ -438,33 +320,20 @@ class CookieMonitor: f"失败: {fail_count} 个网站", f"", ] - + for website in user_result['websites']: status = "✓ 成功" if website['success'] else f"✗ 失败: {website['error']}" content_lines.append(f" {website['name']}: {status}") - + content = "\n".join(content_lines) - - print(f"\n{'='*60}") - print(f"发送用户通知: {user_name}") - print(f"{'='*60}") - print(f"\n发送内容:") - print(f"Title: {title}") - print(f"Content:\n{content}") - print(f"\n发送方式: 爱语飞飞 API (GET 请求)") - print(f"API URL: https://iyuu.cn/{iyuu_token}.send") - print(f"参数: text={title}, desp={content[:50]}...") - print(f"\n正在发送...") - + logger.info(f"\n发送用户通知: {user_name}") logger.info(f"Title: {title}") - + if notifier.send(title, content): - print(f"\n发送结果: ✓ 成功") logger.info(f"✓ 用户 {user_name} 通知发送成功") mark_notification_sent(user_name) else: - print(f"\n发送结果: ✗ 失败") logger.error(f"✗ 用户 {user_name} 通知发送失败") def run(self): @@ -474,7 +343,7 @@ class CookieMonitor: logger.info(f"时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") logger.info("="*50) - users = self.config.get('users', []) + users = self.config_manager.get_users() if not users: logger.error("未找到用户配置") @@ -486,15 +355,14 @@ class CookieMonitor: except Exception as e: logger.error(f"处理用户失败: {e}") - for user_result in self.all_results: - pass - logger.info("\n" + "="*50) logger.info("Cookie 监控结束") logger.info("="*50) def main(): + """主函数""" + from logger import setup_logging setup_logging() monitor = CookieMonitor() monitor.run() diff --git a/notification.py b/notification.py new file mode 100644 index 0000000..a0fa744 --- /dev/null +++ b/notification.py @@ -0,0 +1,101 @@ +"""信息发送管理模块""" +import requests +from logger import get_logger + +logger = get_logger(__name__) + + +class NotificationManager: + """通知管理器""" + + def __init__(self, token: str = "", provider: str = "iyuu"): + """ + 初始化通知管理器 + + Args: + token: 通知token + provider: 通知服务提供商 (iyuu, wechat, dingtalk等) + """ + self.token = token + self.provider = provider + self.api_url = f"https://iyuu.cn/{token}.send" if provider == "iyuu" else "" + + def send(self, title: str, content: str) -> bool: + """ + 发送通知 + + Args: + title: 通知标题 + content: 通知内容 + + Returns: + 是否发送成功 + """ + if not self.token: + logger.warning("未配置通知token，跳过发送") + return False + + if self.provider == "iyuu": + return self._send_iyuu(title, content) + else: + logger.warning(f"不支持的通知提供商: {self.provider}") + return False + + def _send_iyuu(self, title: str, content: str) -> bool: + """ + 通过爱语飞飞发送通知 + + Args: + title: 通知标题 + content: 通知内容 + + Returns: + 是否发送成功 + """ + try: + logger.info(f"发送通知: {title}") + + response = requests.post( + self.api_url, + json={ + 'text': title, + 'desp': content + }, + timeout=10 + ) + + response.raise_for_status() + result = response.json() + + if result.get('errcode') == 0: + logger.info("通知发送成功") + return True + else: + logger.error(f"通知发送失败: {result.get('errmsg')}") + return False + + except requests.RequestException as e: + logger.error(f"通知请求失败: {e}") + return False + except Exception as e: + logger.error(f"通知发送异常: {e}") + return False + + +def create_notification_manager(config: dict) -> NotificationManager: + """ + 根据配置创建通知管理器 + + Args: + config: 通知配置字典 + + Returns: + 通知管理器实例 + """ + token = config.get('iyuu_token', '') + provider = config.get('provider', 'iyuu') + + if token: + return NotificationManager(token=token, provider=provider) + + return None diff --git a/notifier.py b/notifier.py deleted file mode 100644 index f428885..0000000 --- a/notifier.py +++ /dev/null @@ -1,54 +0,0 @@ -"""消息通知模块 - 通过爱语飞飞发送通知""" -import requests -from applogger import get_logger - -logger = get_logger(__name__) - - -class IYUUNotifier: - """爱语飞飞通知客户端""" - - def __init__(self, token: str): - self.token = token - self.api_url = f"https://iyuu.cn/{token}.send" - - def send(self, title: str, content: str) -> bool: - print(f"\n{'='*50}") - print(f"[测试] 准备发送通知") - print(f"[测试] API URL: {self.api_url}") - print(f"[测试] Title: {title}") - print(f"[测试] Content: {content[:100]}...") - print(f"{'='*50}\n") - - try: - response = requests.get( - self.api_url, - params={ - 'text': title, - 'desp': content - }, - timeout=10 - ) - - print(f"[测试] 响应状态码: {response.status_code}") - print(f"[测试] 响应内容: {response.text}") - - response.raise_for_status() - result = response.json() - - print(f"[测试] 解析结果: {result}") - - if result.get('errcode') == 0: - print(f"[测试] ✓ 发送成功!") - return True - else: - print(f"[测试] ✗ 发送失败: {result.get('errmsg')}") - logger.error(f"发送通知失败: {result.get('errmsg')}") - return False - - except requests.RequestException as e: - logger.error(f"发送通知请求失败: {e}") - return False - except Exception as e: - logger.error(f"发送通知异常: {e}") - return False diff --git a/send_test.py b/send_test.py deleted file mode 100644 index 465f265..0000000 --- a/send_test.py +++ /dev/null @@ -1,62 +0,0 @@ -"""直接发送测试消息给两个用户""" -from notifier import IYUUNotifier -from datetime import datetime - -iyuu_token = "IYUU37629Tc1d371c7ce99a49ff9778e196286b7e4592be191" -notifier = IYUUNotifier(iyuu_token) - -print("="*60) -print("发送测试消息给用户A") -print("="*60) - -title_a = "【测试】用户A - Cookie监控测试" -content_a = f"""**用户A Cookie监控测试报告** - -测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} -测试类型: 直接发送测试消息 - -这是一条测试消息，用于验证爱语飞飞通知功能是否正常工作。 - -如果您收到此消息，说明通知功能正常。""" - -print(f"\n发送内容:") -print(f"Title: {title_a}") -print(f"Content:\n{content_a}") -print(f"\n发送方式: 爱语飞飞 API (GET 请求)") -print(f"API URL: https://iyuu.cn/{iyuu_token}.send") -print(f"参数: text={title_a}, desp={content_a}") -print("\n正在发送...") - -result_a = notifier.send(title_a, content_a) -print(f"\n发送结果: {'✓ 成功' if result_a else '✗ 失败'}") - -print("\n" + "="*60) -print("发送测试消息给用户B") -print("="*60) - -title_b = "【测试】用户B - Cookie监控测试" -content_b = f"""**用户B Cookie监控测试报告** - -测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} -测试类型: 直接发送测试消息 - -这是一条测试消息，用于验证爱语飞飞通知功能是否正常工作。 - -如果您收到此消息，说明通知功能正常。""" - -print(f"\n发送内容:") -print(f"Title: {title_b}") -print(f"Content:\n{content_b}") -print(f"\n发送方式: 爱语飞飞 API (GET 请求)") -print(f"API URL: https://iyuu.cn/{iyuu_token}.send") -print(f"参数: text={title_b}, desp={content_b}") -print("\n正在发送...") - -result_b = notifier.send(title_b, content_b) -print(f"\n发送结果: {'✓ 成功' if result_b else '✗ 失败'}") - -print("\n" + "="*60) -print("测试完成") -print("="*60) -print(f"用户A: {'✓ 发送成功' if result_a else '✗ 发送失败'}") -print(f"用户B: {'✓ 发送成功' if result_b else '✗ 发送失败'}") diff --git a/test_notify.py b/test_notify.py deleted file mode 100644 index 2a16f0c..0000000 --- a/test_notify.py +++ /dev/null @@ -1,42 +0,0 @@ -"""测试爱语飞飞通知""" -import requests - -iyuu_token = "IYUU37629Tc1d371c7ce99a49ff9778e196286b7e4592be191" -api_url = f"https://iyuu.cn/{iyuu_token}.send" - -title = "【测试】Cookie监控通知测试" -content = """这是一条测试消息。 - -测试时间: 2026-03-04 -测试内容: 验证爱语飞飞通知是否正常工作 - -如果您收到此消息，说明通知功能正常。""" - -print(f"API URL: {api_url}") -print(f"Title: {title}") -print(f"Content: {content}") -print("\n正在发送请求...") - -try: - response = requests.get( - api_url, - params={ - 'text': title, - 'desp': content - }, - timeout=10 - ) - - print(f"状态码: {response.status_code}") - print(f"响应内容: {response.text}") - - result = response.json() - print(f"\n解析结果: {result}") - - if result.get('errcode') == 0: - print("\n✓ 发送成功！") - else: - print(f"\n✗ 发送失败: {result.get('errmsg')}") - -except Exception as e: - print(f"\n✗ 请求异常: {e}")