小红书批量发布协议, 抖音自动批量发布软件脚本,笔记作品视频自动发布工具【python】

简介: 这个工具框架包含了小红书和抖音的批量发布功能,支持图片和视频处理、定时发布等功能

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

这个工具框架包含了小红书和抖音的批量发布功能,支持图片和视频处理、定时发布等功能。使用时需要先配置config.json文件,填入从平台获取的API凭证。代码包含了错误处理和请求间隔,避免被封禁。

import os
import time
import json
import requests
from typing import List, Dict
from datetime import datetime
from PIL import Image, ImageFilter
from moviepy.editor import VideoFileClip

class SocialMediaPoster:
def init(self, config_path: str = "config.json"):
self.config = self._load_config(config_path)
self.xiaohongshu_api = "https://api.xiaohongshu.com"
self.douyin_api = "https://open.douyin.com"
self.session = requests.Session()

def _load_config(self, path: str) -> Dict:
    with open(path, 'r', encoding='utf-8') as f:
        return json.load(f)

def _get_xiaohongshu_token(self) -> str:
    url = f"{self.xiaohongshu_api}/oauth/access_token"
    payload = {
        "client_id": self.config["xiaohongshu"]["client_id"],
        "client_secret": self.config["xiaohongshu"]["client_secret"],
        "grant_type": "client_credentials"
    }
    response = self.session.post(url, data=payload)
    return response.json().get("access_token")

def _get_douyin_token(self) -> str:
    url = f"{self.douyin_api}/oauth/access_token/"
    params = {
        "client_key": self.config["douyin"]["client_key"],
        "client_secret": self.config["douyin"]["client_secret"],
        "grant_type": "client_credential"
    }
    response = self.session.get(url, params=params)
    return response.json().get("access_token")

def _process_image(self, image_path: str) -> str:
    img = Image.open(image_path)
    if img.mode != 'RGB':
        img = img.convert('RGB')
    img = img.filter(ImageFilter.SHARPEN)
    output_path = f"processed_{os.path.basename(image_path)}"
    img.save(output_path)
    return output_path

def _process_video(self, video_path: str) -> str:
    clip = VideoFileClip(video_path)
    if clip.duration > 60:  # 超过60秒的视频需要裁剪
        clip = clip.subclip(0, 60)
    output_path = f"processed_{os.path.basename(video_path)}"
    clip.write_videofile(output_path)
    return output_path

def post_to_xiaohongshu(self, content: str, media_paths: List[str], 
                       scheduled_time: datetime = None) -> Dict:
    token = self._get_xiaohongshu_token()
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

    processed_media = []
    for path in media_paths:
        if path.lower().endswith(('.png', '.jpg', '.jpeg')):
            processed_media.append(self._process_image(path))
        elif path.lower().endswith(('.mp4', '.mov')):
            processed_media.append(self._process_video(path))

    payload = {
        "content": content,
        "media_urls": processed_media,
        "visibility": self.config["xiaohongshu"].get("visibility", "PUBLIC")
    }

    if scheduled_time:
        payload["scheduled_publish_time"] = scheduled_time.isoformat()

    url = f"{self.xiaohongshu_api}/api/v1/notes"
    response = self.session.post(url, headers=headers, json=payload)
    return response.json()

def post_to_douyin(self, content: str, video_path: str, 
                  scheduled_time: datetime = None) -> Dict:
    token = self._get_douyin_token()
    headers = {
        "access-token": token,
        "Content-Type": "application/json"
    }

    processed_video = self._process_video(video_path)

    payload = {
        "text": content,
        "video_url": processed_video,
        "privacy_level": self.config["douyin"].get("privacy_level", 0)
    }

    if scheduled_time:
        payload["publish_time"] = scheduled_time.timestamp()

    url = f"{self.douyin_api}/api/v1/video/upload"
    response = self.session.post(url, headers=headers, json=payload)
    return response.json()

def batch_post(self, platform: str, posts: List[Dict]) -> List[Dict]:
    results = []
    for post in posts:
        try:
            if platform.lower() == "xiaohongshu":
                result = self.post_to_xiaohongshu(
                    post["content"],
                    post["media_paths"],
                    post.get("scheduled_time")
                )
            elif platform.lower() == "douyin":
                result = self.post_to_douyin(
                    post["content"],
                    post["video_path"],
                    post.get("scheduled_time")
                )
            else:
                raise ValueError(f"Unsupported platform: {platform}")

            results.append({
                "status": "success",
                "platform": platform,
                "post_id": result.get("id"),
                "response": result
            })
            time.sleep(5)  # 防止请求过于频繁
        except Exception as e:
            results.append({
                "status": "failed",
                "platform": platform,
                "error": str(e)
            })
    return results


‘from datetime import datetime, timedelta
from social_media_poster import SocialMediaPoster

def main():
poster = SocialMediaPoster()

# 小红书批量发布
xhs_posts = [
    {
        "content": "今天分享一个美食教程 #美食 #教程",
        "media_paths": ["food1.jpg", "food2.jpg"],
        "scheduled_time": datetime.now() + timedelta(hours=1)
    },
    {
        "content": "旅行日记 #旅行 #摄影",
        "media_paths": ["travel1.jpg", "travel2.jpg", "travel_video.mp4"]
    }
]
xhs_results = poster.batch_post("xiaohongshu", xhs_posts)
print("小红书发布结果:", xhs_results)

# 抖音批量发布
dy_posts = [
    {
        "content": "搞笑视频来啦 #搞笑 #娱乐",
        "video_path": "funny_video.mp4",
        "scheduled_time": datetime.now() + timedelta(hours=2)
    }
]
dy_results = poster.batch_post("douyin", dy_posts)
print("抖音发布结果:", dy_results)

if name == "main":
main()

from datetime import datetime, timedelta
from social_media_poster import SocialMediaPoster

def main():
poster = SocialMediaPoster()

# 小红书批量发布
xhs_posts = [
    {
        "content": "今天分享一个美食教程 #美食 #教程",
        "media_paths": ["food1.jpg", "food2.jpg"],
        "scheduled_time": datetime.now() + timedelta(hours=1)
    },
    {
        "content": "旅行日记 #旅行 #摄影",
        "media_paths": ["travel1.jpg", "travel2.jpg", "travel_video.mp4"]
    }
]
xhs_results = poster.batch_post("xiaohongshu", xhs_posts)
print("小红书发布结果:", xhs_results)

# 抖音批量发布
dy_posts = [
    {
        "content": "搞笑视频来啦 #搞笑 #娱乐",
        "video_path": "funny_video.mp4",
        "scheduled_time": datetime.now() + timedelta(hours=2)
    }
]
dy_results = poster.batch_post("douyin", dy_posts)
print("抖音发布结果:", dy_results)

if name == "main":
main()

相关文章
|
26天前
|
存储 缓存 测试技术
理解Python装饰器:简化代码的强大工具
理解Python装饰器:简化代码的强大工具
|
2月前
|
程序员 测试技术 开发者
Python装饰器:简化代码的强大工具
Python装饰器:简化代码的强大工具
158 92
|
14天前
|
机器学习/深度学习 编解码 Python
Python图片上采样工具 - RealESRGANer
Real-ESRGAN基于深度学习实现图像超分辨率放大,有效改善传统PIL缩放的模糊问题。支持多种模型版本,推荐使用魔搭社区提供的预训练模型,适用于将小图高质量放大至大图,放大倍率越低效果越佳。
|
24天前
|
算法 安全 数据安全/隐私保护
Python随机数函数全解析:5个核心工具的实战指南
Python的random模块不仅包含基础的随机数生成函数,还提供了如randint()、choice()、shuffle()和sample()等实用工具,适用于游戏开发、密码学、统计模拟等多个领域。本文深入解析这些函数的用法、底层原理及最佳实践,帮助开发者高效利用随机数,提升代码质量与安全性。
121 0
|
存储 监控 API
Python笔记2(函数参数、面向对象、装饰器、高级函数、捕获异常、dir)
Python笔记2(函数参数、面向对象、装饰器、高级函数、捕获异常、dir)
135 0
|
Python
Python基础 笔记(九) 函数及进阶
Python基础 笔记(九) 函数及进阶
86 6
|
存储 Python
Python笔记8 函数
本文是作者的Python复习笔记第八篇,全面介绍了Python中的函数定义与使用,包括函数的参数传递(位置参数、关键字参数、默认参数、列表参数、任意数量参数和关键字参数)、函数的返回值以及如何创建和调用函数库(模块),并提供了丰富的示例代码。
75 0
|
Python
【python笔记】使用zip函数迭代多个可迭代对象
【python笔记】使用zip函数迭代多个可迭代对象
101 0
|
Python
Python基础 笔记(三) 标识符、输入输出函数
Python基础 笔记(三) 标识符、输入输出函数
122 7
|
算法 数据处理 索引
【Python 小白到精通 | 课程笔记】第三章:数据处理就像侦探游戏(函数和包)
【Python 小白到精通 | 课程笔记】第三章:数据处理就像侦探游戏(函数和包)
320 0
【Python 小白到精通 | 课程笔记】第三章:数据处理就像侦探游戏(函数和包)

推荐镜像

更多