multi-platform-media-manage/server/python/platforms/xiaohongshu.py

# -*- 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="发布完成"
        )