import os import time from typing import Dict, List, Any, Optional from DrissionPage import ChromiumPage, ChromiumOptions, Chromium from logger_config import global_logger as logger from config import ConfigManager class LoginManager: """ 登录管理器，负责处理网站登录和会话维护 """ def __init__(self): """ 初始化登录管理器，创建浏览器对象 """ self.browser = None self.config_manager = ConfigManager() self.sites = [] # 存储包含网站和页面信息的字典列表 # 初始化浏览器对象 try: # 创建浏览器配置对象 co = ChromiumOptions() # 从配置文件中读取浏览器路径 config = self.config_manager.load_config() edge_path = config.get('browser_path', r'C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe') if os.path.exists(edge_path): # 设置浏览器路径 co.set_browser_path(edge_path) logger.info(f"已配置浏览器路径: {edge_path}") else: logger.warning(f"浏览器路径不存在: {edge_path}") logger.info("将使用默认配置，让DrissionPage自动查找浏览器") # 用该配置创建浏览器对象 self.browser = Chromium(addr_or_opts=co) logger.info("浏览器对象初始化成功") except Exception as e: logger.error(f"初始化浏览器失败: {e}") self.browser = None def save_cookies(self) -> List[Dict[str, Any]]: """ 保存所有打开网站的cookies到配置文件 Returns: List[Dict[str, Any]]: 包含所有网站cookies的列表 """ logger.info("开始保存所有网站的cookies到配置文件") all_sites_cookies = [] if not self.sites: logger.warning("没有打开的网站，无法获取cookies") return all_sites_cookies try: # 加载配置文件 config = self.config_manager.load_config() # 遍历所有网站获取cookies for index, site_info in enumerate(self.sites): try: page = site_info['page'] site_name = site_info.get('name', f'网站_{index+1}') site_url = getattr(page, 'url', site_info.get('url', f'网站_{index+1}')) page_title = getattr(page, 'title', f'网站_{index+1}') logger.info(f"获取网站 {index+1}/{len(self.sites)} 的cookies: {site_name} - {site_url}") # 获取页面的cookies cookies_list = self._get_page_cookies(page, site_name, site_url) # 将网站信息和cookies添加到结果列表 if cookies_list: all_sites_cookies.append({ 'url': site_url, 'title': page_title, 'cookies': cookies_list }) logger.info(f"成功获取网站 {site_url} 的 {len(cookies_list)} 条cookies记录") # 更新配置文件中的cookies - 使用name属性作为匹配条件 for config_site in config.get('sites', []): if config_site.get('name') == site_name: config_site['cookies'] = cookies_list logger.info(f"已更新配置文件中网站 {site_name} 的cookies") break else: logger.warning(f"未能获取网站 {site_name} - {site_url} 的cookies") except Exception as e: logger.error(f"处理网站时发生错误: {e}") continue # 使用ConfigManager保存更新后的配置 if self.config_manager.save_config(config): logger.info("已将更新后的配置保存到配置文件") else: logger.error("保存配置文件失败") except Exception as e: logger.error(f"保存cookies到配置文件时发生错误: {e}") logger.info(f"成功获取并保存 {len(all_sites_cookies)} 个网站的cookies") return all_sites_cookies def close(self, quit_browser: bool = None) -> None: """ 关闭浏览器页面，可选择是否同时关闭浏览器对象 Args: quit_browser (bool): 是否关闭浏览器对象，None表示根据配置决定 """ # 关闭所有页面 if self.sites: for page_info in reversed(self.sites): try: page = page_info['page'] if hasattr(page, 'close'): page.close() logger.info("一个标签页已关闭") except Exception as e: logger.error(f"关闭浏览器页面时发生错误: {e}") # 清空sites列表 self.sites = [] else: logger.warning("没有活动的浏览器页面需要关闭") # 决定是否关闭浏览器对象 should_quit = quit_browser # 只有当调用者没有明确指定时，才从配置中读取 if should_quit is None: try: config = self.config_manager.load_config() should_quit = config.get('AutoQuit', False) except Exception as e: logger.error(f"读取配置时出错: {e}") should_quit = False # 如果需要关闭浏览器对象 if should_quit and self.browser: try: self.browser.quit() logger.info("浏览器对象已关闭") self.browser = None except Exception as e: logger.error(f"关闭浏览器对象时发生错误: {e}") def _add_page_metadata(self, page: Any, name: str, url: str, cookies: Optional[List[Dict[str, Any]]] = None) -> None: """ 添加页面元数据到sites列表 Args: page: 页面对象 name: 网站名称 url: 网站URL cookies: cookies列表 """ self.sites.append({ 'page': page, 'name': name, 'url': url, 'original_cookies': cookies if cookies else [] }) def _set_cookies_to_page(self, page: Any, cookies: List[Dict[str, Any]] or Dict[str, str], site_name: str) -> bool: """ 为页面对象设置cookies Args: page: 页面对象 cookies: cookies列表或字典 site_name: 网站名称 Returns: bool: 设置是否成功 """ logger.info(f"为网站 '{site_name}' 设置 {len(cookies)} 条cookies记录") try: # 检查cookies类型，支持列表格式和字典格式 if isinstance(cookies, dict): # 兼容旧的字典格式，转换为列表格式 cookies_list = [] for key, value in cookies.items(): cookies_list.append({ 'name': key, 'value': value, 'domain': '', 'path': '/', 'expires': 0, 'secure': False, 'httpOnly': False }) cookies = cookies_list logger.info(f"已将字典格式cookies转换为列表格式") # 尝试多种方式设置cookies if hasattr(page, 'set') and hasattr(page.set, 'cookies'): if hasattr(page.set.cookies, 'clear'): page.set.cookies.clear() page.set.cookies(cookies) logger.info(f"成功使用set.cookies()方法设置cookies") elif hasattr(page, 'set_cookies'): page.set_cookies(cookies) logger.info("成功使用set_cookies方法设置cookies") elif hasattr(page, 'cookies'): page.cookies = cookies logger.info("成功使用page.cookies属性设置cookies") else: logger.error("无法找到设置cookies的有效方法") return False # 刷新页面以应用cookies page.refresh() logger.info(f"网站 '{site_name}' 页面已刷新以应用cookies") return True except Exception as e: logger.error(f"为网站 '{site_name}' 设置cookies失败: {e}") return False def _get_page_cookies(self, page: Any, site_name: str, page_url: Optional[str] = None) -> List[Dict[str, Any]]: """ 获取指定页面的cookies Args: page: 页面对象 site_name: 网站名称 page_url: 页面URL（可选） Returns: List[Dict[str, Any]]: cookies列表 """ cookies_list = [] cookies = None # 构建日志标识 log_id = f"网站 '{site_name}'" if not page_url else f"页面 {site_name} - {page_url}" # 尝试多种方式获取cookies try: if hasattr(page, 'get_cookies'): cookies = page.get_cookies() elif hasattr(page, 'cookies'): if callable(page.cookies): cookies = page.cookies() else: cookies = page.cookies except Exception as api_e: logger.warning(f"获取{log_id}的cookies时出错: {api_e}") # 处理获取到的cookies if cookies is not None: if isinstance(cookies, list): for cookie in cookies: if 'name' in cookie and 'value' in cookie: standard_cookie = { 'name': cookie.get('name'), 'value': cookie.get('value'), 'domain': cookie.get('domain', ''), 'path': cookie.get('path', '/'), 'expires': cookie.get('expires', 0), 'secure': cookie.get('secure', False), 'httpOnly': cookie.get('httpOnly', False) } cookies_list.append(standard_cookie) elif isinstance(cookies, dict): for key, value in cookies.items(): cookies_list.append({ 'name': key, 'value': value, 'domain': '', 'path': '/', 'expires': 0, 'secure': False, 'httpOnly': False }) # 如果还是没有获取到cookies，尝试使用JavaScript if not cookies_list: try: if hasattr(page, 'run_js'): js_cookies = page.run_js('return document.cookie') elif hasattr(page, 'execute_script'): js_cookies = page.execute_script('return document.cookie') else: js_cookies = None if js_cookies: cookie_pairs = js_cookies.split(';') for pair in cookie_pairs: pair = pair.strip() if '=' in pair: try: key, value = pair.split('=', 1) cookies_list.append({ 'name': key.strip(), 'value': value.strip(), 'domain': '', 'path': '/', 'expires': 0, 'secure': False, 'httpOnly': False }) except Exception as parse_e: logger.warning(f"解析cookie对失败: {pair}, 错误: {parse_e}") except Exception as js_e: logger.warning(f"使用JavaScript获取{log_id}的cookies时出错: {js_e}") return cookies_list def get_duration(self) -> int: """ 获取配置中的延时时间 Returns: int: 延时秒数 """ try: return self.config_manager.load_config().get('duration', 30) except Exception as e: logger.error(f"获取duration配置时发生错误: {e}") return 30 # 返回默认值 def run(self) -> None: """ 从配置文件中读取网站信息，为每个网站新建标签页并加载cookies """ logger.info("开始运行自动登录流程") # 检查浏览器对象是否初始化 if not self.browser: logger.error("浏览器对象未初始化，无法运行自动登录流程") return try: # 从配置文件中读取网站信息 config = self.config_manager.load_config() sites = config.get('sites', []) logger.info(f"获取到 {len(sites)} 个网站配置") if not sites: logger.warning("配置文件中没有网站信息") return logger.info(f"找到 {len(sites)} 个网站需要登录") # 存储所有创建的标签页 tabs = [] # 为每个网站创建标签页并设置cookies for site in sites: try: site_name = site.get('name', '未命名网站') site_url = site.get('url', '') site_cookies = site.get('cookies', []) if not site_url: logger.warning(f"网站 '{site_name}' 缺少URL，跳过") continue logger.info(f"开始处理网站: {site_name} ({site_url})") # 创建新标签页 try: new_tab = self.browser.new_tab(site_url) logger.info(f"为网站 '{site_name}' 创建了新标签页") tabs.append(new_tab) # 添加网站元数据 self._add_page_metadata(new_tab, site_name, site_url, site_cookies) except Exception as e: logger.error(f"为网站 '{site_name}' 创建标签页失败: {e}") continue # 设置cookies if site_cookies: self._set_cookies_to_page(new_tab, site_cookies, site_name) # 等待页面加载完成 logger.info("页面加载中...") time.sleep(2) # 仅保留短暂延时确保页面加载完成 logger.info(f"网站 '{site_name}' 处理完成") except Exception as e: logger.error(f"处理网站时发生错误: {e}") continue logger.info(f"所有网站处理完成，共创建了 {len(tabs)} 个标签页") # 所有标签页已通过_add_page_metadata添加到sites列表中 if tabs: logger.info("已创建并记录所有标签页") except Exception as e: logger.error(f"运行自动登录流程时发生错误: {e}")