# -*- coding: utf-8 -*-
|
from playwright.async_api import Playwright, Page
|
import asyncio
|
import random
|
|
from auto_media_publisher.uploaders.uploader_base import UploaderBase
|
|
CONTENT_UPLOAD_URL="https://baijiahao.baidu.com/builder/rc/edit?type=videoV2"
|
VIDEO_PUBLISH_SECCESSFULL_URL = "https://baijiahao.baidu.com/builder/rc/clue**"
|
|
|
class UploaderBaijiahao(UploaderBase):
|
|
async def set_schedule_time(self, page):
|
"""
|
TODO 时间选择，日后在处理 百家号的时间选择不准确，目前是随机
|
"""
|
publish_datetime_day = f"{self.publish_datetime.month}月{self.publish_datetime.day}日" if self.publish_datetime.day >9 else f"{self.publish_datetime.month}月0{self.publish_datetime.day}日"
|
publish_datetime_hour = f"{self.publish_datetime.hour}点"
|
publish_datetime_min = f"{self.publish_datetime.minute}分"
|
await page.locator("div.cheetah-modal-confirm-body-wrapper").wait_for(state='visible',timeout=30000)
|
|
try:
|
await page.locator('div.select-wrap').nth(0).click()
|
await asyncio.sleep(1)
|
target = page.locator(f'div[title="{publish_datetime_day}"]')
|
await target.wait_for(state='visible')
|
await target.click()
|
await asyncio.sleep(1)
|
await page.locator('div.select-wrap').nth(1).click()
|
await asyncio.sleep(1)
|
target = page.locator(f'div.cheetah-select-item-option-content:has-text("{publish_datetime_hour}")')
|
await target.wait_for(state='visible')
|
await target.click()
|
await asyncio.sleep(1)
|
await page.locator('div.select-wrap').nth(2).click()
|
await asyncio.sleep(1)
|
target = page.locator(f'div.cheetah-select-item-option-content:has-text("{publish_datetime_min}")')
|
await target.wait_for(state='visible')
|
await target.click()
|
await asyncio.sleep(1)
|
|
except Exception as e:
|
self.logger.error(f"[Baidu BaiJiaHao] 设置时间发生错误.{e}")
|
|
await page.locator("div.cheetah-modal-confirm-btns > button").nth(1).click()
|
|
async def handle_upload_error(self, page):
|
# 日后实现，目前没遇到
|
return
|
print("视频出错了，重新上传中")
|
|
async def uploading_video(self, page):
|
while True:
|
upload_failed = await page.locator('div .cover-overlay:has-text("上传失败")').count()
|
if upload_failed:
|
self.logger.error("发现上传出错了...")
|
# await self.handle_upload_error(page) # 假设这是处理上传错误的函数
|
return False
|
|
uploading = await page.locator('div .cover-overlay:has-text("上传中")').count()
|
if uploading:
|
self.logger.info("正在上传视频中...")
|
await asyncio.sleep(2) # 等待2秒再次检查
|
continue
|
|
# 检查上传是否成功
|
if not uploading and not upload_failed:
|
self.logger.info("视频上传完毕")
|
return True
|
|
async def set_schedule_publish(self, page):
|
while True:
|
try:
|
self.logger.info("开始点击发布定时...")
|
buttons = page.get_by_role("button", name="定时发布")
|
count = await buttons.count()
|
for i in range(count):
|
btn = buttons.nth(i)
|
if await btn.is_visible():
|
await btn.click()
|
break
|
await asyncio.sleep(2) # 等待2秒再次检查
|
await self.set_schedule_time(page)
|
except Exception as e:
|
self.logger.error(f"定时发布失败: {e}")
|
raise # 重新抛出异常，让重试装饰器捕获
|
|
async def publish_video(self, page: Page):
|
if self.publish_datetime != 0:
|
# 定时发布
|
await self.set_schedule_publish(page)
|
else:
|
# 立即发布
|
await self.direct_publish(page)
|
|
async def direct_publish(self, page):
|
try:
|
publish_button = page.locator("button >> text=发布")
|
if await publish_button.count():
|
await publish_button.click()
|
except Exception as e:
|
self.logger.error(f"直接发布视频失败: {e}")
|
raise # 重新抛出异常，让重试装饰器捕获
|
|
async def add_title_tags(self, page):
|
title_page = page.get_by_placeholder('添加标题获得更多推荐')
|
if len(self.title) <= 8:
|
self.title += " 你不知道的"
|
await title_page.fill(self.title[:30])
|
|
|
async def _upload_core(self) -> bool:
|
try:
|
|
page = await self.context.new_page()
|
await page.goto(CONTENT_UPLOAD_URL, timeout=120000)
|
self.logger.info(f"正在上传-------{self.path_video}")
|
await page.wait_for_url(CONTENT_UPLOAD_URL, timeout=120000)
|
|
# 点击 "上传视频" 按钮
|
await page.locator("div[class^='video-main-page'] input").set_input_files(self.path_video)
|
|
# 等待页面跳转到指定的 URL
|
while True:
|
# 判断是是否进入视频发布页面，没进入，则自动等待到超时
|
try:
|
await page.locator("div#formMain:visible").wait_for(timeout=30000)
|
break
|
except:
|
self.logger.info("正在等待进入视频发布页面...")
|
await asyncio.sleep(0.1)
|
|
# 填充标题和话题
|
# 这里为了避免页面变化，故使用相对位置定位：作品标题父级右侧第一个元素的input子元素
|
await asyncio.sleep(1)
|
self.logger.info("正在填充标题和话题...")
|
await self.add_title_tags(page)
|
|
upload_status = await self.uploading_video(page)
|
if not upload_status:
|
self.logger.error(f"发现上传出错了... 文件:{self.path_video}")
|
raise
|
|
# 判断视频封面图是否生成成功
|
while True:
|
self.logger.info("正在确认封面完成, 准备去点击定时/发布...")
|
if await page.locator("div.cheetah-spin-page img").count():
|
self.logger.info("封面已完成，点击定时/发布...")
|
break
|
else:
|
self.logger.info("等待封面生成...")
|
await asyncio.sleep(3)
|
|
await self.publish_video(page)
|
await page.wait_for_timeout(2000)
|
if await page.locator('div.passMod_dialog-page >> text=百度安全验证:visible').count():
|
self.logger.error("出现验证，退出")
|
raise Exception("出现验证，退出")
|
await page.wait_for_url(VIDEO_PUBLISH_SECCESSFULL_URL, timeout=20000)
|
self.logger.info("视频发布成功")
|
|
await self.context.storage_state(path=self.cookie_path) # 保存cookie
|
self.logger.info('cookie更新完毕！')
|
await asyncio.sleep(2) # 这里延迟是为了方便眼睛直观的观看
|
|
return True
|
except Exception as e:
|
self.logger.error(f"{self.user} upload video {self.path_video} fails, error is :{e}")
|
return False
|
