python脚本

用Python搭建自动化写作+SEO优化+排期发布流水线

发表于 2025年04月21日阅读 98 评论 0

大家好，我是何三，80后老猿，独立开发者

作为一名内容创作者，我深知每天产出高质量文章的艰辛。从选题到写作，从SEO优化到发布排期，每一个环节都需要投入大量时间。直到有一天，我决定用Python搭建一套全自动化流水线，彻底解放我的生产力。

想象一下这样的场景：每天早上醒来，你的WordPress博客已经自动发布了一篇经过SEO优化的原创文章，而你只需要在前一天晚上设置好关键词和排期时间。这不是魔法，而是Python自动化带来的奇迹。

自动化写作流水线全貌

让我们先看看整个自动化流程的架构图：

graph TD
    A[输入关键词] --> B[调用DeepSeek API生成标题]
    B --> C[调用DeepSeek API生成文章]
    C --> D[SEO优化处理]
    D --> E[排期设置]
    E --> F[WordPress自动发布]

这个流程看似简单，但每一个环节都蕴含着技术细节。下面我将逐步拆解每个步骤的实现方法。

第一步：调用DeepSeek API生成优质标题

标题是文章的门面，一个好的标题能显著提高点击率。我使用DeepSeek的API来批量生成多个候选标题，然后从中挑选最合适的。

import requests
import json

def generate_titles(keyword, num_titles=5):
    url = "https://api.deepseek.com/v1/title_generation"
    headers = {
        "Authorization": "Bearer YOUR_DEEPSEEK_API_KEY",
        "Content-Type": "application/json"
    }
    payload = {
        "keyword": keyword,
        "num_titles": num_titles,
        "tone": "professional",
        "creativity": 0.7
    }

    response = requests.post(url, headers=headers, data=json.dumps(payload))
    if response.status_code == 200:
        return response.json().get("titles", [])
    else:
        print(f"Error generating titles: {response.text}")
        return []

# 示例使用
keyword = "Python自动化写作"
titles = generate_titles(keyword)
print("生成的标题候选：")
for i, title in enumerate(titles, 1):
    print(f"{i}. {title}")

这段代码会基于你提供的关键词，生成5个不同风格的标题。在实际应用中，我会根据历史数据选择点击率高的标题模式，比如"如何..."、"X个技巧..."等。

第二步：批量生成高质量文章内容

有了标题后，下一步是生成文章正文。这里我同样使用DeepSeek的API，但加入了更多控制参数确保内容质量。

def generate_article(title, word_count=1500):
    url = "https://api.deepseek.com/v1/article_generation"
    headers = {
        "Authorization": "Bearer YOUR_DEEPSEEK_API_KEY",
        "Content-Type": "application/json"
    }
    payload = {
        "title": title,
        "word_count": word_count,
        "style": "blog_post",
        "keywords": ["自动化", "Python", "SEO"],
        "avoid_plagiarism": True,
        "include_subheadings": True
    }

    response = requests.post(url, headers=headers, data=json.dumps(payload))
    if response.status_code == 200:
        return response.json().get("article", "")
    else:
        print(f"Error generating article: {response.text}")
        return ""

# 示例使用
selected_title = titles[0]  # 选择第一个标题
article_content = generate_article(selected_title)
print(f"生成的文章内容预览：\n{article_content[:500]}...")

为了确保文章质量，我通常会设置以下参数：

word_count：控制文章长度
style：保持一致的博客风格
keywords：确保包含目标关键词
avoid_plagiarism：避免抄袭
include_subheadings：包含小标题提高可读性

第三步：自动化SEO优化

生成的内容虽然不错，但还需要针对SEO进行优化。我开发了一个简单的SEO优化模块：

from bs4 import BeautifulSoup
import re

def optimize_seo(content, primary_keyword, secondary_keywords=None):
    if secondary_keywords is None:
        secondary_keywords = []

    # 分析内容并生成优化建议
    soup = BeautifulSoup(content, 'html.parser')
    text = soup.get_text()

    # 计算关键词密度
    total_words = len(re.findall(r'\w+', text))
    primary_count = len(re.findall(re.escape(primary_keyword), text, re.IGNORECASE))
    primary_density = (primary_count / total_words) * 100

    # 生成优化报告
    optimization_report = {
        "primary_keyword": primary_keyword,
        "primary_keyword_density": f"{primary_density:.2f}%",
        "recommended_density": "1-2%",
        "word_count": total_words,
        "subheadings": len(soup.find_all(['h2', 'h3'])),
        "images": len(soup.find_all('img')),
        "links": len(soup.find_all('a'))
    }

    # 自动优化内容
    if primary_density < 1:
        # 在适当位置添加关键词
        paragraphs = soup.find_all('p')
        for i in range(0, len(paragraphs), 3):
            if i < len(paragraphs):
                original_text = paragraphs[i].get_text()
                paragraphs[i].string = f"{original_text} {primary_keyword}."

    # 确保第一个段落包含关键词
    first_paragraph = soup.find('p')
    if first_paragraph and primary_keyword.lower() not in first_paragraph.get_text().lower():
        first_paragraph.insert(0, f"{primary_keyword} ")

    return str(soup), optimization_report

# 示例使用
optimized_content, report = optimize_seo(article_content, "Python自动化")
print("SEO优化报告：")
for key, value in report.items():
    print(f"{key}: {value}")

这个SEO优化模块会： 1. 分析关键词密度 2. 检查标题和副标题结构 3. 确保第一段包含主关键词 4. 自动调整关键词分布 5. 生成详细的优化报告

第四步：WordPress自动发布

最后一步是将优化后的内容自动发布到WordPress。我使用python-wordpress-xmlrpc库来实现这一功能：

from wordpress_xmlrpc import Client, WordPressPost
from wordpress_xmlrpc.methods.posts import NewPost
from wordpress_xmlrpc.methods.media import UploadFile
import datetime

def publish_to_wordpress(title, content, keywords, schedule_date=None):
    # 连接WordPress
    wp = Client(
        'https://your-site.com/xmlrpc.php',
        'username',
        'password'
    )

    # 创建文章对象
    post = WordPressPost()
    post.title = title
    post.content = content
    post.post_status = 'publish' if schedule_date is None else 'future'
    post.terms_names = {
        'post_tag': keywords,
        'category': ['技术', 'Python']
    }

    if schedule_date:
        post.date = schedule_date

    # 发布文章
    post_id = wp.call(NewPost(post))
    return post_id

# 示例使用 - 立即发布
publish_to_wordpress(
    title=selected_title,
    content=optimized_content,
    keywords=["Python", "自动化", "SEO"]
)

# 示例使用 - 排期发布
schedule_time = datetime.datetime.now() + datetime.timedelta(days=1)
publish_to_wordpress(
    title=selected_title,
    content=optimized_content,
    keywords=["Python", "自动化", "SEO"],
    schedule_date=schedule_time
)

完整流水线整合

现在，我们将所有步骤整合成一个完整的自动化流水线：

import datetime
from typing import List, Dict

class AutomatedBloggingPipeline:
    def __init__(self, deepseek_api_key: str, wp_username: str, wp_password: str):
        self.deepseek_api_key = deepseek_api_key
        self.wp_username = wp_username
        self.wp_password = wp_password

    def generate_titles(self, keyword: str, num_titles: int = 5) -> List[str]:
        """生成多个标题候选"""
        # 实现代码同上...
        pass

    def generate_article(self, title: str, word_count: int = 1500) -> str:
        """根据标题生成文章内容"""
        # 实现代码同上...
        pass

    def optimize_seo(self, content: str, primary_keyword: str, secondary_keywords: List[str] = None) -> tuple:
        """SEO优化内容"""
        # 实现代码同上...
        pass

    def publish_to_wordpress(self, title: str, content: str, keywords: List[str], schedule_date: datetime.datetime = None) -> int:
        """发布到WordPress"""
        # 实现代码同上...
        pass

    def run_pipeline(self, keyword: str, schedule_days: List[int] = None):
        """运行完整流水线"""
        if schedule_days is None:
            schedule_days = [0]  # 默认当天发布

        # 生成标题
        titles = self.generate_titles(keyword)
        print(f"为关键词 '{keyword}' 生成了 {len(titles)} 个标题")

        # 为每个排期日生成并发布文章
        for i, day_offset in enumerate(schedule_days):
            if i >= len(titles):
                break

            title = titles[i]
            print(f"\n处理标题: {title}")

            # 生成文章
            content = self.generate_article(title)

            # SEO优化
            optimized_content, report = self.optimize_seo(content, keyword)
            print(f"SEO优化完成 - 关键词密度: {report['primary_keyword_density']}")

            # 设置排期时间
            schedule_date = None
            if day_offset > 0:
                schedule_date = datetime.datetime.now() + datetime.timedelta(days=day_offset)
                print(f"文章将排期到: {schedule_date}")

            # 发布文章
            post_id = self.publish_to_wordpress(
                title=title,
                content=optimized_content,
                keywords=[keyword, "自动化", "Python"],
                schedule_date=schedule_date
            )
            print(f"文章发布成功! ID: {post_id}")

# 使用示例
if __name__ == "__main__":
    pipeline = AutomatedBloggingPipeline(
        deepseek_api_key="YOUR_DEEPSEEK_API_KEY",
        wp_username="YOUR_WP_USERNAME",
        wp_password="YOUR_WP_PASSWORD"
    )

    # 为"Python自动化"关键词生成3篇文章，分别在今天、明天和后天发布
    pipeline.run_pipeline(
        keyword="Python自动化",
        schedule_days=[0, 1, 2]
    )

进阶功能：批量处理关键词

为了实现真正的规模化，我添加了批量处理多个关键词的功能：

def batch_process_keywords(self, keywords: List[str], posts_per_keyword: int = 1):
    """批量处理多个关键词"""
    for keyword in keywords:
        print(f"\n开始处理关键词: {keyword}")
        try:
            # 为每个关键词生成指定数量的文章
            self.run_pipeline(
                keyword=keyword,
                schedule_days=list(range(posts_per_keyword))
            print(f"关键词 '{keyword}' 处理完成")
        except Exception as e:
            print(f"处理关键词 '{keyword}' 时出错: {str(e)}")

# 使用示例
keywords_to_process = [
    "Python自动化",
    "SEO优化技巧",
    "内容营销",
    "博客写作"
]

pipeline.batch_process_keywords(
    keywords=keywords_to_process,
    posts_per_keyword=2  # 每个关键词生成2篇文章
)

错误处理与日志记录

在生产环境中，完善的错误处理和日志记录至关重要：

import logging
from logging.handlers import RotatingFileHandler

def setup_logging():
    """配置日志记录"""
    logger = logging.getLogger("AutomatedBlogging")
    logger.setLevel(logging.INFO)

    # 创建文件处理器
    file_handler = RotatingFileHandler(
        'automated_blogging.log',
        maxBytes=1024*1024,
        backupCount=5
    )
    file_handler.setFormatter(logging.Formatter(
        '%(asctime)s - %(levelname)s - %(message)s'
    ))

    # 创建控制台处理器
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(logging.Formatter(
        '%(levelname)s - %(message)s'
    ))

    logger.addHandler(file_handler)
    logger.addHandler(console_handler)
    return logger

# 在类中添加日志记录
class AutomatedBloggingPipeline:
    def __init__(self, ...):
        self.logger = setup_logging()

    def run_pipeline(self, ...):
        try:
            self.logger.info(f"开始处理关键词: {keyword}")
            # ...原有代码...
            self.logger.info(f"文章发布成功! ID: {post_id}")
        except Exception as e:
            self.logger.error(f"处理过程中出错: {str(e)}", exc_info=True)
            raise

性能优化与速率限制

当处理大量内容时，需要注意API调用限制：

import time
from ratelimit import limits, sleep_and_retry

class AutomatedBloggingPipeline:
    # 限制为每分钟30次API调用
    @sleep_and_retry
    @limits(calls=30, period=60)
    def generate_titles(self, ...):
        # ...原有代码...

    @sleep_and_retry
    @limits(calls=30, period=60)
    def generate_article(self, ...):
        # ...原有代码...

    def run_pipeline(self, ...):
        start_time = time.time()
        # ...原有代码...
        end_time = time.time()
        self.logger.info(f"流水线执行时间: {end_time-start_time:.2f}秒")