From fd467d9f60b70beb68a55ef38ecb7ae98855c6d2 Mon Sep 17 00:00:00 2001 From: mzl Date: Mon, 12 Jan 2026 16:58:09 +0800 Subject: [PATCH] ts --- app/web.py | 154 ++++++++++++++++++----------------------------------- 1 file changed, 51 insertions(+), 103 deletions(-) diff --git a/app/web.py b/app/web.py index dfca37c..8ce91b3 100644 --- a/app/web.py +++ b/app/web.py @@ -462,7 +462,7 @@ async def check_login_status(): """检查小红书登录状态（直接验证）""" try: # 如果浏览器可用，直接使用浏览器验证 - if hasattr(app_state, 'browser') and app_state.browser: + if app_state.browser: try: page = await app_state.browser.new_page() await page.goto("https://www.xiaohongshu.com/", timeout=10000) @@ -473,58 +473,43 @@ async def check_login_status(): page.close() # 检查是否有小红书相关的session cookie - web_session = None + session_cookies = [] for cookie in cookies: - if 'web_session' in cookie.get('name', '').lower(): - web_session = cookie - break + cookie_name = cookie.get('name', '').lower() + if any(keyword in cookie_name for keyword in ['session', 'token', 'login', 'auth']): + 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 { "success": True, "message": "已登录小红书", - "logged_in": True + "logged_in": True, + "cookies_count": len(session_cookies) } else: + print("[DEBUG] 未找到session cookies") return { "success": False, "message": "未找到登录状态，请先登录" } except Exception as e: + print(f"[DEBUG] 浏览器检查失败: {str(e)}") pass - # 如果浏览器不可用，尝试访问登录页面检查 - try: - # 使用 requests 作为备选方案 - import asyncio - import aiohttp - - 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 { - "success": True, - "message": "已登录小红书", - "logged_in": True - } - - except Exception as e: - # 如果 aiohttp 不可用，跳过此检查 - pass + # 检查本地是否有登录状态文件 + auth_file = os.path.join(config.DATA_DIR, "auth_state.json") + if os.path.exists(auth_file): + with open(auth_file, 'r', encoding='utf-8') as f: + auth_data = json.load(f) + return { + "success": True, + "message": "本地登录状态有效", + "logged_in": True, + "login_time": auth_data.get("login_time"), + "expire_time": auth_data.get("expire_time") + } return { "success": False, @@ -542,92 +527,53 @@ async def check_login_status(): async def verify_login(): """验证登录状态""" try: - # 检查浏览器是否已初始化 - print(f"[DEBUG] 浏览器状态: {app_state.browser is not None}") - - # 优先检查浏览器的实际登录状态 + # 直接检查浏览器中的cookie if app_state.browser: try: - # 创建新页面检查登录状态 - page = await app_state.browser.new_page() - print("[DEBUG] 成功创建新页面") + # 获取所有cookies + cookies = await app_state.browser.context.cookies() - # 访问小红书主页 - await page.goto("https://www.xiaohongshu.com/", timeout=10000) - print("[DEBUG] 成功访问小红书主页") + print(f"[DEBUG] 找到 {len(cookies)} 个cookies") - # 等待页面加载 - await asyncio.sleep(3) + # 检查是否有小红书的session cookie + session_cookies = [] + for cookie in cookies: + cookie_name = cookie.get('name', '').lower() + cookie_value = cookie.get('value', '') - # 获取页面标题和URL - title = await page.title() - current_url = page.url + print(f"[DEBUG] Cookie: {cookie_name} = {cookie_value[:20] if cookie_value else 'EMPTY'}") - print(f"[DEBUG] 页面标题: {title}") - print(f"[DEBUG] 当前URL: {current_url}") + if any(keyword in cookie_name for keyword in ['session', 'token', 'login', 'auth']) and cookie_value: + session_cookies.append(cookie) + print(f"[DEBUG] 找到session cookie: {cookie_name}") - # 如果被重定向到登录页面或包含登录字样 - if "登录" in title or "login" in title.lower() or "login" in current_url.lower(): - print("[DEBUG] 检测到登录页面，用户未登录") - await page.close() - return { - "success": False, - "message": "未找到登录状态，请先登录" + if session_cookies: + # 创建本地状态文件 + auth_file = os.path.join(config.DATA_DIR, "auth_state.json") + auth_data = { + "login_time": datetime.now().isoformat(), + "expire_time": (datetime.now() + timedelta(hours=24)).isoformat() } + 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 { "success": True, "message": "登录成功", "logged_in": True, - "login_time": datetime.now().isoformat(), - "expire_time": (datetime.now() + timedelta(hours=24)).isoformat(), + "login_time": auth_data["login_time"], + "expire_time": auth_data["expire_time"], "expire_in_hours": 24 } else: - print("[DEBUG] 未找到用户标识，但页面访问正常，认为已登录") - return { - "success": True, - "message": "登录成功", - "logged_in": True, - "login_time": datetime.now().isoformat(), - "expire_time": (datetime.now() + timedelta(hours=24)).isoformat(), - "expire_in_hours": 24 - } + print("[DEBUG] 未找到任何session cookies") except Exception as e: - print(f"[DEBUG] 浏览器访问出错: {str(e)}") - # 如果浏览器访问失败，回退到本地文件检查 + print(f"[DEBUG] 浏览器cookie检查失败: {str(e)}") pass - else: - print("[DEBUG] 浏览器未初始化") - # 检查本地登录状态文件（兼容处理） + # 检查本地登录状态文件 auth_file = os.path.join(config.DATA_DIR, "auth_state.json") - if os.path.exists(auth_file): with open(auth_file, 'r', encoding='utf-8') as f: auth_data = json.load(f) @@ -654,12 +600,14 @@ async def verify_login(): "expire_in_hours": 24 } + print("[DEBUG] 所有检查都失败了") return { "success": False, "message": "未找到登录状态，请先登录" } except Exception as e: + print(f"[DEBUG] 验证登录异常: {str(e)}") return { "success": False, "message": f"验证失败: {str(e)}"