# -*- 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="百家号评论功能暂未实现" )