| 1 | # -*- coding: utf-8 -*- |
| 2 | import re |
| 3 | from datetime import datetime |
| 4 | |
| 5 | from playwright.async_api import Playwright, async_playwright |
| 6 | import os |
| 7 | import asyncio |
| 8 | |
| 9 | from conf import LOCAL_CHROME_PATH, LOCAL_CHROME_HEADLESS |
| 10 | from uploader.tk_uploader.tk_config import Tk_Locator |
| 11 | from utils.base_social_media import set_init_script |
| 12 | from utils.files_times import get_absolute_path |
| 13 | from utils.log import tiktok_logger |
| 14 | |
| 15 | |
| 16 | async def cookie_auth(account_file): |
| 17 | async with async_playwright() as playwright: |
| 18 | browser = await playwright.chromium.launch(headless=LOCAL_CHROME_HEADLESS) |
| 19 | context = await browser.new_context(storage_state=account_file) |
| 20 | context = await set_init_script(context) |
| 21 | # 创建一个新的页面 |
| 22 | page = await context.new_page() |
| 23 | # 访问指定的 URL |
| 24 | await page.goto("https://www.tiktok.com/tiktokstudio/upload?lang=en") |
| 25 | await page.wait_for_load_state('networkidle') |
| 26 | try: |
| 27 | # 选择所有的 select 元素 |
| 28 | select_elements = await page.query_selector_all('select') |
| 29 | for element in select_elements: |
| 30 | class_name = await element.get_attribute('class') |
| 31 | # 使用正则表达式匹配特定模式的 class 名称 |
| 32 | if re.match(r'tiktok-.*-SelectFormContainer.*', class_name): |
| 33 | tiktok_logger.error("[+] cookie expired") |
| 34 | return False |
| 35 | tiktok_logger.success("[+] cookie valid") |
| 36 | return True |
| 37 | except: |
| 38 | tiktok_logger.success("[+] cookie valid") |
| 39 | return True |
| 40 | |
| 41 | |
| 42 | async def tiktok_setup(account_file, handle=False): |
| 43 | account_file = get_absolute_path(account_file, "tk_uploader") |
| 44 | if not os.path.exists(account_file) or not await cookie_auth(account_file): |
| 45 | if not handle: |
| 46 | return False |
| 47 | tiktok_logger.info('[+] cookie file is not existed or expired. Now open the browser auto. Please login with your way(gmail phone, whatever, the cookie file will generated after login') |
| 48 | await get_tiktok_cookie(account_file) |
| 49 | return True |
| 50 | |
| 51 | |
| 52 | async def get_tiktok_cookie(account_file): |
| 53 | async with async_playwright() as playwright: |
| 54 | options = { |
| 55 | 'args': [ |
| 56 | '--lang en-GB', |
| 57 | ], |
| 58 | 'headless': LOCAL_CHROME_HEADLESS, # Set headless option here |
| 59 | } |
| 60 | # Make sure to run headed. |
| 61 | browser = await playwright.chromium.launch(**options) |
| 62 | # Setup context however you like. |
| 63 | context = await browser.new_context() # Pass any options |
| 64 | context = await set_init_script(context) |
| 65 | # Pause the page, and start recording manually. |
| 66 | page = await context.new_page() |
| 67 | await page.goto("https://www.tiktok.com/login?lang=en") |
| 68 | await page.pause() |
| 69 | # 点击调试器的继续，保存cookie |
| 70 | await context.storage_state(path=account_file) |
| 71 | |
| 72 | |
| 73 | class TiktokVideo(object): |
| 74 | def __init__(self, title, file_path, tags, publish_date, account_file, thumbnail_path=None): |
| 75 | self.title = title |
| 76 | self.file_path = file_path |
| 77 | self.tags = tags |
| 78 | self.publish_date = publish_date |
| 79 | self.thumbnail_path = thumbnail_path |
| 80 | self.account_file = account_file |
| 81 | self.local_executable_path = LOCAL_CHROME_PATH |
| 82 | self.headless = LOCAL_CHROME_HEADLESS |
| 83 | self.locator_base = None |
| 84 | |
| 85 | async def set_schedule_time(self, page, publish_date): |
| 86 | schedule_input_element = self.locator_base.get_by_label('Schedule') |
| 87 | await schedule_input_element.wait_for(state='visible') # 确保按钮可见 |
| 88 | |
| 89 | await schedule_input_element.click(force=True) |
| 90 | if await self.locator_base.locator('div.TUXButton-content >> text=Allow').count(): |
| 91 | await self.locator_base.locator('div.TUXButton-content >> text=Allow').click() |
| 92 | |
| 93 | scheduled_picker = self.locator_base.locator('div.scheduled-picker') |
| 94 | await scheduled_picker.locator('div.TUXInputBox').nth(1).click() |
| 95 | |
| 96 | calendar_month = await self.locator_base.locator( |
| 97 | 'div.calendar-wrapper span.month-title').inner_text() |
| 98 | |
| 99 | n_calendar_month = datetime.strptime(calendar_month, '%B').month |
| 100 | |
| 101 | schedule_month = publish_date.month |
| 102 | |
| 103 | if n_calendar_month != schedule_month: |
| 104 | if n_calendar_month < schedule_month: |
| 105 | arrow = self.locator_base.locator('div.calendar-wrapper span.arrow').nth(-1) |
| 106 | else: |
| 107 | arrow = self.locator_base.locator('div.calendar-wrapper span.arrow').nth(0) |
| 108 | await arrow.click() |
| 109 | |
| 110 | # day set |
| 111 | valid_days_locator = self.locator_base.locator( |
| 112 | 'div.calendar-wrapper span.day.valid') |
| 113 | valid_days = await valid_days_locator.count() |
| 114 | for i in range(valid_days): |
| 115 | day_element = valid_days_locator.nth(i) |
| 116 | text = await day_element.inner_text() |
| 117 | if text.strip() == str(publish_date.day): |
| 118 | await day_element.click() |
| 119 | break |
| 120 | # time set |
| 121 | await scheduled_picker.locator('div.TUXInputBox').nth(0).click() |
| 122 | |
| 123 | hour_str = publish_date.strftime("%H") |
| 124 | correct_minute = int(publish_date.minute / 5) |
| 125 | minute_str = f"{correct_minute:02d}" |
| 126 | |
| 127 | hour_selector = f"span.tiktok-timepicker-left:has-text('{hour_str}')" |
| 128 | minute_selector = f"span.tiktok-timepicker-right:has-text('{minute_str}')" |
| 129 | |
| 130 | # pick hour first |
| 131 | await page.wait_for_timeout(1000) # 等待500毫秒 |
| 132 | await self.locator_base.locator(hour_selector).click() |
| 133 | # click time button again |
| 134 | await page.wait_for_timeout(1000) # 等待500毫秒 |
| 135 | # pick minutes after |
| 136 | await self.locator_base.locator(minute_selector).click() |
| 137 | |
| 138 | # click title to remove the focus. |
| 139 | # await self.locator_base.locator("h1:has-text('Upload video')").click() |
| 140 | |
| 141 | async def handle_upload_error(self, page): |
| 142 | tiktok_logger.info("video upload error retrying.") |
| 143 | select_file_button = self.locator_base.locator('button[aria-label="Select file"]') |
| 144 | async with page.expect_file_chooser() as fc_info: |
| 145 | await select_file_button.click() |
| 146 | file_chooser = await fc_info.value |
| 147 | await file_chooser.set_files(self.file_path) |
| 148 | |
| 149 | async def upload(self, playwright: Playwright) -> None: |
| 150 | browser = await playwright.chromium.launch(headless=self.headless, executable_path=self.local_executable_path) |
| 151 | context = await browser.new_context(storage_state=f"{self.account_file}") |
| 152 | # context = await set_init_script(context) |
| 153 | page = await context.new_page() |
| 154 | |
| 155 | # change language to eng first |
| 156 | await self.change_language(page) |
| 157 | await page.goto("https://www.tiktok.com/tiktokstudio/upload") |
| 158 | tiktok_logger.info(f'[+]Uploading-------{self.title}.mp4') |
| 159 | |
| 160 | await page.wait_for_url("https://www.tiktok.com/tiktokstudio/upload", timeout=10000) |
| 161 | |
| 162 | try: |
| 163 | await page.wait_for_selector('iframe[data-tt="Upload_index_iframe"], div.upload-container', timeout=10000) |
| 164 | tiktok_logger.info("Either iframe or div appeared.") |
| 165 | except Exception as e: |
| 166 | tiktok_logger.error("Neither iframe nor div appeared within the timeout.") |
| 167 | |
| 168 | await self.choose_base_locator(page) |
| 169 | |
| 170 | upload_button = self.locator_base.locator( |
| 171 | 'button:has-text("Select video"):visible') |
| 172 | await upload_button.wait_for(state='visible') # 确保按钮可见 |
| 173 | |
| 174 | async with page.expect_file_chooser() as fc_info: |
| 175 | await upload_button.click() |
| 176 | file_chooser = await fc_info.value |
| 177 | await file_chooser.set_files(self.file_path) |
| 178 | |
| 179 | await self.add_title_tags(page) |
| 180 | # detect upload status |
| 181 | await self.detect_upload_status(page) |
| 182 | if self.thumbnail_path: |
| 183 | tiktok_logger.info(f'[+] Uploading thumbnail file {self.title}.png') |
| 184 | await self.upload_thumbnails(page) |
| 185 | |
| 186 | if self.publish_date != 0: |
| 187 | await self.set_schedule_time(page, self.publish_date) |
| 188 | |
| 189 | await self.click_publish(page) |
| 190 | tiktok_logger.success(f"video_id: {await self.get_last_video_id(page)}") |
| 191 | |
| 192 | await context.storage_state(path=f"{self.account_file}") # save cookie |
| 193 | tiktok_logger.info(' [-] update cookie！') |
| 194 | await asyncio.sleep(2) # close delay for look the video status |
| 195 | # close all |
| 196 | await context.close() |
| 197 | await browser.close() |
| 198 | |
| 199 | async def add_title_tags(self, page): |
| 200 | |
| 201 | editor_locator = self.locator_base.locator('div.public-DraftEditor-content') |
| 202 | await editor_locator.click() |
| 203 | |
| 204 | await page.keyboard.press("End") |
| 205 | |
| 206 | await page.keyboard.press("Control+A") |
| 207 | |
| 208 | await page.keyboard.press("Delete") |
| 209 | |
| 210 | await page.keyboard.press("End") |
| 211 | |
| 212 | await page.wait_for_timeout(1000) # 等待1秒 |
| 213 | |
| 214 | await page.keyboard.insert_text(self.title) |
| 215 | await page.wait_for_timeout(1000) # 等待1秒 |
| 216 | await page.keyboard.press("End") |
| 217 | |
| 218 | await page.keyboard.press("Enter") |
| 219 | |
| 220 | # tag part |
| 221 | for index, tag in enumerate(self.tags, start=1): |
| 222 | tiktok_logger.info("Setting the %s tag" % index) |
| 223 | await page.keyboard.press("End") |
| 224 | await page.wait_for_timeout(1000) # 等待1秒 |
| 225 | await page.keyboard.insert_text("#" + tag + " ") |
| 226 | await page.keyboard.press("Space") |
| 227 | await page.wait_for_timeout(1000) # 等待1秒 |
| 228 | |
| 229 | await page.keyboard.press("Backspace") |
| 230 | await page.keyboard.press("End") |
| 231 | |
| 232 | async def upload_thumbnails(self, page): |
| 233 | await self.locator_base.locator(".cover-container").click() |
| 234 | await self.locator_base.locator(".cover-edit-container >> text=Upload cover").click() |
| 235 | async with page.expect_file_chooser() as fc_info: |
| 236 | await self.locator_base.locator(".upload-image-upload-area").click() |
| 237 | file_chooser = await fc_info.value |
| 238 | await file_chooser.set_files(self.thumbnail_path) |
| 239 | await self.locator_base.locator('div.cover-edit-panel:not(.hide-panel)').get_by_role( |
| 240 | "button", name="Confirm").click() |
| 241 | await page.wait_for_timeout(3000) # wait 3s, fix it later |
| 242 | |
| 243 | async def change_language(self, page): |
| 244 | # set the language to english |
| 245 | await page.goto("https://www.tiktok.com") |
| 246 | await page.wait_for_load_state('domcontentloaded') |
| 247 | await page.wait_for_selector('[data-e2e="nav-more-menu"]') |
| 248 | # 已经设置为英文, 省略这个步骤 |
| 249 | if await page.locator('[data-e2e="nav-more-menu"]').text_content() == "More": |
| 250 | return |
| 251 | |
| 252 | await page.locator('[data-e2e="nav-more-menu"]').click() |
| 253 | await page.locator('[data-e2e="language-select"]').click() |
| 254 | await page.locator('#creator-tools-selection-menu-header >> text=English (US)').click() |
| 255 | |
| 256 | async def click_publish(self, page): |
| 257 | success_flag_div = 'div.common-modal-confirm-modal' |
| 258 | while True: |
| 259 | try: |
| 260 | publish_button = self.locator_base.locator('div.button-group button').nth(0) |
| 261 | if await publish_button.count(): |
| 262 | await publish_button.click() |
| 263 | |
| 264 | await page.wait_for_url("https://www.tiktok.com/tiktokstudio/content", timeout=3000) |
| 265 | tiktok_logger.success(" [-] video published success") |
| 266 | break |
| 267 | except Exception as e: |
| 268 | tiktok_logger.exception(f" [-] Exception: {e}") |
| 269 | tiktok_logger.info(" [-] video publishing") |
| 270 | await asyncio.sleep(0.5) |
| 271 | |
| 272 | async def get_last_video_id(self, page): |
| 273 | await page.wait_for_selector('div[data-tt="components_PostTable_Container"]') |
| 274 | video_list_locator = self.locator_base.locator('div[data-tt="components_PostTable_Container"] div[data-tt="components_PostInfoCell_Container"] a') |
| 275 | if await video_list_locator.count(): |
| 276 | first_video_obj = await video_list_locator.nth(0).get_attribute('href') |
| 277 | video_id = re.search(r'video/(\d+)', first_video_obj).group(1) if first_video_obj else None |
| 278 | return video_id |
| 279 | |
| 280 | |
| 281 | async def detect_upload_status(self, page): |
| 282 | while True: |
| 283 | try: |
| 284 | # if await self.locator_base.locator('div.btn-post > button').get_attribute("disabled") is None: |
| 285 | if await self.locator_base.locator( |
| 286 | 'div.button-group > button >> text=Post').get_attribute("disabled") is None: |
| 287 | tiktok_logger.info(" [-]video uploaded.") |
| 288 | break |
| 289 | else: |
| 290 | tiktok_logger.info(" [-] video uploading...") |
| 291 | await asyncio.sleep(2) |
| 292 | if await self.locator_base.locator( |
| 293 | 'button[aria-label="Select file"]').count(): |
| 294 | tiktok_logger.info(" [-] found some error while uploading now retry...") |
| 295 | await self.handle_upload_error(page) |
| 296 | except: |
| 297 | tiktok_logger.info(" [-] video uploading...") |
| 298 | await asyncio.sleep(2) |
| 299 | |
| 300 | async def choose_base_locator(self, page): |
| 301 | # await page.wait_for_selector('div.upload-container') |
| 302 | if await page.locator('iframe[data-tt="Upload_index_iframe"]').count(): |
| 303 | self.locator_base = page.frame_locator(Tk_Locator.tk_iframe) |
| 304 | else: |
| 305 | self.locator_base = page.locator(Tk_Locator.default) |
| 306 | |
| 307 | async def main(self): |
| 308 | async with async_playwright() as playwright: |
| 309 | await self.upload(playwright) |
| 310 |