baoyu图文blog生成技能：从零搭建自动化内容生产流水线

1次阅读

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

手动维护技术博客时，我们常遇到以下问题：

排版耗时 ：每次发布内容都需要手动调整格式，Markdown 语法虽简单，但处理复杂布局（如多图混排）仍需大量时间
素材管理混乱 ：图片分散在不同文件夹，引用路径容易出错，历史版本难以追溯
内容更新低效 ：技术文档需要频繁修订时，涉及多处联动的修改往往遗漏
风格不统一 ：多人协作时，不同作者的写作习惯导致博客整体风格割裂

方案	开发成本	灵活性	性能	学习曲线	适用场景
Markdown 原生	低	差	高	低	简单文档
Jinja2 模板	中	中	中	中	需要部分动态内容的场景
baoyu 生成器	高	高	高	高	自动化内容生产系统

+-------------------+       +-----------------+       +-----------------+
|   ContentBuilder  |<>---->|  TemplateEngine  |<>---->|   ImageProcessor |
+-------------------+       +-----------------+       +-----------------+
| +build_article()  |       | +render()       |       | +optimize()     |
| +validate()       |       | +compile()      |       | +resize()       |
+-------------------+       +-----------------+       +-----------------+
          ^                         ^                          ^
          |                         |                          |
+-------------------+       +-----------------+       +-----------------+
|   MarkdownParser  |       |    ASTVisitor   |       |  CacheManager   |
+-------------------+       +-----------------+       +-----------------+

from typing import List, Dict
from dataclasses import dataclass

@dataclass
class ImageNode:
    url: str
    alt_text: str
    width: int = 800

class ContentProcessor:
    def __init__(self, max_width: int = 1200):
        self.max_width = max_width

    def process_images(self, nodes: List[ImageNode]) -> List[Dict]:
        """ 处理图片节点并生成响应式 HTML

        Args:
            nodes: 图片节点列表

        Returns:
            处理后的图片属性字典列表
        """
        processed = []
        for node in nodes:
            try:
                ratio = min(1, self.max_width / node.width)
                processed.append({
                    'src': node.url,
                    'alt': node.alt_text,
                    'style': f"width:{ratio*100}%;max-width:{node.width}px"
                })
            except ZeroDivisionError:
                processed.append({'src': node.url, 'alt': node.alt_text})

        return processed

import aiohttp
import asyncio
from pathlib import Path

async def download_image(url: str, save_path: Path):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            if response.status == 200:
                with open(save_path, 'wb') as f:
                    while True:
                        chunk = await response.content.read(1024)
                        if not chunk:
                            break
                        f.write(chunk)

async def batch_download(urls: List[str], output_dir: Path):
    tasks = []
    os.makedirs(output_dir, exist_ok=True)

    for idx, url in enumerate(urls):
        save_path = output_dir / f"img_{idx}{Path(url).suffix}"
        tasks.append(download_image(url, save_path))

    await asyncio.gather(*tasks, return_exceptions=True)