# -*- 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