# -*- coding: utf-8 -*- """ 小红书视频发布器 参考: matrix/xhs_uploader/main.py 使用 xhs SDK API 方式发布，更稳定 """ import asyncio import os import sys from pathlib import Path from .base import BasePublisher, PublishParams, PublishResult # 添加 matrix 项目路径，用于导入签名脚本 MATRIX_PATH = Path(__file__).parent.parent.parent.parent / "matrix" sys.path.insert(0, str(MATRIX_PATH)) # 尝试导入 xhs SDK try: from xhs import XhsClient XHS_SDK_AVAILABLE = True except ImportError: print("[Warning] xhs 库未安装，请运行: pip install xhs") XhsClient = None XHS_SDK_AVAILABLE = False # 签名脚本路径 STEALTH_JS_PATH = MATRIX_PATH / "xhs-api" / "js" / "stealth.min.js" class XiaohongshuPublisher(BasePublisher): """ 小红书视频发布器 优先使用 xhs SDK API 方式发布 """ platform_name = "xiaohongshu" login_url = "https://creator.xiaohongshu.com/" publish_url = "https://creator.xiaohongshu.com/publish/publish" cookie_domain = ".xiaohongshu.com" async def get_sign(self, uri: str, data=None, a1: str = "", web_session: str = ""): """获取小红书 API 签名""" from playwright.async_api import async_playwright try: async with async_playwright() as playwright: browser = await playwright.chromium.launch(headless=True) browser_context = await browser.new_context() if STEALTH_JS_PATH.exists(): await browser_context.add_init_script(path=str(STEALTH_JS_PATH)) page = await browser_context.new_page() await page.goto("https://www.xiaohongshu.com") await asyncio.sleep(1) await page.reload() await asyncio.sleep(1) if a1: await browser_context.add_cookies([ {'name': 'a1', 'value': a1, 'domain': ".xiaohongshu.com", 'path': "/"} ]) await page.reload() await asyncio.sleep(0.5) encrypt_params = await page.evaluate( "([url, data]) => window._webmsxyw(url, data)", [uri, data] ) await browser_context.close() await browser.close() return { "x-s": encrypt_params["X-s"], "x-t": str(encrypt_params["X-t"]) } except Exception as e: import traceback traceback.print_exc() raise Exception(f"签名失败: {e}") def sign_sync(self, uri, data=None, a1="", web_session=""): """同步签名函数，供 XhsClient 使用""" return asyncio.run(self.get_sign(uri, data, a1, web_session)) async def publish_via_api(self, cookies: str, params: PublishParams) -> PublishResult: """通过 API 发布视频""" if not XHS_SDK_AVAILABLE: raise Exception("xhs SDK 未安装，请运行: pip install xhs") self.report_progress(10, "正在通过 API 发布...") print(f"[{self.platform_name}] 使用 XHS SDK API 发布...") print(f"[{self.platform_name}] 视频路径: {params.video_path}") print(f"[{self.platform_name}] 标题: {params.title}") # 转换 cookie 格式 cookie_list = self.parse_cookies(cookies) cookie_string = self.cookies_to_string(cookie_list) if cookie_list else cookies print(f"[{self.platform_name}] Cookie 长度: {len(cookie_string)}") self.report_progress(20, "正在上传视频...") # 创建客户端 xhs_client = XhsClient(cookie_string, sign=self.sign_sync) print(f"[{self.platform_name}] 开始调用 create_video_note...") # 发布视频 try: result = xhs_client.create_video_note( title=params.title, desc=params.description or params.title, topics=params.tags or [], post_time=params.publish_date.strftime("%Y-%m-%d %H:%M:%S") if params.publish_date else None, video_path=params.video_path, cover_path=params.cover_path if params.cover_path and os.path.exists(params.cover_path) else None ) print(f"[{self.platform_name}] SDK 返回结果: {result}") except Exception as e: import traceback traceback.print_exc() print(f"[{self.platform_name}] SDK 调用失败: {e}") raise Exception(f"XHS SDK 发布失败: {e}") # 验证返回结果 if not result: raise Exception("XHS SDK 返回空结果") # 检查是否有错误 if isinstance(result, dict): if result.get("code") and result.get("code") != 0: raise Exception(f"发布失败: {result.get('msg', '未知错误')}") if result.get("success") == False: raise Exception(f"发布失败: {result.get('msg', result.get('error', '未知错误'))}") note_id = result.get("note_id", "") if isinstance(result, dict) else "" video_url = result.get("url", "") if isinstance(result, dict) else "" if not note_id: print(f"[{self.platform_name}] 警告: 未获取到 note_id，返回结果: {result}") self.report_progress(100, "发布成功") print(f"[{self.platform_name}] 发布成功! note_id={note_id}, url={video_url}") return PublishResult( success=True, platform=self.platform_name, video_id=note_id, video_url=video_url, message="发布成功" ) async def publish(self, cookies: str, params: PublishParams) -> PublishResult: """发布视频到小红书""" print(f"\n{'='*60}") print(f"[{self.platform_name}] 开始发布视频") print(f"[{self.platform_name}] 视频路径: {params.video_path}") print(f"[{self.platform_name}] 标题: {params.title}") print(f"[{self.platform_name}] XHS SDK 可用: {XHS_SDK_AVAILABLE}") print(f"{'='*60}") # 检查视频文件 if not os.path.exists(params.video_path): raise Exception(f"视频文件不存在: {params.video_path}") print(f"[{self.platform_name}] 视频文件存在，大小: {os.path.getsize(params.video_path)} bytes") self.report_progress(5, "正在准备发布...") # 临时禁用 API 方式，直接使用 Playwright（更稳定） # TODO: 后续优化 API 方式的返回值验证 # if XHS_SDK_AVAILABLE: # try: # result = await self.publish_via_api(cookies, params) # print(f"[{self.platform_name}] API 发布完成: {result}") # return result # except Exception as e: # import traceback # traceback.print_exc() # print(f"[{self.platform_name}] API 发布失败: {e}") # print(f"[{self.platform_name}] 尝试使用 Playwright 方式...") # 使用 Playwright 方式发布（更可靠） print(f"[{self.platform_name}] 使用 Playwright 方式发布...") return await self.publish_via_playwright(cookies, params) async def publish_via_playwright(self, cookies: str, params: PublishParams) -> PublishResult: """通过 Playwright 发布视频""" self.report_progress(10, "正在初始化浏览器...") print(f"[{self.platform_name}] Playwright 方式开始...") await self.init_browser() cookie_list = self.parse_cookies(cookies) print(f"[{self.platform_name}] 设置 {len(cookie_list)} 个 cookies") await self.set_cookies(cookie_list) if not self.page: raise Exception("Page not initialized") self.report_progress(15, "正在打开发布页面...") # 直接访问视频发布页面 publish_url = "https://creator.xiaohongshu.com/publish/publish?source=official" print(f"[{self.platform_name}] 打开页面: {publish_url}") await self.page.goto(publish_url) await asyncio.sleep(3) current_url = self.page.url print(f"[{self.platform_name}] 当前 URL: {current_url}") # 检查登录状态 if "login" in current_url or "passport" in current_url: screenshot_path = f"debug_login_required_{self.platform_name}.png" await self.page.screenshot(path=screenshot_path) raise Exception(f"登录已过期，请重新登录（截图: {screenshot_path}）") self.report_progress(20, "正在上传视频...") # 等待页面加载 await asyncio.sleep(2) # 上传视频 upload_triggered = False # 方法1: 直接设置隐藏的 file input print(f"[{self.platform_name}] 尝试方法1: 设置 file input") file_inputs = self.page.locator('input[type="file"]') input_count = await file_inputs.count() print(f"[{self.platform_name}] 找到 {input_count} 个 file input") if input_count > 0: # 找到接受视频的 input for i in range(input_count): input_el = file_inputs.nth(i) accept = await input_el.get_attribute('accept') or '' print(f"[{self.platform_name}] Input {i} accept: {accept}") if 'video' in accept or '*' in accept or not accept: await input_el.set_input_files(params.video_path) upload_triggered = True print(f"[{self.platform_name}] 视频文件已设置到 input {i}") break # 方法2: 点击上传区域触发文件选择器 if not upload_triggered: print(f"[{self.platform_name}] 尝试方法2: 点击上传区域") try: upload_area = self.page.locator('[class*="upload-wrapper"], [class*="upload-area"], .upload-input').first if await upload_area.count() > 0: async with self.page.expect_file_chooser(timeout=5000) as fc_info: await upload_area.click() file_chooser = await fc_info.value await file_chooser.set_files(params.video_path) upload_triggered = True print(f"[{self.platform_name}] 通过点击上传区域上传成功") except Exception as e: print(f"[{self.platform_name}] 方法2失败: {e}") if not upload_triggered: screenshot_path = f"debug_upload_failed_{self.platform_name}.png" await self.page.screenshot(path=screenshot_path) raise Exception(f"无法上传视频文件（截图: {screenshot_path}）") self.report_progress(40, "等待视频上传完成...") print(f"[{self.platform_name}] 等待视频上传和处理...") # 等待上传完成（检测页面变化） upload_complete = False for i in range(60): # 最多等待3分钟 await asyncio.sleep(3) # 检查是否有标题输入框（上传完成后出现） title_input_count = await self.page.locator('input[placeholder*="标题"], input[placeholder*="填写标题"]').count() # 或者检查编辑器区域 editor_count = await self.page.locator('[class*="ql-editor"], [contenteditable="true"]').count() # 检查发布按钮是否可见 publish_btn_count = await self.page.locator('.publishBtn, button:has-text("发布")').count() print(f"[{self.platform_name}] 检测 {i+1}: 标题框={title_input_count}, 编辑器={editor_count}, 发布按钮={publish_btn_count}") if title_input_count > 0 or (editor_count > 0 and publish_btn_count > 0): upload_complete = True print(f"[{self.platform_name}] 视频上传完成!") break if not upload_complete: screenshot_path = f"debug_upload_timeout_{self.platform_name}.png" await self.page.screenshot(path=screenshot_path) raise Exception(f"视频上传超时（截图: {screenshot_path}）") await asyncio.sleep(2) self.report_progress(60, "正在填写笔记信息...") print(f"[{self.platform_name}] 填写标题: {params.title[:20]}") # 填写标题 title_filled = False title_selectors = [ 'input[placeholder*="标题"]', 'input[placeholder*="填写标题"]', '[class*="title"] input', '.c-input_inner', ] for selector in title_selectors: title_input = self.page.locator(selector).first if await title_input.count() > 0: await title_input.click() await title_input.fill('') # 先清空 await title_input.fill(params.title[:20]) title_filled = True print(f"[{self.platform_name}] 标题已填写，使用选择器: {selector}") break if not title_filled: print(f"[{self.platform_name}] 警告: 未找到标题输入框") # 填写描述和标签 if params.description or params.tags: desc_filled = False desc_selectors = [ '[class*="ql-editor"]', '[class*="content-input"] [contenteditable="true"]', '[class*="editor"] [contenteditable="true"]', '.ql-editor', ] for selector in desc_selectors: desc_input = self.page.locator(selector).first if await desc_input.count() > 0: await desc_input.click() await asyncio.sleep(0.5) if params.description: await self.page.keyboard.type(params.description, delay=20) print(f"[{self.platform_name}] 描述已填写") if params.tags: # 添加标签 await self.page.keyboard.press("Enter") for tag in params.tags[:5]: # 最多5个标签 await self.page.keyboard.type(f"#{tag}", delay=20) await asyncio.sleep(0.3) await self.page.keyboard.press("Space") print(f"[{self.platform_name}] 标签已填写: {params.tags[:5]}") desc_filled = True break if not desc_filled: print(f"[{self.platform_name}] 警告: 未找到描述输入框") await asyncio.sleep(2) self.report_progress(80, "正在发布...") await asyncio.sleep(2) # 滚动到页面底部确保发布按钮可见 await self.page.evaluate("window.scrollTo(0, document.body.scrollHeight)") await asyncio.sleep(1) print(f"[{self.platform_name}] 查找发布按钮...") # 点击发布 publish_selectors = [ 'button.publishBtn', '.publishBtn', 'button.d-button.red', 'button:has-text("发布"):not(:has-text("定时发布"))', '[class*="publish"][class*="btn"]', ] publish_clicked = False for selector in publish_selectors: try: btn = self.page.locator(selector).first if await btn.count() > 0: is_visible = await btn.is_visible() is_enabled = await btn.is_enabled() print(f"[{self.platform_name}] 按钮 {selector}: visible={is_visible}, enabled={is_enabled}") if is_visible and is_enabled: box = await btn.bounding_box() if box: print(f"[{self.platform_name}] 点击发布按钮: {selector}, 位置: ({box['x']}, {box['y']})") # 使用真实鼠标点击 await self.page.mouse.click(box['x'] + box['width']/2, box['y'] + box['height']/2) publish_clicked = True break except Exception as e: print(f"[{self.platform_name}] 选择器 {selector} 错误: {e}") if not publish_clicked: # 保存截图用于调试 screenshot_path = f"debug_publish_failed_{self.platform_name}.png" await self.page.screenshot(path=screenshot_path, full_page=True) print(f"[{self.platform_name}] 未找到发布按钮，截图保存到: {screenshot_path}") # 打印页面 HTML 结构用于调试 buttons = await self.page.query_selector_all('button') print(f"[{self.platform_name}] 页面上共有 {len(buttons)} 个按钮") for i, btn in enumerate(buttons[:10]): text = await btn.text_content() or '' cls = await btn.get_attribute('class') or '' print(f" 按钮 {i}: text='{text.strip()[:30]}', class='{cls[:50]}'") raise Exception("未找到发布按钮") print(f"[{self.platform_name}] 已点击发布按钮，等待发布完成...") self.report_progress(90, "等待发布结果...") # 等待发布完成（检测 URL 变化或成功提示） publish_success = False for i in range(20): # 最多等待 20 秒 await asyncio.sleep(1) current_url = self.page.url # 检查是否跳转到发布成功页面或内容管理页面 if "published=true" in current_url or "success" in current_url or "content" in current_url: publish_success = True print(f"[{self.platform_name}] 发布成功! 跳转到: {current_url}") break # 检查是否有成功提示 try: success_msg = await self.page.locator('[class*="success"], .toast-success, [class*="Toast"]').first.is_visible() if success_msg: publish_success = True print(f"[{self.platform_name}] 检测到成功提示!") break except: pass # 检查是否有错误提示 try: error_elements = self.page.locator('[class*="error"], .toast-error, [class*="fail"]') if await error_elements.count() > 0: error_text = await error_elements.first.text_content() if error_text and len(error_text.strip()) > 0: raise Exception(f"发布失败: {error_text.strip()}") except Exception as e: if "发布失败" in str(e): raise # 如果没有明确的成功标志，保存截图 if not publish_success: final_url = self.page.url print(f"[{self.platform_name}] 发布结果不确定，当前 URL: {final_url}") screenshot_path = f"debug_publish_result_{self.platform_name}.png" await self.page.screenshot(path=screenshot_path, full_page=True) print(f"[{self.platform_name}] 截图保存到: {screenshot_path}") # 如果 URL 还是发布页面，可能发布失败 if "publish/publish" in final_url: raise Exception(f"发布可能失败，仍停留在发布页面（截图: {screenshot_path}）") self.report_progress(100, "发布完成") print(f"[{self.platform_name}] Playwright 方式发布完成!") return PublishResult( success=True, platform=self.platform_name, message="发布完成" )