1. tk(chrome support)
2. tk add thumbnail support
This commit is contained in:
pookz@stme
@@ -67,7 +67,7 @@ meta_file 内容(content)：
|
||||
```
|
||||
|
||||
### Usage
|
||||
1. 设置conf 文件中的 `LOCAL_CHROME_PATH`(在douyin 或者视频号可能出现chromium 不兼容的各种问题，建议设置本地的chrome)
|
||||
1. 设置conf 文件中的 `LOCAL_CHROME_PATH`(在douyin、视频号 tiktok可能出现chromium 不兼容的各种问题，建议设置本地的chrome)
|
||||
2. 这里分割出来3条路
|
||||
- 可自行研究源码，免费、任意 穿插在自己的项目中
|
||||
- 可参考下面的各个平台的使用指南，`examples`文件夹中有各种示例代码
|
||||
@@ -141,7 +141,7 @@ generate_schedule_time_next_day 默认从第二天开始（此举为避免选择
|
||||
> 解决方案：
|
||||
> 1. 下载chrome
|
||||
> 2. 找到chrome的安装目录
|
||||
> 3. 将目录配置到DouYinVideo类中 self.local_executable_path = "xxx/xxx/chrome.exe"
|
||||
> 3. 将本地chrome 路径conf文件中 LOCAL_CHROME_PATH = "xxx/xxx/chrome.exe"
|
||||
|
||||
|
||||
参考项目：
|
||||
@@ -168,7 +168,7 @@ generate_schedule_time_next_day 默认从第二天开始（此举为避免选择
|
||||
> 解决方案：
|
||||
> 1. 下载chrome
|
||||
> 2. 找到chrome的安装目录
|
||||
> 3. 将目录配置到TencentVideo类中 self.local_executable_path = "xxx/xxx/chrome.exe"
|
||||
> 3. 将本地chrome 路径conf文件中 LOCAL_CHROME_PATH = "xxx/xxx/chrome.exe"
|
||||
|
||||
---
|
||||
|
||||
@@ -234,11 +234,16 @@ bilibili cookie 长期有效（至少我运行2年以来是这样的）
|
||||
### 5. tiktok
|
||||
使用playwright模拟浏览器行为(Simulating Browser Behavior with playwright)
|
||||
1. 准备视频目录结构(Prepare the video directory structure)
|
||||
2. cookie获取(generate your cookie)：get_tk_cookie.py
|
||||
2. 将本地chrome路径配置到conf文件中 LOCAL_CHROME_PATH = "xxx/xxx/chrome.exe"(Configure the local chrome path to the conf file)
|
||||
3. cookie获取(generate your cookie)：get_tk_cookie.py
|
||||
|
||||
3. 上传视频(upload video)：upload_video_to_tiktok.py
|
||||
|
||||
4. 上传视频(upload video)：upload_video_to_tiktok.py
|
||||
|
||||
24.7.3 update local chrome support:
|
||||
- you can upload your custom thumbnail(place `png` file at the path of `videos`)
|
||||
- if not, the system will take the first frame of the video as the thumbnail.
|
||||
- chrome drive can't login by gmail oauth. the google don't support it. (if you want login by google oauth you can use the `tk_uploader/main.py` old firefox.)
|
||||
- before the upload process change the language to eng.
|
||||
|
||||
其他部分解释：
|
||||
```
|
||||
|
||||
+1
-1
@@ -7,7 +7,7 @@ from pathlib import Path
|
||||
from conf import BASE_DIR
|
||||
from douyin_uploader.main import douyin_setup, DouYinVideo
|
||||
from tencent_uploader.main import weixin_setup, TencentVideo
|
||||
from tk_uploader.main import tiktok_setup, TiktokVideo
|
||||
from tk_uploader.main_chrome import tiktok_setup, TiktokVideo
|
||||
from utils.base_social_media import get_supported_social_media, get_cli_action, SOCIAL_MEDIA_DOUYIN, \
|
||||
SOCIAL_MEDIA_TENCENT, SOCIAL_MEDIA_TIKTOK
|
||||
from utils.constant import TencentZoneTypes
|
||||
|
||||
@@ -2,7 +2,7 @@ import asyncio
|
||||
from pathlib import Path
|
||||
|
||||
from conf import BASE_DIR
|
||||
from tk_uploader.main import tiktok_setup
|
||||
from tk_uploader.main_chrome import tiktok_setup
|
||||
|
||||
if __name__ == '__main__':
|
||||
account_file = Path(BASE_DIR / "tk_uploader" / "account.json")
|
||||
|
||||
@@ -2,7 +2,8 @@ import asyncio
|
||||
from pathlib import Path
|
||||
|
||||
from conf import BASE_DIR
|
||||
from tk_uploader.main import tiktok_setup, TiktokVideo
|
||||
# from tk_uploader.main import tiktok_setup, TiktokVideo
|
||||
from tk_uploader.main_chrome import tiktok_setup, TiktokVideo
|
||||
from utils.files_times import generate_schedule_time_next_day, get_title_and_hashtags
|
||||
|
||||
|
||||
@@ -17,8 +18,13 @@ if __name__ == '__main__':
|
||||
cookie_setup = asyncio.run(tiktok_setup(account_file, handle=True))
|
||||
for index, file in enumerate(files):
|
||||
title, tags = get_title_and_hashtags(str(file))
|
||||
thumbnail_path = file.with_suffix('.png')
|
||||
print(f"video_file_name：{file}")
|
||||
print(f"video_title：{title}")
|
||||
print(f"video_hashtag：{tags}")
|
||||
app = TiktokVideo(title, file, tags, publish_datetimes[index], account_file)
|
||||
if thumbnail_path.exists():
|
||||
print(f"thumbnail_file_name：{thumbnail_path}")
|
||||
app = TiktokVideo(title, file, tags, publish_datetimes[index], account_file, thumbnail_path)
|
||||
else:
|
||||
app = TiktokVideo(title, file, tags, publish_datetimes[index], account_file)
|
||||
asyncio.run(app.main(), debug=False)
|
||||
|
||||
@@ -0,0 +1,286 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
import re
|
||||
from datetime import datetime
|
||||
|
||||
from playwright.async_api import Playwright, async_playwright
|
||||
import os
|
||||
import asyncio
|
||||
|
||||
from conf import LOCAL_CHROME_PATH
|
||||
from tk_uploader.tk_config import Tk_Locator
|
||||
from utils.base_social_media import set_init_script
|
||||
from utils.files_times import get_absolute_path
|
||||
from utils.log import tiktok_logger
|
||||
|
||||
|
||||
async def cookie_auth(account_file):
|
||||
async with async_playwright() as playwright:
|
||||
browser = await playwright.chromium.launch(headless=True)
|
||||
context = await browser.new_context(storage_state=account_file)
|
||||
context = await set_init_script(context)
|
||||
# 创建一个新的页面
|
||||
page = await context.new_page()
|
||||
# 访问指定的 URL
|
||||
await page.goto("https://www.tiktok.com/tiktokstudio/upload?lang=en")
|
||||
await page.wait_for_load_state('networkidle')
|
||||
try:
|
||||
# 选择所有的 select 元素
|
||||
select_elements = await page.query_selector_all('select')
|
||||
for element in select_elements:
|
||||
class_name = await element.get_attribute('class')
|
||||
# 使用正则表达式匹配特定模式的 class 名称
|
||||
if re.match(r'tiktok-.*-SelectFormContainer.*', class_name):
|
||||
tiktok_logger.error("[+] cookie expired")
|
||||
return False
|
||||
tiktok_logger.success("[+] cookie valid")
|
||||
return True
|
||||
except:
|
||||
tiktok_logger.success("[+] cookie valid")
|
||||
return True
|
||||
|
||||
|
||||
async def tiktok_setup(account_file, handle=False):
|
||||
account_file = get_absolute_path(account_file, "tk_uploader")
|
||||
if not os.path.exists(account_file) or not await cookie_auth(account_file):
|
||||
if not handle:
|
||||
return False
|
||||
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')
|
||||
await get_tiktok_cookie(account_file)
|
||||
return True
|
||||
|
||||
|
||||
async def get_tiktok_cookie(account_file):
|
||||
async with async_playwright() as playwright:
|
||||
options = {
|
||||
'args': [
|
||||
'--lang en-GB',
|
||||
],
|
||||
'headless': False, # Set headless option here
|
||||
}
|
||||
# Make sure to run headed.
|
||||
browser = await playwright.chromium.launch(**options)
|
||||
# Setup context however you like.
|
||||
context = await browser.new_context() # Pass any options
|
||||
context = await set_init_script(context)
|
||||
# Pause the page, and start recording manually.
|
||||
page = await context.new_page()
|
||||
await page.goto("https://www.tiktok.com/login?lang=en")
|
||||
await page.pause()
|
||||
# 点击调试器的继续，保存cookie
|
||||
await context.storage_state(path=account_file)
|
||||
|
||||
|
||||
class TiktokVideo(object):
|
||||
def __init__(self, title, file_path, tags, publish_date: datetime, account_file, thumbnail_path=None):
|
||||
self.title = title
|
||||
self.file_path = file_path
|
||||
self.tags = tags
|
||||
self.publish_date = publish_date
|
||||
self.thumbnail_path = thumbnail_path
|
||||
self.account_file = account_file
|
||||
self.local_executable_path = LOCAL_CHROME_PATH
|
||||
|
||||
async def set_schedule_time(self, page, publish_date):
|
||||
schedule_input_element = page.frame_locator(Tk_Locator.tk_iframe).get_by_label('Schedule')
|
||||
await schedule_input_element.wait_for(state='visible') # 确保按钮可见
|
||||
|
||||
await schedule_input_element.click()
|
||||
if await page.frame_locator(Tk_Locator.tk_iframe).locator('div.TUXButton-content >> text=Allow').count():
|
||||
await page.frame_locator(Tk_Locator.tk_iframe).locator('div.TUXButton-content >> text=Allow').click()
|
||||
|
||||
scheduled_picker = page.frame_locator(Tk_Locator.tk_iframe).locator('div.scheduled-picker')
|
||||
await scheduled_picker.locator('div.TUXInputBox').nth(1).click()
|
||||
|
||||
calendar_month = await page.frame_locator(Tk_Locator.tk_iframe).locator(
|
||||
'div.calendar-wrapper span.month-title').inner_text()
|
||||
|
||||
n_calendar_month = datetime.strptime(calendar_month, '%B').month
|
||||
|
||||
schedule_month = publish_date.month
|
||||
|
||||
if n_calendar_month != schedule_month:
|
||||
if n_calendar_month < schedule_month:
|
||||
arrow = page.frame_locator(Tk_Locator.tk_iframe).locator('div.calendar-wrapper span.arrow').nth(-1)
|
||||
else:
|
||||
arrow = page.frame_locator(Tk_Locator.tk_iframe).locator('div.calendar-wrapper span.arrow').nth(0)
|
||||
await arrow.click()
|
||||
|
||||
# day set
|
||||
valid_days_locator = page.frame_locator(Tk_Locator.tk_iframe).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(publish_date.day):
|
||||
await day_element.click()
|
||||
break
|
||||
# time set
|
||||
await scheduled_picker.locator('div.TUXInputBox').nth(0).click()
|
||||
|
||||
hour_str = publish_date.strftime("%H")
|
||||
correct_minute = int(publish_date.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(500) # 等待500毫秒
|
||||
await page.frame_locator(Tk_Locator.tk_iframe).locator(hour_selector).click()
|
||||
# click time button again
|
||||
await page.wait_for_timeout(500) # 等待500毫秒
|
||||
await scheduled_picker.locator('div.TUXInputBox').nth(0).click()
|
||||
await page.wait_for_timeout(500) # 等待500毫秒
|
||||
# pick minutes after
|
||||
await page.frame_locator(Tk_Locator.tk_iframe).locator(minute_selector).click()
|
||||
|
||||
# click title to remove the focus.
|
||||
# await page.frame_locator(Tk_Locator.tk_iframe).locator("h1:has-text('Upload video')").click()
|
||||
|
||||
async def handle_upload_error(self, page):
|
||||
tiktok_logger.info("video upload error retrying.")
|
||||
select_file_button = page.frame_locator(Tk_Locator.tk_iframe).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.file_path)
|
||||
|
||||
async def upload(self, playwright: Playwright) -> None:
|
||||
browser = await playwright.chromium.launch(headless=False, executable_path=self.local_executable_path)
|
||||
context = await browser.new_context(storage_state=f"{self.account_file}")
|
||||
context = await set_init_script(context)
|
||||
page = await context.new_page()
|
||||
|
||||
# change language to eng first
|
||||
await self.change_language(page)
|
||||
await page.goto("https://www.tiktok.com/tiktokstudio/upload")
|
||||
tiktok_logger.info(f'[+]Uploading-------{self.title}.mp4')
|
||||
|
||||
await page.wait_for_url("https://www.tiktok.com/tiktokstudio/upload", timeout=10000)
|
||||
await page.wait_for_selector('iframe[data-tt="Upload_index_iframe"]')
|
||||
upload_button = page.frame_locator(Tk_Locator.tk_iframe).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.file_path)
|
||||
|
||||
await self.add_title_tags(page)
|
||||
# detact upload status
|
||||
await self.detact_upload_status(page)
|
||||
if self.thumbnail_path:
|
||||
tiktok_logger.info(f'[+] Uploading thumbnail file {self.title}.png')
|
||||
await self.upload_thumbnails(page)
|
||||
|
||||
if self.publish_date != 0:
|
||||
await self.set_schedule_time(page, self.publish_date)
|
||||
|
||||
await self.click_publish(page)
|
||||
|
||||
await context.storage_state(path=f"{self.account_file}") # save cookie
|
||||
tiktok_logger.info(' [-] update cookie！')
|
||||
await asyncio.sleep(2) # close delay for look the video status
|
||||
# close all
|
||||
await context.close()
|
||||
await browser.close()
|
||||
|
||||
async def add_title_tags(self, page):
|
||||
|
||||
editor_locator = page.frame_locator(Tk_Locator.tk_iframe).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):
|
||||
tiktok_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 page.frame_locator(Tk_Locator.tk_iframe).locator(".cover-selector-image-container").click()
|
||||
await page.frame_locator(Tk_Locator.tk_iframe).locator(".cover-edit-container >> text=Upload cover").click()
|
||||
async with page.expect_file_chooser() as fc_info:
|
||||
await page.frame_locator(Tk_Locator.tk_iframe).locator(".upload-image-upload-area").click()
|
||||
file_chooser = await fc_info.value
|
||||
await file_chooser.set_files(self.thumbnail_path)
|
||||
await page.frame_locator(Tk_Locator.tk_iframe).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("https://www.tiktok.com")
|
||||
await page.wait_for_url("https://www.tiktok.com/", timeout=100000)
|
||||
await page.wait_for_selector('#header-more-menu-icon')
|
||||
|
||||
await page.locator('#header-more-menu-icon').hover()
|
||||
await page.locator('[data-e2e="language-select"]').click()
|
||||
await page.locator('#lang-setting-popup-list >> text=English').click()
|
||||
|
||||
async def click_publish(self, page):
|
||||
success_flag_div = 'div.common-modal-confirm-modal'
|
||||
while True:
|
||||
try:
|
||||
publish_button = page.frame_locator(Tk_Locator.tk_iframe).locator('div.button-group button').nth(0)
|
||||
if await publish_button.count():
|
||||
await publish_button.click()
|
||||
|
||||
await page.frame_locator(Tk_Locator.tk_iframe).locator(success_flag_div).wait_for(state="visible", timeout=1500)
|
||||
tiktok_logger.success(" [-] video published success")
|
||||
break
|
||||
except Exception as e:
|
||||
if await page.frame_locator(Tk_Locator.tk_iframe).locator(success_flag_div).count():
|
||||
tiktok_logger.success(" [-]video published success")
|
||||
break
|
||||
else:
|
||||
tiktok_logger.exception(f" [-] Exception: {e}")
|
||||
tiktok_logger.info(" [-] video publishing")
|
||||
await page.screenshot(full_page=True)
|
||||
await asyncio.sleep(0.5)
|
||||
|
||||
async def detact_upload_status(self, page):
|
||||
while True:
|
||||
try:
|
||||
# if await page.frame_locator(Tk_Locator.tk_iframe).locator('div.btn-post > button').get_attribute("disabled") is None:
|
||||
if await page.frame_locator(Tk_Locator.tk_iframe).locator(
|
||||
'div.button-group > button >> text=Post').get_attribute("disabled") is None:
|
||||
tiktok_logger.info(" [-]video uploaded.")
|
||||
break
|
||||
else:
|
||||
tiktok_logger.info(" [-] video uploading...")
|
||||
await asyncio.sleep(2)
|
||||
if await page.frame_locator(Tk_Locator.tk_iframe).locator(
|
||||
'button[aria-label="Select file"]').count():
|
||||
tiktok_logger.info(" [-] found some error while uploading now retry...")
|
||||
await self.handle_upload_error(page)
|
||||
except:
|
||||
tiktok_logger.info(" [-] video uploading...")
|
||||
await asyncio.sleep(2)
|
||||
|
||||
async def main(self):
|
||||
async with async_playwright() as playwright:
|
||||
await self.upload(playwright)
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 103 KiB |
Reference in New Issue
Block a user