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 发布...")
        
        # 转换 cookie 格式
        cookie_list = self.parse_cookies(cookies)
        cookie_string = self.cookies_to_string(cookie_list) if cookie_list else cookies
        
        self.report_progress(20, "正在上传视频...")
        
        # 创建客户端
        xhs_client = XhsClient(cookie_string, sign=self.sign_sync)
        
        # 发布视频
        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
        )
        
        self.report_progress(100, "发布成功")
        
        return PublishResult(
            success=True,
            platform=self.platform_name,
            video_id=result.get("note_id", ""),
            video_url=result.get("url", ""),
            message="发布成功"
        )
    
    async def publish(self, cookies: str, params: PublishParams) -> PublishResult:
        """发布视频到小红书"""
        # 检查视频文件
        if not os.path.exists(params.video_path):
            raise Exception(f"视频文件不存在: {params.video_path}")
        
        self.report_progress(5, "正在准备发布...")
        
        # 优先使用 API 方式
        if XHS_SDK_AVAILABLE:
            try:
                return await self.publish_via_api(cookies, params)
            except Exception as e:
                print(f"[{self.platform_name}] API 发布失败: {e}")
                print(f"[{self.platform_name}] 尝试使用 Playwright 方式...")
        
        # 回退到 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, "正在初始化浏览器...")
        
        await self.init_browser()
        
        cookie_list = self.parse_cookies(cookies)
        await self.set_cookies(cookie_list)
        
        if not self.page:
            raise Exception("Page not initialized")
        
        self.report_progress(15, "正在打开发布页面...")
        
        await self.page.goto(self.publish_url)
        await asyncio.sleep(3)
        
        # 检查登录状态
        if "login" in self.page.url or "passport" in self.page.url:
            raise Exception("登录已过期，请重新登录")
        
        self.report_progress(20, "正在上传视频...")
        
        # 尝试点击视频标签
        try:
            video_tab = self.page.locator('div.tab:has-text("上传视频"), span:has-text("上传视频")').first
            if await video_tab.count() > 0:
                await video_tab.click()
                await asyncio.sleep(1)
        except:
            pass
        
        # 上传视频
        upload_triggered = False
        
        # 方法1: 点击上传按钮
        try:
            upload_btn = self.page.locator('button:has-text("上传视频")').first
            if await upload_btn.count() > 0:
                async with self.page.expect_file_chooser() as fc_info:
                    await upload_btn.click()
                file_chooser = await fc_info.value
                await file_chooser.set_files(params.video_path)
                upload_triggered = True
        except:
            pass
        
        # 方法2: 直接设置 file input
        if not upload_triggered:
            file_input = await self.page.$('input[type="file"]')
            if file_input:
                await file_input.set_input_files(params.video_path)
                upload_triggered = True
        
        if not upload_triggered:
            raise Exception("无法上传视频文件")
        
        self.report_progress(40, "等待视频上传完成...")
        
        # 等待上传完成
        for _ in range(100):
            await asyncio.sleep(3)
            # 检查标题输入框是否出现
            title_input = await self.page.locator('input[placeholder*="标题"]').count()
            if title_input > 0:
                break
        
        self.report_progress(60, "正在填写笔记信息...")
        
        # 填写标题
        title_selectors = [
            'input[placeholder*="标题"]',
            '[class*="title"] input',
        ]
        for selector in title_selectors:
            title_input = self.page.locator(selector).first
            if await title_input.count() > 0:
                await title_input.fill(params.title[:20])
                break
        
        # 填写描述和标签
        if params.description or params.tags:
            desc_selectors = [
                '[class*="content-input"] [contenteditable="true"]',
                '[class*="editor"] [contenteditable="true"]',
            ]
            for selector in desc_selectors:
                desc_input = self.page.locator(selector).first
                if await desc_input.count() > 0:
                    await desc_input.click()
                    if params.description:
                        await self.page.keyboard.type(params.description, delay=30)
                    if params.tags:
                        await self.page.keyboard.press("Enter")
                        for tag in params.tags:
                            await self.page.keyboard.type(f"#{tag} ", delay=30)
                    break
        
        self.report_progress(80, "正在发布...")
        
        await asyncio.sleep(2)
        
        # 点击发布
        publish_selectors = [
            'button.publishBtn',
            '.publishBtn',
            'button:has-text("发布")',
        ]
        
        for selector in publish_selectors:
            btn = self.page.locator(selector).first
            if await btn.count() > 0 and await btn.is_visible():
                box = await btn.bounding_box()
                if box:
                    await self.page.mouse.click(box['x'] + box['width']/2, box['y'] + box['height']/2)
                    break
        
        await asyncio.sleep(5)
        
        self.report_progress(100, "发布完成")
        
        return PublishResult(
            success=True,
            platform=self.platform_name,
            message="发布完成"
        )