# -*- coding: utf-8 -*-
|
from datetime import datetime
|
|
from playwright.async_api import Page
|
import asyncio
|
from auto_media_publisher.uploaders.uploader_base import UploaderBase
|
|
HOMEPAGE_URL="https://www.tiktok.com"
|
CONTENT_UPLOAD_URL="https://www.tiktok.com/tiktokstudio/upload?from=webapp"
|
|
VIDEO_PUBLISH_SECCESSFULL_URL = "https://www.tiktok.com/tiktokstudio/content"
|
|
Tk_Locator_tk_iframe = '[data-tt="Upload_index_iframe"]'
|
Tk_Locator_default = 'body'
|
|
class UploaderTiktok(UploaderBase):
|
|
async def set_schedule_time(self, page):
|
schedule_input_element = self.locator_base.get_by_label('Schedule')
|
await schedule_input_element.wait_for(state='visible') # 确保按钮可见
|
|
await schedule_input_element.click(force=True)
|
if await self.locator_base.locator('div.TUXButton-content >> text=Allow').count():
|
await self.locator_base.locator('div.TUXButton-content >> text=Allow').click()
|
|
scheduled_picker = self.locator_base.locator('div.scheduled-picker')
|
await scheduled_picker.locator('div.TUXInputBox').nth(1).click()
|
|
calendar_month = await self.locator_base.locator(
|
'div.calendar-wrapper span.month-title').inner_text()
|
|
n_calendar_month = datetime.strptime(calendar_month, '%B').month
|
|
schedule_month = self.scheduled_time.month
|
|
if n_calendar_month != schedule_month:
|
if n_calendar_month < schedule_month:
|
arrow = self.locator_base.locator('div.calendar-wrapper span.arrow').nth(-1)
|
else:
|
arrow = self.locator_base.locator('div.calendar-wrapper span.arrow').nth(0)
|
await arrow.click()
|
|
# day set
|
valid_days_locator = self.locator_base.locator(
|
'div.calendar-wrapper span.day.valid')
|
valid_days = await valid_days_locator.count()
|
for i in range(valid_days):
|
day_element = valid_days_locator.nth(i)
|
text = await day_element.inner_text()
|
if text.strip() == str(self.scheduled_time.day):
|
await day_element.click()
|
break
|
# time set
|
await scheduled_picker.locator('div.TUXInputBox').nth(0).click()
|
|
hour_str = self.scheduled_time.strftime("%H")
|
correct_minute = int(self.scheduled_time.minute / 5)
|
minute_str = f"{correct_minute:02d}"
|
|
hour_selector = f"span.tiktok-timepicker-left:has-text('{hour_str}')"
|
minute_selector = f"span.tiktok-timepicker-right:has-text('{minute_str}')"
|
|
# pick hour first
|
await page.wait_for_timeout(1000) # 等待500毫秒
|
await self.locator_base.locator(hour_selector).click()
|
# click time button again
|
await page.wait_for_timeout(1000) # 等待500毫秒
|
# pick minutes after
|
await self.locator_base.locator(minute_selector).click()
|
|
# click title to remove the focus.
|
# await self.locator_base.locator("h1:has-text('Upload video')").click()
|
|
async def handle_upload_error(self, page):
|
self.logger.info("video upload error retrying.")
|
select_file_button = self.locator_base.locator('button[aria-label="Select file"]')
|
async with page.expect_file_chooser() as fc_info:
|
await select_file_button.click()
|
file_chooser = await fc_info.value
|
await file_chooser.set_files(self.path_media)
|
|
async def add_title_tags(self, page):
|
|
editor_locator = self.locator_base.locator('div.public-DraftEditor-content')
|
await editor_locator.click()
|
await page.keyboard.press("End")
|
await page.keyboard.press("Control+A")
|
await page.keyboard.press("Delete")
|
await page.keyboard.press("End")
|
await page.wait_for_timeout(1000) # 等待1秒
|
await page.keyboard.insert_text(self.title)
|
await page.wait_for_timeout(1000) # 等待1秒
|
await page.keyboard.press("End")
|
await page.keyboard.press("Enter")
|
|
# tag part
|
for index, tag in enumerate(self.tags, start=1):
|
self.logger.info("Setting the %s tag" % index)
|
await page.keyboard.press("End")
|
await page.wait_for_timeout(1000) # 等待1秒
|
await page.keyboard.insert_text("#" + tag + " ")
|
await page.keyboard.press("Space")
|
await page.wait_for_timeout(1000) # 等待1秒
|
|
await page.keyboard.press("Backspace")
|
await page.keyboard.press("End")
|
|
async def upload_thumbnails(self, page):
|
await self.locator_base.locator(".cover-container").click()
|
await self.locator_base.locator(".cover-edit-container >> text=Upload cover").click()
|
async with page.expect_file_chooser() as fc_info:
|
await self.locator_base.locator(".upload-image-upload-area").click()
|
file_chooser = await fc_info.value
|
await file_chooser.set_files(self.path_cover)
|
await self.locator_base.locator('div.cover-edit-panel:not(.hide-panel)').get_by_role(
|
"button", name="Confirm").click()
|
await page.wait_for_timeout(3000) # wait 3s, fix it later
|
|
async def change_language(self, page):
|
# set the language to english
|
await page.goto(HOMEPAGE_URL)
|
await page.wait_for_load_state('domcontentloaded')
|
await page.wait_for_selector('[data-e2e="nav-more-menu"]')
|
# 已经设置为英文, 省略这个步骤
|
if await page.locator('[data-e2e="nav-more-menu"]').text_content() == "More":
|
return
|
|
await page.locator('[data-e2e="nav-more-menu"]').click()
|
await page.locator('[data-e2e="language-select"]').click()
|
await page.locator('#creator-tools-selection-menu-header >> text=English').nth(1).click()
|
|
async def click_publish(self, page):
|
success_flag_div = 'div.common-modal-confirm-modal'
|
while True:
|
try:
|
publish_button = self.locator_base.locator('div.button-group button').nth(0)
|
if await publish_button.count():
|
await publish_button.click()
|
|
await page.wait_for_url(VIDEO_PUBLISH_SECCESSFULL_URL, timeout=3000)
|
self.logger.info(" [-] video published success")
|
break
|
except Exception as e:
|
self.logger.exception(f" [-] Exception: {e}")
|
self.logger.info(" [-] video publishing")
|
await asyncio.sleep(0.5)
|
|
async def detect_upload_status(self, page):
|
while True:
|
try:
|
# if await self.locator_base.locator('div.btn-post > button').get_attribute("disabled") is None:
|
if await self.locator_base.locator(
|
'div.button-group > button >> text=Post').get_attribute("disabled") is None:
|
self.logger.info(" [-]video uploaded.")
|
break
|
else:
|
self.logger.info(" [-] video uploading...")
|
await asyncio.sleep(2)
|
if await self.locator_base.locator(
|
'button[aria-label="Select file"]').count():
|
self.logger.info(" [-] found some error while uploading now retry...")
|
await self.handle_upload_error(page)
|
except:
|
self.logger.info(" [-] video uploading...")
|
await asyncio.sleep(2)
|
|
async def choose_base_locator(self, page):
|
# await page.wait_for_selector('div.upload-container')
|
if await page.locator('iframe[data-tt="Upload_index_iframe"]').count():
|
self.locator_base = page.frame_locator(Tk_Locator_tk_iframe)
|
else:
|
self.locator_base = page.locator(Tk_Locator_default)
|
|
async def _upload_core(self) -> bool:
|
try:
|
|
page = await self.context.new_page()
|
await self.change_language(page)
|
await page.goto(CONTENT_UPLOAD_URL, timeout=120000)
|
self.logger.info(f'[+]Uploading-------{self.title}.mp4')
|
|
await page.wait_for_url(CONTENT_UPLOAD_URL, timeout=120000)
|
|
try:
|
await page.wait_for_selector('iframe[data-tt="Upload_index_iframe"], div.upload-container', timeout=10000)
|
self.logger.info("Either iframe or div appeared.")
|
except Exception as e:
|
self.logger.error("Neither iframe nor div appeared within the timeout.")
|
|
await self.choose_base_locator(page)
|
|
upload_button = self.locator_base.locator(
|
'button:has-text("Select video"):visible')
|
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 self.add_title_tags(page)
|
# detect upload status
|
await self.detect_upload_status(page)
|
if self.path_cover:
|
self.logger.info(f'[+] Uploading thumbnail file {self.title}.png')
|
await self.upload_thumbnails(page)
|
|
if self.scheduled_time != 0:
|
await self.set_schedule_time(page)
|
|
await self.click_publish(page)
|
|
await self.context.storage_state(path=f"{self.cookie_path}") # save cookie
|
self.logger.info('tiktok [-] update cookie！')
|
await asyncio.sleep(2) # close delay for look the video status
|
|
return True
|
except Exception as e:
|
self.logger.error(f"{self.user} upload video {self.path_media} fails, error is :{e}")
|
return False
|
