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

# -*- coding: utf-8 -*-
"""
抖音视频发布器
参考: matrix/douyin_uploader/main.py
"""

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


class DouyinPublisher(BasePublisher):
    """
    抖音视频发布器
    使用 Playwright 自动化操作抖音创作者中心
    """
    
    platform_name = "douyin"
    login_url = "https://creator.douyin.com/"
    publish_url = "https://creator.douyin.com/creator-micro/content/upload"
    cookie_domain = ".douyin.com"
    
    async def set_schedule_time(self, publish_date: datetime):
        """设置定时发布"""
        if not self.page:
            return
        
        # 选择定时发布
        label_element = self.page.locator("label.radio-d4zkru: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 handle_upload_error(self, video_path: str):
        """处理上传错误，重新上传"""
        if not self.page:
            return
        
        print(f"[{self.platform_name}] 视频出错了，重新上传中...")
        await self.page.locator('div.progress-div [class^="upload-btn-input"]').set_input_files(video_path)
    
    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}] Headless: {self.headless}")
        print(f"{'='*60}")
        
        self.report_progress(5, "正在初始化浏览器...")
        
        # 初始化浏览器
        await self.init_browser()
        print(f"[{self.platform_name}] 浏览器初始化完成")
        
        # 解析并设置 cookies
        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")
        
        # 检查视频文件
        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(10, "正在打开上传页面...")
        
        # 访问上传页面
        await self.page.goto(self.publish_url)
        await self.page.wait_for_url(self.publish_url, timeout=30000)
        
        self.report_progress(15, "正在选择视频文件...")
        
        # 点击上传区域
        upload_div = self.page.locator("div[class*='container-drag']").first
        async with self.page.expect_file_chooser() as fc_info:
            await upload_div.click()
        file_chooser = await fc_info.value
        await file_chooser.set_files(params.video_path)
        
        # 等待跳转到发布页面
        self.report_progress(20, "等待进入发布页面...")
        for _ in range(60):
            try:
                await self.page.wait_for_url(
                    "https://creator.douyin.com/creator-micro/content/post/video*",
                    timeout=2000
                )
                break
            except:
                await asyncio.sleep(1)
        
        await asyncio.sleep(2)
        self.report_progress(30, "正在填充标题和话题...")
        
        # 填写标题
        title_input = self.page.get_by_text('作品标题').locator("..").locator(
            "xpath=following-sibling::div[1]").locator("input")
        if await title_input.count():
            await title_input.fill(params.title[:30])
        else:
            # 备用方式
            title_container = self.page.locator(".notranslate")
            await title_container.click()
            await self.page.keyboard.press("Control+KeyA")
            await self.page.keyboard.press("Delete")
            await self.page.keyboard.type(params.title)
            await self.page.keyboard.press("Enter")
        
        # 添加话题标签
        if params.tags:
            css_selector = ".zone-container"
            for tag in params.tags:
                print(f"[{self.platform_name}] 添加话题: #{tag}")
                await self.page.type(css_selector, "#" + tag)
                await self.page.press(css_selector, "Space")
        
        self.report_progress(40, "等待视频上传完成...")
        
        # 等待视频上传完成
        for _ in range(120):
            try:
                count = await self.page.locator("div").filter(has_text="重新上传").count()
                if count > 0:
                    print(f"[{self.platform_name}] 视频上传完毕")
                    break
                
                # 检查上传错误
                if await self.page.locator('div.progress-div > div:has-text("上传失败")').count():
                    await self.handle_upload_error(params.video_path)
                
                await asyncio.sleep(3)
            except:
                await asyncio.sleep(3)
        
        self.report_progress(60, "处理视频设置...")
        
        # 关闭弹窗
        known_btn = self.page.get_by_role("button", name="我知道了")
        if await known_btn.count() > 0:
            await known_btn.first.click()
        
        await asyncio.sleep(2)
        
        # 设置位置
        try:
            await self.page.locator('div.semi-select span:has-text("输入地理位置")').click()
            await asyncio.sleep(1)
            await self.page.keyboard.press("Control+KeyA")
            await self.page.keyboard.press("Delete")
            await self.page.keyboard.type(params.location)
            await asyncio.sleep(1)
            await self.page.locator('div[role="listbox"] [role="option"]').first.click()
        except Exception as e:
            print(f"[{self.platform_name}] 设置位置失败: {e}")
        
        # 开启头条/西瓜同步
        try:
            third_part_element = '[class^="info"] > [class^="first-part"] div div.semi-switch'
            if await self.page.locator(third_part_element).count():
                class_name = await self.page.eval_on_selector(
                    third_part_element, 'div => div.className')
                if 'semi-switch-checked' not in class_name:
                    await self.page.locator(third_part_element).locator(
                        'input.semi-switch-native-control').click()
        except:
            pass
        
        # 定时发布
        if params.publish_date:
            self.report_progress(70, "设置定时发布...")
            await self.set_schedule_time(params.publish_date)
        
        self.report_progress(80, "正在发布...")
        print(f"[{self.platform_name}] 查找发布按钮...")
        
        # 点击发布
        publish_clicked = False
        for i in range(30):
            try:
                publish_btn = self.page.get_by_role('button', name="发布", exact=True)
                btn_count = await publish_btn.count()
                print(f"[{self.platform_name}] 发布按钮数量: {btn_count}")
                
                if btn_count > 0:
                    print(f"[{self.platform_name}] 点击发布按钮...")
                    await publish_btn.click()
                    publish_clicked = True
                
                await self.page.wait_for_url(
                    "https://creator.douyin.com/creator-micro/content/manage",
                    timeout=5000
                )
                self.report_progress(100, "发布成功")
                print(f"[{self.platform_name}] 发布成功! 已跳转到内容管理页面")
                return PublishResult(
                    success=True,
                    platform=self.platform_name,
                    message="发布成功"
                )
            except Exception as e:
                current_url = self.page.url
                print(f"[{self.platform_name}] 尝试 {i+1}/30, 当前URL: {current_url}")
                
                if "content/manage" in current_url:
                    self.report_progress(100, "发布成功")
                    print(f"[{self.platform_name}] 发布成功! 已在内容管理页面")
                    return PublishResult(
                        success=True,
                        platform=self.platform_name,
                        message="发布成功"
                    )
                
                # 检查是否有错误提示
                try:
                    error_toast = self.page.locator('[class*="toast"][class*="error"], [class*="error-tip"]')
                    if await error_toast.count() > 0:
                        error_text = await error_toast.first.text_content()
                        if error_text:
                            print(f"[{self.platform_name}] 检测到错误提示: {error_text}")
                            raise Exception(f"发布失败: {error_text}")
                except:
                    pass
                
                await asyncio.sleep(1)
        
        # 发布超时，保存截图
        screenshot_path = f"debug_publish_timeout_{self.platform_name}.png"
        await self.page.screenshot(path=screenshot_path, full_page=True)
        print(f"[{self.platform_name}] 发布超时，截图保存到: {screenshot_path}")
        raise Exception(f"发布超时（截图: {screenshot_path}）")