"""
登录功能实现
"""
import asyncio
from typing import Tuple, Optional
from playwright.async_api import Page
from loguru import logger
class LoginAction:
"""登录操作"""
def __init__(self, page: Page):
self.page = page
async def check_login_status(self) -> bool:
"""检查登录状态"""
logger.info("Checking login status...")
try:
await self.page.goto("https://www.xiaohongshu.com/explore")
await self.page.wait_for_load_state("load")
# 等待一小段时间让页面完全加载
await asyncio.sleep(1)
# 检查是否存在登录后的用户元素
user_element = await self.page.query_selector(
".main-container .user .link-wrapper .channel"
)
is_logged_in = user_element is not None
logger.info(f"Login status: {'logged in' if is_logged_in else 'not logged in'}")
return is_logged_in
except Exception as e:
logger.error(f"Failed to check login status: {e}")
return False
async def fetch_qrcode_image(self) -> Tuple[Optional[str], bool]:
"""
获取登录二维码图片
Returns:
Tuple[Optional[str], bool]: (二维码图片 URL, 是否已登录)
"""
logger.info("Fetching QR code image...")
try:
await self.page.goto("https://www.xiaohongshu.com/explore")
await self.page.wait_for_load_state("load")
# 等待一小段时间让页面完全加载
await asyncio.sleep(2)
# 检查是否已经登录
user_element = await self.page.query_selector(
".main-container .user .link-wrapper .channel"
)
if user_element:
logger.info("Already logged in")
return None, True
# 获取二维码图片
qrcode_element = await self.page.wait_for_selector(
".login-container .qrcode-img",
timeout=10000
)
src = await qrcode_element.get_attribute("src")
if not src:
logger.error("QR code src is empty")
return None, False
logger.info("QR code image fetched successfully")
return src, False
except Exception as e:
logger.error(f"Failed to fetch QR code image: {e}")
return None, False
async def wait_for_login(self, timeout: int = 120) -> bool:
"""
等待用户扫码登录
Args:
timeout: 超时时间（秒）
Returns:
bool: 是否登录成功
"""
logger.info(f"Waiting for login (timeout: {timeout}s)...")
try:
# 等待登录成功的元素出现
await self.page.wait_for_selector(
".main-container .user .link-wrapper .channel",
timeout=timeout * 1000
)
logger.info("Login successful!")
return True
except Exception as e:
logger.error(f"Login timeout or failed: {e}")
return False
async def login(self, timeout: int = 120) -> bool:
"""
执行登录流程
Args:
timeout: 超时时间（秒）
Returns:
bool: 是否登录成功
"""
logger.info("Starting login process...")
# 检查是否已登录
if await self.check_login_status():
logger.info("Already logged in")
return True
# 获取二维码
qr_url, is_logged_in = await self.fetch_qrcode_image()
if is_logged_in:
return True
if not qr_url:
logger.error("Failed to get QR code")
return False
logger.info(f"QR code URL: {qr_url}")
logger.info("Please scan the QR code to login...")
# 等待登录
return await self.wait_for_login(timeout)
