# -*- coding: utf-8 -*-
|
from playwright.async_api import Playwright, Page
|
import asyncio
|
import datetime
|
|
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 virtual_scroll_select(self,page,text, wrapper_index = 0,panel_selector="div.rc-virtual-list-holder-inner",
|
item_selector_template='div.cheetah-select-item-option-content:has-text("{text}")',
|
step=200, # 每次滚动像素
|
max_attempts=15,
|
direction="down" # "down" 或 "up"
|
):
|
await page.locator('div.select-wrap').nth(wrapper_index).click()
|
await asyncio.sleep(1)
|
|
panel = page.locator(panel_selector).nth(wrapper_index)
|
item_selector = item_selector_template.format(text=text)
|
|
|
for i in range(max_attempts):
|
# 尝试查找
|
items = page.locator(item_selector)
|
if await items.count() > 0:
|
# 找到了，滚下去保证可见，再点击
|
await items.first.scroll_into_view_if_needed()
|
await items.first.click()
|
return
|
|
# 没找到就滚一轮
|
delta = step if direction == "down" else -step
|
await panel.evaluate(f"(el) => el.scrollTop += {delta}")
|
await asyncio.sleep(0.2)
|
await asyncio.sleep(1)
|
|
|
async def set_schedule_time(self, page):
|
"""
|
TODO 时间选择，日后在处理 百家号的时间选择不准确，目前是随机
|
"""
|
scheduled_time_day = f"{self.scheduled_time.month}月{self.scheduled_time.day}日" if self.scheduled_time.day >9 else f"{self.scheduled_time.month}月0{self.scheduled_time.day}日"
|
scheduled_time_hour = f"{self.scheduled_time.hour}点"
|
scheduled_time_min = f"{self.scheduled_time.minute}分"
|
await page.locator("div.cheetah-modal-confirm-body-wrapper").wait_for(state='visible',timeout=30000)
|
|
current_hour = datetime.datetime.now().hour
|
current_min = datetime.datetime.now().minute
|
direction_hour = "down" if self.scheduled_time.hour > current_hour else "up"
|
direction_minute = "down" if self.scheduled_time.minute > current_min else "up"
|
|
print(direction_hour)
|
print(dir)
|
|
|
try:
|
await page.locator('div.select-wrap').nth(0).click()
|
await asyncio.sleep(1)
|
target = page.locator(f'div[title="{scheduled_time_day}"]')
|
#await target.wait_for(state='visible')
|
await target.click()
|
await asyncio.sleep(1)
|
await self.virtual_scroll_select(page,scheduled_time_hour,wrapper_index=1,direction=direction_hour)
|
await self.virtual_scroll_select(page,scheduled_time_min,wrapper_index=2,direction=direction_minute)
|
#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("{scheduled_time_min}")')
|
#await target.wait_for(state='visible')
|
#await target.scroll_into_view_if_needed()
|
#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.scheduled_time != 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_media}")
|
await page.wait_for_url(CONTENT_UPLOAD_URL, timeout=120000)
|
|
# 点击 "上传视频" 按钮
|
upload_div = page.locator(".PL3VsCg3MxIPY5UYiCS_")
|
await upload_div.wait_for(state="visible", timeout=10000)
|
async with page.expect_file_chooser() as fc_info:
|
await upload_div.click()
|
file_chooser = await fc_info.value
|
await file_chooser.set_files(self.path_media)
|
await asyncio.sleep(2)
|
|
# 等待页面跳转到指定的 URL
|
for i in range(10):
|
try:
|
await page.locator("div#formMain:visible").wait_for(timeout=30000)
|
break
|
except:
|
self.logger.info(f"第 {i+1} 次尝试：正在等待进入视频发布页面...")
|
await asyncio.sleep(0.1)
|
else:
|
self.logger.warning("超过最大尝试次数，仍未进入视频发布页面。")
|
|
# 填充标题和话题
|
# 这里为了避免页面变化，故使用相对位置定位：作品标题父级右侧第一个元素的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_media}")
|
raise
|
|
# 判断视频封面图是否生成成功
|
while True:
|
self.logger.info("正在确认封面完成, 准备去点击定时/发布...")
|
if await page.locator("div.erVanTLKMOvnU7OgFaoD 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_media} fails, error is :{e}")
|
return False
|
