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

# -*- coding: utf-8 -*-
"""
快手视频发布器
参考: matrix/ks_uploader/main.py
"""

import asyncio
import os
from datetime import datetime
from .base import BasePublisher, PublishParams, PublishResult


class KuaishouPublisher(BasePublisher):
    """
    快手视频发布器
    使用 Playwright 自动化操作快手创作者中心
    """
    
    platform_name = "kuaishou"
    login_url = "https://cp.kuaishou.com/"
    publish_url = "https://cp.kuaishou.com/article/publish/video"
    cookie_domain = ".kuaishou.com"
    
    async def set_schedule_time(self, publish_date: datetime):
        """设置定时发布"""
        if not self.page:
            return
        
        # 选择定时发布
        label_element = self.page.locator("label.radio--4Gpx6:has-text('定时发布')")
        await label_element.click()
        await asyncio.sleep(1)
        
        # 输入时间
        publish_date_str = publish_date.strftime("%Y-%m-%d %H:%M")
        await self.page.locator('.semi-input[placeholder="日期和时间"]').click()
        await self.page.keyboard.press("Control+KeyA")
        await self.page.keyboard.type(str(publish_date_str))
        await self.page.keyboard.press("Enter")
        await asyncio.sleep(1)
    
    async def upload_cover(self, cover_path: str):
        """上传封面图"""
        if not self.page or not cover_path or not os.path.exists(cover_path):
            return
        
        try:
            await self.page.get_by_role("button", name="编辑封面").click()
            await asyncio.sleep(1)
            await self.page.get_by_role("tab", name="上传封面").click()
            
            preview_div = self.page.get_by_role("tabpanel", name="上传封面").locator("div").nth(1)
            async with self.page.expect_file_chooser() as fc_info:
                await preview_div.click()
            preview_chooser = await fc_info.value
            await preview_chooser.set_files(cover_path)
            
            await self.page.get_by_role("button", name="确认").click()
            await asyncio.sleep(3)
            
            print(f"[{self.platform_name}] 封面上传成功")
        except Exception as e:
            print(f"[{self.platform_name}] 封面上传失败: {e}")
    
    async def publish(self, cookies: str, params: PublishParams) -> PublishResult:
        """发布视频到快手"""
        self.report_progress(5, "正在初始化浏览器...")
        
        # 初始化浏览器
        await self.init_browser()
        
        # 解析并设置 cookies
        cookie_list = self.parse_cookies(cookies)
        await self.set_cookies(cookie_list)
        
        if not self.page:
            raise Exception("Page not initialized")
        
        # 检查视频文件
        if not os.path.exists(params.video_path):
            raise Exception(f"视频文件不存在: {params.video_path}")
        
        self.report_progress(10, "正在打开上传页面...")
        
        # 访问上传页面
        await self.page.goto(self.publish_url)
        await self.page.wait_for_url(self.publish_url, timeout=30000)
        
        self.report_progress(15, "正在选择视频文件...")
        
        # 点击上传按钮
        upload_btn = self.page.get_by_role("button", name="上传视频")
        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)
        
        await asyncio.sleep(1)
        
        # 关闭可能的弹窗
        known_btn = self.page.get_by_role("button", name="我知道了")
        if await known_btn.count():
            await known_btn.click()
        
        self.report_progress(20, "正在填充标题...")
        
        # 填写标题
        await asyncio.sleep(1)
        title_input = self.page.get_by_placeholder('添加合适的话题和描述，作品能获得更多推荐～')
        if await title_input.count():
            await title_input.click()
            await title_input.fill(params.title[:30])
        
        self.report_progress(30, "等待视频上传完成...")
        
        # 等待上传完成
        for _ in range(120):
            try:
                count = await self.page.locator('span:has-text("上传成功")').count()
                if count > 0:
                    print(f"[{self.platform_name}] 视频上传完毕")
                    break
                await asyncio.sleep(3)
            except:
                await asyncio.sleep(3)
        
        self.report_progress(50, "正在上传封面...")
        
        # 上传封面
        await self.upload_cover(params.cover_path)
        
        # 定时发布（快手暂不支持或选择器有变化）
        # if params.publish_date:
        #     await self.set_schedule_time(params.publish_date)
        
        self.report_progress(80, "正在发布...")
        
        # 点击发布
        for _ in range(30):
            try:
                publish_btn = self.page.get_by_role('button', name="发布", exact=True)
                if await publish_btn.count():
                    await publish_btn.click()
                await self.page.wait_for_url(
                    "https://cp.kuaishou.com/article/manage/video*",
                    timeout=5000
                )
                self.report_progress(100, "发布成功")
                return PublishResult(
                    success=True,
                    platform=self.platform_name,
                    message="发布成功"
                )
            except:
                current_url = self.page.url
                if "manage/video" in current_url:
                    self.report_progress(100, "发布成功")
                    return PublishResult(
                        success=True,
                        platform=self.platform_name,
                        message="发布成功"
                    )
                await asyncio.sleep(1)
        
        raise Exception("发布超时")