做技术博客今年第三年了。一开始每篇都手写，从构思到发布，一篇至少要两三个小时。后来内容多了，几个平台同步发，时间根本不够用。

我本身是写Python的后端，就想能不能用代码把重复劳动自动化掉。

花了几周的业余时间，把AI生成、选题发现、排版适配、多平台分发串成了一条流水线。目前跑了几个月，效果还行。这篇文章就拆开来聊聊每个环节我是怎么实现的，代码贴在下面。



整个系统分四个模块：


选题采集 → AI生成初稿 → 排版适配 → 多平台分发


每个模块可以独立运行，也可以串起来全自动跑。

技术栈：Python 3.10+，AI调用DeepSeek API，数据处理用Pandas + jieba，浏览器自动化用Playwright。

---

模块一：选题发现

写了个TopicFinder，从RSS和技术社区的榜单抓取热门话题，用jieba做关键词提取，然后组合成选题建议。

```python
import feedparser
import jieba
import jieba.analyse
from collections import Counter
from datetime import datetime, timedelta
from typing import List, Dict

class TopicFinder:
    """从RSS和技术社区抓取热点话题，提取关键词，生成选题建议"""
   
    def __init__(self):
        self.feeds = [
            "https://news.ycombinator.com/rss",
            "https://www.infoq.cn/feed",
            "https://feeds.feedburner.com/ruanyifeng",
        ]
        self.seed_tags = [
            "Python", "Go", "Rust", "微服务", "Docker",
            "Kubernetes", "AI", "大模型", "数据库", "性能优化"
        ]
   
    def fetch_recent(self, hours=48) -> List[Dict]:
        """从RSS源抓取近期热门文章"""
        articles = []
        deadline = datetime.now() - timedelta(hours=hours)
       
        for url in self.feeds:
            try:
                feed = feedparser.parse(url)
                for entry in feed.entries:
                    pub = entry.get("published_parsed")
                    if pub and datetime(*pub[:6]) >= deadline:
                        articles.append({
                            "title": entry.get("title", ""),
                            "link": entry.get("link", ""),
                            "source": url,
                        })
            except Exception as e:
                print(f"[WARN] 抓取失败 {url}: {e}")
        return articles
   
    def extract_keywords(self, texts: List[str], top_n=15):
        """用TF-IDF和TextRank提取关键词，合并排序"""
        words = []
        for t in texts:
            words.extend(jieba.analyse.extract_tags(t, topK=10, withWeight=True))
            words.extend(jieba.analyse.textrank(t, topK=10, withWeight=True))
       
        counter = Counter()
        for w, s in words:
            counter[w] += s
        return counter.most_common(top_n)
   
    def suggest_topics(self, hot_keywords: List[tuple]) -> List[str]:
        """基于热词生成选题"""
        topics = []
        for word, _ in hot_keywords[:8]:
            if any(tag.lower() in word.lower() for tag in self.seed_tags):
                topics.append(f"从零开始搞懂{word}：实战向教程")
            else:
                topics.append(f"当Python遇上{word}：一个自动化的思路")
       
        # 组合选题
        for i in range(min(4, len(hot_keywords) - 1)):
            w1, w2 = hot_keywords[i][0], hot_keywords[i+1][0]
            topics.append(f"{w1} + {w2}：整合方案与踩坑记录")
       
        return topics
```

运行效果类似这样：

```
抓到 37 篇文章
热门关键词:
  AI Agent: 0.89
  RAG: 0.75
  向量数据库: 0.59
  LLM微调: 0.61

推荐选题:
  1. 从零开始搞懂AI Agent：实战向教程
  2. 当Python遇上RAG：一个自动化的思路
  3. AI Agent + RAG：整合方案与踩坑记录
```

---

模块二：AI生成初稿

ContentGenerator封装了对大模型的调用，传入主题和配置参数，返回Markdown格式的初稿。

```python
import os, time, json
import requests
from dataclasses import dataclass, field
from typing import List, Optional

@dataclass
class ArticleConfig:
    topic: str
    platform: str = "csdn"       # csdn / juejin / weixin
    word_count: int = 2000
    style: str = "技术教程"       # 技术教程 / 原理分析 / 实践记录
    include_code: bool = True
    keywords: List[str] = field(default_factory=list)

class ContentGenerator:
    """调用大模型API生成文章初稿，支持OpenAI兼容接口"""
   
    def __init__(self, api_key: str = "", base_url: str = "",
                 model: str = "deepseek-chat"):
        self.api_key = api_key or os.getenv("LLM_API_KEY", "")
        self.base_url = base_url or os.getenv("LLM_BASE_URL",
                            "https://api.deepseek.com/v1")
        self.model = model
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        })
   
    def generate(self, config: ArticleConfig, retries=3) -> Optional[str]:
        prompt = self._build_prompt(config)
        payload = {
            "model": self.model,
            "messages": [
                {"role": "system", "content": prompt},
                {"role": "user", "content": f"写一篇关于{config.topic}的技术文章"}
            ],
            "temperature": 0.7,
            "max_tokens": 4096,
        }
       
        for i in range(retries):
            try:
                resp = self.session.post(
                    f"{self.base_url}/chat/completions",
                    json=payload, timeout=120
                )
                resp.raise_for_status()
                content = resp.json()["choices"][0]["message"]["content"]
                if len(content) >= config.word_count * 0.6:
                    return content
                payload["temperature"] += 0.1
            except Exception as e:
                print(f"第{i+1}次失败: {e}")
                time.sleep(2 ** i)
        return None
   
    def _build_prompt(self, cfg: ArticleConfig) -> str:
        return f"""你是一位技术博主，写技术文章。
目标读者是开发者，关注实战和原理。
文章要求：
- 内容务实，有可运行的代码示例
- 结构清晰，分小节
- 避免空泛的描述，每段都要有具体信息输出
- 总字数约{cfg.word_count}字
- Markdown格式，代码块标注语言

文章主题：{cfg.topic}
风格：{cfg.style}
"""
```

---

模块三：排版适配

同一个Markdown在不同平台渲染效果不一样。我写了个PostProcessor做格式统一处理——自动分段、补空行、各平台特殊规则适配。

```python
import re

class PostProcessor:
    """Markdown后期处理：统一排版 + 多平台格式适配"""
   
    def __init__(self):
        self.rules = {
            "csdn": {"code_lang_fallback": "text"},
            "juejin": {"escape_html": True},
            "weixin": {"heading_prefix": True},
        }
   
    def format(self, md: str, platform: str = "csdn") -> str:
        text = md
        # 表格前后加空行（有些渲染器不识别紧贴的表格）
        text = re.sub(r'([^\n])\n(\|)', r'\1\n\n\2', text)
        text = re.sub(r'(\|[^\n]+\|)\n([^\n])', r'\1\n\n\2', text)
        # 代码块前后加空行
        text = re.sub(r'([^\n])\n(```)', r'\1\n\n\2', text)
        text = re.sub(r'(```)\n([^\n])', r'\1\n\n\2', text)
        # 处理超长段落
        paras = text.split('\n\n')
        result = []
        for p in paras:
            lines = p.split('\n')
            if any(l.startswith(x) for x in ('```', '#', '|', '-', '*', '>') for l in lines):
                result.append(p)
                continue
            if len(lines) <= 6:
                result.append(p)
                continue
            # 按句号拆长段
            chunks, cur = [], ""
            import re as re2
            sents = re2.split(r'(？|！|。|\. )', p)
            for s in sents:
                cur += s
                if len(cur) >= 250:
                    chunks.append(cur)
                    cur = ""
            if cur:
                chunks.append(cur)
            result.extend(chunks)
       
        text = '\n\n'.join(result)
       
        # 平台特定处理
        if platform == "csdn":
            text = re.sub(r'```(\s*)$', r'```text', text, flags=re.MULTILINE)
        elif platform == "juejin":
            import html
            text = re.sub(r'`([^`]+)`', lambda m: f'`{html.escape(m.group(1))}`', text)
        
        return text
```

---

模块四：半自动发布

发布这步我没全自动化。Playwright打开浏览器填好内容，让我看一眼再点发送——防止手滑发错。

```python
from playwright.sync_api import sync_playwright
import json, time
from pathlib import Path

class Publisher:
    """文章发布器：打开编辑器填入内容，人工确认后发布"""
   
    def __init__(self, cookie_dir="./cookies"):
        self.cookie_dir = Path(cookie_dir)
        self.cookie_dir.mkdir(exist_ok=True)
        self.sites = {
            "csdn": {
                "url": "https://mp.csdn.net/mp_blog/creation/editor",
                "title": "#article-title",
                "content": "#editor-content",
                "publish": ".publish-btn",
            },
            "juejin": {
                "url": "https://juejin.cn/editor/drafts/new",
                "title": ".editor-title",
                "content": ".markdown-editor",
                "publish": ".publish-btn",
            },
        }
   
    def publish(self, site: str, title: str, md_content: str, tags=None):
        cfg = self.sites.get(site)
        if not cfg:
            raise ValueError(f"未适配的平台: {site}")
       
        with sync_playwright() as pw:
            browser = pw.chromium.launch(headless=False)
            ctx = browser.new_context(viewport={"w": 1400, "h": 900})
            self._try_load_cookies(ctx, site)
           
            page = ctx.new_page()
            page.goto(cfg["url"])
           
            # 若需要登录，扫码一次后续自动复用cookie
            if "login" in page.url or "passport" in page.url:
                print("需要扫码登录...")
                page.wait_for_url(lambda u: "login" not in u and "passport" not in u,
                                  timeout=120000)
                self._save_cookies(ctx, site)
           
            page.fill(cfg["title"], title)
            page.fill(cfg["content"], md_content)
           
            if tags:
                tag_el = page.query_selector(".tag-input, .tag-editor")
                if tag_el:
                    for t in tags:
                        tag_el.fill(t)
                        page.keyboard.press("Enter")
                        time.sleep(0.3)
           
            print(f"文章已填入编辑器，请检查: {title}")
            if input("发布？(y/n): ").strip().lower() == 'y':
                page.click(cfg["publish"])
                print("已发布")
           
            browser.close()
   
    def _try_load_cookies(self, ctx, site):
        f = self.cookie_dir / f"{site}.json"
        if f.exists():
            ctx.add_cookies(json.loads(f.read_text()))
   
    def _save_cookies(self, ctx, site):
        f = self.cookie_dir / f"{site}.json"
        f.write_text(json.dumps(ctx.cookies()))
```

---

碰到的几个坑

**1. AI生成的文字太规整，一眼能看出来**
解决方法是生成后跑一遍后处理脚本，把"首先"、"值得注意的是"、"综上所述"这类高频词替换掉，句式打散重排。

**2. 不同平台的Markdown方言不一样**
CSDN的代码块语言标注缺失会显示纯文本，掘金的HTML标签会失效。PostProcessor里针对各平台做了特殊处理。

**3. 相同内容在不同平台被判定重复**
每个平台的版本要在开头和结尾做差异化改写，发布时间间隔至少半天。

**4. 全自动发布有风险**
所以我保留了人工确认环节——脚本填入内容后等我确认才发出去。

---

完整启动步骤

```bash
 环境准备
git clone <项目地址>
cd content-factory
python -m venv venv && source venv/bin/activate
pip install feedparser requests jieba playwright apscheduler
playwright install chromium

配置
export LLM_API_KEY="sk-xxx"
export LLM_BASE_URL="https://api.deepseek.com/v1"

运行一次完整流程
python -c "
from topic_finder import TopicFinder
from content_generator import ContentGenerator, ArticleConfig

finder = TopicFinder()
gen = ContentGenerator()

articles = finder.fetch_recent()
titles = [a['title'] for a in articles]
hot = finder.extract_keywords(titles)
topics = finder.suggest_topics(hot)

config = ArticleConfig(topic=topics[0], platform='csdn')
content = gen.generate(config)

with open(f'output/{topics[0][:20]}.md', 'w') as f:
    f.write(content)
print(f'生成完成，{len(content)}字')
"
```

---

整个过程从选题到初稿大概十几分钟，排版和发布另算。目前跑下来，写技术文章的效率确实提高不少。后面打算加上自动配图和A/B标题测试，不过那是下一步的事了。

有问题欢迎评论区交流，看到了会回。