自动化发布系统设计
系统概述
实现6个新平台的自动化发布：小红书、快手、抖音、知乎、网易、微信公众号。
从EP37开始，每天早上6点自动发布1期，逐步补齐到36期。
发布时间表
| 日期 |
集数 |
状态 |
| 2026-04-13 周一 06:00 |
EP37 |
待发布 |
| 2026-04-14 周二 06:00 |
EP38 |
待发布 |
| 2026-04-15 周三 06:00 |
EP39 |
待发布 |
| 2026-04-16 周四 06:00 |
EP40 |
待发布 |
| 2026-04-17 周五 06:00 |
EP41 |
待发布 |
| 2026-04-18 周六 06:00 |
EP42 |
待发布 |
| 2026-04-19 周日 06:00 |
EP43 |
待发布 |
| ... |
... |
... |
系统架构
┌─────────────────────────────────────────────────────────┐
│ Scheduler (调度器) │
│ - APScheduler (每天06:00触发) │
│ - Cron表达式: 0 6 * * * │
└────────────────────┬────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Task Manager (任务管理器) │
│ - 任务队列 (Redis/SQLite) │
│ - 任务调度 (优先级、重试策略) │
│ - 状态追踪 (pending/scheduled/published/failed) │
└────────────────────┬────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Publisher Manager (发布管理器) │
│ - 并发控制 (最多同时发布N个平台) │
│ - 失败重试 (指数退避) │
│ - 结果聚合 (返回各平台发布结果) │
└────────────────────┬────────────────────────────────────┘
│
┌────────────┼────────────┐
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ 小红书 │ │ 抖音 │ │ 快手 │
│ Publisher│ │ Publisher│ │ Publisher│
└──────────┘ └──────────┘ └──────────┘
┌──────────┐ ┌──────────┐ ┌──────────┐
│ 知乎 │ │ 网易 │ │ 微信公众号│
│ Publisher│ │ Publisher│ │ Publisher│
└──────────┘ └──────────┘ └──────────┘
│
▼
┌─────────────────────────────────────────────────────────┐
│ Storage (数据存储) │
│ - 发布历史 (published_history.json) │
│ - 任务状态 (task_status.json) │
│ - 平台配置 (platform_config.json) │
└─────────────────────────────────────────────────────────┘
平台技术方案
1. 小红书 (Xiaohongshu)
- 方案: Playwright/Selenium + 无头浏览器
- 原因: 没有官方API，必须Web UI自动化
- 挑战: 登录验证、滑动验证码、反爬
- 应对: 使用undetected_chromedriver，人工登录一次后保存cookies
2. 抖音 (Douyin)
- 方案A: 官方Open API（需要企业认证）
- 方案B: Playwright + Web UI自动化（备选）
- 原因: 有官方API但门槛高，Web UI更灵活
- 挑战: 视频上传速度、审核等待时间
- 应对: 异步上传，后台轮询状态
3. 快手 (Kuaishou)
- 方案: Playwright + Web UI自动化
- 原因: API文档不完善，Web UI更可靠
- 挑战: 登录流程、视频格式要求
- 应对: 视频预处理，符合快手规范
4. 知乎 (Zhihu)
- 方案: Playwright + 文章编辑器
- 原因: 知乎需要富文本编辑器，不适合API
- 挑战: Markdown转换、图片上传、引用格式
- 应对: 使用markdown库转换，图片单独上传
5. 网易云音乐 (NetEase)
- 方案: HTTP API + 签名算法
- 原因: 有音频上传API
- 挑战: 签名算法、登录认证
- 应对: 研究API签名，实现认证流程
6. 微信公众号 (WeChat MP)
- 方案: 官方API（已实现基础）
- 状态: 已有weixin_mp.py，需要完善
- 挑战: token刷新、素材上传限制
- 应对: 自动token管理，分批上传素材
任务优先级
| 平台 |
优先级 |
依赖 |
预计时间 |
| 微信公众号 |
P0 |
已有基础 |
1天 |
| 小红书 |
P0 |
需要UI自动化 |
2天 |
| 抖音 |
P0 |
需要UI自动化 |
2天 |
| 快手 |
P1 |
需要UI自动化 |
1天 |
| 知乎 |
P1 |
需要文章编辑器 |
1天 |
| 网易 |
P2 |
需要API研究 |
2天 |
数据结构
发布任务 (PublishTask)
{
"task_id": "task_20260413_ep37",
"episode_id": "EP37",
"video_path": "/path/to/EP37_final.mp4",
"audio_path": "/path/to/EP37_edge.mp3",
"title": "情人节深夜的第一行代码",
"description": "...",
"tags": ["AI", "气功"],
"platforms": ["xiaohongshu", "douyin", "kuaishou", "zhihu", "netease", "wechat"],
"scheduled_time": "2026-04-13T06:00:00",
"status": "pending", # pending/scheduled/publishing/published/failed
"created_at": "2026-04-11T12:00:00",
"updated_at": "2026-04-11T12:00:00"
}
平台发布结果 (PublishResult)
{
"task_id": "task_20260413_ep37",
"platform": "xiaohongshu",
"status": "published", # published/failed
"video_id": "xiaohongshu_123456",
"url": "https://www.xiaohongshu.com/explore/123456",
"published_at": "2026-04-13T06:15:00",
"error": null # 失败时记录错误信息
}
调度器配置
APScheduler配置
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
scheduler = AsyncIOScheduler()
# 每天早上6点执行
scheduler.add_job(
publish_scheduled_episode,
trigger=CronTrigger(hour=6, minute=0),
id='daily_publish',
replace_existing=True
)
scheduler.start()
发布函数
async def publish_scheduled_episode():
"""每天早上6点执行发布"""
# 1. 查找当天需要发布的剧集
episode = get_todays_episode()
if not episode:
logger.info("今天没有需要发布的剧集")
return
# 2. 创建发布任务
task = create_publish_task(episode)
# 3. 并发发布到各平台
results = await multi_platform_publish(task)
# 4. 记录结果
save_publish_results(results)
并发控制
import asyncio
async def multi_platform_publish(task: PublishTask):
"""并发发布到多个平台"""
platforms = task.platforms
# 并发发布（最多同时3个平台）
semaphore = asyncio.Semaphore(3)
async def publish_with_limit(platform):
async with semaphore:
return await publish_to_platform(task, platform)
results = await asyncio.gather(
*[publish_with_limit(p) for p in platforms],
return_exceptions=True
)
return results
错误处理
重试策略
from tenacity import retry, stop_after_attempt, wait_exponential
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=4, max=10)
)
async def publish_with_retry(platform, task):
"""带重试的发布"""
return await publish_to_platform(platform, task)
失败处理
- 立即记录失败日志
- 发送通知（邮件/钉钉/企业微信）
- 标记任务为failed，人工介入
- 支持手动重试
监控和报警
监控指标
- 任务成功率
- 平台响应时间
- 发布队列长度
- 失败原因统计
报警方式
- 失败超过2个平台：发送报警
- 连续3天失败：发送紧急报警
- 每日发布摘要：定时发送日报
部署方案
开发环境
- 本地测试：手动触发调度器
- 日志输出：console
生产环境
- Systemd服务：持久化运行
- 日志输出：/var/log/lingtongask/publisher.log
- 进程监控：supervisor或pm2
Systemd配置
[Unit]
Description=LingTongAsk Auto Publisher
After=network.target
[Service]
Type=simple
User=ai
WorkingDirectory=/home/ai/lingtongask
ExecStart=/usr/bin/python3 -m scripts.auto_publisher
Restart=always
RestartSec=10
[Install]
WantedBy=multi-user.target
安全考虑
凭证管理
- API密钥存储在.env文件（不提交到git）
- Cookies加密存储
- 定期刷新token
反爬应对
- User-Agent轮换
- 请求间隔随机化（2-5秒）
- IP代理（如有必要）
- 人机行为模拟
测试方案
单元测试
集成测试
- 完整发布流程（EP37）
- 多平台并发发布
- 失败重试机制
灰度发布
性能优化
视频预处理
- 统一视频格式：MP4, H.264编码
- 优化视频大小：每个平台有不同限制
- 预生成多个尺寸版本
并发控制
- 限制同时上传的平台数（避免被限流）
- 使用连接池复用HTTP连接
缓存策略
后续扩展
新平台接入
- 遵循PlatformPublisher接口
- 实现authenticate/upload/publish方法
- 添加到PUBLISHERS映射表
功能增强
- 定时发布编辑界面
- 发布历史查询
- 手动重试失败任务
- 批量发布（一次发布多集）