# -*- coding: utf-8 -*-
|
from playwright.async_api import Page
|
import asyncio
|
import traceback
|
|
from auto_media_publisher.uploaders.uploader_base import UploaderBase
|
|
CONTENT_UPLOAD_URL="https://creator.douyin.com/creator-micro/content/upload"
|
VIDEO_PUBLISH_SECCESSFULL_URL = "https://creator.douyin.com/creator-micro/content/manage?enter_from=publish"
|
|
|
class UploaderDouyin(UploaderBase):
|
|
async def set_schedule_time_douyin(self, page):
|
try:
|
# 选择包含特定文本内容的 label 元素
|
label_element = page.locator("[class^='radio']:has-text('定时发布')")
|
# 在选中的 label 元素下点击 checkbox
|
await label_element.click()
|
await asyncio.sleep(1)
|
scheduled_time_hour = self.scheduled_time.strftime("%Y-%m-%d %H:%M")
|
await asyncio.sleep(1)
|
await page.locator('.semi-input[placeholder="日期和时间"]').click()
|
await page.keyboard.press("Control+KeyA")
|
await page.keyboard.type(str(scheduled_time_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 _try_set_cover_image(self, page: Page):
|
try:
|
if self.path_cover:
|
await page.click('text="选择封面"')
|
await page.locator("#dy-creator-content-modal-body").wait_for(timeout=30000)
|
await page.click('text="设置竖封面"')
|
await page.wait_for_timeout(2000) # 等待2秒
|
# 定位到上传区域并点击
|
await page.locator("div[class^='semi-upload upload'] >> input.semi-upload-hidden-input").set_input_files(self.path_cover)
|
await page.wait_for_timeout(3000) # 等待2秒
|
await page.locator("button:visible:has-text('完成')").click()
|
except Exception as e:
|
self.logger.error(f'[ERROR ] {self.user} , {self.platform} ,设置 封面 发生错误：{e}')
|
|
async def _try_set_location(self, page: Page):
|
try:
|
|
# 1. 打开下拉框
|
await page.locator(".anchor-container-hgj7gj > div > .semi-select-arrow").click()
|
await asyncio.sleep(1) # 这段可以保留，也可以换成 wait_for
|
await page.get_by_role("option", name="位置").click()
|
await page.locator('div.semi-select span:has-text("输入地理位置")').click()
|
await page.keyboard.press("Control+A") # 推荐比按Backspace更可靠
|
await page.keyboard.press("Backspace")
|
|
# 4. 输入地理位置
|
await page.wait_for_timeout(1000)
|
await page.keyboard.type(self.location)
|
await page.wait_for_timeout(2000) # 可以尝试替换成更精确的 wait_for() 检查 listbox 是否加载
|
listbox = page.get_by_role("listbox")
|
|
target = listbox.locator("div").filter(has_text=self.location).nth(2)
|
await target.click()
|
|
except Exception as e:
|
self.logger.error(f'[ERROR ] {self.user} , {self.platform} ,设置 位置 发生错误：{e}')
|
|
async def handle_upload_error(self, page):
|
self.logger.info('视频出错了，重新上传中')
|
await page.locator('div.progress-div [class^="upload-btn-input"]').set_input_files(self.path_media)
|
|
async def _try_upload_mp4_and_wait(self,page):
|
try:
|
await page.locator("div[class^='container'] input").set_input_files(self.path_media)
|
while True:
|
# 判断重新上传按钮是否存在，如果不存在，代表视频正在上传，则等待
|
try:
|
number = await page.locator('[class^="long-card"] div:has-text("重新上传")').count()
|
if number > 0:
|
self.logger.info(" [-]视频上传完毕")
|
break
|
else:
|
self.logger.info(" [-] 正在上传视频中...")
|
await asyncio.sleep(2)
|
|
if await page.locator('div.progress-div > div:has-text("上传失败")').count():
|
self.logger.error(" [-] 发现上传出错了... 准备重试")
|
await self.handle_upload_error(page)
|
except Exception as e:
|
self.logger.error(f" [exception] 正在上传视频中... 错误：{e}")
|
self.logger.debug(traceback.format_exc()) # 打印堆栈信息（如果 logger 支持 debug）
|
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
|
|
async def _try_set_tilte_and_description(self,page):
|
try:
|
await asyncio.sleep(1)
|
self.logger.info(f' [-] 正在填充标题和话题...')
|
await page.get_by_placeholder("填写作品标题，为作品获得更多流量").click()
|
await page.get_by_placeholder("填写作品标题，为作品获得更多流量").fill(self.title[:30])
|
await page.locator(".zone-container").click()
|
description_tags = self.description + ' ' + ' '.join(f"#{tag}" for tag in self.tags)
|
await page.locator(".zone-container").fill(description_tags)
|
self.logger.info(f'总共添加{len(self.tags)}个话题')
|
except Exception as e:
|
self.logger.error(f'[ERROR ] {self.user} , {self.platform} ,设置title 发生错误：{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(f'[+]正在上传-------{self.title}.mp4')
|
await page.wait_for_url(CONTENT_UPLOAD_URL, timeout=120000)
|
|
# 1、上传并等待视频上传结束
|
await self._try_upload_mp4_and_wait(page)
|
# 2、设置标题和描述
|
await self._try_set_tilte_and_description(page)
|
# 3、设置封面
|
await self._try_set_cover_image(page)
|
# 4、设置位置 和 商品
|
await self._try_set_location(page)
|
third_part_element = '[class^="info"] > [class^="first-part"] div div.semi-switch'
|
# 定位是否有第三方平台
|
if await page.locator(third_part_element).count():
|
# 检测是否是已选中状态
|
if 'semi-switch-checked' not in await page.eval_on_selector(third_part_element, 'div => div.className'):
|
await page.locator(third_part_element).locator('input.semi-switch-native-control').click()
|
|
# 5、设置定时发布
|
if self.scheduled_time != 0:
|
await self.set_schedule_time_douyin(page)
|
|
|
|
# 6 发布并判断视频是否发布成功
|
while True:
|
# 判断视频是否发布成功
|
try:
|
publish_button = page.get_by_role('button', name="发布", exact=True)
|
if await publish_button.count():
|
await publish_button.click()
|
await page.wait_for_url(VIDEO_PUBLISH_SECCESSFULL_URL,timeout=3000) # 如果自动跳转到作品页面，则代表发布成功
|
self.logger.info(" [-]视频发布成功")
|
break
|
except:
|
self.logger.info(" [-] 视频正在发布中...")
|
await page.screenshot(full_page=True)
|
await asyncio.sleep(0.5)
|
|
# 7、 更新Cookie信息，退出
|
await self.context.storage_state(path=self.cookie_path) # 保存cookie
|
self.logger.info(f' {self.user} , {self.platform} [-]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
|
