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

# -*- coding: utf-8 -*-
"""
百家号视频发布器
"""

import asyncio
import json
from typing import List
from datetime import datetime
from .base import (
    BasePublisher, PublishParams, PublishResult,
    WorkItem, WorksResult, CommentItem, CommentsResult
)


class BaijiahaoPublisher(BasePublisher):
    """
    百家号视频发布器
    使用 Playwright 自动化操作百家号创作者中心
    """
    
    platform_name = "baijiahao"
    login_url = "https://baijiahao.baidu.com/"
    publish_url = "https://baijiahao.baidu.com/builder/rc/edit?type=video"
    cookie_domain = ".baidu.com"
    
    # 登录检测配置
    login_check_url = "https://baijiahao.baidu.com/builder/rc/home"
    login_indicators = ["passport.baidu.com", "/login", "wappass.baidu.com"]
    login_selectors = ['text="登录"', 'text="请登录"', '[class*="login-btn"]']
    
    async def get_account_info(self, cookies: str) -> dict:
        """
        获取百家号账号信息
        通过调用 settingInfo API 获取用户信息
        """
        print(f"\n{'='*60}")
        print(f"[{self.platform_name}] 获取账号信息")
        print(f"{'='*60}")
        
        try:
            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")
            
            # 访问百家号后台首页
            print(f"[{self.platform_name}] 访问后台首页...")
            await self.page.goto(self.login_check_url, wait_until="domcontentloaded", timeout=30000)
            await asyncio.sleep(3)
            
            # 检查登录状态
            current_url = self.page.url
            print(f"[{self.platform_name}] 当前 URL: {current_url}")
            
            for indicator in self.login_indicators:
                if indicator in current_url:
                    print(f"[{self.platform_name}] 检测到登录页面，Cookie 已失效")
                    return {
                        "success": False,
                        "error": "Cookie 已失效，需要重新登录",
                        "need_login": True
                    }
            
            # 调用 settingInfo API 获取用户信息
            print(f"[{self.platform_name}] 调用 settingInfo API...")
            api_result = await self.page.evaluate('''
                async () => {
                    try {
                        const response = await fetch('https://baijiahao.baidu.com/user-ui/cms/settingInfo', {
                            method: 'GET',
                            credentials: 'include',
                            headers: {
                                'Accept': 'application/json, text/plain, */*'
                            }
                        });
                        return await response.json();
                    } catch (e) {
                        return { error: e.message };
                    }
                }
            ''')
            
            print(f"[{self.platform_name}] API 响应: errno={api_result.get('errno')}")
            
            if api_result.get('error'):
                return {
                    "success": False,
                    "error": api_result.get('error')
                }
            
            if api_result.get('errno') == 0 and api_result.get('data'):
                data = api_result['data']
                account_info = {
                    "success": True,
                    "account_id": str(data.get('new_uc_id', '')) or f"baijiahao_{int(datetime.now().timestamp() * 1000)}",
                    "account_name": data.get('name', '') or '百家号账号',
                    "avatar_url": data.get('avatar', ''),
                    "fans_count": 0,  # 百家号 API 不直接返回粉丝数
                    "works_count": 0,
                }
                print(f"[{self.platform_name}] 获取成功: {account_info['account_name']}")
                return account_info
            else:
                error_msg = api_result.get('errmsg', '未知错误')
                print(f"[{self.platform_name}] API 返回错误: {error_msg}")
                
                # 如果是登录相关错误，标记需要重新登录
                if api_result.get('errno') in [10000010, 10001401]:
                    return {
                        "success": False,
                        "error": error_msg,
                        "need_login": True
                    }
                
                return {
                    "success": False,
                    "error": error_msg
                }
            
        except Exception as e:
            import traceback
            traceback.print_exc()
            return {
                "success": False,
                "error": str(e)
            }
        finally:
            await self.close_browser()
    
    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"{'='*60}")
        
        # TODO: 实现百家号视频发布逻辑
        return PublishResult(
            success=False,
            platform=self.platform_name,
            error="百家号发布功能暂未实现"
        )
    
    async def get_works(self, cookies: str, page: int = 0, page_size: int = 20) -> WorksResult:
        """获取百家号作品列表"""
        print(f"\n{'='*60}")
        print(f"[{self.platform_name}] 获取作品列表")
        print(f"[{self.platform_name}] page={page}, page_size={page_size}")
        print(f"{'='*60}")
        
        works: List[WorkItem] = []
        total = 0
        has_more = False
        
        try:
            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")
            
            # 访问内容管理页面
            await self.page.goto("https://baijiahao.baidu.com/builder/rc/content", wait_until="domcontentloaded", timeout=30000)
            await asyncio.sleep(3)
            
            # 检查登录状态
            current_url = self.page.url
            for indicator in self.login_indicators:
                if indicator in current_url:
                    raise Exception("Cookie 已过期，请重新登录")
            
            # 调用作品列表 API
            cursor = page * page_size
            api_result = await self.page.evaluate(f'''
                async () => {{
                    try {{
                        const response = await fetch('https://baijiahao.baidu.com/pcui/article/lists?start={cursor}&count={page_size}&article_type=video', {{
                            method: 'GET',
                            credentials: 'include',
                            headers: {{
                                'Accept': 'application/json'
                            }}
                        }});
                        return await response.json();
                    }} catch (e) {{
                        return {{ error: e.message }};
                    }}
                }}
            ''')
            
            print(f"[{self.platform_name}] API 响应: {json.dumps(api_result, ensure_ascii=False)[:200]}")
            
            if api_result.get('errno') == 0:
                article_list = api_result.get('data', {}).get('article_list', [])
                has_more = api_result.get('data', {}).get('has_more', False)
                
                for article in article_list:
                    work_id = str(article.get('article_id', ''))
                    if not work_id:
                        continue
                    
                    works.append(WorkItem(
                        work_id=work_id,
                        title=article.get('title', ''),
                        cover_url=article.get('cover_images', [''])[0] if article.get('cover_images') else '',
                        duration=0,
                        status='published',
                        publish_time=article.get('publish_time', ''),
                        play_count=int(article.get('read_count', 0)),
                        like_count=int(article.get('like_count', 0)),
                        comment_count=int(article.get('comment_count', 0)),
                        share_count=int(article.get('share_count', 0)),
                    ))
                
                total = len(works)
            
            print(f"[{self.platform_name}] 获取到 {total} 个作品")
            
        except Exception as e:
            import traceback
            traceback.print_exc()
            return WorksResult(
                success=False,
                platform=self.platform_name,
                error=str(e)
            )
        finally:
            await self.close_browser()
        
        return WorksResult(
            success=True,
            platform=self.platform_name,
            works=works,
            total=total,
            has_more=has_more
        )
    
    async def get_comments(self, cookies: str, work_id: str, cursor: str = "") -> CommentsResult:
        """获取百家号作品评论"""
        # TODO: 实现评论获取逻辑
        return CommentsResult(
            success=False,
            platform=self.platform_name,
            work_id=work_id,
            error="百家号评论功能暂未实现"
        )