# -*- coding: utf-8 -*-
|
from playwright.async_api import Page
|
import asyncio
|
from auto_media_publisher.uploaders.uploader_base import UploaderBase
|
|
CONTENT_UPLOAD_URL="https://member.bilibili.com/platform/upload/video/frame"
|
VIDEO_PUBLISH_SECCESSFULL_URL = "https://cp.kuaishou.com/article/manage/video?status=2&from=publish"
|
|
|
class UploaderBilibili(UploaderBase):
|
|
async def set_schedule_time(self, page):
|
try:
|
await page.locator(".switch-roll").first.click()
|
await asyncio.sleep(1)
|
|
scheduled_date = self.scheduled_time.strftime("%Y-%m-%d")
|
scheduled_month = self.scheduled_time.strftime("%Y年%m月")
|
exact_day = str(self.scheduled_time.day)
|
|
await page.locator(".date-show-icon").first.click()
|
await asyncio.sleep(1)
|
if scheduled_month != await page.locator("p.date-picker-nav-title").text_content():
|
await page.locator(".next-btn-day.icon-sprite.icon-sprite-ic_into").click()
|
await asyncio.sleep(1)
|
await page.get_by_text(exact_day, exact=True).click()
|
await asyncio.sleep(1)
|
|
exact_hour = f"{self.scheduled_time.hour:02d}"
|
exact_min = f"{self.scheduled_time.minute:02d}"
|
await page.locator(".date-picker-timer > .date-show-icon").click()
|
await asyncio.sleep(0.5)
|
await page.locator('div.time-picker-panel-select-wrp').nth(0).locator(f'span:nth-child({self.scheduled_time.hour + 1})').click()
|
minute_index = ((self.scheduled_time.minute // 5) * 5 // 5) + 1 # 序号从1开始
|
await page.locator('div.time-picker-panel-select-wrp').nth(1).locator(f'span:nth-child({minute_index})').click()
|
|
await asyncio.sleep(1)
|
except Exception as e:
|
self.logger.error(f'[ERROR ] {self.user} , {self.platform} ,设置 发布时间 发生错误：{e}')
|
|
async def set_cover_image(self,page):
|
try:
|
# 通过点击 div.content 触发文件选择框的打开
|
upload_div = page.locator("div.cover-main-upload div.content")
|
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_cover)
|
# 确保上传完成
|
await page.locator('button:has-text("完成")').click(timeout=30000) # 点击“完成”按钮，确保上传完成
|
except Exception as e:
|
self.logger.error(f'[ERROR ] {self.user} , {self.platform} ,设置 封面 发生错误：{e}')
|
|
async def upload_media_and_wait(self,page):
|
try:
|
upload_div = page.get_by_text("拖拽到此处也可上传上传视频")
|
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)
|
|
max_retries = 60 # 设置最大重试次数,最大等待时间为 2 分钟
|
retry_count = 0
|
|
while retry_count < max_retries:
|
try:
|
number = await page.locator('span.success', has_text="上传完成").count()
|
|
if number > 0:
|
self.logger.info("视频上传完毕")
|
break
|
else:
|
if retry_count % 5 == 0:
|
self.logger.info("正在上传视频中...")
|
await asyncio.sleep(2)
|
except Exception as e:
|
self.logger.error(f"检查上传状态时发生错误: {e}")
|
await asyncio.sleep(2) # 等待 2 秒后重试
|
retry_count += 1
|
|
if retry_count == max_retries:
|
self.logger.warning("超过最大重试次数，视频上传可能未完成。")
|
|
except Exception as e:
|
self.logger.error(f'[ERROR ] {self.user} , {self.platform} ,设置 封面 发生错误：{e}')
|
|
async def set_title_and_description(self,page):
|
try:
|
self.logger.info("正在填充标题和话题...")
|
await page.get_by_placeholder("请输入稿件标题").click()
|
await page.get_by_placeholder("请输入稿件标题").press("ControlOrMeta+a")
|
await page.get_by_placeholder("请输入稿件标题").fill(self.title)
|
await page.locator('div[data-placeholder="填写更全面的相关信息，让更多的人能找到你的视频吧"]').click()
|
description_tags = self.description + ' ' + ' '.join(f"#{tag}" for tag in self.tags)
|
await page.locator('div[data-placeholder="填写更全面的相关信息，让更多的人能找到你的视频吧"]').fill(description_tags)
|
|
# 快手只能添加3个话题
|
for index, tag in enumerate(self.tags[:7], start=1):
|
self.logger.info("正在添加第%s个话题" % index)
|
await page.get_by_placeholder("按回车键Enter创建标签").nth(0).click()
|
await page.keyboard.type(tag)
|
await page.keyboard.press("Enter")
|
await asyncio.sleep(1)
|
except Exception as e:
|
self.logger.error(f'[ERROR ] {self.user} , {self.platform} ,设置 标题和描述 发生错误：{e}')
|
|
async def _upload_core(self) -> bool:
|
try:
|
# 创建一个新的页面
|
page = await self.context.new_page()
|
# 访问指定的 URL
|
await page.goto(CONTENT_UPLOAD_URL, timeout=120000)
|
self.logger.info('正在上传-------{}.mp4'.format(self.title))
|
await page.wait_for_url(CONTENT_UPLOAD_URL, timeout=120000)
|
|
|
await self.upload_media_and_wait(page)
|
await self.set_title_and_description(page)
|
await self.set_cover_image(page)
|
|
|
# select CLass
|
if self.category:
|
try:
|
await page.locator(".select-controller").first.click()
|
await asyncio.sleep(1)
|
await page.get_by_title(self.category).click()
|
except Exception as e:
|
self.logger.error(f'[ bilibili ] setting catalog error. {e}')
|
|
# 定时发布
|
if self.scheduled_time != 0:
|
await self.set_schedule_time(page)
|
|
# 判断视频是否发布成功
|
while True:
|
try:
|
publish_button = page.get_by_text("立即投稿", exact=True)
|
if await publish_button.count() > 0:
|
await publish_button.click()
|
await asyncio.sleep(1)
|
|
await page.get_by_role("button", name="再投一个").wait_for(timeout=10000)
|
self.logger.info("视频发布成功")
|
break
|
except Exception as e:
|
self.logger.info(f"视频正在发布中... 错误: {e}")
|
await page.screenshot(full_page=True)
|
await asyncio.sleep(1)
|
|
await self.context.storage_state(path=self.cookie_path) # 保存cookie
|
self.logger.info('cookie更新完毕！')
|
await asyncio.sleep(3) # 这里延迟是为了方便眼睛直观的观看
|
|
return True
|
except Exception as e:
|
self.logger.error(f"{self.user} upload video {self.path_media} fails, error is :{e}")
|
return False
|
