awaioi/EveryPublish
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 将所有后端文件都迁移到 sau_backend 文件夹内

Browse Source
This commit is contained in:
Edan's Win11
2025-06-03 15:28:08 +08:00
parent 794e8feaea
commit 5c91a68d8c
60 changed files with 73 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sau_backend/uploader/__init__.py
+5
View File
@@ -0,0 +1,5 @@
from pathlib import Path
from sau_backend.conf import BASE_DIR
Path(BASE_DIR / "cookies").mkdir(exist_ok=True)
sau_backend/uploader/baijiahao_uploader/__init__.py
View File
sau_backend/uploader/baijiahao_uploader/main.py
+505
View File
@@ -0,0 +1,505 @@
# -*- coding: utf-8 -*-
import random
from datetime import datetime
from playwright.async_api import Playwright, async_playwright, Page
import os
import time
import asyncio
from sau_backend.conf import LOCAL_CHROME_PATH
from sau_backend.utils.base_social_media import set_init_script
from sau_backend.utils.log import baijiahao_logger
from sau_backend.utils.network import async_retry
async def baijiahao_cookie_gen(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://baijiahao.baidu.com/builder/theme/bjh/login")
await page.pause()
# 点击调试器的继续，保存cookie
await context.storage_state(path=account_file)
baijiahao_logger.success("cookie saved")
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://baijiahao.baidu.com/builder/rc/home")
await page.wait_for_timeout(timeout=5000)
if await page.get_by_text('注册/登录百家号').count():
baijiahao_logger.error("等待5秒 cookie 失效")
return False
else:
baijiahao_logger.success("[+] cookie 有效")
return True
async def baijiahao_setup(account_file, handle=False):
if not os.path.exists(account_file) or not await cookie_auth(account_file):
if not handle:
return False
baijiahao_logger.error("cookie文件不存在或已失效，即将自动打开浏览器，请扫码登录，登陆后会自动生成cookie文件")
await baijiahao_cookie_gen(account_file)
return True
class BaiJiaHaoVideo(object):
def __init__(self, title, file_path, tags, publish_date: datetime, account_file, proxy_setting=None):
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
self.proxy_setting = proxy_setting
async def set_schedule_time(self, page, publish_date):
"""
todo 时间选择，日后在处理 百家号的时间选择不准确，目前是随机
"""
publish_date_day = f"{publish_date.month}月{publish_date.day}日" if publish_date.day >9 else f"{publish_date.month}月0{publish_date.day}日"
publish_date_hour = f"{publish_date.hour}点"
publish_date_min = f"{publish_date.minute}分"
await page.wait_for_selector('div.select-wrap', timeout=5000)
for _ in range(3):
try:
await page.locator('div.select-wrap').nth(0).click()
await page.wait_for_selector('div.rc-virtual-list div.cheetah-select-item', timeout=5000)
break
except:
await page.locator('div.select-wrap').nth(0).click()
# page.locator(f'div.rc-virtual-list-holder-inner >> text={publish_date_day}').click()
await page.wait_for_timeout(2000)
await page.locator(f'div.rc-virtual-list div.cheetah-select-item >> text={publish_date_day}').click()
await page.wait_for_timeout(2000)
# 改为随机点击一个 hour
for _ in range(3):
try:
await page.locator('div.select-wrap').nth(1).click()
await page.wait_for_selector('div.rc-virtual-list div.rc-virtual-list-holder-inner:visible', timeout=5000)
break
except:
await page.locator('div.select-wrap').nth(1).click()
await page.wait_for_timeout(2000)
current_choice_hour = await page.locator('div.rc-virtual-list:visible div.cheetah-select-item-option').count()
await page.wait_for_timeout(2000)
await page.locator('div.rc-virtual-list:visible div.cheetah-select-item-option').nth(
random.randint(1, current_choice_hour-3)).click()
# 2024.08.05 current_choice_hour的获取可能有问题，页面有7，这里获取了10，暂时硬编码至6
await page.wait_for_timeout(2000)
await page.locator("button >> text=定时发布").click()
async def handle_upload_error(self, page):
# 日后实现，目前没遇到
return
print("视频出错了，重新上传中")
async def upload(self, playwright: Playwright) -> None:
# 使用 Chromium 浏览器启动一个浏览器实例
browser = await playwright.chromium.launch(headless=False, executable_path=self.local_executable_path, proxy=self.proxy_setting)
# 创建一个浏览器上下文，使用指定的 cookie 文件
context = await browser.new_context(storage_state=f"{self.account_file}", user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.4324.150 Safari/537.36')
# context = await set_init_script(context)
await context.grant_permissions(['geolocation'])
# 创建一个新的页面
page = await context.new_page()
# 访问指定的 URL
await page.goto("https://baijiahao.baidu.com/builder/rc/edit?type=videoV2", timeout=60000)
baijiahao_logger.info(f"正在上传-------{self.title}.mp4")
# 等待页面跳转到指定的 URL，没进入，则自动等待到超时
baijiahao_logger.info('正在打开主页...')
await page.wait_for_url("https://baijiahao.baidu.com/builder/rc/edit?type=videoV2", timeout=60000)
# 点击 "上传视频" 按钮
await page.locator("div[class^='video-main-container'] input").set_input_files(self.file_path)
# 等待页面跳转到指定的 URL
while True:
# 判断是是否进入视频发布页面，没进入，则自动等待到超时
try:
await page.wait_for_selector("div#formMain:visible")
break
except:
baijiahao_logger.info("正在等待进入视频发布页面...")
await asyncio.sleep(0.1)
# 填充标题和话题
# 这里为了避免页面变化，故使用相对位置定位：作品标题父级右侧第一个元素的input子元素
await asyncio.sleep(1)
baijiahao_logger.info("正在填充标题和话题...")
await self.add_title_tags(page)
upload_status = await self.uploading_video(page)
if not upload_status:
baijiahao_logger.error(f"发现上传出错了... 文件:{self.file_path}")
raise
# 判断视频封面图是否生成成功
while True:
baijiahao_logger.info("正在确认封面完成, 准备去点击定时/发布...")
if await page.locator("div.cheetah-spin-container img").count():
baijiahao_logger.info("封面已完成，点击定时/发布...")
break
else:
baijiahao_logger.info("等待封面生成...")
await asyncio.sleep(3)
await self.publish_video(page, self.publish_date)
await page.wait_for_timeout(2000)
if await page.locator('div.passMod_dialog-container >> text=百度安全验证:visible').count():
baijiahao_logger.error("出现验证，退出")
raise Exception("出现验证，退出")
await page.wait_for_url("https://baijiahao.baidu.com/builder/rc/clue**", timeout=5000)
baijiahao_logger.success("视频发布成功")
await context.storage_state(path=self.account_file) # 保存cookie
baijiahao_logger.info('cookie更新完毕！')
await asyncio.sleep(2) # 这里延迟是为了方便眼睛直观的观看
# 关闭浏览器上下文和浏览器实例
await context.close()
await browser.close()
@async_retry(timeout=300) # 例如，最多重试3次，超时时间为180秒
async def uploading_video(self, page):
while True:
upload_failed = await page.locator('div .cover-overlay:has-text("上传失败")').count()
if upload_failed:
baijiahao_logger.error("发现上传出错了...")
# await self.handle_upload_error(page) # 假设这是处理上传错误的函数
return False
uploading = await page.locator('div .cover-overlay:has-text("上传中")').count()
if uploading:
baijiahao_logger.info("正在上传视频中...")
await asyncio.sleep(2) # 等待2秒再次检查
continue
# 检查上传是否成功
if not uploading and not upload_failed:
baijiahao_logger.success("视频上传完毕")
return True
async def set_schedule_publish(self, page, publish_date):
while True:
schedule_element = page.locator("div.op-btn-outter-content >> text=定时发布").locator("..").locator(
'button')
try:
await schedule_element.click()
await page.wait_for_selector('div.select-wrap:visible', timeout=3000)
await page.wait_for_timeout(timeout=2000)
baijiahao_logger.info("开始点击发布定时...")
await self.set_schedule_time(page, publish_date)
break
except Exception as e:
baijiahao_logger.error(f"定时发布失败: {e}")
raise # 重新抛出异常，让重试装饰器捕获
@async_retry(timeout=300) # 例如，最多重试3次，超时时间为180秒
async def publish_video(self, page: Page, publish_date):
if publish_date != 0:
# 定时发布
await self.set_schedule_publish(page, publish_date)
else:
# 立即发布
await self.direct_publish(page)
async def direct_publish(self, page):
try:
publish_button = page.locator("button >> text=发布")
if await publish_button.count():
await publish_button.click()
except Exception as e:
baijiahao_logger.error(f"直接发布视频失败: {e}")
raise # 重新抛出异常，让重试装饰器捕获
async def add_title_tags(self, page):
title_container = page.get_by_placeholder('添加标题获得更多推荐')
if len(self.title) <= 8:
self.title += " 你不知道的"
await title_container.fill(self.title[:30])
async def main(self):
async with async_playwright() as playwright:
await self.upload(playwright)
# 使用 AI成片 功能
async def ai2video(self, playwright: Playwright) -> None:
# 使用 Chromium 浏览器启动一个浏览器实例
browser = await playwright.chromium.launch(headless=False, executable_path=self.local_executable_path, proxy=self.proxy_setting)
# 创建一个浏览器上下文，使用指定的 cookie 文件
context = await browser.new_context(
viewport={"width": 1600, "height": 900},
storage_state=f"{self.account_file}",
user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/127.0.4324.150 Safari/537.36'
)
# context = await set_init_script(context)
await context.grant_permissions(['geolocation'])
# 创建一个新的页面
page = await context.new_page()
# 访问指定的 URL
await page.goto("https://aigc.baidu.com/make", timeout=60000)
# 等待页面跳转到指定的 URL，没进入，则自动等待到超时
baijiahao_logger.info('正在打开主页...')
await page.wait_for_url("https://aigc.baidu.com/make", timeout=60000)
# 点击"全网"标签
await page.locator('div.rounded-lg.border:has-text("全网")').click()
await asyncio.sleep(1) # 这里延迟是为了方便眼睛直观的观看
# 点击 "上传视频" 按钮
# await page.locator("div[class^='video-main-container'] input").set_input_files(self.file_path)
# region 操作处
# 生成日期时间键名（格式：ai2video_YYYYMMDDHHMM）
now = datetime.now()
datetime_str = now.strftime("%Y%m%d%H%M")
processed_key = "ai2video_processed_titles"
batch_key = f"ai2video_{datetime_str}"
# 初始化LocalStorage
await page.evaluate(f"""
if (!localStorage.getItem("{processed_key}")) {{
localStorage.setItem("{processed_key}", JSON.stringify([]));
}}
if (!localStorage.getItem("{batch_key}")) {{
localStorage.setItem("{batch_key}", JSON.stringify([]));
}}
""")
# 定位新闻列表容器（转义特殊CSS字符）
container_selector = '.overflow-auto.flex-grow.h-0.saas-scrollbar.mt\-\[-4px\].pl\-\[24px\].pr\-\[10px\].pb\-\[18px\]'
news_items = await page.locator(container_selector).locator('div.py\-\[6px\].group.cursor-pointer').all()
for item in news_items:
try:
# 获取新闻标题
title_elem = item.locator('div.flex.text-gray-darker.items-center.relative.pr\-\[56px\] > span')
title = await title_elem.text_content()
if not title:
continue
# 检查是否已处理过
is_processed = await page.evaluate(
f"""title => {{
const processedList = JSON.parse(localStorage.getItem("{processed_key}") || "[]");
return processedList.includes(title);
}}""",
title
)
if is_processed:
print(f"[跳过] {title}")
continue
# 悬停显示按钮（根据HTML结构，按钮在悬停时显示）
await item.hover()
# 点击生成文案按钮
button = item.locator('button:has-text("生成文案")')
await button.click()
print(f"[点击] {title}")
# 等待30秒
# await page.wait_for_timeout(30000)
print(f"[等待完成] {title}")
# 监听"一键成片"按钮
print(f"[开始监听] 一键成片按钮")
should_exit_while_loop = False # 添加标志变量
while True:
# 定位"一键成片"按钮
one_key_button = page.locator("button:has-text('一键成片')")
# 检查按钮是否存在
if await one_key_button.count() > 0:
# 检查按钮是否有disabled属性
is_disabled = await one_key_button.get_attribute("disabled")
if is_disabled is None:
# 按钮不再被禁用，点击它
print(f"[发现可点击按钮] 一键成片")
await one_key_button.click() # 先点击一键成片按钮
# 等待可能出现的"温馨提示"窗口
print(f"[检查] 是否出现温馨提示窗口")
await page.wait_for_timeout(2000) # 等待2秒，让窗口有时间显示
try:
# 检查是否存在"温馨提示"窗口，设置较短的超时时间
tip_window = page.locator("div:has-text('温馨提示') >> visible=true")
if await tip_window.count() > 0:
print(f"[发现] 温馨提示窗口")
# 定位并点击"知道了"按钮，设置较短的超时时间
know_button = page.locator("button:has-text('知道了')")
if await know_button.count() > 0:
try:
# 设置较短的超时时间进行点击
await know_button.click(timeout=5000)
print(f"[已点击] 知道了按钮")
except Exception as e:
print(f"[警告] 点击知道了按钮时出错: {str(e)}")
else:
print(f"[警告] 未找到知道了按钮")
else:
print(f"[信息] 未出现温馨提示窗口，继续执行")
except Exception as e:
print(f"[警告] 处理温馨提示窗口时出错: {str(e)}")
# 继续执行，不要因为这个错误中断流程
# 记录到LocalStorage前打印日志
print(f"[开始记录] 准备将标题 '{title}' 记录到LocalStorage")
# 记录到LocalStorage
await page.evaluate(
f"""
(title, processedKey, batchKey) => {{
// 更新已处理列表
const processedList = JSON.parse(localStorage.getItem(processedKey) || "[]");
if (!processedList.includes(title)) {{
processedList.push(title);
localStorage.setItem(processedKey, JSON.stringify(processedList));
}}
// 更新当前批次记录
const batchList = JSON.parse(localStorage.getItem(batchKey) || "[]");
if (!batchList.includes(title)) {{
batchList.push(title);
localStorage.setItem(batchKey, JSON.stringify(batchList));
}}
}}
""",
title, processed_key, batch_key
)
# 记录完成后打印日志
print(f"[记录完成] 标题 '{title}' 已成功记录到LocalStorage")
print(f"[记录完成] {title}")
# 监听新打开的标签页
print(f"[监听] 等待新标签页打开")
# 获取当前所有页面
current_pages = context.pages
current_page_count = len(current_pages)
# 等待新标签页打开（最多等待10秒）
new_page = None
max_wait_time = 10 # 最大等待时间（秒）
start_time = time.time()
while time.time() - start_time < max_wait_time:
# 获取最新的页面列表
pages = context.pages
# 如果页面数量增加，说明新标签页已打开
if len(pages) > current_page_count:
# 获取最新打开的页面（通常是列表中的最后一个）
new_page = pages[-1]
print(f"[发现] 新标签页已打开")
break
# 短暂等待后再次检查
await asyncio.sleep(0.5)
# 如果找到新标签页，获取其标题和URL并保存
if new_page:
# 等待页面加载完成
try:
await new_page.wait_for_load_state("domcontentloaded", timeout=5000)
# 获取页面标题和URL
page_title = await new_page.title()
page_url = new_page.url
print(f"[获取] 标题: {page_title}")
print(f"[获取] URL: {page_url}")
# 将标题和URL保存到url.txt文件
with open("url.txt", "a", encoding="utf-8") as f:
f.write(f"{page_title}\n{page_url}\n\n")
print(f"[保存] 标题和URL已保存到url.txt")
# 等待5秒后关闭新标签页
print(f"[等待] 5秒后将关闭新标签页")
await asyncio.sleep(5)
await new_page.close()
print(f"[关闭] 新标签页已关闭")
except Exception as e:
print(f"[错误] 处理新标签页时出错: {str(e)}")
try:
# 尝试关闭页面，即使出错
await new_page.close()
print(f"[关闭] 新标签页已关闭（出错后）")
except:
pass
else:
print(f"[警告] 未检测到新标签页打开")
# 跳出整个while循环
print(f"[操作] 跳出所有循环，不再处理其他新闻")
should_exit_while_loop = True # 设置标志变量
break # 跳出while循环
# 检查是否需要跳出while循环
if should_exit_while_loop:
break
# 每秒检查一次按钮状态
await page.wait_for_timeout(1000)
# 检查是否需要跳出for循环
if should_exit_while_loop:
print(f"[操作] 跳出for循环，完全结束处理")
break # 跳出for循环
except Exception as e:
print(f"处理新闻时出错: {str(e)}")
continue
# endregion 操作处
print(f"[循环完成] 准备关闭浏览器")
# 暂停 1000s
await asyncio.sleep(1000) # 这里延迟是为了方便眼睛直观的观看
# 退出前保存 storage 信息
await context.storage_state(path=self.account_file) # 保存cookie
baijiahao_logger.info('cookie更新完毕！')
await asyncio.sleep(2) # 这里延迟是为了方便眼睛直观的观看
# 关闭浏览器上下文和浏览器实例
await context.close()
await browser.close()
async def mainAi(self):
async with async_playwright() as playwright:
await self.ai2video(playwright)
sau_backend/uploader/bilibili_uploader/__init__.py
+5
View File
@@ -0,0 +1,5 @@
from pathlib import Path
from sau_backend.conf import BASE_DIR
Path(BASE_DIR / "cookies" / "bilibili_uploader").mkdir(exist_ok=True)
sau_backend/uploader/bilibili_uploader/biliup.exe
BIN
View File
Binary file not shown.
sau_backend/uploader/bilibili_uploader/main.py
+79
View File
@@ -0,0 +1,79 @@
import json
import pathlib
import random
from biliup.plugins.bili_webup import BiliBili, Data
from sau_backend.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
sau_backend/uploader/douyin_uploader/__init__.py
+5
View File
@@ -0,0 +1,5 @@
from pathlib import Path
from sau_backend.conf import BASE_DIR
Path(BASE_DIR / "cookies" / "douyin_uploader").mkdir(exist_ok=True)
sau_backend/uploader/douyin_uploader/main.py
+249
View File
@@ -0,0 +1,249 @@
# -*- coding: utf-8 -*-
from datetime import datetime
from playwright.async_api import Playwright, async_playwright, Page
import os
import asyncio
from sau_backend.conf import LOCAL_CHROME_PATH
from sau_backend.utils.base_social_media import set_init_script
from sau_backend.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")
try:
await page.wait_for_url("https://creator.douyin.com/creator-micro/content/upload", timeout=5000)
except:
print("[+] 等待5秒 cookie 失效")
await context.close()
await browser.close()
return False
# 2024.06.17 抖音创作者中心改版
if await page.get_by_text('手机号登录').count() or 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, thumbnail_path=None):
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
self.thumbnail_path = thumbnail_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("div[class^='container'] input").set_input_files(self.file_path)
# 等待页面跳转到指定的 URL 2025.01.08修改在原有基础上兼容两种页面
while True:
try:
# 尝试等待第一个 URL
await page.wait_for_url(
"https://creator.douyin.com/creator-micro/content/publish?enter_from=publish_page", timeout=3000)
douyin_logger.info("[+] 成功进入version_1发布页面!")
break # 成功进入页面后跳出循环
except Exception:
try:
# 如果第一个 URL 超时，再尝试等待第二个 URL
await page.wait_for_url(
"https://creator.douyin.com/creator-micro/content/post/video?enter_from=publish_page",
timeout=3000)
douyin_logger.info("[+] 成功进入version_2发布页面!")
break # 成功进入页面后跳出循环
except:
print(" [-] 超时未进入视频发布页面，重新尝试...")
await asyncio.sleep(0.5) # 等待 0.5 秒后重新尝试
# 填充标题和话题
# 检查是否存在包含输入框的元素
# 这里为了避免页面变化，故使用相对位置定位：作品标题父级右侧第一个元素的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('[class^="long-card"] 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 self.set_thumbnail(page, self.thumbnail_path)
# 更换可见元素
await self.set_location(page, "杭州市")
# 頭條/西瓜
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 set_thumbnail(self, page: Page, thumbnail_path: str):
if thumbnail_path:
await page.click('text="选择封面"')
await page.wait_for_selector("div.semi-modal-content:visible")
await page.click('text="设置竖封面"')
await page.wait_for_timeout(2000) # 等待2秒
# 定位到上传区域并点击
await page.locator("div[class^='semi-upload upload'] >> input.semi-upload-hidden-input").set_input_files(thumbnail_path)
await page.wait_for_timeout(2000) # 等待2秒
await page.locator("div[class^='extractFooter'] button:visible:has-text('完成')").click()
# finish_confirm_element = page.locator("div[class^='confirmBtn'] >> div:has-text('完成')")
# if await finish_confirm_element.count():
# await finish_confirm_element.click()
# await page.locator("div[class^='footer'] button:has-text('完成')").click()
async def set_location(self, page: Page, location: str = "杭州市"):
# todo supoort location later
# await page.get_by_text('添加标签').locator("..").locator("..").locator("xpath=following-sibling::div").locator(
# "div.semi-select-single").nth(0).click()
await page.locator('div.semi-select span:has-text("输入地理位置")').click()
await page.keyboard.press("Backspace")
await page.wait_for_timeout(2000)
await page.keyboard.type(location)
await page.wait_for_selector('div[role="listbox"] [role="option"]', timeout=5000)
await page.locator('div[role="listbox"] [role="option"]').first.click()
async def main(self):
async with async_playwright() as playwright:
await self.upload(playwright)
sau_backend/uploader/ks_uploader/__init__.py
+5
View File
@@ -0,0 +1,5 @@
from pathlib import Path
from sau_backend.conf import BASE_DIR
Path(BASE_DIR / "cookies" / "ks_uploader").mkdir(exist_ok=True)
sau_backend/uploader/ks_uploader/main.py
+214
View File
@@ -0,0 +1,214 @@
# -*- coding: utf-8 -*-
from datetime import datetime
from playwright.async_api import Playwright, async_playwright
import os
import asyncio
from sau_backend.conf import LOCAL_CHROME_PATH
from sau_backend.utils.base_social_media import set_init_script
from sau_backend.utils.files_times import get_absolute_path
from sau_backend.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'
self.local_executable_path = LOCAL_CHROME_PATH
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 浏览器启动一个浏览器实例
print(self.local_executable_path)
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)
context.on("close", lambda: context.storage_state(path=self.account_file))
# 创建一个新的页面
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")
# 点击 "上传视频" 按钮
upload_button = page.locator("button[class^='_upload-btn']")
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 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.keyboard.type(f"#{tag} ")
await asyncio.sleep(2)
max_retries = 60 # 设置最大重试次数,最大等待时间为 2 分钟
retry_count = 0
while retry_count < max_retries:
try:
# 获取包含 '上传中' 文本的元素数量
number = await page.locator("text=上传中").count()
if number == 0:
kuaishou_logger.success("视频上传完毕")
break
else:
if retry_count % 5 == 0:
kuaishou_logger.info("正在上传视频中...")
await asyncio.sleep(2)
except Exception as e:
kuaishou_logger.error(f"检查上传状态时发生错误: {e}")
await asyncio.sleep(2) # 等待 2 秒后重试
retry_count += 1
if retry_count == max_retries:
kuaishou_logger.warning("超过最大重试次数，视频上传可能未完成。")
# 定时任务
if self.publish_date != 0:
await self.set_schedule_time(page, self.publish_date)
# 判断视频是否发布成功
while True:
try:
publish_button = page.get_by_text("发布", exact=True)
if await publish_button.count() > 0:
await publish_button.click()
await asyncio.sleep(1)
confirm_button = page.get_by_text("确认发布")
if await confirm_button.count() > 0:
await confirm_button.click()
# 等待页面跳转，确认发布成功
await page.wait_for_url(
"https://cp.kuaishou.com/article/manage/video?status=2&from=publish",
timeout=5000,
)
kuaishou_logger.success("视频发布成功")
break
except Exception as e:
kuaishou_logger.info(f"视频正在发布中... 错误: {e}")
await page.screenshot(full_page=True)
await asyncio.sleep(1)
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)
sau_backend/uploader/tencent_uploader/__init__.py
+5
View File
@@ -0,0 +1,5 @@
from pathlib import Path
from sau_backend.conf import BASE_DIR
Path(BASE_DIR / "cookies" / "tencent_uploader").mkdir(exist_ok=True)
sau_backend/uploader/tencent_uploader/main.py
+267
View File
@@ -0,0 +1,267 @@
# -*- coding: utf-8 -*-
from datetime import datetime
from playwright.async_api import Playwright, async_playwright
import os
import asyncio
from sau_backend.conf import LOCAL_CHROME_PATH
from sau_backend.utils.base_social_media import set_init_script
from sau_backend.utils.files_times import get_absolute_path
from sau_backend.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)
sau_backend/uploader/tk_uploader/__init__.py
+5
View File
@@ -0,0 +1,5 @@
from pathlib import Path
from sau_backend.conf import BASE_DIR
Path(BASE_DIR / "cookies" / "tk_uploader").mkdir(exist_ok=True)
sau_backend/uploader/tk_uploader/main.py
+265
View File
@@ -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 sau_backend.uploader.tk_uploader.tk_config import Tk_Locator
from sau_backend.utils.base_social_media import set_init_script
from sau_backend.utils.files_times import get_absolute_path
from sau_backend.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)
sau_backend/uploader/tk_uploader/main_chrome.py
+298
View File
@@ -0,0 +1,298 @@
# -*- coding: utf-8 -*-
import re
from datetime import datetime
from playwright.async_api import Playwright, async_playwright
import os
import asyncio
from sau_backend.conf import LOCAL_CHROME_PATH
from sau_backend.uploader.tk_uploader.tk_config import Tk_Locator
from sau_backend.utils.base_social_media import set_init_script
from sau_backend.utils.files_times import get_absolute_path
from sau_backend.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(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 = 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(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):
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-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_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').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("https://www.tiktok.com/tiktokstudio/content", timeout=3000)
tiktok_logger.success(" [-] video published success")
break
except Exception as e:
tiktok_logger.exception(f" [-] Exception: {e}")
tiktok_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:
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)
sau_backend/uploader/tk_uploader/tk_config.py
+4
View File
@@ -0,0 +1,4 @@
class Tk_Locator(object):
tk_iframe = '[data-tt="Upload_index_iframe"]'
default = 'body'
sau_backend/uploader/xhs_uploader/__init__.py
View File
sau_backend/uploader/xhs_uploader/accounts.ini
+2
View File
@@ -0,0 +1,2 @@
[account1]
cookies = changeme
sau_backend/uploader/xhs_uploader/main.py
+58
View File
@@ -0,0 +1,58 @@
import configparser
import json
import pathlib
from time import sleep
import requests
from playwright.sync_api import sync_playwright
from sau_backend.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))
sau_backend/uploader/xhs_uploader/xhs_login_qrcode.py
+34
View File
@@ -0,0 +1,34 @@
import datetime
import json
import qrcode
from time import sleep
from xhs import XhsClient
from sau_backend.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))