本文目录导读：
我来详细介绍如何编写自动发布文章的脚本,包括多种方案和完整示例。
整体架构设计
graph LR
A[文章内容源] --> B[内容处理]
B --> C[发布平台适配]
C --> D[自动发布]
D --> E[发布监控]
Python完整示例
基础发布脚本
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
自动发布文章脚本
支持：WordPress、CSDN、掘金、知乎
"""
import os
import time
import json
import logging
import requests
from datetime import datetime
from typing import Dict, List, Optional
import argparse
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('publish_log.txt', encoding='utf-8'),
logging.StreamHandler()
]
)
logger = logging.getLogger(__name__)
class ArticlePublisher:
"""文章发布器基类"""
def __init__(self, config: Dict):
self.config = config
self.session = requests.Session()
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
})
def publish(self, article: Dict) -> bool:
"""发布文章（子类实现）"""
raise NotImplementedError
class WordPressPublisher(ArticlePublisher):
"""WordPress发布器"""
def publish(self, article: Dict) -> bool:
"""通过REST API发布到WordPress"""
try:
url = f"{self.config['site_url']}/wp-json/wp/v2/posts"
auth = (self.config['username'], self.config['password'])
data = {
'title': article['title'],
'content': article['content'],
'status': 'publish', # draft, publish, pending
'categories': article.get('categories', []),
'tags': article.get('tags', []),
}
response = self.session.post(url, json=data, auth=auth)
response.raise_for_status()
logger.info(f"✅ WordPress发布成功: {article['title']}")
return True
except Exception as e:
logger.error(f"❌ WordPress发布失败: {e}")
return False
class CSDNPublisher(ArticlePublisher):
"""CSDN发布器（需要Cookie）"""
def publish(self, article: Dict) -> bool:
"""发布到CSDN"""
try:
# 准备CSDN的API接口
url = "https://blog.csdn.net/community/home-api/v1/article"
headers = {
'Cookie': self.config['cookie'],
'Content-Type': 'application/json',
'Referer': 'https://editor.csdn.net/',
'Origin': 'https://editor.csdn.net'
}
data = {
'title': article['title'],
'markdowncontent': article['content'],
'categories': article.get('categories', []),
'tags': article.get('tags', ['']),
'readType': 'public',
'hideOriginal': False,
'type': 'original',
}
response = self.session.post(url, json=data, headers=headers)
response.raise_for_status()
logger.info(f"✅ CSDN发布成功: {article['title']}")
return True
except Exception as e:
logger.error(f"❌ CSDN发布失败: {e}")
return False
class ArticleManager:
"""文章管理器"""
def __init__(self, config_path: str = 'config.json'):
with open(config_path, 'r', encoding='utf-8') as f:
self.config = json.load(f)
self.publishers = self._init_publishers()
def _init_publishers(self) -> Dict:
"""初始化各平台发布器"""
publishers = {}
if 'wordpress' in self.config:
publishers['wordpress'] = WordPressPublisher(self.config['wordpress'])
if 'csdn' in self.config:
publishers['csdn'] = CSDNPublisher(self.config['csdn'])
return publishers
def read_articles(self, path: str) -> List[Dict]:
"""读取文章目录"""
articles = []
for filename in os.listdir(path):
if filename.endswith(('.md', '.txt', '.html')):
filepath = os.path.join(path, filename)
with open(filepath, 'r', encoding='utf-8') as f:
content = f.read()
# 解析文章信息
article = self._parse_article(content, filename)
if article:
articles.append(article)
return articles
def _parse_article(self, content: str, filename: str) -> Optional[Dict]:
"""解析文章内容"""
# 支持简单的前置元数据格式
if content.startswith('---'):
# 解析YAML front matter
import re
match = re.match(r'^---\s*\n(.*?)\n---\s*\n(.*)', content, re.DOTALL)
if match:
meta_text = match.group(1)
content_body = match.group(2)
# 简单解析 meta
meta = {}
for line in meta_text.split('\n'):
if ':' in line:
key, value = line.split(':', 1)
meta[key.strip()] = value.strip()
return {
'title': meta.get('title', filename),
'content': content_body,
'categories': meta.get('categories', '').split(',') if meta.get('categories') else [],
'tags': meta.get('tags', '').split(',') if meta.get('tags') else [],
}
# 默认处理
return {
'title': filename.split('.')[0],
'content': content,
'categories': [],
'tags': []
}
def publish_to_all(self, article: Dict) -> Dict:
"""发布到所有平台"""
results = {}
for platform, publisher in self.publishers.items():
results[platform] = publisher.publish(article)
time.sleep(2) # 间隔2秒，避免过快
return results
def batch_publish(self, articles_dir: str = 'articles') -> Dict:
"""批量发布文章"""
all_results = {}
articles = self.read_articles(articles_dir)
logger.info(f"📝 找到 {len(articles)} 篇文章")
for article in articles:
logger.info(f"📄 正在发布: {article['title']}")
result = self.publish_to_all(article)
all_results[article['title']] = result
# 保存成功/失败状态
if all(result.values()):
self._mark_published(article['title'])
else:
logger.warning(f"⚠️ 部分平台发布失败: {result}")
# 随机延迟，3-5分钟
delay = 180 + int(time.time() % 120)
logger.info(f"⏳ 等待 {delay/60:.1f} 分钟后下一篇")
time.sleep(delay)
return all_results
def _mark_published(self, title: str):
"""标记文章已发布"""
state_file = 'publish_state.json'
state = {}
if os.path.exists(state_file):
with open(state_file, 'r') as f:
state = json.load(f)
state[title] = datetime.now().isoformat()
with open(state_file, 'w') as f:
json.dump(state, f, ensure_ascii=False, indent=2)
def schedule_publish(self, articles_dir: str, interval_hours: int = 24):
"""定时发布"""
import schedule
def job():
self.batch_publish(articles_dir)
# 每天在指定时间发布
schedule.every(interval_hours).hours.do(job)
# 立即执行第一次
job()
while True:
schedule.run_pending()
time.sleep(60)
def main():
"""主函数"""
parser = argparse.ArgumentParser(description='自动发布文章脚本')
parser.add_argument('--action', choices=['publish', 'schedule', 'test'],
default='publish', help='操作类型')
parser.add_argument('--dir', default='articles', help='文章目录')
parser.add_argument('--config', default='config.json', help='配置文件路径')
parser.add_argument('--interval', type=int, default=24, help='发布时间间隔（小时）')
args = parser.parse_args()
# 创建发布管理器
manager = ArticleManager(args.config)
if args.action == 'publish':
# 立即发布
results = manager.batch_publish(args.dir)
logger.info(f"📊 发布结果: {results}")
elif args.action == 'schedule':
# 定时发布
manager.schedule_publish(args.dir, args.interval)
elif args.action == 'test':
# 测试连接
test_article = {
'title': '测试文章',
'content': '这是一篇测试文章。',
'categories': ['测试'],
'tags': ['test']
}
results = manager.publish_to_all(test_article)
logger.info(f"🧪 测试结果: {results}")
if __name__ == '__main__':
main()
配置文件示例 (config.json)
{
"wordpress": {
"site_url": "https://your-site.com",
"username": "your-username",
"password": "your-application-password"
},
"csdn": {
"cookie": "your-csdn-cookie"
},
"juejin": {
"cookie": "your-juejin-cookie"
}
}
文章模板示例
--- 我的新文章 categories: 技术,Python tags: 编程,自动化 --- 内容...
其他语言版本
Node.js 版本
// publish.js
const axios = require('axios');
const fs = require('fs');
const path = require('path');
const cron = require('node-cron');
class Publisher {
constructor(config) {
this.config = config;
}
async publishToWordPress(article) {
try {
const response = await axios.post(
`${this.config.wordpress.url}/posts`,
{
title: article.title,
content: article.content,
status: 'publish'
},
{
auth: {
username: this.config.wordpress.username,
password: this.config.wordpress.password
}
}
);
console.log(`✅ WordPress published: ${article.title}`);
return true;
} catch (error) {
console.error('❌ WordPress publish failed:', error.message);
return false;
}
}
// 从文件读取文章
readArticles(directory) {
const articles = [];
const files = fs.readdirSync(directory);
for (const file of files) {
if (file.endsWith('.md') || file.endsWith('.txt')) {
const content = fs.readFileSync(path.join(directory, file), 'utf8');
articles.push({
title: file.replace(/\.(md|txt)$/, ''),
content: content
});
}
}
return articles;
}
}
// 定时任务
async function schedulePublish() {
const publisher = new Publisher(config);
// 每天8点发布
cron.schedule('0 8 * * *', async () => {
console.log('⏰ Publishing scheduled articles...');
const articles = publisher.readArticles('./articles');
for (const article of articles) {
await publisher.publishToWordPress(article);
await sleep(300000); // 5分钟间隔
}
});
}
function sleep(ms) {
return new Promise(resolve => setTimeout(resolve, ms));
}
// 主入口
const config = require('./config.json');
schedulePublish();
Bash 脚本版本
#!/bin/bash
# publish.sh - 简单的发布脚本
PUBLISH_LOG="publish.log"
ARTICLES_DIR="./articles"
CONFIG_FILE="./publish.conf"
log() {
echo "[$(date '+%Y-%m-%d %H:%M:%S')] $1" | tee -a $PUBLISH_LOG
}
# 加载配置
source $CONFIG_FILE
# 检查文章目录
if [ ! -d "$ARTICLES_DIR" ]; then
log "❌ 文章目录不存在: $ARTICLES_DIR"
exit 1
fi
# 发布单篇文章
publish_article() {
local file="$1"
local title=$(basename "$file" | sed 's/\.[^.]*$//')
local content_file="$file"
# 使用curl发布到WordPress
curl -X POST "${WP_URL}/wp-json/wp/v2/posts" \
-u "${WP_USER}:${WP_PASS}" \
-H "Content-Type: application/json" \
-d "{
\"title\": \"$title\",
\"content\": \"$(cat $content_file | python3 -c 'import sys,json; print(json.dumps(sys.stdin.read()))')\",
\"status\": \"publish\"
}" >> $PUBLISH_LOG 2>&1
if [ $? -eq 0 ]; then
log "✅ 发布成功: $title"
# 移动文章到已发布目录
mkdir -p "$ARTICLES_DIR/published"
mv "$file" "$ARTICLES_DIR/published/"
else
log "❌ 发布失败: $title"
fi
}
# 批量发布
publish_all() {
for file in $ARTICLES_DIR/*.{md,txt,html}; do
if [ -f "$file" ]; then
publish_article "$file"
sleep 300 # 5分钟间隔
fi
done
}
# 主逻辑
case "$1" in
publish)
publish_all
;;
schedule)
# 使用cron定时执行
log "📝 计划任务启动"
while true; do
publish_all
sleep 86400 # 24小时
done
;;
*)
echo "用法: $0 {publish|schedule}"
exit 1
;;
esac
使用步骤
- 安装依赖
# Python pip install requests python-schedule pyyaml
Node.js
npm install axios node-cron
2. **配置文件**
```bash
cp config.example.json config.json
# 编辑配置，填入各平台信息-
准备文章
mkdir articles # 将markdown文件放入articles目录
-
运行脚本
# 测试连接 python publish.py --action test
立即发布
python publish.py --action publish --dir articles
定时发布（每天）
python publish.py --action schedule --interval 24
## 五、注意事项
1. **API限制**：各平台API调用频率有限制
2. **安全认证**：使用API密钥而不是明文密码
3. **错误处理**：完善重试机制和日志记录
4. **合规性**：遵守各平台的内容政策
5. **频率控制**：设置合理的发布间隔，避免被屏蔽
这个脚本可以根据需求扩展支持更多平台和功能。
