批量上传发布视频的软件，小红书抖音快手哔哩哔哩，自动发布上传作品工具【python】

下载地址：https://www.pan38.com/yun/share.php?code=JCnzE 提取密码：h7789

这个项目包含完整的视频批量上传功能，支持多个平台，包含视频处理、配置管理和错误处理等功能。使用时需要安装moviepy、requests、pyyaml等依赖库。

import os
import time
from platforms import (
XiaohongshuUploader,
DouyinUploader,
KuaishouUploader,
BilibiliUploader
)
from video_processor import VideoProcessor
from config_loader import load_config

def batch_upload_videos(config_path):

# 加载配置文件
config = load_config(config_path)

# 初始化视频处理器
processor = VideoProcessor(
    max_duration=config['video']['max_duration'],
    watermark_path=config['video']['watermark']
)

# 初始化各平台上传器
platforms = {
    'xiaohongshu': XiaohongshuUploader(config['xiaohongshu']),
    'douyin': DouyinUploader(config['douyin']),
    'kuaishou': KuaishouUploader(config['kuaishou']),
    'bilibili': BilibiliUploader(config['bilibili'])
}

# 处理视频目录
video_dir = config['video']['source_dir']
for video_file in os.listdir(video_dir):
    if video_file.endswith(('.mp4', '.mov', '.avi')):
        video_path = os.path.join(video_dir, video_file)

        # 处理视频
        processed_path = processor.process(video_path)

        # 上传到各平台
        for platform_name, uploader in platforms.items():
            if config['platforms'][platform_name]['enable']:
                try:
                    print(f"开始上传 {video_file} 到 {platform_name}")
                    uploader.upload(
                        video_path=processed_path,
                        title=config['video']['title_template'].format(
                            filename=os.path.splitext(video_file)[0]
                        ),
                        description=config['video']['description_template'],
                        tags=config['video']['tags']
                    )
                    print(f"{video_file} 上传到 {platform_name} 成功")
                    time.sleep(config['platforms'][platform_name]['interval'])
                except Exception as e:
                    print(f"上传到 {platform_name} 失败: {str(e)}")

print("所有视频上传完成")

if name == "main":
batch_upload_videos("config.yaml")

import requests
import json
from abc import ABC, abstractmethod

class BaseUploader(ABC):
def init(self, config):
self.config = config
self.session = requests.Session()
self._login()

@abstractmethod
def _login(self):
    pass

@abstractmethod
def upload(self, video_path, title, description, tags):
    pass

class XiaohongshuUploader(BaseUploader):
def _login(self):
login_url = "https://edith.xiaohongshu.com/api/sns/web/user/login"
payload = {
"mobile": self.config['username'],
"password": self.config['password'],
"deviceId": self.config['device_id']
}
response = self.session.post(login_url, json=payload)
if response.status_code != 200:
raise Exception("小红书登录失败")
self.cookies = response.cookies

def upload(self, video_path, title, description, tags):
    # 获取上传凭证
    token_url = "https://edith.xiaohongshu.com/api/sns/web/upload/token"
    token_resp = self.session.get(token_url, cookies=self.cookies)
    upload_token = token_resp.json()['data']['token']

    # 上传视频
    upload_url = "https://upload.xiaohongshu.com/api/media/upload"
    headers = {
        "Authorization": f"Bearer {upload_token}",
        "Content-Type": "multipart/form-data"
    }
    with open(video_path, 'rb') as f:
        files = {'file': (os.path.basename(video_path), f)}
        upload_resp = self.session.post(upload_url, headers=headers, files=files)

    if upload_resp.status_code != 200:
        raise Exception("视频上传失败")

    video_id = upload_resp.json()['data']['id']

    # 发布视频
    publish_url = "https://edith.xiaohongshu.com/api/sns/web/post/create"
    payload = {
        "type": "video",
        "title": title,
        "description": description,
        "tags": tags,
        "video_id": video_id,
        "visibility": self.config['visibility']
    }
    publish_resp = self.session.post(publish_url, json=payload, cookies=self.cookies)

    if publish_resp.status_code != 200:
        raise Exception("视频发布失败")

    return publish_resp.json()

class DouyinUploader(BaseUploader):

# 类似实现，省略部分代码...
pass

class KuaishouUploader(BaseUploader):

# 类似实现，省略部分代码...
pass

class BilibiliUploader(BaseUploader):

# 类似实现，省略部分代码...
pass

os
import subprocess
from moviepy.editor import VideoFileClip

class VideoProcessor:
def init(self, max_duration=300, watermark_path=None):
self.max_duration = max_duration
self.watermark_path = watermark_path

def process(self, video_path):
    # 检查视频时长
    clip = VideoFileClip(video_path)
    if clip.duration > self.max_duration:
        clip = clip.subclip(0, self.max_duration)

    # 添加水印
    if self.watermark_path and os.path.exists(self.watermark_path):
        watermark = (VideoFileClip(self.watermark_path)
                    .set_duration(clip.duration)
                    .resize(height=50)
                    .set_position(("right", "bottom")))
        clip = CompositeVideoClip([clip, watermark])

    # 保存处理后的视频
    output_path = os.path.join(
        os.path.dirname(video_path),
        "processed_" + os.path.basename(video_path)
    )
    clip.write_videofile(output_path, codec="libx264", audio_codec="aac")

    return output_path

def compress_video(self, input_path, output_path, crf=28):
    cmd = [
        'ffmpeg',
        '-i', input_path,
        '-vcodec', 'libx264',
        '-crf', str(crf),
        '-preset', 'fast',
        '-acodec', 'copy',
        output_path
    ]
    subprocess.run(cmd, check=True)
    return output_path

import yaml
from pathlib import Path

def load_config(config_path):
with open(config_path, 'r', encoding='utf-8') as f:
config = yaml.safe_load(f)

# 验证必要配置项
required_keys = [
    'video.source_dir',
    'video.title_template',
    'platforms.xiaohongshu.enable',
    'platforms.douyin.enable',
    'platforms.kuaishou.enable',
    'platforms.bilibili.enable'
]

for key in required_keys:
    keys = key.split('.')
    current = config
    for k in keys:
        if k not in current:
            raise ValueError(f"缺少必要配置项: {key}")
        current = current[k]

return config