# -*- coding: utf-8 -*-
|
from playwright.async_api import Page
|
import asyncio
|
from auto_media_publisher.uploaders.uploader_base import UploaderBase
|
|
CONTENT_UPLOAD_URL="https://cp.kuaishou.com/article/publish/video"
|
VIDEO_PUBLISH_SECCESSFULL_URL = "https://cp.kuaishou.com/article/manage/video?status=2&from=publish"
|
|
|
class UploaderKuaishou(UploaderBase):
|
|
async def set_schedule_time(self, page):
|
try:
|
self.logger.info("{self.user}, {self.platform} click schedule time......")
|
scheduled_date_hour = self.scheduled_time.strftime("%Y-%m-%d %H:%M:%S")
|
await page.locator("label:text('发布时间')").locator('xpath=following-sibling::div').locator(
|
'.ant-radio-input').nth(1).click()
|
await asyncio.sleep(1)
|
|
await page.locator('div.ant-picker-input input[placeholder="选择日期时间"]').click()
|
await asyncio.sleep(1)
|
|
await page.keyboard.press("Control+KeyA")
|
await page.keyboard.type(str(scheduled_date_hour))
|
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_media_and_wait(self,page):
|
try:
|
upload_button = page.locator("button[class^='_upload-btn']")
|
await upload_button.wait_for(state='visible') # 确保按钮可见
|
|
async with page.expect_file_chooser() as fc_info:
|
await upload_button.click()
|
file_chooser = await fc_info.value
|
await file_chooser.set_files(self.path_media)
|
await asyncio.sleep(3)
|
|
# 等待按钮可交互
|
new_feature_button = page.locator('button[type="button"] span:text("我知道了")')
|
if await new_feature_button.count() > 0:
|
await new_feature_button.click()
|
|
max_retries = 60 # 设置最大重试次数,最大等待时间为 5 分钟
|
retry_count = 0
|
|
while retry_count < max_retries:
|
try:
|
number = await page.locator("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(5) # 等待 5 秒后重试
|
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_text("描述").locator("xpath=following-sibling::div").click()
|
self.logger.info("clear existing title")
|
await page.keyboard.press("Backspace")
|
await page.keyboard.press("Control+KeyA")
|
await page.keyboard.press("Delete")
|
self.logger.info("filling new title")
|
await page.keyboard.type(self.title+" "+self.description)
|
await page.keyboard.press("Enter")
|
|
# 快手只能添加3个话题
|
for index, tag in enumerate(self.tags[:3], start=1):
|
self.logger.info("正在添加第%s个话题" % index)
|
await page.keyboard.type(f"#{tag} ")
|
await asyncio.sleep(2)
|
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()
|
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)
|
# 1、上传视频 并等待
|
await self.upload_media_and_wait(page)
|
# 2、 设置标题和描述
|
await self.set_title_and_description(page)
|
# 定时任务
|
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)
|
confirm_button = page.get_by_text("确认发布")
|
if await confirm_button.count() > 0:
|
await confirm_button.click()
|
|
# 等待页面跳转，确认发布成功
|
await page.wait_for_url(VIDEO_PUBLISH_SECCESSFULL_URL,timeout=5000)
|
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
|
