294 lines
14 KiB
Python
294 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
||
from datetime import datetime
|
||
|
||
from playwright.async_api import Playwright, async_playwright, Page
|
||
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
|
||
from utils.anti_detection import create_stealth_browser, create_stealth_context, setup_stealth_page
|
||
from utils.human_typing_wrapper import create_human_typer
|
||
|
||
|
||
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, headless=True):
|
||
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
|
||
self.headless = headless # 是否使用无头模式，默认为True
|
||
|
||
async def set_schedule_time_douyin(self, page, publish_date):
|
||
# 创建人类化输入包装器用于定时发布
|
||
human_typer = create_human_typer(page)
|
||
|
||
# 选择包含特定文本内容的 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:
|
||
# 使用统一的反检测工具
|
||
browser = await create_stealth_browser(
|
||
playwright=playwright,
|
||
headless=self.headless,
|
||
executable_path=self.local_executable_path
|
||
)
|
||
|
||
context = await create_stealth_context(
|
||
browser=browser,
|
||
account_file=self.account_file,
|
||
headless=self.headless
|
||
)
|
||
|
||
page = await setup_stealth_page(context, "https://creator.douyin.com/creator-micro/content/upload")
|
||
|
||
# 创建人类化输入包装器
|
||
human_typer = create_human_typer(page)
|
||
|
||
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.locator(
|
||
'input.semi-input.semi-input-default[placeholder="填写作品标题，为作品获得更多流量"]' #我自己改的，他之前是相对位置，但是没写上
|
||
)
|
||
if await title_container.count():
|
||
# 使用人类化输入填充标题
|
||
success = await human_typer.type_text_human(
|
||
'input.semi-input.semi-input-default[placeholder="填写作品标题，为作品获得更多流量"]',
|
||
self.title[:30],
|
||
clear_first=True
|
||
)
|
||
if not success:
|
||
douyin_logger.warning("人类化标题输入失败，使用传统方式")
|
||
await title_container.fill(self.title[:30])
|
||
else:
|
||
# 使用人类化输入的备用方案
|
||
success = await human_typer.type_text_human(".notranslate", self.title, clear_first=True)
|
||
if not success:
|
||
douyin_logger.warning("人类化标题输入失败，使用传统方式")
|
||
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"
|
||
# 使用人类化输入添加标签
|
||
tags_text = " ".join([f"#{tag}" for tag in self.tags]) + " " # 最后也加一个空格
|
||
success = await human_typer.type_text_human(css_selector, tags_text, clear_first=False)
|
||
|
||
if not success:
|
||
douyin_logger.warning("人类化标签输入失败，使用传统方式")
|
||
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)}个话题')
|
||
# 调试用：暂停10秒，让你有时间观察页面上的标签是否正确输入
|
||
# await page.wait_for_timeout(10000)
|
||
# douyin_logger.info("观察时间结束，中断程序")
|
||
|
||
# # 强制中断（后续代码不执行）
|
||
# raise SystemExit("调试中断：标签输入流程完成")
|
||
|
||
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)
|
||
|
||
|