下载地址:https://www.pan38.com/yun/share.php?code=JCnzE 提取密码:aa98
这个批量发布工具支持快手、抖音、B站和YouTube平台,使用Selenium实现自动化上传。主程序包含视频信息获取、多平台上传和结果记录功能。使用时需要安装Chrome浏览器和对应驱动。
import os
import time
import requests
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from configparser import ConfigParser
import json
import hashlib
import logging
from datetime import datetime
class MultiPlatformUploader:
def init(self):
self.config = ConfigParser()
self.config.read('config.ini')
self.setup_logging()
self.driver = None
self.platform_apis = {
'kuaishou': self.upload_to_kuaishou,
'douyin': self.upload_to_douyin,
'bilibili': self.upload_to_bilibili,
'youtube': self.upload_to_youtube
}
def setup_logging(self):
logging.basicConfig(
filename='uploader.log',
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
self.logger = logging.getLogger(__name__)
def init_driver(self):
options = webdriver.ChromeOptions()
options.add_argument('--disable-notifications')
options.add_argument('--disable-infobars')
options.add_argument('--disable-extensions')
options.add_argument('--disable-gpu')
options.add_argument('--no-sandbox')
options.add_argument('--headless')
self.driver = webdriver.Chrome(options=options)
self.driver.implicitly_wait(10)
def upload_to_kuaishou(self, video_path, title, tags):
try:
self.driver.get('https://www.kuaishou.com/upload')
upload_btn = self.driver.find_element(By.XPATH, '//input[@type="file"]')
upload_btn.send_keys(os.path.abspath(video_path))
title_field = WebDriverWait(self.driver, 30).until(
EC.presence_of_element_located((By.XPATH, '//textarea[@placeholder="添加标题"]'))
)
title_field.send_keys(title)
if tags:
tag_field = self.driver.find_element(By.XPATH, '//input[@placeholder="添加标签"]')
for tag in tags:
tag_field.send_keys(tag)
tag_field.send_keys(Keys.ENTER)
time.sleep(0.5)
publish_btn = self.driver.find_element(By.XPATH, '//button[contains(text(),"发布")]')
publish_btn.click()
WebDriverWait(self.driver, 60).until(
EC.url_contains('profile')
)
self.logger.info(f'Successfully uploaded to Kuaishou: {title}')
return True
except Exception as e:
self.logger.error(f'Error uploading to Kuaishou: {str(e)}')
return False
def upload_to_douyin(self, video_path, title, tags):
try:
self.driver.get('https://creator.douyin.com/creator-micro/content/upload')
upload_btn = self.driver.find_element(By.XPATH, '//input[@type="file"]')
upload_btn.send_keys(os.path.abspath(video_path))
title_field = WebDriverWait(self.driver, 30).until(
EC.presence_of_element_located((By.XPATH, '//textarea[@placeholder="填写标题"]'))
)
title_field.send_keys(title)
if tags:
tag_field = self.driver.find_element(By.XPATH, '//input[@placeholder="添加标签"]')
for tag in tags:
tag_field.send_keys(tag)
tag_field.send_keys(Keys.ENTER)
time.sleep(0.5)
publish_btn = self.driver.find_element(By.XPATH, '//button[contains(text(),"发布")]')
publish_btn.click()
WebDriverWait(self.driver, 60).until(
EC.url_contains('content/manage')
)
self.logger.info(f'Successfully uploaded to Douyin: {title}')
return True
except Exception as e:
self.logger.error(f'Error uploading to Douyin: {str(e)}')
return False
def upload_to_bilibili(self, video_path, title, tags):
try:
self.driver.get('https://member.bilibili.com/platform/upload/video/frame')
upload_btn = self.driver.find_element(By.XPATH, '//input[@type="file"]')
upload_btn.send_keys(os.path.abspath(video_path))
title_field = WebDriverWait(self.driver, 30).until(
EC.presence_of_element_located((By.XPATH, '//input[@placeholder="填写标题"]'))
)
title_field.send_keys(title)
if tags:
tag_field = self.driver.find_element(By.XPATH, '//input[@placeholder="添加标签"]')
for tag in tags:
tag_field.send_keys(tag)
tag_field.send_keys(Keys.ENTER)
time.sleep(0.5)
publish_btn = self.driver.find_element(By.XPATH, '//button[contains(text(),"发布")]')
publish_btn.click()
WebDriverWait(self.driver, 60).until(
EC.url_contains('content/manage')
)
self.logger.info(f'Successfully uploaded to Bilibili: {title}')
return True
except Exception as e:
self.logger.error(f'Error uploading to Bilibili: {str(e)}')
return False
def upload_to_youtube(self, video_path, title, tags):
try:
self.driver.get('https://www.youtube.com/upload')
upload_btn = self.driver.find_element(By.XPATH, '//input[@type="file"]')
upload_btn.send_keys(os.path.abspath(video_path))
title_field = WebDriverWait(self.driver, 30).until(
EC.presence_of_element_located((By.XPATH, '//div[@id="textbox"]'))
)
title_field.send_keys(title)
if tags:
tag_field = self.driver.find_element(By.XPATH, '//input[@placeholder="添加标签"]')
for tag in tags:
tag_field.send_keys(tag)
tag_field.send_keys(Keys.ENTER)
time.sleep(0.5)
next_btn = self.driver.find_element(By.XPATH, '//ytcp-button[@id="next-button"]')
for _ in range(3):
next_btn.click()
time.sleep(1)
publish_btn = self.driver.find_element(By.XPATH, '//ytcp-button[@id="done-button"]')
publish_btn.click()
WebDriverWait(self.driver, 60).until(
EC.url_contains('upload')
)
self.logger.info(f'Successfully uploaded to YouTube: {title}')
return True
except Exception as e:
self.logger.error(f'Error uploading to YouTube: {str(e)}')
return False
def batch_upload(self, video_path, title, tags=None, platforms=None):
if not platforms:
platforms = self.config['DEFAULT']['platforms'].split(',')
results = {}
self.init_driver()
try:
for platform in platforms:
if platform.strip() in self.platform_apis:
start_time = datetime.now()
success = self.platform_apis[platform.strip()](video_path, title, tags)
end_time = datetime.now()
duration = (end_time - start_time).total_seconds()
results[platform] = {
'status': 'success' if success else 'failed',
'time_elapsed': duration
}
else:
results[platform] = {
'status': 'unsupported',
'time_elapsed': 0
}
finally:
if self.driver:
self.driver.quit()
return results
def get_video_info(self, video_path):
file_size = os.path.getsize(video_path) / (1024 * 1024) # MB
duration = self.get_video_duration(video_path)
return {
'file_size': round(file_size, 2),
'duration': duration,
'md5': self.get_file_md5(video_path)
}
def get_video_duration(self, video_path):
try:
import cv2
cap = cv2.VideoCapture(video_path)
fps = cap.get(cv2.CAP_PROP_FPS)
frame_count = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))
duration = frame_count / fps
cap.release()
return round(duration, 2)
except:
return 0
def get_file_md5(self, file_path):
hash_md5 = hashlib.md5()
with open(file_path, "rb") as f:
for chunk in iter(lambda: f.read(4096), b""):
hash_md5.update(chunk)
return hash_md5.hexdigest()
if name == 'main':
uploader = MultiPlatformUploader()
video_path = input('Enter video file path: ')
title = input('Enter video title: ')
tags = input('Enter tags (comma separated): ').split(',')
video_info = uploader.get_video_info(video_path)
print(f"\nVideo Info:")
print(f"Size: {video_info['file_size']} MB")
print(f"Duration: {video_info['duration']} seconds")
print(f"MD5: {video_info['md5']}\n")
print("Available platforms: kuaishou, douyin, bilibili, youtube")
platforms = input('Enter platforms to upload (comma separated, leave empty for all): ')
if platforms:
platforms = platforms.split(',')
else:
platforms = None
print("\nStarting upload process...")
results = uploader.batch_upload(video_path, title, tags, platforms)
print("\nUpload Results:")
for platform, result in results.items():
print(f"{platform}: {result['status']} ({result['time_elapsed']:.2f}s)")