""" 小红书认证实现 """ import asyncio from typing import Optional from playwright.async_api import Page from .base_auth import BaseAuth from ..core.models import PlatformType, AccountInfo from ..config.platform_config import get_platform_url, get_selectors, get_wait_times from ..utils.logger import get_platform_logger from ..utils.exceptions import LoginFailedError, AuthenticationError class XiaoHongShuAuth(BaseAuth): """小红书认证实现""" def __init__(self): super().__init__(PlatformType.XIAOHONGSHU) self.login_url = get_platform_url(PlatformType.XIAOHONGSHU, "login") self.logger = get_platform_logger("xiaohongshu") async def login(self, account_info: AccountInfo, headless: bool = False) -> bool: """ 执行小红书登录流程 Args: account_info: 账号信息 headless: 是否使用无头模式 Returns: 登录是否成功 """ try: self.logger.info(f"开始小红书登录: {account_info.username}") # 创建页面 page = await self.create_page(account_info, headless) # 尝试加载现有Cookie cookies_loaded = await self.load_cookies(page, account_info) if cookies_loaded: # 验证Cookie是否有效 if await self._validate_existing_session(page): self.logger.info("使用现有Cookie登录成功") return True else: self.logger.info("Cookie已失效，需要重新登录") # 执行扫码登录 success = await self._perform_qr_login(page, account_info) if success: # 保存新的Cookie await self.save_cookies(page, account_info) self.logger.success("小红书登录完成") else: self.logger.failure("小红书登录失败") # 关闭页面 await page.close() return success except Exception as e: self.logger.error(f"小红书登录异常: {e}") raise LoginFailedError(f"小红书登录失败: {e}", platform="xiaohongshu") async def check_login_status(self, page: Page) -> bool: """ 检查小红书登录状态 Args: page: Playwright页面对象 Returns: 是否已登录 """ try: selectors = get_selectors(PlatformType.XIAOHONGSHU, "login", "login_success_indicators") for selector in selectors: try: element = await page.wait_for_selector(selector, timeout=3000) if element and await element.is_visible(): self.logger.debug(f"检测到登录状态指示器: {selector}") return True except: continue # 检查URL是否包含登录相关页面 current_url = page.url if "login" not in current_url.lower() and "auth" not in current_url.lower(): # 尝试检测用户信息元素 user_selectors = [ "[data-testid='user-info']", ".user-name", ".avatar img", "[class*='user'] [class*='avatar']" ] for selector in user_selectors: try: element = await page.wait_for_selector(selector, timeout=2000) if element and await element.is_visible(): self.logger.debug(f"检测到用户信息元素: {selector}") return True except: continue return False except Exception as e: self.logger.warning(f"登录状态检查失败: {e}") return False async def _validate_existing_session(self, page: Page) -> bool: """ 验证现有会话是否有效 Args: page: Playwright页面对象 Returns: 会话是否有效 """ try: # 导航到创作者中心首页 await self.safe_goto(page, self.login_url) await asyncio.sleep(3) # 检查是否需要登录 if await self.check_login_status(page): self.logger.debug("现有会话有效") return True else: self.logger.debug("现有会话无效，需要重新登录") return False except Exception as e: self.logger.warning(f"会话验证失败: {e}") return False async def _perform_qr_login(self, page: Page, account_info: AccountInfo) -> bool: """ 执行扫码登录流程 Args: page: Playwright页面对象 account_info: 账号信息 Returns: 登录是否成功 """ try: self.logger.info("开始扫码登录流程") # 导航到登录页面 if not await self.safe_goto(page, self.login_url): raise LoginFailedError("无法访问登录页面") # 等待二维码加载 qr_selectors = get_selectors(PlatformType.XIAOHONGSHU, "login", "qr_code") qr_code_loaded = False for selector in qr_selectors: if await self.wait_for_element(page, selector, timeout=15000): qr_code_loaded = True self.logger.info("二维码已加载，请使用小红书App扫码") break if not qr_code_loaded: # 尝试其他可能的二维码选择器 alternative_selectors = [ "img[src*='qr']", "canvas", ".qrcode", "[class*='qr']", "[id*='qr']" ] for selector in alternative_selectors: if await self.wait_for_element(page, selector, timeout=5000): qr_code_loaded = True self.logger.info(f"检测到二维码元素: {selector}") break if not qr_code_loaded: self.logger.warning("未检测到二维码，可能已经登录") # 检查是否已经登录 if await self.check_login_status(page): return True else: raise LoginFailedError("二维码加载失败") # 等待用户扫码登录 wait_times = get_wait_times(PlatformType.XIAOHONGSHU, "login") login_timeout = wait_times.get("login_timeout", 300) success = await self.wait_for_login_success(page, timeout=login_timeout) if success: self.logger.info("扫码登录成功") return True else: raise LoginFailedError("扫码登录超时") except Exception as e: self.logger.error(f"扫码登录失败: {e}") return False async def force_relogin(self, account_info: AccountInfo, headless: bool = False) -> bool: """ 强制重新登录 Args: account_info: 账号信息 headless: 是否使用无头模式 Returns: 登录是否成功 """ try: self.logger.info("强制重新登录") # 删除现有Cookie self.delete_cookies(account_info) # 执行登录 return await self.login(account_info, headless) except Exception as e: self.logger.error(f"强制重新登录失败: {e}") return False async def test_login(self, account_info: AccountInfo) -> bool: """ 测试登录状态 Args: account_info: 账号信息 Returns: 登录是否有效 """ try: self.logger.info("测试登录状态") # 创建页面 page = await self.create_page(account_info, headless=True) # 加载Cookie cookies_loaded = await self.load_cookies(page, account_info) if not cookies_loaded: self.logger.info("没有找到Cookie文件") await page.close() return False # 验证登录状态 login_valid = await self._validate_existing_session(page) await page.close() if login_valid: self.logger.info("登录状态有效") return True else: self.logger.info("登录状态无效") return False except Exception as e: self.logger.error(f"登录状态测试失败: {e}") return False