Clawbot 生成公众号图文实战指南：从零搭建自动化内容生产流水线

1次阅读

没有评论

共计 2637 个字符，预计需要花费 7 分钟才能阅读完成。

手动制作公众号图文一直是内容运营者的噩梦，尤其在以下三个方面效率极低：

数据采集耗时 ：每天需要人工从知乎、微博等平台复制粘贴热点内容，不仅费时还容易遗漏重要信息。
排版一致性差 ：不同编辑制作的图文风格不统一，影响品牌形象。
多平台分发困难 ：同一内容需要在多个平台发布时，需要重复编辑和调整格式。

在选择自动化工具时，我们对比了三种主流方案：

Clawbot：
优势：内置动态渲染和反爬规避机制，适合处理 JavaScript 渲染的页面。
劣势：学习曲线较陡，需要一定的 Python 基础。
Puppeteer：
优势：强大的动态页面处理能力。
劣势：需要 Node.js 环境，不适合纯 Python 技术栈。
BeautifulSoup：
优势：简单易用，适合静态页面。
劣势：无法处理动态渲染的内容。

最终选择 Clawbot，因其在动态渲染和反爬规避方面的综合表现最佳。

使用 Clawbot 抓取知乎热榜数据，并通过 XPath 选择器提取关键信息：

from clawbot import Clawbot
import json

def fetch_zhihu_hot() -> list[dict]:
    """抓取知乎热榜数据"""
    bot = Clawbot()
    try:
        bot.get("https://www.zhihu.com/hot")
        items = bot.find_all("//div[@class='HotItem-content']", timeout=10)
        results = []
        for item in items:
            title = item.find(".//h2").text
            url = item.find(".//a").get("href")
            results.append({"title": title, "url": url})
        return results
    except Exception as e:
        print(f"抓取失败: {e}")
        return []
    finally:
        bot.close()

使用 Jinja2 将采集的数据渲染为 Markdown 格式：

from jinja2 import Environment, FileSystemLoader

def render_markdown(data: list[dict]) -> str:
    """渲染 Markdown 模板"""
    env = Environment(loader=FileSystemLoader("./templates"))
    template = env.get_template("wechat.md.j2")
    return template.render(articles=data)

模板文件示例（wechat.md.j2）：

# 今日热点

{% for article in articles %}
## {{article.title}}

[阅读原文]({{article.url}})

---
{% endfor %}

通过 wxpy 库自动上传到公众号后台：

import requests

def upload_to_wechat(content: str, title: str) -> bool:
    """上传内容到公众号"""
    url = "https://api.weixin.qq.com/cgi-bin/material/add_news"
    headers = {"Content-Type": "multipart/form-data"}
    data = {
        "title": title,
        "content": content,
        "digest": "今日热点精选"
    }
    try:
        response = requests.post(url, headers=headers, data=data)
        return response.json().get("errcode") == 0
    except Exception as e:
        print(f"上传失败: {e}")
        return False

使用 Redis 实现请求去重和失败重试机制：

import redis
from tenacity import retry, stop_after_attempt

redis_pool = redis.ConnectionPool(host="localhost", port=6379, db=0)

@retry(stop=stop_after_attempt(3))
def fetch_with_retry(url: str) -> str:
    """带重试的请求"""
    r = redis.Redis(connection_pool=redis_pool)
    if r.sismember("visited_urls", url):
        return ""

    # 实际请求代码...
    r.sadd("visited_urls", url)
    return "fetched content"

基于 Prometheus 的监控埋点：

from prometheus_client import Counter, start_http_server

REQUEST_COUNT = Counter("wechat_publish_count", "Number of articles published")

def upload_with_metrics(content: str) -> bool:
    """带监控的上传"""
    success = upload_to_wechat(content)
    if success:
        REQUEST_COUNT.inc()
    return success

start_http_server(8000)  # 启动监控服务器