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

1次阅读

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

手动制作公众号图文时，运营人员通常面临三大效率瓶颈：

数据采集耗时 ：需要人工从知乎、微博等平台复制内容，无法批量获取结构化数据
排版风格混乱 ：不同编辑器的样式不兼容，反复调整行距 / 字体消耗大量时间
多平台分发困难 ：同一内容需在微信 / 头条等平台重复排版，适配各平台规范

以某教育类公众号为例，编辑每天花费 3 小时处理 20 篇图文，其中 60% 时间消耗在复制粘贴和格式调整上。

对比常见爬虫方案在公众号场景的适应性：

方案	动态渲染支持	反爬规避能力	代码复杂度
Clawbot	✅ 内置无头浏览器	✅ 自动轮换 UA/IP	低
Puppeteer	✅	❌ 需自行实现	中
BeautifulSoup	❌ 仅静态解析	❌	低

选择 Clawbot 的核心优势在于其：

开箱即用的动态页面渲染
内置智能重试和代理中间件
与 Python 生态无缝集成

from clawbot import Spider
from typing import List, Dict

class ZhihuHotSpider(Spider):
    def start_requests(self) -> List[Dict]:
        return [{
            'url': 'https://www.zhihu.com/billboard',
            'callback': self.parse_hotlist,
            'headers': {'Referer': 'https://www.zhihu.com/'}
        }]

    def parse_hotlist(self, response) -> List[Dict]:
        items = []
        # 使用 CSS 选择器获取热榜条目（比 XPath 更易维护）for card in response.css('.HotList-item'):
            try:
                items.append({'title': card.css('.HotList-itemTitle::text').get().strip(),
                    'metrics': card.css('.HotList-itemMetrics::text').get()})
            except AttributeError as e:
                self.logger.warning(f'解析异常: {e}')
        return items

创建 Jinja2 模板文件 article.md.j2：

# {{title}}

> 热度: {{metrics}}

{% for paragraph in content %}
{{paragraph}}
{% endfor %}

渲染引擎封装：

from jinja2 import Environment, FileSystemLoader
import os

class MarkdownRenderer:
    def __init__(self, template_dir: str):
        self.env = Environment(loader=FileSystemLoader(template_dir),
            autoescape=True
        )

    def render(self, template_name: str, **context) -> str:
        template = self.env.get_template(template_name)
        return template.render(**context)

import requests
from requests_toolbelt.multipart import encoder

class WechatPublisher:
    def __init__(self, token: str):
        self.session = requests.Session()
        self.token = token

    def upload_media(self, file_path: str) -> str:
        url = f'https://api.weixin.qq.com/cgi-bin/media/upload?access_token={self.token}'

        with open(file_path, 'rb') as f:
            form = encoder.MultipartEncoder({'media': (os.path.basename(file_path), f)
            })

            resp = self.session.post(
                url,
                data=form,
                headers={'Content-Type': form.content_type}
            )

        return resp.json().get('media_id')

import redis
from hashlib import md5

class DedupMiddleware:
    def __init__(self, redis_conn):
        self.redis = redis_conn
        self.EXPIRE_HOURS = 48

    def process_request(self, request):
        # 生成请求指纹
        fp = md5(f'{request["url"]}{request.get("data","")}'.encode()).hexdigest()

        if self.redis.get(fp):
            raise DropRequest(f'Duplicate request: {request["url"]}')

        self.redis.setex(fp, self.EXPIRE_HOURS*3600, 1)

from prometheus_client import Counter, Histogram

REQUEST_COUNT = Counter(
    'clawbot_requests_total', 
    'Total request count',
    ['domain', 'status']
)

RESPONSE_TIME = Histogram(
    'clawbot_response_seconds',
    'Response time distribution',
    ['endpoint']
)

# 在请求回调中埋点
with RESPONSE_TIME.labels(endpoint='wechat_api').time():
    resp = make_request()
    REQUEST_COUNT.labels(
        domain='weixin.qq.com',
        status=resp.status_code
    ).inc()

严格遵守单个 access_token 每日 2000 次调用限制
实现令牌桶算法控制请求速率：

from ratelimit import limits, sleep_and_retry

class WechatAPI:
    @sleep_and_retry
    @limits(calls=5, period=1)  # 每秒 5 次
    def call_api(self):
        pass

import html

def sanitize_content(text: str) -> str:
    # 转义 HTML 特殊字符
    cleaned = html.escape(text)
    # 处理微信 emoji 编码
    return cleaned.replace('[ 表情]', '[emoji]')

扩展多平台发布可考虑：

抽象发布器接口

from abc import ABC, abstractmethod

class PlatformPublisher(ABC):
    @abstractmethod
    def upload(self, content): pass

class WechatPublisher(PlatformPublisher): ...
class ToutiaoPublisher(PlatformPublisher): ...