如何编写自动发布文章的脚本

wen 实用脚本 2026-08-01 246

本文目录导读：

如何编写自动发布文章的脚本

Node.js
立即发布
定时发布（每天）

我来详细介绍如何编写自动发布文章的脚本,包括多种方案和完整示例。

整体架构设计

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. **频率控制**：设置合理的发布间隔，避免被屏蔽
这个脚本可以根据需求扩展支持更多平台和功能。

本文地址： https://www.bmjup.com/post/35583.html