1. add kuaishou uploader
2. refactor the code
This commit is contained in:
pookz@stme
@@ -0,0 +1,5 @@
|
||||
from pathlib import Path
|
||||
|
||||
from conf import BASE_DIR
|
||||
|
||||
Path(BASE_DIR / "cookies").mkdir(exist_ok=True)
|
||||
@@ -0,0 +1,5 @@
|
||||
from pathlib import Path
|
||||
|
||||
from conf import BASE_DIR
|
||||
|
||||
Path(BASE_DIR / "cookies" / "bilibili_uploader").mkdir(exist_ok=True)
|
||||
Binary file not shown.
@@ -0,0 +1,79 @@
|
||||
import json
|
||||
import pathlib
|
||||
import random
|
||||
from biliup.plugins.bili_webup import BiliBili, Data
|
||||
|
||||
from utils.log import bilibili_logger
|
||||
|
||||
|
||||
def extract_keys_from_json(data):
|
||||
"""Extract specified keys from the provided JSON data."""
|
||||
keys_to_extract = ["SESSDATA", "bili_jct", "DedeUserID__ckMd5", "DedeUserID", "access_token"]
|
||||
extracted_data = {}
|
||||
|
||||
# Extracting cookie data
|
||||
for cookie in data['cookie_info']['cookies']:
|
||||
if cookie['name'] in keys_to_extract:
|
||||
extracted_data[cookie['name']] = cookie['value']
|
||||
|
||||
# Extracting access_token
|
||||
if "access_token" in data['token_info']:
|
||||
extracted_data['access_token'] = data['token_info']['access_token']
|
||||
|
||||
return extracted_data
|
||||
|
||||
|
||||
def read_cookie_json_file(filepath: pathlib.Path):
|
||||
with open(filepath, 'r', encoding='utf-8') as file:
|
||||
content = json.load(file)
|
||||
return content
|
||||
|
||||
|
||||
def random_emoji():
|
||||
emoji_list = ["🍏", "🍎", "🍊", "🍋", "🍌", "🍉", "🍇", "🍓", "🍈", "🍒", "🍑", "🍍", "🥭", "🥥", "🥝",
|
||||
"🍅", "🍆", "🥑", "🥦", "🥒", "🥬", "🌶", "🌽", "🥕", "🥔", "🍠", "🥐", "🍞", "🥖", "🥨", "🥯", "🧀", "🥚", "🍳", "🥞",
|
||||
"🥓", "🥩", "🍗", "🍖", "🌭", "🍔", "🍟", "🍕", "🥪", "🥙", "🌮", "🌯", "🥗", "🥘", "🥫", "🍝", "🍜", "🍲", "🍛", "🍣",
|
||||
"🍱", "🥟", "🍤", "🍙", "🍚", "🍘", "🍥", "🥮", "🥠", "🍢", "🍡", "🍧", "🍨", "🍦", "🥧", "🍰", "🎂", "🍮", "🍭", "🍬",
|
||||
"🍫", "🍿", "🧂", "🍩", "🍪", "🌰", "🥜", "🍯", "🥛", "🍼", "☕️", "🍵", "🥤", "🍶", "🍻", "🥂", "🍷", "🥃", "🍸", "🍹",
|
||||
"🍾", "🥄", "🍴", "🍽", "🥣", "🥡", "🥢"]
|
||||
return random.choice(emoji_list)
|
||||
|
||||
|
||||
class BilibiliUploader(object):
|
||||
def __init__(self, cookie_data, file: pathlib.Path, title, desc, tid, tags, dtime):
|
||||
self.upload_thread_num = 3
|
||||
self.copyright = 1
|
||||
self.lines = 'AUTO'
|
||||
self.cookie_data = cookie_data
|
||||
self.file = file
|
||||
self.title = title
|
||||
self.desc = desc
|
||||
self.tid = tid
|
||||
self.tags = tags
|
||||
self.dtime = dtime
|
||||
self._init_data()
|
||||
|
||||
def _init_data(self):
|
||||
self.data = Data()
|
||||
self.data.copyright = self.copyright
|
||||
self.data.title = self.title
|
||||
self.data.desc = self.desc
|
||||
self.data.tid = self.tid
|
||||
self.data.set_tag(self.tags)
|
||||
self.data.dtime = self.dtime
|
||||
|
||||
def upload(self):
|
||||
with BiliBili(self.data) as bili:
|
||||
bili.login_by_cookies(self.cookie_data)
|
||||
bili.access_token = self.cookie_data.get('access_token')
|
||||
video_part = bili.upload_file(str(self.file), lines=self.lines,
|
||||
tasks=self.upload_thread_num) # 上传视频，默认线路AUTO自动选择，线程数量3。
|
||||
video_part['title'] = self.title
|
||||
self.data.append(video_part)
|
||||
ret = bili.submit() # 提交视频
|
||||
if ret.get('code') == 0:
|
||||
bilibili_logger.success(f'[+] {self.file.name}上传 成功')
|
||||
return True
|
||||
else:
|
||||
bilibili_logger.error(f'[-] {self.file.name}上传 失败, error messge: {ret.get("message")}')
|
||||
return False
|
||||
@@ -0,0 +1,5 @@
|
||||
from pathlib import Path
|
||||
|
||||
from conf import BASE_DIR
|
||||
|
||||
Path(BASE_DIR / "cookies" / "douyin_uploader").mkdir(exist_ok=True)
|
||||
@@ -0,0 +1,211 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
from datetime import datetime
|
||||
|
||||
from playwright.async_api import Playwright, async_playwright
|
||||
import os
|
||||
import asyncio
|
||||
|
||||
from conf import LOCAL_CHROME_PATH
|
||||
from utils.base_social_media import set_init_script
|
||||
from utils.log import douyin_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://creator.douyin.com/creator-micro/content/upload")
|
||||
# 2024.06.17 抖音创作者中心改版
|
||||
if await page.get_by_text('手机号登录').count():
|
||||
print("[+] 等待5秒 cookie 失效")
|
||||
return False
|
||||
else:
|
||||
print("[+] cookie 有效")
|
||||
return True
|
||||
|
||||
|
||||
async def douyin_setup(account_file, handle=False):
|
||||
if not os.path.exists(account_file) or not await cookie_auth(account_file):
|
||||
if not handle:
|
||||
# Todo alert message
|
||||
return False
|
||||
douyin_logger.info('[+] cookie文件不存在或已失效，即将自动打开浏览器，请扫码登录，登陆后会自动生成cookie文件')
|
||||
await douyin_cookie_gen(account_file)
|
||||
return True
|
||||
|
||||
|
||||
async def douyin_cookie_gen(account_file):
|
||||
async with async_playwright() as playwright:
|
||||
options = {
|
||||
'headless': False
|
||||
}
|
||||
# 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://creator.douyin.com/")
|
||||
await page.pause()
|
||||
# 点击调试器的继续，保存cookie
|
||||
await context.storage_state(path=account_file)
|
||||
|
||||
|
||||
class DouYinVideo(object):
|
||||
def __init__(self, title, file_path, tags, publish_date: datetime, account_file):
|
||||
self.title = title # 视频标题
|
||||
self.file_path = file_path
|
||||
self.tags = tags
|
||||
self.publish_date = publish_date
|
||||
self.account_file = account_file
|
||||
self.date_format = '%Y年%m月%d日 %H:%M'
|
||||
self.local_executable_path = LOCAL_CHROME_PATH
|
||||
|
||||
async def set_schedule_time_douyin(self, page, publish_date):
|
||||
# 选择包含特定文本内容的 label 元素
|
||||
label_element = page.locator("[class^='radio']:has-text('定时发布')")
|
||||
# 在选中的 label 元素下点击 checkbox
|
||||
await label_element.click()
|
||||
await asyncio.sleep(1)
|
||||
publish_date_hour = publish_date.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(publish_date_hour))
|
||||
await page.keyboard.press("Enter")
|
||||
|
||||
await asyncio.sleep(1)
|
||||
|
||||
async def handle_upload_error(self, page):
|
||||
douyin_logger.info('视频出错了，重新上传中')
|
||||
await page.locator('div.progress-div [class^="upload-btn-input"]').set_input_files(self.file_path)
|
||||
|
||||
async def upload(self, playwright: Playwright) -> None:
|
||||
# 使用 Chromium 浏览器启动一个浏览器实例
|
||||
if self.local_executable_path:
|
||||
browser = await playwright.chromium.launch(headless=False, executable_path=self.local_executable_path)
|
||||
else:
|
||||
browser = await playwright.chromium.launch(headless=False)
|
||||
# 创建一个浏览器上下文，使用指定的 cookie 文件
|
||||
context = await browser.new_context(storage_state=f"{self.account_file}")
|
||||
context = await set_init_script(context)
|
||||
|
||||
# 创建一个新的页面
|
||||
page = await context.new_page()
|
||||
# 访问指定的 URL
|
||||
await page.goto("https://creator.douyin.com/creator-micro/content/upload")
|
||||
douyin_logger.info(f'[+]正在上传-------{self.title}.mp4')
|
||||
# 等待页面跳转到指定的 URL，没进入，则自动等待到超时
|
||||
douyin_logger.info(f'[-] 正在打开主页...')
|
||||
await page.wait_for_url("https://creator.douyin.com/creator-micro/content/upload")
|
||||
# 点击 "上传视频" 按钮
|
||||
await page.locator("[name='upload-btn']").set_input_files(self.file_path)
|
||||
|
||||
# 等待页面跳转到指定的 URL
|
||||
while True:
|
||||
# 判断是是否进入视频发布页面，没进入，则自动等待到超时
|
||||
try:
|
||||
await page.wait_for_url(
|
||||
"https://creator.douyin.com/creator-micro/content/publish?enter_from=publish_page")
|
||||
break
|
||||
except:
|
||||
douyin_logger.info(f' [-] 正在等待进入视频发布页面...')
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
# 填充标题和话题
|
||||
# 检查是否存在包含输入框的元素
|
||||
# 这里为了避免页面变化，故使用相对位置定位：作品标题父级右侧第一个元素的input子元素
|
||||
await asyncio.sleep(1)
|
||||
douyin_logger.info(f' [-] 正在填充标题和话题...')
|
||||
title_container = page.get_by_text('作品标题').locator("..").locator("xpath=following-sibling::div[1]").locator("input")
|
||||
if await title_container.count():
|
||||
await title_container.fill(self.title[:30])
|
||||
else:
|
||||
titlecontainer = page.locator(".notranslate")
|
||||
await titlecontainer.click()
|
||||
await page.keyboard.press("Backspace")
|
||||
await page.keyboard.press("Control+KeyA")
|
||||
await page.keyboard.press("Delete")
|
||||
await page.keyboard.type(self.title)
|
||||
await page.keyboard.press("Enter")
|
||||
css_selector = ".zone-container"
|
||||
for index, tag in enumerate(self.tags, start=1):
|
||||
await page.type(css_selector, "#" + tag)
|
||||
await page.press(css_selector, "Space")
|
||||
douyin_logger.info(f'总共添加{len(self.tags)}个话题')
|
||||
|
||||
while True:
|
||||
# 判断重新上传按钮是否存在，如果不存在，代表视频正在上传，则等待
|
||||
try:
|
||||
# 新版：定位重新上传
|
||||
number = await page.locator('div label+div:has-text("重新上传")').count()
|
||||
if number > 0:
|
||||
douyin_logger.success(" [-]视频上传完毕")
|
||||
break
|
||||
else:
|
||||
douyin_logger.info(" [-] 正在上传视频中...")
|
||||
await asyncio.sleep(2)
|
||||
|
||||
if await page.locator('div.progress-div > div:has-text("上传失败")').count():
|
||||
douyin_logger.error(" [-] 发现上传出错了... 准备重试")
|
||||
await self.handle_upload_error(page)
|
||||
except:
|
||||
douyin_logger.info(" [-] 正在上传视频中...")
|
||||
await asyncio.sleep(2)
|
||||
|
||||
# 更换可见元素
|
||||
await page.locator('div.semi-select span:has-text("输入地理位置")').click()
|
||||
await page.keyboard.press("Backspace")
|
||||
await page.wait_for_timeout(2000)
|
||||
# await page.keyboard.press("Control+KeyA")
|
||||
# await page.keyboard.press("Delete")
|
||||
await page.keyboard.type("杭州市")
|
||||
# await asyncio.sleep(1)
|
||||
await page.wait_for_timeout(1000)
|
||||
await page.locator('div[role="listbox"] [role="option"]').first.click()
|
||||
|
||||
# 頭條/西瓜
|
||||
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()
|
||||
|
||||
if self.publish_date != 0:
|
||||
await self.set_schedule_time_douyin(page, self.publish_date)
|
||||
|
||||
# 判断视频是否发布成功
|
||||
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("https://creator.douyin.com/creator-micro/content/manage**",
|
||||
timeout=3000) # 如果自动跳转到作品页面，则代表发布成功
|
||||
douyin_logger.success(" [-]视频发布成功")
|
||||
break
|
||||
except:
|
||||
douyin_logger.info(" [-] 视频正在发布中...")
|
||||
await page.screenshot(full_page=True)
|
||||
await asyncio.sleep(0.5)
|
||||
|
||||
await context.storage_state(path=self.account_file) # 保存cookie
|
||||
douyin_logger.success(' [-]cookie更新完毕！')
|
||||
await asyncio.sleep(2) # 这里延迟是为了方便眼睛直观的观看
|
||||
# 关闭浏览器上下文和浏览器实例
|
||||
await context.close()
|
||||
await browser.close()
|
||||
|
||||
async def main(self):
|
||||
async with async_playwright() as playwright:
|
||||
await self.upload(playwright)
|
||||
|
||||
|
||||
@@ -0,0 +1,5 @@
|
||||
from pathlib import Path
|
||||
|
||||
from conf import BASE_DIR
|
||||
|
||||
Path(BASE_DIR / "cookies" / "ks_uploader").mkdir(exist_ok=True)
|
||||
@@ -0,0 +1,189 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
from datetime import datetime
|
||||
|
||||
from playwright.async_api import Playwright, async_playwright
|
||||
import os
|
||||
import asyncio
|
||||
|
||||
from utils.base_social_media import set_init_script
|
||||
from utils.files_times import get_absolute_path
|
||||
from utils.log import kuaishou_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://cp.kuaishou.com/article/publish/video")
|
||||
try:
|
||||
await page.wait_for_selector("div.names div.container div.name:text('机构服务')", timeout=5000) # 等待5秒
|
||||
|
||||
kuaishou_logger.info("[+] 等待5秒 cookie 失效")
|
||||
return False
|
||||
except:
|
||||
kuaishou_logger.success("[+] cookie 有效")
|
||||
return True
|
||||
|
||||
|
||||
async def ks_setup(account_file, handle=False):
|
||||
account_file = get_absolute_path(account_file, "ks_uploader")
|
||||
if not os.path.exists(account_file) or not await cookie_auth(account_file):
|
||||
if not handle:
|
||||
return False
|
||||
kuaishou_logger.info('[+] cookie文件不存在或已失效，即将自动打开浏览器，请扫码登录，登陆后会自动生成cookie文件')
|
||||
await get_ks_cookie(account_file)
|
||||
return True
|
||||
|
||||
|
||||
async def get_ks_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://cp.kuaishou.com")
|
||||
await page.pause()
|
||||
# 点击调试器的继续，保存cookie
|
||||
await context.storage_state(path=account_file)
|
||||
|
||||
|
||||
class KSVideo(object):
|
||||
def __init__(self, title, file_path, tags, publish_date: datetime, account_file):
|
||||
self.title = title # 视频标题
|
||||
self.file_path = file_path
|
||||
self.tags = tags
|
||||
self.publish_date = publish_date
|
||||
self.account_file = account_file
|
||||
self.date_format = '%Y-%m-%d %H:%M'
|
||||
|
||||
async def handle_upload_error(self, page):
|
||||
kuaishou_logger.error("视频出错了，重新上传中")
|
||||
await page.locator('div.progress-div [class^="upload-btn-input"]').set_input_files(self.file_path)
|
||||
|
||||
async def upload(self, playwright: Playwright) -> None:
|
||||
# 使用 Chromium 浏览器启动一个浏览器实例
|
||||
browser = await playwright.chromium.launch(headless=False)
|
||||
# 创建一个浏览器上下文，使用指定的 cookie 文件
|
||||
context = await browser.new_context(storage_state=f"{self.account_file}")
|
||||
context = await set_init_script(context)
|
||||
|
||||
# 创建一个新的页面
|
||||
page = await context.new_page()
|
||||
# 访问指定的 URL
|
||||
await page.goto("https://cp.kuaishou.com/article/publish/video")
|
||||
kuaishou_logger.info('正在上传-------{}.mp4'.format(self.title))
|
||||
# 等待页面跳转到指定的 URL，没进入，则自动等待到超时
|
||||
kuaishou_logger.info('正在打开主页...')
|
||||
await page.wait_for_url("https://cp.kuaishou.com/article/publish/video")
|
||||
# 点击 "上传视频" 按钮
|
||||
await page.locator("div.vVExjn9O3UQ- input").set_input_files(self.file_path)
|
||||
|
||||
await asyncio.sleep(2)
|
||||
|
||||
if not await page.get_by_text("封面编辑").count():
|
||||
raise Exception("似乎没有跳转到到编辑页面")
|
||||
|
||||
await asyncio.sleep(1)
|
||||
|
||||
# 等待按钮可交互
|
||||
new_feature_button = page.locator('button[type="button"] span:text("我知道了")')
|
||||
if await new_feature_button.count() > 0:
|
||||
await new_feature_button.click()
|
||||
|
||||
kuaishou_logger.info("正在填充标题和话题...")
|
||||
await page.get_by_text('填写描述').locator("xpath=following-sibling::div").click()
|
||||
kuaishou_logger.info("clear existing title")
|
||||
await page.keyboard.press("Backspace")
|
||||
await page.keyboard.press("Control+KeyA")
|
||||
await page.keyboard.press("Delete")
|
||||
kuaishou_logger.info("filling new title")
|
||||
await page.keyboard.type(self.title)
|
||||
await page.keyboard.press("Enter")
|
||||
|
||||
# 快手只能添加3个话题
|
||||
for index, tag in enumerate(self.tags[:3], start=1):
|
||||
kuaishou_logger.info("正在添加第%s个话题" % index)
|
||||
await page.locator('span:text("#话题")').click()
|
||||
await page.type('div.clGhv3UpdEo-', tag, delay=100)
|
||||
await asyncio.sleep(2)
|
||||
await page.locator('div.FZcv90s7kFs- > div').nth(0).click()
|
||||
|
||||
while True:
|
||||
try:
|
||||
number = await page.locator('div > span:text("上传成功")').count()
|
||||
if number > 0:
|
||||
kuaishou_logger.success("视频上传完毕")
|
||||
break
|
||||
else:
|
||||
kuaishou_logger.info("正在上传视频中...")
|
||||
await asyncio.sleep(2)
|
||||
except:
|
||||
kuaishou_logger.info("正在上传视频中...")
|
||||
await asyncio.sleep(2)
|
||||
|
||||
# 定时任务
|
||||
if self.publish_date != 0:
|
||||
await self.set_schedule_time(page, self.publish_date)
|
||||
|
||||
# 判断视频是否发布成功
|
||||
while True:
|
||||
# 判断视频是否发布成功
|
||||
try:
|
||||
publish_button = page.get_by_role('button', name="发布", exact=True)
|
||||
if await publish_button.count():
|
||||
await publish_button.click()
|
||||
|
||||
await asyncio.sleep(1)
|
||||
confirm_button = page.locator("button > span:text('确认发布')")
|
||||
if await confirm_button.count():
|
||||
await page.locator("button > span:text('确认发布')").click()
|
||||
|
||||
await page.wait_for_url("https://cp.kuaishou.com/article/manage/video?status=2&from=publish",
|
||||
timeout=1500)
|
||||
kuaishou_logger.success("视频发布成功")
|
||||
break
|
||||
except:
|
||||
kuaishou_logger.info("视频正在发布中...")
|
||||
await page.screenshot(full_page=True)
|
||||
await asyncio.sleep(0.5)
|
||||
|
||||
await context.storage_state(path=self.account_file) # 保存cookie
|
||||
kuaishou_logger.info('cookie更新完毕！')
|
||||
await asyncio.sleep(2) # 这里延迟是为了方便眼睛直观的观看
|
||||
# 关闭浏览器上下文和浏览器实例
|
||||
await context.close()
|
||||
await browser.close()
|
||||
|
||||
async def main(self):
|
||||
async with async_playwright() as playwright:
|
||||
await self.upload(playwright)
|
||||
|
||||
async def set_schedule_time(self, page, publish_date):
|
||||
kuaishou_logger.info("click schedule")
|
||||
publish_date_hour = publish_date.strftime("%Y-%m-%d %H:%M:%S")
|
||||
await page.locator("label:text('发布时间')").locator('xpath=following-sibling::div').locator(
|
||||
'.ant-radio-input').nth(1).click()
|
||||
await asyncio.sleep(1)
|
||||
|
||||
await page.locator('div.ant-picker-input input[placeholder="选择日期时间"]').click()
|
||||
await asyncio.sleep(1)
|
||||
|
||||
await page.keyboard.press("Control+KeyA")
|
||||
await page.keyboard.type(str(publish_date_hour))
|
||||
await page.keyboard.press("Enter")
|
||||
await asyncio.sleep(1)
|
||||
|
||||
|
||||
@@ -0,0 +1,5 @@
|
||||
from pathlib import Path
|
||||
|
||||
from conf import BASE_DIR
|
||||
|
||||
Path(BASE_DIR / "cookies" / "tencent_uploader").mkdir(exist_ok=True)
|
||||
@@ -0,0 +1,267 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
from datetime import datetime
|
||||
|
||||
from playwright.async_api import Playwright, async_playwright
|
||||
import os
|
||||
import asyncio
|
||||
|
||||
from conf import LOCAL_CHROME_PATH
|
||||
from utils.base_social_media import set_init_script
|
||||
from utils.files_times import get_absolute_path
|
||||
from utils.log import tencent_logger
|
||||
|
||||
|
||||
def format_str_for_short_title(origin_title: str) -> str:
|
||||
# 定义允许的特殊字符
|
||||
allowed_special_chars = "《》“”:+?%°"
|
||||
|
||||
# 移除不允许的特殊字符
|
||||
filtered_chars = [char if char.isalnum() or char in allowed_special_chars else ' ' if char == ',' else '' for
|
||||
char in origin_title]
|
||||
formatted_string = ''.join(filtered_chars)
|
||||
|
||||
# 调整字符串长度
|
||||
if len(formatted_string) > 16:
|
||||
# 截断字符串
|
||||
formatted_string = formatted_string[:16]
|
||||
elif len(formatted_string) < 6:
|
||||
# 使用空格来填充字符串
|
||||
formatted_string += ' ' * (6 - len(formatted_string))
|
||||
|
||||
return formatted_string
|
||||
|
||||
|
||||
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://channels.weixin.qq.com/platform/post/create")
|
||||
try:
|
||||
await page.wait_for_selector('div.title-name:has-text("视频号小店")', timeout=5000) # 等待5秒
|
||||
tencent_logger.error("[+] 等待5秒 cookie 失效")
|
||||
return False
|
||||
except:
|
||||
tencent_logger.success("[+] cookie 有效")
|
||||
return True
|
||||
|
||||
|
||||
async def get_tencent_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
|
||||
# Pause the page, and start recording manually.
|
||||
context = await set_init_script(context)
|
||||
page = await context.new_page()
|
||||
await page.goto("https://channels.weixin.qq.com")
|
||||
await page.pause()
|
||||
# 点击调试器的继续，保存cookie
|
||||
await context.storage_state(path=account_file)
|
||||
|
||||
|
||||
async def weixin_setup(account_file, handle=False):
|
||||
account_file = get_absolute_path(account_file, "tencent_uploader")
|
||||
if not os.path.exists(account_file) or not await cookie_auth(account_file):
|
||||
if not handle:
|
||||
# Todo alert message
|
||||
return False
|
||||
tencent_logger.info('[+] cookie文件不存在或已失效，即将自动打开浏览器，请扫码登录，登陆后会自动生成cookie文件')
|
||||
await get_tencent_cookie(account_file)
|
||||
return True
|
||||
|
||||
|
||||
class TencentVideo(object):
|
||||
def __init__(self, title, file_path, tags, publish_date: datetime, account_file, category=None):
|
||||
self.title = title # 视频标题
|
||||
self.file_path = file_path
|
||||
self.tags = tags
|
||||
self.publish_date = publish_date
|
||||
self.account_file = account_file
|
||||
self.category = category
|
||||
self.local_executable_path = LOCAL_CHROME_PATH
|
||||
|
||||
async def set_schedule_time_tencent(self, page, publish_date):
|
||||
label_element = page.locator("label").filter(has_text="定时").nth(1)
|
||||
await label_element.click()
|
||||
|
||||
await page.click('input[placeholder="请选择发表时间"]')
|
||||
|
||||
str_month = str(publish_date.month) if publish_date.month > 9 else "0" + str(publish_date.month)
|
||||
current_month = str_month + "月"
|
||||
# 获取当前的月份
|
||||
page_month = await page.inner_text('span.weui-desktop-picker__panel__label:has-text("月")')
|
||||
|
||||
# 检查当前月份是否与目标月份相同
|
||||
if page_month != current_month:
|
||||
await page.click('button.weui-desktop-btn__icon__right')
|
||||
|
||||
# 获取页面元素
|
||||
elements = await page.query_selector_all('table.weui-desktop-picker__table a')
|
||||
|
||||
# 遍历元素并点击匹配的元素
|
||||
for element in elements:
|
||||
if 'weui-desktop-picker__disabled' in await element.evaluate('el => el.className'):
|
||||
continue
|
||||
text = await element.inner_text()
|
||||
if text.strip() == str(publish_date.day):
|
||||
await element.click()
|
||||
break
|
||||
|
||||
# 输入小时部分（假设选择11小时）
|
||||
await page.click('input[placeholder="请选择时间"]')
|
||||
await page.keyboard.press("Control+KeyA")
|
||||
await page.keyboard.type(str(publish_date.hour))
|
||||
|
||||
# 选择标题栏（令定时时间生效）
|
||||
await page.locator("div.input-editor").click()
|
||||
|
||||
async def handle_upload_error(self, page):
|
||||
tencent_logger.info("视频出错了，重新上传中")
|
||||
await page.locator('div.media-status-content div.tag-inner:has-text("删除")').click()
|
||||
await page.get_by_role('button', name="删除", exact=True).click()
|
||||
file_input = page.locator('input[type="file"]')
|
||||
await file_input.set_input_files(self.file_path)
|
||||
|
||||
async def upload(self, playwright: Playwright) -> None:
|
||||
# 使用 Chromium (这里使用系统内浏览器，用chromium 会造成h264错误
|
||||
browser = await playwright.chromium.launch(headless=False, executable_path=self.local_executable_path)
|
||||
# 创建一个浏览器上下文，使用指定的 cookie 文件
|
||||
context = await browser.new_context(storage_state=f"{self.account_file}")
|
||||
context = await set_init_script(context)
|
||||
|
||||
# 创建一个新的页面
|
||||
page = await context.new_page()
|
||||
# 访问指定的 URL
|
||||
await page.goto("https://channels.weixin.qq.com/platform/post/create")
|
||||
tencent_logger.info(f'[+]正在上传-------{self.title}.mp4')
|
||||
# 等待页面跳转到指定的 URL，没进入，则自动等待到超时
|
||||
await page.wait_for_url("https://channels.weixin.qq.com/platform/post/create")
|
||||
# await page.wait_for_selector('input[type="file"]', timeout=10000)
|
||||
file_input = page.locator('input[type="file"]')
|
||||
await file_input.set_input_files(self.file_path)
|
||||
# 填充标题和话题
|
||||
await self.add_title_tags(page)
|
||||
# 添加商品
|
||||
# await self.add_product(page)
|
||||
# 合集功能
|
||||
await self.add_collection(page)
|
||||
# 原创选择
|
||||
await self.add_original(page)
|
||||
# 检测上传状态
|
||||
await self.detect_upload_status(page)
|
||||
if self.publish_date != 0:
|
||||
await self.set_schedule_time_tencent(page, self.publish_date)
|
||||
# 添加短标题
|
||||
await self.add_short_title(page)
|
||||
|
||||
await self.click_publish(page)
|
||||
|
||||
await context.storage_state(path=f"{self.account_file}") # 保存cookie
|
||||
tencent_logger.success(' [-]cookie更新完毕！')
|
||||
await asyncio.sleep(2) # 这里延迟是为了方便眼睛直观的观看
|
||||
# 关闭浏览器上下文和浏览器实例
|
||||
await context.close()
|
||||
await browser.close()
|
||||
|
||||
async def add_short_title(self, page):
|
||||
short_title_element = page.get_by_text("短标题", exact=True).locator("..").locator(
|
||||
"xpath=following-sibling::div").locator(
|
||||
'span input[type="text"]')
|
||||
if await short_title_element.count():
|
||||
short_title = format_str_for_short_title(self.title)
|
||||
await short_title_element.fill(short_title)
|
||||
|
||||
async def click_publish(self, page):
|
||||
while True:
|
||||
try:
|
||||
publish_buttion = page.locator('div.form-btns button:has-text("发表")')
|
||||
if await publish_buttion.count():
|
||||
await publish_buttion.click()
|
||||
await page.wait_for_url("https://channels.weixin.qq.com/platform/post/list", timeout=1500)
|
||||
tencent_logger.success(" [-]视频发布成功")
|
||||
break
|
||||
except Exception as e:
|
||||
current_url = page.url
|
||||
if "https://channels.weixin.qq.com/platform/post/list" in current_url:
|
||||
tencent_logger.success(" [-]视频发布成功")
|
||||
break
|
||||
else:
|
||||
tencent_logger.exception(f" [-] Exception: {e}")
|
||||
tencent_logger.info(" [-] 视频正在发布中...")
|
||||
await asyncio.sleep(0.5)
|
||||
|
||||
async def detect_upload_status(self, page):
|
||||
while True:
|
||||
# 匹配删除按钮，代表视频上传完毕，如果不存在，代表视频正在上传，则等待
|
||||
try:
|
||||
# 匹配删除按钮，代表视频上传完毕
|
||||
if "weui-desktop-btn_disabled" not in await page.get_by_role("button", name="发表").get_attribute(
|
||||
'class'):
|
||||
tencent_logger.info(" [-]视频上传完毕")
|
||||
break
|
||||
else:
|
||||
tencent_logger.info(" [-] 正在上传视频中...")
|
||||
await asyncio.sleep(2)
|
||||
# 出错了视频出错
|
||||
if await page.locator('div.status-msg.error').count() and await page.locator(
|
||||
'div.media-status-content div.tag-inner:has-text("删除")').count():
|
||||
tencent_logger.error(" [-] 发现上传出错了...准备重试")
|
||||
await self.handle_upload_error(page)
|
||||
except:
|
||||
tencent_logger.info(" [-] 正在上传视频中...")
|
||||
await asyncio.sleep(2)
|
||||
|
||||
async def add_title_tags(self, page):
|
||||
await page.locator("div.input-editor").click()
|
||||
await page.keyboard.type(self.title)
|
||||
await page.keyboard.press("Enter")
|
||||
for index, tag in enumerate(self.tags, start=1):
|
||||
await page.keyboard.type("#" + tag)
|
||||
await page.keyboard.press("Space")
|
||||
tencent_logger.info(f"成功添加hashtag: {len(self.tags)}")
|
||||
|
||||
async def add_collection(self, page):
|
||||
collection_elements = page.get_by_text("添加到合集").locator("xpath=following-sibling::div").locator(
|
||||
'.option-list-wrap > div')
|
||||
if await collection_elements.count() > 1:
|
||||
await page.get_by_text("添加到合集").locator("xpath=following-sibling::div").click()
|
||||
await collection_elements.first.click()
|
||||
|
||||
async def add_original(self, page):
|
||||
if await page.get_by_label("视频为原创").count():
|
||||
await page.get_by_label("视频为原创").check()
|
||||
# 检查 "我已阅读并同意 《视频号原创声明使用条款》" 元素是否存在
|
||||
label_locator = await page.locator('label:has-text("我已阅读并同意 《视频号原创声明使用条款》")').is_visible()
|
||||
if label_locator:
|
||||
await page.get_by_label("我已阅读并同意 《视频号原创声明使用条款》").check()
|
||||
await page.get_by_role("button", name="声明原创").click()
|
||||
# 2023年11月20日 wechat更新: 可能新账号或者改版账号，出现新的选择页面
|
||||
if await page.locator('div.label span:has-text("声明原创")').count() and self.category:
|
||||
# 因处罚无法勾选原创，故先判断是否可用
|
||||
if not await page.locator('div.declare-original-checkbox input.ant-checkbox-input').is_disabled():
|
||||
await page.locator('div.declare-original-checkbox input.ant-checkbox-input').click()
|
||||
if not await page.locator(
|
||||
'div.declare-original-dialog label.ant-checkbox-wrapper.ant-checkbox-wrapper-checked:visible').count():
|
||||
await page.locator('div.declare-original-dialog input.ant-checkbox-input:visible').click()
|
||||
if await page.locator('div.original-type-form > div.form-label:has-text("原创类型"):visible').count():
|
||||
await page.locator('div.form-content:visible').click() # 下拉菜单
|
||||
await page.locator(
|
||||
f'div.form-content:visible ul.weui-desktop-dropdown__list li.weui-desktop-dropdown__list-ele:has-text("{self.category}")').first.click()
|
||||
await page.wait_for_timeout(1000)
|
||||
if await page.locator('button:has-text("声明原创"):visible').count():
|
||||
await page.locator('button:has-text("声明原创"):visible').click()
|
||||
|
||||
async def main(self):
|
||||
async with async_playwright() as playwright:
|
||||
await self.upload(playwright)
|
||||
@@ -0,0 +1,5 @@
|
||||
from pathlib import Path
|
||||
|
||||
from conf import BASE_DIR
|
||||
|
||||
Path(BASE_DIR / "cookies" / "tk_uploader").mkdir(exist_ok=True)
|
||||
@@ -0,0 +1,265 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
import re
|
||||
from datetime import datetime
|
||||
|
||||
from playwright.async_api import Playwright, async_playwright
|
||||
import os
|
||||
import asyncio
|
||||
from uploader.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.firefox.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.firefox.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, account_file):
|
||||
self.title = title
|
||||
self.file_path = file_path
|
||||
self.tags = tags
|
||||
self.publish_date = publish_date
|
||||
self.account_file = account_file
|
||||
self.locator_base = None
|
||||
|
||||
|
||||
async def set_schedule_time(self, page, publish_date):
|
||||
schedule_input_element = self.locator_base.get_by_label('Schedule')
|
||||
await schedule_input_element.wait_for(state='visible') # 确保按钮可见
|
||||
|
||||
await schedule_input_element.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 = publish_date.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(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 self.locator_base.locator(hour_selector).click()
|
||||
# click time button again
|
||||
# 等待某个特定的元素出现或状态变化，表明UI已更新
|
||||
await page.wait_for_timeout(1000) # 等待500毫秒
|
||||
await scheduled_picker.locator('div.TUXInputBox').nth(0).click()
|
||||
# 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):
|
||||
tiktok_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.file_path)
|
||||
|
||||
async def upload(self, playwright: Playwright) -> None:
|
||||
browser = await playwright.firefox.launch(headless=False)
|
||||
context = await browser.new_context(storage_state=f"{self.account_file}")
|
||||
context = await set_init_script(context)
|
||||
page = await context.new_page()
|
||||
|
||||
await page.goto("https://www.tiktok.com/creator-center/upload")
|
||||
tiktok_logger.info(f'[+]Uploading-------{self.title}.mp4')
|
||||
|
||||
await page.wait_for_url("https://www.tiktok.com/tiktokstudio/upload", timeout=10000)
|
||||
|
||||
try:
|
||||
await page.wait_for_selector('iframe[data-tt="Upload_index_iframe"], div.upload-container', timeout=10000)
|
||||
tiktok_logger.info("Either iframe or div appeared.")
|
||||
except Exception as e:
|
||||
tiktok_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.file_path)
|
||||
|
||||
await self.add_title_tags(page)
|
||||
# detact upload status
|
||||
await self.detect_upload_status(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 = 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):
|
||||
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 click_publish(self, page):
|
||||
success_flag_div = '#\\:r9\\:'
|
||||
while True:
|
||||
try:
|
||||
publish_button = self.locator_base.locator('div.btn-post')
|
||||
if await publish_button.count():
|
||||
await publish_button.click()
|
||||
|
||||
await self.locator_base.locator(success_flag_div).wait_for(state="visible", timeout=3000)
|
||||
tiktok_logger.success(" [-] video published success")
|
||||
break
|
||||
except Exception as e:
|
||||
if await self.locator_base.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 detect_upload_status(self, page):
|
||||
while True:
|
||||
try:
|
||||
if await self.locator_base.locator('div.btn-post > button').get_attribute("disabled") is None:
|
||||
tiktok_logger.info(" [-]video uploaded.")
|
||||
break
|
||||
else:
|
||||
tiktok_logger.info(" [-] video uploading...")
|
||||
await asyncio.sleep(2)
|
||||
if await self.locator_base.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 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 = self.locator_base
|
||||
else:
|
||||
self.locator_base = page.locator(Tk_Locator.default)
|
||||
|
||||
async def main(self):
|
||||
async with async_playwright() as playwright:
|
||||
await self.upload(playwright)
|
||||
|
||||
@@ -0,0 +1,302 @@
|
||||
# -*- 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 uploader.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, 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
|
||||
self.locator_base = None
|
||||
|
||||
async def set_schedule_time(self, page, publish_date):
|
||||
schedule_input_element = self.locator_base.get_by_label('Schedule')
|
||||
await schedule_input_element.wait_for(state='visible') # 确保按钮可见
|
||||
|
||||
await schedule_input_element.click()
|
||||
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 = publish_date.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(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 self.locator_base.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 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):
|
||||
tiktok_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.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)
|
||||
|
||||
try:
|
||||
await page.wait_for_selector('iframe[data-tt="Upload_index_iframe"], div.upload-container', timeout=10000)
|
||||
tiktok_logger.info("Either iframe or div appeared.")
|
||||
except Exception as e:
|
||||
tiktok_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.file_path)
|
||||
|
||||
await self.add_title_tags(page)
|
||||
# detect upload status
|
||||
await self.detect_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 = 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):
|
||||
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 self.locator_base.locator(".cover-selector-image-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.thumbnail_path)
|
||||
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("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 = self.locator_base.locator('div.button-group button').nth(0)
|
||||
if await publish_button.count():
|
||||
await publish_button.click()
|
||||
|
||||
await self.locator_base.locator(success_flag_div).wait_for(state="visible", timeout=3000)
|
||||
tiktok_logger.success(" [-] video published success")
|
||||
break
|
||||
except Exception as e:
|
||||
if await self.locator_base.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 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:
|
||||
tiktok_logger.info(" [-]video uploaded.")
|
||||
break
|
||||
else:
|
||||
tiktok_logger.info(" [-] video uploading...")
|
||||
await asyncio.sleep(2)
|
||||
if await self.locator_base.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 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 main(self):
|
||||
async with async_playwright() as playwright:
|
||||
await self.upload(playwright)
|
||||
@@ -0,0 +1,4 @@
|
||||
|
||||
class Tk_Locator(object):
|
||||
tk_iframe = '[data-tt="Upload_index_iframe"]'
|
||||
default = 'body'
|
||||
@@ -0,0 +1,2 @@
|
||||
[account1]
|
||||
cookies = changeme
|
||||
@@ -0,0 +1,58 @@
|
||||
import configparser
|
||||
import json
|
||||
import pathlib
|
||||
from time import sleep
|
||||
|
||||
import requests
|
||||
from playwright.sync_api import sync_playwright
|
||||
|
||||
from conf import BASE_DIR, XHS_SERVER
|
||||
|
||||
config = configparser.RawConfigParser()
|
||||
config.read('accounts.ini')
|
||||
|
||||
|
||||
def sign_local(uri, data=None, a1="", web_session=""):
|
||||
for _ in range(10):
|
||||
try:
|
||||
with sync_playwright() as playwright:
|
||||
stealth_js_path = pathlib.Path(BASE_DIR / "utils/stealth.min.js")
|
||||
chromium = playwright.chromium
|
||||
|
||||
# 如果一直失败可尝试设置成 False 让其打开浏览器，适当添加 sleep 可查看浏览器状态
|
||||
browser = chromium.launch(headless=True)
|
||||
|
||||
browser_context = browser.new_context()
|
||||
browser_context.add_init_script(path=stealth_js_path)
|
||||
context_page = browser_context.new_page()
|
||||
context_page.goto("https://www.xiaohongshu.com")
|
||||
browser_context.add_cookies([
|
||||
{'name': 'a1', 'value': a1, 'domain': ".xiaohongshu.com", 'path': "/"}]
|
||||
)
|
||||
context_page.reload()
|
||||
# 这个地方设置完浏览器 cookie 之后，如果这儿不 sleep 一下签名获取就失败了，如果经常失败请设置长一点试试
|
||||
sleep(2)
|
||||
encrypt_params = context_page.evaluate("([url, data]) => window._webmsxyw(url, data)", [uri, data])
|
||||
return {
|
||||
"x-s": encrypt_params["X-s"],
|
||||
"x-t": str(encrypt_params["X-t"])
|
||||
}
|
||||
except Exception:
|
||||
# 这儿有时会出现 window._webmsxyw is not a function 或未知跳转错误，因此加一个失败重试趴
|
||||
pass
|
||||
raise Exception("重试了这么多次还是无法签名成功，寄寄寄")
|
||||
|
||||
|
||||
def sign(uri, data=None, a1="", web_session=""):
|
||||
# 填写自己的 flask 签名服务端口地址
|
||||
res = requests.post(f"{XHS_SERVER}/sign",
|
||||
json={"uri": uri, "data": data, "a1": a1, "web_session": web_session})
|
||||
signs = res.json()
|
||||
return {
|
||||
"x-s": signs["x-s"],
|
||||
"x-t": signs["x-t"]
|
||||
}
|
||||
|
||||
|
||||
def beauty_print(data: dict):
|
||||
print(json.dumps(data, ensure_ascii=False, indent=2))
|
||||
@@ -0,0 +1,34 @@
|
||||
import datetime
|
||||
import json
|
||||
import qrcode
|
||||
from time import sleep
|
||||
|
||||
from xhs import XhsClient
|
||||
|
||||
from uploader.xhs_uploader.main import sign
|
||||
|
||||
# pip install qrcode
|
||||
if __name__ == '__main__':
|
||||
xhs_client = XhsClient(sign=sign, timeout=60)
|
||||
print(datetime.datetime.now())
|
||||
qr_res = xhs_client.get_qrcode()
|
||||
qr_id = qr_res["qr_id"]
|
||||
qr_code = qr_res["code"]
|
||||
|
||||
qr = qrcode.QRCode(version=1, error_correction=qrcode.ERROR_CORRECT_L,
|
||||
box_size=50,
|
||||
border=1)
|
||||
qr.add_data(qr_res["url"])
|
||||
qr.make()
|
||||
qr.print_ascii()
|
||||
|
||||
while True:
|
||||
check_qrcode = xhs_client.check_qrcode(qr_id, qr_code)
|
||||
print(check_qrcode)
|
||||
sleep(1)
|
||||
if check_qrcode["code_status"] == 2:
|
||||
print(json.dumps(check_qrcode["login_info"], indent=4))
|
||||
print("当前 cookie：" + xhs_client.cookie)
|
||||
break
|
||||
|
||||
print(json.dumps(xhs_client.get_self_info(), indent=4))
|
||||
Reference in New Issue
Block a user