276 lines
8.8 KiB
Python
276 lines
8.8 KiB
Python
"""
|
||
小红书认证实现
|
||
"""
|
||
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 |