xinyouhe/xhs_crawler_claude
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ts

Browse Source
This commit is contained in:
mzl
2026-01-12 16:58:09 +08:00
parent 03ddd4c628
commit fd467d9f60
1 changed files with 51 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/web.py
+48 -100
View File
@@ -462,7 +462,7 @@ async def check_login_status():
"""检查小红书登录状态（直接验证）""" """检查小红书登录状态（直接验证）"""
try: try:
# 如果浏览器可用，直接使用浏览器验证 # 如果浏览器可用，直接使用浏览器验证
if hasattr(app_state, 'browser') and app_state.browser: if app_state.browser:
try: try:
page = await app_state.browser.new_page() page = await app_state.browser.new_page()
await page.goto("https://www.xiaohongshu.com/", timeout=10000) await page.goto("https://www.xiaohongshu.com/", timeout=10000)
@@ -473,59 +473,44 @@ async def check_login_status():
page.close() page.close()
# 检查是否有小红书相关的session cookie # 检查是否有小红书相关的session cookie
web_session = None session_cookies = []
for cookie in cookies: for cookie in cookies:
if 'web_session' in cookie.get('name', '').lower(): cookie_name = cookie.get('name', '').lower()
web_session = cookie if any(keyword in cookie_name for keyword in ['session', 'token', 'login', 'auth']):
break session_cookies.append(cookie)
print(f"[DEBUG] 找到session cookie: {cookie.get('name')} = {cookie.get('value', '')[:20]}...")
if web_session and web_session.get('value'): if session_cookies:
return { return {
"success": True, "success": True,
"message": "已登录小红书", "message": "已登录小红书",
"logged_in": True "logged_in": True,
"cookies_count": len(session_cookies)
} }
else: else:
print("[DEBUG] 未找到session cookies")
return { return {
"success": False, "success": False,
"message": "未找到登录状态，请先登录" "message": "未找到登录状态，请先登录"
} }
except Exception as e: except Exception as e:
print(f"[DEBUG] 浏览器检查失败: {str(e)}")
pass pass
# 如果浏览器不可用，尝试访问登录页面检查 # 检查本地是否有登录状态文件
try: auth_file = os.path.join(config.DATA_DIR, "auth_state.json")
# 使用 requests 作为备选方案 if os.path.exists(auth_file):
import asyncio with open(auth_file, 'r', encoding='utf-8') as f:
import aiohttp auth_data = json.load(f)
async with aiohttp.ClientSession() as session:
async with session.get("https://www.xiaohongshu.com/") as response:
if response.status == 200:
# 检查响应中是否有登录相关内容
content = await response.text()
# 如果页面包含登录标识，说明未登录
if any(keyword in content for keyword in ['登录', 'login', 'signin']):
return {
"success": False,
"message": "未找到登录状态，请先登录"
}
# 尝试获取cookies
cookies = session.cookie_jar
if cookies:
return { return {
"success": True, "success": True,
"message": "已登录小红书", "message": "本地登录状态有效",
"logged_in": True "logged_in": True,
"login_time": auth_data.get("login_time"),
"expire_time": auth_data.get("expire_time")
} }
except Exception as e:
# 如果 aiohttp 不可用，跳过此检查
pass
return { return {
"success": False, "success": False,
"message": "未找到登录状态，请先登录" "message": "未找到登录状态，请先登录"
@@ -542,92 +527,53 @@ async def check_login_status():
async def verify_login(): async def verify_login():
"""验证登录状态""" """验证登录状态"""
try: try:
# 检查浏览器是否已初始化 # 直接检查浏览器中的cookie
print(f"[DEBUG] 浏览器状态: {app_state.browser is not None}")
# 优先检查浏览器的实际登录状态
if app_state.browser: if app_state.browser:
try: try:
# 创建新页面检查登录状态 # 获取所有cookies
page = await app_state.browser.new_page() cookies = await app_state.browser.context.cookies()
print("[DEBUG] 成功创建新页面")
# 访问小红书主页 print(f"[DEBUG] 找到 {len(cookies)} 个cookies")
await page.goto("https://www.xiaohongshu.com/", timeout=10000)
print("[DEBUG] 成功访问小红书主页")
# 等待页面加载 # 检查是否有小红书的session cookie
await asyncio.sleep(3) session_cookies = []
for cookie in cookies:
cookie_name = cookie.get('name', '').lower()
cookie_value = cookie.get('value', '')
# 获取页面标题和URL print(f"[DEBUG] Cookie: {cookie_name} = {cookie_value[:20] if cookie_value else 'EMPTY'}")
title = await page.title()
current_url = page.url
print(f"[DEBUG] 页面标题: {title}") if any(keyword in cookie_name for keyword in ['session', 'token', 'login', 'auth']) and cookie_value:
print(f"[DEBUG] 当前URL: {current_url}") session_cookies.append(cookie)
print(f"[DEBUG] 找到session cookie: {cookie_name}")
# 如果被重定向到登录页面或包含登录字样 if session_cookies:
if "登录" in title or "login" in title.lower() or "login" in current_url.lower(): # 创建本地状态文件
print("[DEBUG] 检测到登录页面，用户未登录") auth_file = os.path.join(config.DATA_DIR, "auth_state.json")
await page.close() auth_data = {
return { "login_time": datetime.now().isoformat(),
"success": False, "expire_time": (datetime.now() + timedelta(hours=24)).isoformat()
"message": "未找到登录状态，请先登录"
} }
with open(auth_file, 'w', encoding='utf-8') as f:
json.dump(auth_data, f, ensure_ascii=False, indent=2)
# 尝试查找已登录的用户标识
selectors = [
'img[src*="avatar"]',
'[class*="avatar"]',
'[class*="user-avatar"]',
'[class*="profile-avatar"]',
'.user-name',
'[class*="username"]',
'.user-nick-name',
'[class*="nick-name"]'
]
found_user_elements = False
for selector in selectors:
elements = await page.query_selector_all(selector)
if elements:
print(f"[DEBUG] 找到用户标识: {selector}, 数量: {len(elements)}")
found_user_elements = True
break
await page.close()
if found_user_elements:
print("[DEBUG] 检测到用户标识，认为已登录")
return { return {
"success": True, "success": True,
"message": "登录成功", "message": "登录成功",
"logged_in": True, "logged_in": True,
"login_time": datetime.now().isoformat(), "login_time": auth_data["login_time"],
"expire_time": (datetime.now() + timedelta(hours=24)).isoformat(), "expire_time": auth_data["expire_time"],
"expire_in_hours": 24 "expire_in_hours": 24
} }
else: else:
print("[DEBUG] 未找到用户标识，但页面访问正常，认为已登录") print("[DEBUG] 未找到任何session cookies")
return {
"success": True,
"message": "登录成功",
"logged_in": True,
"login_time": datetime.now().isoformat(),
"expire_time": (datetime.now() + timedelta(hours=24)).isoformat(),
"expire_in_hours": 24
}
except Exception as e: except Exception as e:
print(f"[DEBUG] 浏览器访问出错: {str(e)}") print(f"[DEBUG] 浏览器cookie检查失败: {str(e)}")
# 如果浏览器访问失败，回退到本地文件检查
pass pass
else:
print("[DEBUG] 浏览器未初始化")
# 检查本地登录状态文件（兼容处理） # 检查本地登录状态文件
auth_file = os.path.join(config.DATA_DIR, "auth_state.json") auth_file = os.path.join(config.DATA_DIR, "auth_state.json")
if os.path.exists(auth_file): if os.path.exists(auth_file):
with open(auth_file, 'r', encoding='utf-8') as f: with open(auth_file, 'r', encoding='utf-8') as f:
auth_data = json.load(f) auth_data = json.load(f)
@@ -654,12 +600,14 @@ async def verify_login():
"expire_in_hours": 24 "expire_in_hours": 24
} }
print("[DEBUG] 所有检查都失败了")
return { return {
"success": False, "success": False,
"message": "未找到登录状态，请先登录" "message": "未找到登录状态，请先登录"
} }
except Exception as e: except Exception as e:
print(f"[DEBUG] 验证登录异常: {str(e)}")
return { return {
"success": False, "success": False,
"message": f"验证失败: {str(e)}" "message": f"验证失败: {str(e)}"