import os import time from typing import Dict, List, Any from DrissionPage import ChromiumPage, ChromiumOptions from logger_config import global_logger as logger from config import ConfigManager class LoginManager: """ 登录管理器，负责处理网站登录和会话维护 """ def __init__(self): """ 初始化登录管理器 """ self.page = None self.config_manager = ConfigManager() def login(self, site_url: str, cookies: List[Dict[str, Any]] = None) -> bool: """ 登录网站 Args: site_url (str): 网站URL cookies (List[Dict[str, Any]], optional): 预先设置的cookies记录列表 Returns: bool: 登录是否成功 """ logger.info(f"开始登录网站: {site_url}") try: # 根据DrissionPage 4.x文档，使用ChromiumOptions配置浏览器 logger.info("正在配置浏览器...") # 创建浏览器配置对象 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): # 根据文档，使用set_browser_path方法设置浏览器路径 co.set_browser_path(edge_path) logger.info(f"已配置Edge浏览器路径: {edge_path}") else: logger.warning(f"Edge浏览器路径不存在: {edge_path}") logger.info("将使用默认配置，让DrissionPage自动查找浏览器") # 使用配置创建ChromiumPage实例 # 根据文档，在DrissionPage 4.x中，应该使用正确的方式初始化 self.page = ChromiumPage(co) # 访问网站 logger.info(f"正在加载网站: {site_url}") self.page.get(site_url) # 如果提供了cookies，设置它们 if cookies: logger.info(f"设置 {len(cookies)} 条cookies记录") # 检查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 try: # 首先尝试使用set.cookies()方法（DrissionPage 4.x推荐方式） if hasattr(self.page, 'set') and hasattr(self.page.set, 'cookies'): if hasattr(self.page.set.cookies, 'clear'): self.page.set.cookies.clear() self.page.set.cookies(cookies) logger.info(f"成功使用set.cookies()方法设置cookies") # 备选方案1：使用set_cookies方法 elif hasattr(self.page, 'set_cookies'): self.page.set_cookies(cookies) logger.info("成功使用set_cookies方法设置cookies") # 备选方案2：使用page.cookies属性 elif hasattr(self.page, 'cookies'): self.page.cookies = cookies logger.info("成功使用page.cookies属性设置cookies") else: logger.error("无法找到设置cookies的有效方法") except Exception as e: logger.error(f"设置cookies失败: {e}") # 刷新页面以应用cookies try: self.page.refresh() except Exception as e: logger.warning(f"刷新页面时出错: {e}") # 等待页面加载完成 time.sleep(2) # 验证是否登录成功 current_url = self.page.url logger.info(f"当前URL: {current_url}") # 如果能成功加载页面，暂时认为登录成功 return True except Exception as e: error_msg = str(e) logger.error(f"登录网站时发生错误: {error_msg}") # 针对浏览器路径错误提供详细指导 if "无法找到浏览器可执行文件路径" in error_msg: logger.warning("===== 浏览器配置错误指导 =====") logger.warning("DrissionPage无法找到系统中的浏览器") logger.warning("请尝试以下解决方法:") logger.warning("1. 确保Microsoft Edge浏览器已正确安装") logger.warning("2. Edge浏览器默认安装路径: C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe") logger.warning("3. 检查Edge浏览器是否在系统PATH环境变量中") logger.warning("4. 或者尝试安装Chrome浏览器") logger.warning("5. 确保用户有足够的权限运行浏览器") logger.warning("==========================") # 安全关闭浏览器 if self.page: try: self.page.close() except: pass self.page = None return False def maintain_login(self, duration: int = 120) -> None: """ 保持登录状态 Args: duration (int): 保持时间（秒） """ if not self.page: logger.warning("没有活动的浏览器页面，无法保持登录状态") return logger.info(f"保持登录状态 {duration} 秒...") try: # 定期刷新页面以保持会话活跃 interval = min(30, duration) # 最多每30秒刷新一次 elapsed = 0 while elapsed < duration: time.sleep(interval) elapsed += interval # 刷新页面 try: self.page.refresh() logger.info(f"页面已刷新，已保持登录 {elapsed} 秒") except Exception as e: logger.error(f"刷新页面时发生错误: {e}") break except KeyboardInterrupt: logger.info("保持登录状态被用户中断") except Exception as e: logger.error(f"保持登录状态时发生错误: {e}") def get_cookies(self) -> List[Dict[str, Any]]: """ 获取当前页面的cookies Returns: List[Dict[str, Any]]: cookies记录列表 """ if not self.page: logger.warning("没有活动的浏览器页面，无法获取cookies") return [] try: cookies_list = [] # 根据DrissionPage文档，尝试多种方式获取cookies cookies = None # 首先尝试使用标准API获取cookies try: # 尝试使用get_cookies方法 if hasattr(self.page, 'get_cookies'): logger.info("尝试使用page.get_cookies()方法获取cookies") cookies = self.page.get_cookies() # 尝试使用page.cookies属性 elif hasattr(self.page, 'cookies'): logger.info("尝试使用page.cookies属性获取cookies") # 检查是否是可调用对象 if callable(self.page.cookies): cookies = self.page.cookies() else: cookies = self.page.cookies else: logger.error("无法找到获取cookies的有效方法") return [] except Exception as api_e: logger.warning(f"使用标准API获取cookies失败: {api_e}，尝试其他方法") # 处理获取到的cookies if cookies is not None: if isinstance(cookies, list): # 如果已经是列表格式，确保每条记录包含必要字段 for cookie in cookies: # 确保cookie记录包含必要的字段 if 'name' in cookie and 'value' in cookie: # 标准化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，尝试使用execute_script方法 if not cookies_list: try: logger.info("尝试使用JavaScript获取cookies") if hasattr(self.page, 'run_js'): js_cookies = self.page.run_js('return document.cookie') elif hasattr(self.page, 'execute_script'): js_cookies = self.page.execute_script('return document.cookie') else: logger.warning("无法找到执行JavaScript的方法") js_cookies = None if js_cookies: # 解析JavaScript返回的cookie字符串 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获取cookies失败: {js_e}") logger.info(f"成功获取 {len(cookies_list)} 条cookies记录") return cookies_list except Exception as e: logger.error(f"获取cookies时发生错误: {e}") return [] def close(self) -> None: """ 关闭浏览器页面 """ if self.page: try: self.page.browser.quit() # self.page.close() # self.page = None logger.info("浏览器页面已关闭") except Exception as e: logger.error(f"关闭浏览器页面时发生错误: {e}") else: logger.warning("没有活动的浏览器页面需要关闭")