下载地址：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()
