Vibe Coding 课程-干货集训
07 自动化写作
Skill文件夹结构
.claude/skills/autokafka/
├── SKILL.md # Skill 主文档（工作流、写作规范、配图系统）
└── tools/ # Python 工具集
├── init.py
├── progress_updater.py # 前端进度同步（必须在每个阶段调用）
├── batch_image_generator.py # Gemini API 批量生图
├── prompt_generator.py # 插图 Prompt 生成器
├── research_analyzer.py # Research 分析器
├── weixin_collector.py # 微信公众号采集
├── redbook_collector.py # 小红书采集
└── layout_engine.py # 图文编排引擎Skill详细内容
SKILL.md
---
name: autokafka
description: |
AI科普文章自动化写作系统。用于生成面向AI小白读者的高质量、有温度、无AI味的科普文章。
触发条件：
- 用户说 "/autokafka <主题>" 或 "/autokafka"
- 用户请求写科普文章、AI科普、技术科普
- 用户想要深度研究某个AI/技术话题并生成文章
- 用户提到"写一篇关于xxx的文章"
功能：多源深度研究（Web/公众号/小红书）→ 智能写作 → 7维度自审 → Gemini配图生成
---
# AutoKafka - 大师级自动化科普写作
> 像卡夫卡一样精准，像余华一样朴素，像马尔克斯一样让平凡变得神奇。
## 核心流程
```
用户输入 → INPUT ROUTER → DEEP RESEARCH → WRITING PIPELINE → ILLUSTRATION → OUTPUT
```
## 前端实时同步（强制执行）
**每个阶段必须调用 `progress_updater.py` 更新前端状态！**
工具路径：`.claude/skills/autokafka/tools/progress_updater.py`
### 同步时机和命令
| 阶段 | 命令 | 示例 |
|------|------|------|
| 任务开始 | `init` | `python3 .claude/skills/autokafka/tools/progress_updater.py init "文章标题"` |
| 每个步骤 | `step` | `python3 .claude/skills/autokafka/tools/progress_updater.py step "输入路由" "确定主题"` |
| 状态更新 | `status` | `python3 .claude/skills/autokafka/tools/progress_updater.py status researching 30` |
| 研究完成 | `research` | `python3 .claude/skills/autokafka/tools/progress_updater.py research '{...}'` |
| 文章完成 | `article` | `python3 .claude/skills/autokafka/tools/progress_updater.py article '{...}'` |
| 审核完成 | `review` | `python3 .claude/skills/autokafka/tools/progress_updater.py review '{...}'` |
| 任务结束 | `complete` | `python3 .claude/skills/autokafka/tools/progress_updater.py complete` |
### 进度百分比参考
| 阶段 | 进度范围 |
|------|----------|
| 输入路由 | 0-5% |
| 关键词策略 | 5-10% |
| 多源采集 | 10-25% |
| 信息筛选 | 25-30% |
| 研究报告生成 | 30-35% |
| 用户选择档位 | 35-40% |
| 大纲生成 | 40-50% |
| 草稿撰写 | 50-70% |
| 自我Review | 70-80% |
| 配图规划 | 80-85% |
| 配图生成 | 85-95% |
| 终稿输出 | 95-100% |
### 执行示例
```bash
# 1. 任务开始时立即初始化
python3 .claude/skills/autokafka/tools/progress_updater.py init "个人电脑发展简史"
# 2. 每完成一个步骤，添加 step
python3 .claude/skills/autokafka/tools/progress_updater.py step "输入路由" "✓ 确定主题（演进脉络角度）"
python3 .claude/skills/autokafka/tools/progress_updater.py status researching 5
# 3. 研究阶段
python3 .claude/skills/autokafka/tools/progress_updater.py step "关键词策略" "核心词：个人电脑、PC发展史"
python3 .claude/skills/autokafka/tools/progress_updater.py status researching 10
# 4. 研究完成后更新报告
python3 .claude/skills/autokafka/tools/progress_updater.py research '{"title":"...", "summary":"...", "insights":[...]}'
python3 .claude/skills/autokafka/tools/progress_updater.py status writing 40
# 5. 文章完成后更新
python3 .claude/skills/autokafka/tools/progress_updater.py article '{"title":"...", "content":"...", "wordCount":3000}'
# 6. 任务完成
python3 .claude/skills/autokafka/tools/progress_updater.py complete
```
**重要：前端地址 http://localhost:5173 需要提前运行 `cd frontend && npm run dev`**
---
## 模块一：输入路由
判断输入类型：
**确定主题**（直接进入研究）：
- 包含文章类型词：指南、教程、入门、详解、对比、分析
- 包含限定词：如何、怎么、是什么、vs
- 结构完整：主体 + 角度
**模糊关键词**（先生成3个选题）：
- 仅为概念名词：世界模型、RAG、Agent
- 无明确角度或文章类型
## 模块二：Deep Research
### 多源信息采集
| 信息源 | 工具 | 数量 |
|--------|------|------|
| Web Search | WebSearch 工具 | Top 8 |
| 公众号 | weixin_search_mcp | Top 5 |
| 小红书 | redbook_mcp | Top 5 |
### 关键词策略
- 核心词：从用户输入提取，保持不变
- 细化词：核心词 + 教程/配置/实战/案例/原理/入门
- 禁止：扩大到上位概念
### 筛选规则
- 时效性：仅保留3个月内内容（官方文档6个月）
- 相关性打分：标题命中+3，摘要命中+1，可信源+2
- 最终保留：10篇（直接关键词Top5 + 扩展关键词Top3）
### 可信源列表
公众号：机器之心、量子位、新智元、AI前线、硅星人、极客公园、少数派
网站：sspai.com、36kr.com、jiqizhixin.com、juejin.cn、docs.anthropic.com、openai.com
### Research Report 输出
生成包含以下内容的报告（展示给用户）：
- 信息源概览
- 核心定义（含共识度：★★★★★ 多源共识 / ★★★☆☆ 双源验证 / ★★☆☆☆ 待验证）
- 关键洞察
- 知识图谱
- 源文章列表
### 选题生成（仅模糊关键词触发）
生成3个差异化选题，角度池：
- 入门解惑、原理探秘、实战指南、演进脉络、对比分析、前沿洞察、哲思深挖
## 模块三：科普文写作
### 写作哲学
**核心原则：**
1. 说人话 — 你奶奶能听懂
2. 有温度 — 真实的人在分享
3. 有洞察 — 不只"是什么"，更要"为什么"和"所以呢"
4. 有节制 — 金句在恰当时刻闪光
**禁止清单：**
- "众所周知..." "让我们来看看..." "首先...其次...最后..."
- "这是一个革命性的技术" "简单来说..."
- 过度感叹号、堆砌术语
### 字数档位（询问用户）
| 档位 | 字数 | 金句 | Section |
|------|------|------|---------|
| A. 精炼版 | 1200-2000 | 2-3句 | 2-3个 |
| B. 标准版 | 2000-4000 | 3-5句 | 3-4个 |
| C. 深度版 | 4000-7000 | 5-7句 | 4-6个 |
| D. 长文版 | 7000-10000 | 6-8句 | 5-8个 |
### 文章结构
```
HOOK (100-200字) - 3秒抓住读者
├── 反常识开头 / 场景代入 / 尖锐问题 / 故事开头
BRIDGE (50-100字) - 告诉读者能得到什么
CORE (根据档位)
├── 概念/观点陈述（大白话）
├── 类比/例子（让抽象变具体）
├── 延伸思考
└── [可选] 金句
CLOSE (100-200字) - 留下余韵
├── 回扣开头 / 开放问题 / 视野拉远 / 个人感悟
```
### 类比策略
用熟悉解释陌生，来源池：
- 日常生活：做饭、开车、装修
- 人际关系：交朋友、带孩子
- 自然现象：河流、天气
- 城市场景：外卖、地铁
### 写作流水线
**Stage 1: 大纲生成** → 用户审核
**Stage 2: 草稿生成**（带颜色标注）→ 用户审核
- 🟡 HOOK / 🟢 GOLDEN / 🔵 ANALOGY / 🟣 CLOSE / 🔴 NEED_VERIFY / 🟠 AI_SMELL
**Stage 3: 自我Review**（展示给用户）
- 7维度评分：人话指数、类比质量、逻辑连贯、金句质量、AI味检测、开头吸引力、收尾余韵
**Stage 4: 终稿生成** → 用户审核
## 模块四：插图系统
### 统一视觉风格（强制）
所有配图必须遵循以下视觉系统，确保风格一致性：
```
┌─────────────────────────────────────────────────────────────┐
│ KAFKA 插画风格规范 │
├─────────────────────────────────────────────────────────────┤
│ 风格：纯手绘线条风，极简主义 │
│ 线条：单色细线（深棕 #3D3D3D 或黑色 #1A1A1A） │
│ 有机、略带抖动感的手绘质感，非机械直线 │
│ 背景：统一高端米黄色 #F5F0E6 │
│ 填充：无填充或极淡的同色系晕染 │
│ 文案：可有少量手写风格英文标注，不超过3个单词 │
│ 禁止：渐变、阴影、3D效果、复杂纹理、多色彩 │
└─────────────────────────────────────────────────────────────┘
```
### 占位符命名规范（强制）
为避免图片匹配错误，占位符必须包含图片ID：
```
格式：📍 [插图：{image_id} - {描述}]
示例：
📍 [插图：img_hero - 头图：世界模型概念可视化]
📍 [插图：img_pavlov - 巴甫洛夫的狗 vs 人类脑补]
📍 [插图：img_vmc - 2018 World Models V-M-C 架构图]
```
**规则：**
1. `image_id` 必须与生成的图片文件名一致（不含扩展名）
2. 先规划所有图片ID，再在文章中插入占位符
3. 图片生成时使用相同的 ID 作为文件名
### Prompt 生成流程（三步法）
每张配图的 prompt 必须按以下流程生成：
**Step 1: 理解段落主旨**
- 这个段落在讲什么核心概念？
- 作者想传达什么洞察？
**Step 2: 抽象隐喻元素**
- 将概念转化为视觉符号/隐喻
- 例：「AI外挂技能」→ 机器人 + 环绕大脑的技能包
**Step 3: 描述极简画面**
- 用最少的元素表达核心隐喻
- 明确元素位置关系和连接方式
### Prompt 模板
```
[STYLE PREFIX - 必须放在每个prompt开头]
STRICT STYLE REQUIREMENTS:
- Style: Hand-drawn line art, minimalist sketch style
- Lines: Single-weight organic lines in dark brown (#3D3D3D), slightly wobbly like real pen strokes
- Background: Solid warm cream color (#F5F0E6), no gradients
- Fill: No fill, or very light same-tone wash for emphasis
- Text: Maximum 3 handwritten-style English words as labels (optional)
- Aesthetic: Editorial illustration style, like The New Yorker or Monocle magazine
- NO gradients, NO shadows, NO 3D effects, NO complex textures, NO multiple colors
- Aspect ratio: 16:9
[CONTENT - 具体画面描述]
SCENE DESCRIPTION:
- Main element: [核心视觉元素，如：一个简笔画机器人头部轮廓]
- Secondary elements: [辅助元素，如：3个小方块漂浮在头部周围，用虚线连接到大脑]
- Composition: [构图，如：居中构图，元素紧凑]
- Optional label: [可选标注，如：小字 "Skills" 在角落]
NEGATIVE PROMPT:
photorealistic, 3D render, gradient, shadow, complex texture, colorful, cartoon style, anime, digital art, multiple colors, busy background
```
### 插图数量
| 档位 | 数量 | 必有 |
|------|------|------|
| 精炼版 | 1-2张 | 头图 |
| 标准版 | 3-4张 | 头图+概念+类比 |
| 深度版 | 4-6张 | 头图+概念+类比+流程 |
| 长文版 | 5-8张 | 每个Section至少1张 |
### 插图类型
- HERO: 头图，文章核心概念的视觉隐喻
- CONCEPT: 概念可视化，将抽象概念具象化
- ANALOGY: 类比场景图，配合文中类比
- DIAGRAM: 流程/结构图，用线条连接的元素关系
- MOOD: 氛围图，留白较多的意境图
### 生图工具
使用 `tools/batch_image_generator.py` 调用 Gemini API 批量生成。
配置文件：项目根目录 `config.env`
## 工具脚本
- `tools/progress_updater.py` - **前端进度同步**（必须在每个阶段调用）
- `tools/batch_image_generator.py` - 批量生图（Gemini REST API）
- `tools/prompt_generator.py` - 插图 Prompt 生成器
- `tools/layout_engine.py` - 图文编排引擎
- `tools/research_analyzer.py` - Research 分析器
- `tools/weixin_collector.py` - 公众号采集
- `tools/redbook_collector.py` - 小红书采集
init.py
"""
AutoKafka 工具集
"""
from .weixin_collector import WeixinCollector, collect_weixin_articles
from .redbook_collector import RedbookCollector, collect_redbook_notes
from .research_analyzer import (
ResearchAnalyzer,
ResearchReport,
ConsensusLevel,
create_research_report
)
from .layout_engine import LayoutEngine, generate_layout, IllustrationType, LengthTier
from .prompt_generator import PromptGenerator, generate_image_prompts
from .batch_image_generator import (
GeminiImageGenerator,
MockImageGenerator,
generate_images
)
__all__ = [
# 采集器
'WeixinCollector',
'collect_weixin_articles',
'RedbookCollector',
'collect_redbook_notes',
# Research
'ResearchAnalyzer',
'ResearchReport',
'ConsensusLevel',
'create_research_report',
# 编排
'LayoutEngine',
'generate_layout',
'IllustrationType',
'LengthTier',
# Prompt
'PromptGenerator',
'generate_image_prompts',
# 生图
'GeminiImageGenerator',
'MockImageGenerator',
'generate_images',
]
progress_updater.py
#!/usr/bin/env python3
"""
AutoKafka 前端进度同步工具
用于实时更新 frontend/public/data/tasks.json，
让前端能够展示写作任务的实时进度（进行中 + 历史）。
使用方式（在 Claude 中通过 Bash 调用）：
# 初始化新任务（自动归档旧的进行中任务）
python3 tools/progress_updater.py init "文章标题"
# 添加思考步骤
python3 tools/progress_updater.py step "输入路由" "确定主题（演进脉络角度）"
# 更新状态和进度
python3 tools/progress_updater.py status researching 20
# 更新研究报告
python3 tools/progress_updater.py research '{"title": "...", "summary": "..."}'
# 更新文章内容
python3 tools/progress_updater.py article '{"title": "...", "content": "...", "wordCount": 2000}'
# 更新审核分数
python3 tools/progress_updater.py review '{"overall": 85, "dimensions": [...]}'
# 标记完成
python3 tools/progress_updater.py complete
"""
import json
import os
import sys
from datetime import datetime
from pathlib import Path
# 配置路径
SCRIPT_DIR = Path(__file__).parent
PROJECT_ROOT = SCRIPT_DIR.parent.parent.parent.parent # auto_kafka/
TASKS_FILE = PROJECT_ROOT / "frontend" / "public" / "data" / "tasks.json"
MAX_HISTORY = 20 # 最多保留的历史任务数
def ensure_dir():
"""确保目录存在"""
TASKS_FILE.parent.mkdir(parents=True, exist_ok=True)
def load_tasks():
"""加载任务列表"""
if TASKS_FILE.exists():
with open(TASKS_FILE, 'r', encoding='utf-8') as f:
data = json.load(f)
if isinstance(data, list):
return data
# 兼容旧的单任务格式
elif isinstance(data, dict) and 'id' in data:
return [data]
return []
def save_tasks(tasks):
"""保存任务列表"""
ensure_dir()
with open(TASKS_FILE, 'w', encoding='utf-8') as f:
json.dump(tasks, f, ensure_ascii=False, indent=2)
print(f"[Progress] Updated: {TASKS_FILE}")
def now_iso():
"""返回 ISO 格式时间戳"""
return datetime.now().isoformat() + "Z"
def get_current_task(tasks):
"""获取当前进行中的任务"""
for task in tasks:
if task.get('status') not in ['completed', 'failed']:
return task
return None
def init_task(title: str):
"""初始化新任务"""
tasks = load_tasks()
# 将当前进行中的任务标记为失败（被中断）
for task in tasks:
if task.get('status') not in ['completed', 'failed']:
task['status'] = 'failed'
task['currentPhase'] = '已中断'
task['completedAt'] = now_iso()
# 创建新任务
new_task = {
"id": f"task-{datetime.now().strftime('%Y%m%d%H%M%S')}",
"title": title,
"status": "researching",
"currentPhase": "初始化",
"progress": 0,
"createdAt": now_iso(),
"thinkingSteps": []
}
# 新任务放在列表开头
tasks.insert(0, new_task)
# 限制历史任务数量
if len(tasks) > MAX_HISTORY:
tasks = tasks[:MAX_HISTORY]
save_tasks(tasks)
print(f"[Progress] Initialized task: {title}")
return new_task
def add_step(phase: str, content: str):
"""添加思考步骤"""
tasks = load_tasks()
current = get_current_task(tasks)
if not current:
print("[Error] No active task found. Run 'init' first.")
return
step_id = f"step-{len(current.get('thinkingSteps', [])) + 1}"
step = {
"id": step_id,
"phase": phase,
"content": content,
"timestamp": now_iso(),
"status": "completed"
}
if "thinkingSteps" not in current:
current["thinkingSteps"] = []
current["thinkingSteps"].append(step)
current["currentPhase"] = phase
save_tasks(tasks)
print(f"[Progress] Added step: {phase}")
def update_status(status: str, progress: int):
"""更新状态和进度"""
tasks = load_tasks()
current = get_current_task(tasks)
if not current:
print("[Error] No active task found. Run 'init' first.")
return
current["status"] = status
current["progress"] = min(100, max(0, progress))
# 根据状态更新 currentPhase
phase_map = {
"researching": "深度研究中",
"writing": "文章撰写中",
"reviewing": "质量审核中",
"completed": "全部完成",
"failed": "任务失败"
}
if status in phase_map:
current["currentPhase"] = phase_map[status]
save_tasks(tasks)
print(f"[Progress] Status: {status}, Progress: {progress}%")
def update_research(research_json: str):
"""更新研究报告"""
tasks = load_tasks()
current = get_current_task(tasks)
if not current:
print("[Error] No active task found. Run 'init' first.")
return
try:
research = json.loads(research_json)
current["researchReport"] = research
save_tasks(tasks)
print("[Progress] Updated research report")
except json.JSONDecodeError as e:
print(f"[Error] Invalid JSON: {e}")
def update_article(article_json: str):
"""更新文章内容"""
tasks = load_tasks()
current = get_current_task(tasks)
if not current:
print("[Error] No active task found. Run 'init' first.")
return
try:
article = json.loads(article_json)
current["article"] = article
save_tasks(tasks)
print(f"[Progress] Updated article: {article.get('wordCount', 0)} words")
except json.JSONDecodeError as e:
print(f"[Error] Invalid JSON: {e}")
def update_review(review_json: str):
"""更新审核分数"""
tasks = load_tasks()
current = get_current_task(tasks)
if not current:
print("[Error] No active task found. Run 'init' first.")
return
try:
review = json.loads(review_json)
current["reviewScores"] = review
save_tasks(tasks)
print("[Progress] Updated review scores")
except json.JSONDecodeError as e:
print(f"[Error] Invalid JSON: {e}")
def complete_task():
"""标记任务完成"""
tasks = load_tasks()
current = get_current_task(tasks)
if not current:
print("[Error] No active task found. Run 'init' first.")
return
current["status"] = "completed"
current["progress"] = 100
current["currentPhase"] = "全部完成"
current["completedAt"] = now_iso()
save_tasks(tasks)
print("[Progress] Task completed!")
def main():
if len(sys.argv) < 2:
print(__doc__)
return
cmd = sys.argv[1]
if cmd == "init" and len(sys.argv) >= 3:
init_task(sys.argv[2])
elif cmd == "step" and len(sys.argv) >= 4:
add_step(sys.argv[2], sys.argv[3])
elif cmd == "status" and len(sys.argv) >= 4:
update_status(sys.argv[2], int(sys.argv[3]))
elif cmd == "research" and len(sys.argv) >= 3:
update_research(sys.argv[2])
elif cmd == "article" and len(sys.argv) >= 3:
update_article(sys.argv[2])
elif cmd == "review" and len(sys.argv) >= 3:
update_review(sys.argv[2])
elif cmd == "complete":
complete_task()
else:
print(f"Unknown command: {cmd}")
print(__doc__)
if __name__ == "__main__":
main()
batch_image_generator.py
"""
批量图像生成器
调用 Google Gemini API 批量生成图片
"""
import os
import json
import time
import asyncio
import base64
import requests
from typing import List, Dict, Optional
from dataclasses import dataclass, asdict
from pathlib import Path
@dataclass
class GenerationResult:
"""生成结果"""
id: str
status: str # success / failed / pending
path: Optional[str] = None
position: str = ""
error: Optional[str] = None
@dataclass
class BatchResult:
"""批量生成结果"""
total: int
success: int
failed: int
results: List[GenerationResult]
class GeminiImageGenerator:
"""Google Gemini 图像生成器 (REST API)"""
def __init__(self, config_path: str = None):
"""
初始化生成器
Args:
config_path: 配置文件路径
"""
self.config = self._load_config(config_path)
self.api_key = self.config.get('GEMINI_API_KEY', '')
self.model_name = self.config.get('NANO_BANANA_MODEL', 'gemini-2.0-flash-exp')
# 代理配置
self.http_proxy = self.config.get('HTTP_PROXY', '')
self.https_proxy = self.config.get('HTTPS_PROXY', '')
self.proxies = {}
if self.http_proxy:
self.proxies['http'] = self.http_proxy
if self.https_proxy:
self.proxies['https'] = self.https_proxy
if not self.api_key:
print("警告: Gemini API Key 未配置")
else:
print(f"已配置 Gemini API，模型: {self.model_name}")
if self.proxies:
print(f"使用代理: {self.https_proxy or self.http_proxy}")
def _load_config(self, config_path: str = None) -> Dict:
"""加载配置文件"""
if config_path is None:
project_root = Path(__file__).parent.parent.parent.parent.parent
config_path = project_root / 'config.env'
if not os.path.exists(config_path):
print(f"配置文件不存在: {config_path}")
return {}
try:
config = {}
with open(config_path, 'r', encoding='utf-8') as f:
for line in f:
line = line.strip()
if line and not line.startswith('#') and '=' in line:
key, value = line.split('=', 1)
config[key.strip()] = value.strip().strip('"\'')
return config
except Exception as e:
print(f"加载配置文件失败: {e}")
return {}
def generate_single(
self,
prompt: str,
negative_prompt: str = "",
output_path: str = None,
width: int = 1024,
height: int = 576
) -> Optional[str]:
"""
生成单张图片
Args:
prompt: 正向提示词
negative_prompt: 负向提示词
output_path: 输出路径
width: 图片宽度
height: 图片高度
Returns:
生成的图片路径，失败返回 None
"""
if not self.api_key:
print("API Key 未配置")
return None
try:
# 组合提示词
full_prompt = prompt
if negative_prompt:
full_prompt += f"\n\nAvoid: {negative_prompt}"
# API 端点
url = f"https://generativelanguage.googleapis.com/v1beta/models/{self.model_name}:generateContent?key={self.api_key}"
# 请求数据
data = {
"contents": [{
"parts": [{"text": full_prompt}]
}],
"generationConfig": {
"responseModalities": ["TEXT", "IMAGE"],
"imageConfig": {
"aspectRatio": "16:9"
}
}
}
headers = {
"Content-Type": "application/json"
}
# 发送请求
response = requests.post(
url,
headers=headers,
json=data,
timeout=120,
proxies=self.proxies if self.proxies else None
)
if response.status_code != 200:
print(f"API 错误 ({response.status_code}): {response.text[:500]}")
return None
result = response.json()
# 提取图片数据
candidates = result.get("candidates", [])
if not candidates:
print(f"未生成图片")
return None
parts = candidates[0].get("content", {}).get("parts", [])
for part in parts:
if "inlineData" in part:
image_data = part["inlineData"]["data"]
image_bytes = base64.b64decode(image_data)
if output_path:
# 确保目录存在
output_path = Path(output_path)
output_path.parent.mkdir(parents=True, exist_ok=True)
# 保存图片
with open(output_path, 'wb') as f:
f.write(image_bytes)
return str(output_path)
print("响应中未找到图片数据")
return None
except requests.exceptions.Timeout:
print("请求超时")
return None
except requests.exceptions.RequestException as e:
print(f"请求错误: {e}")
return None
except Exception as e:
print(f"生成图片失败: {e}")
import traceback
traceback.print_exc()
return None
async def generate_batch_async(
self,
prompts: List[Dict],
output_dir: str,
delay: float = 2.0
) -> BatchResult:
"""
异步批量生成图片
Args:
prompts: Prompt 列表
output_dir: 输出目录
delay: 请求间隔（秒）
Returns:
批量生成结果
"""
os.makedirs(output_dir, exist_ok=True)
results = []
for prompt_data in prompts:
prompt_id = prompt_data.get('id', 'img')
prompt_text = prompt_data.get('prompt', '')
negative = prompt_data.get('negative_prompt', '')
position = prompt_data.get('position', '')
output_path = os.path.join(output_dir, f"{prompt_id}.png")
print(f"\n生成 {prompt_id}...")
print(f" Prompt: {prompt_text[:100]}...")
try:
result_path = self.generate_single(
prompt=prompt_text,
negative_prompt=negative,
output_path=output_path
)
if result_path and os.path.exists(result_path):
results.append(GenerationResult(
id=prompt_id,
status='success',
path=result_path,
position=position
))
print(f" ✓ 成功: {result_path}")
else:
results.append(GenerationResult(
id=prompt_id,
status='failed',
position=position,
error='生成失败或文件未保存'
))
print(f" ✗ 失败")
except Exception as e:
results.append(GenerationResult(
id=prompt_id,
status='failed',
position=position,
error=str(e)
))
print(f" ✗ 错误: {e}")
# 请求间隔，避免 rate limit
if delay > 0:
await asyncio.sleep(delay)
success_count = len([r for r in results if r.status == 'success'])
failed_count = len([r for r in results if r.status == 'failed'])
return BatchResult(
total=len(prompts),
success=success_count,
failed=failed_count,
results=results
)
def generate_batch(
self,
prompts: List[Dict],
output_dir: str,
delay: float = 2.0
) -> BatchResult:
"""同步版本的批量生成"""
return asyncio.run(self.generate_batch_async(prompts, output_dir, delay))
class MockImageGenerator:
"""模拟图像生成器（用于测试）"""
def __init__(self):
pass
def generate_batch(
self,
prompts: List[Dict],
output_dir: str,
delay: float = 0.1
) -> BatchResult:
"""批量生成模拟图片"""
os.makedirs(output_dir, exist_ok=True)
results = []
for prompt_data in prompts:
prompt_id = prompt_data.get('id', 'img')
position = prompt_data.get('position', '')
prompt_type = prompt_data.get('type', 'image')
output_path = os.path.join(output_dir, f"{prompt_id}.png")
try:
from PIL import Image, ImageDraw
# 创建渐变背景
width, height = 1024, 576
img = Image.new('RGB', (width, height))
# 简单渐变
for y in range(height):
r = int(200 + 40 * (y / height))
g = int(210 + 30 * (y / height))
b = int(230 + 20 * (y / height))
for x in range(width):
img.putpixel((x, y), (r, g, b))
draw = ImageDraw.Draw(img)
# 添加标识文字
text = f"[{prompt_type.upper()}]\n{prompt_id}"
draw.text((width//2, height//2), text, fill=(100, 100, 120), anchor='mm')
img.save(output_path)
results.append(GenerationResult(
id=prompt_id,
status='success',
path=output_path,
position=position
))
print(f" ✓ Mock 生成: {output_path}")
except ImportError:
# 没有 PIL，创建占位文件
with open(output_path.replace('.png', '.txt'), 'w') as f:
f.write(f"Placeholder for {prompt_id}\n{prompt_data.get('prompt', '')}")
results.append(GenerationResult(
id=prompt_id,
status='success',
path=output_path.replace('.png', '.txt'),
position=position
))
time.sleep(delay)
return BatchResult(
total=len(prompts),
success=len(results),
failed=0,
results=results
)
# 便捷函数
def generate_images(
prompts: List[Dict],
output_dir: str,
config_path: str = None,
use_mock: bool = False
) -> Dict:
"""
生成图片（便捷函数）
Args:
prompts: Prompt 列表
output_dir: 输出目录
config_path: 配置文件路径
use_mock: 是否使用模拟生成器
Returns:
生成结果
"""
if use_mock:
generator = MockImageGenerator()
else:
generator = GeminiImageGenerator(config_path)
result = generator.generate_batch(prompts, output_dir)
return {
'total': result.total,
'success': result.success,
'failed': result.failed,
'results': [asdict(r) for r in result.results]
}
if __name__ == "__main__":
# 测试
test_prompts = [
{
'id': 'img_001',
'type': 'hero',
'prompt': 'Minimalist digital illustration of AI assistant helping human, soft gradients, modern tech aesthetic, clean lines, no text',
'negative_prompt': 'ugly, blurry, text, words',
'position': 'before_paragraph_1'
}
]
print("=== 测试图像生成 ===\n")
# 用真实 API 测试
print("真实 API 测试:")
result = generate_images(
prompts=test_prompts,
output_dir='./test_images_real',
use_mock=False
)
print(json.dumps(result, indent=2, ensure_ascii=False))
prompt_generator.py
"""
插图 Prompt 生成器
根据编排方案生成 Gemini 图像生成 Prompt
"""
from typing import List, Dict, Optional
from dataclasses import dataclass, asdict
import json
import os
@dataclass
class ImagePrompt:
"""图像生成 Prompt"""
id: str
type: str
prompt: str
negative_prompt: str
position: str
context: str
class PromptGenerator:
"""Prompt 生成器"""
# 背景色定义
BACKGROUND_COLOR = "#F5F0E6" # 高端米黄色
# 默认风格基底（纯手绘线条风）
DEFAULT_STYLE_BASE = """
Pure hand-drawn line art illustration, minimalist sketch style,
simple elegant black ink strokes on warm cream background (#{bg_color}),
single-weight fine lines, loose artistic hand-drawn quality,
whitespace as design element, editorial illustration aesthetic,
subtle imperfections that feel human and organic,
16:9 aspect ratio, centered composition with breathing room
""".strip().format(bg_color="F5F0E6")
# 默认负面提示词
DEFAULT_NEGATIVE = """
digital art, 3D render, photorealistic, gradients, shadows,
complex shading, multiple colors, busy composition,
cluttered elements, heavy line weight, perfect symmetry,
corporate clip art, stock photo style, watermark
""".strip()
# 类型特定的提示词模板
# 设计原则：先理解主旨 → 抽象出隐喻/象征元素 → 用极简线条表现
TYPE_TEMPLATES = {
'hero': {
'template': """
{style_base},
[Core visual metaphor]: {metaphor_description},
[Composition]: The main symbolic element is placed at the center,
secondary elements orbit or connect to it with delicate dotted lines,
generous negative space on all sides creating visual breathing room,
[Line quality]: Confident single-stroke outlines, hand-drawn wobble,
minimal interior details, let the silhouette tell the story,
[Optional text]: One short phrase (2-4 words max) in elegant sans-serif,
placed discretely at bottom or corner, acting as a gentle anchor
""",
'negative_extra': "heavy shading, complex backgrounds, multiple focal points"
},
'concept': {
'template': """
{style_base},
[Core visual metaphor]: {metaphor_description},
[Symbolic elements]: Distill the concept into 1-2 iconic objects,
use universally understood symbols (lightbulb=idea, key=access, bridge=connection),
show relationship between elements through spatial arrangement,
[Composition]: Single focal point with supporting elements,
elements connected by thin hand-drawn lines or arrows,
70% whitespace to 30% illustration ratio,
[Line quality]: Clean but imperfect strokes, sketch-like quality,
varying line confidence to show emphasis
""",
'negative_extra': "literal representation, complex diagrams, too many symbols"
},
'analogy': {
'template': """
{style_base},
[Core visual metaphor]: {metaphor_description},
[Scene construction]: Create a mini-narrative with 2-3 symbolic objects,
show the relationship/interaction between the analogy elements,
use familiar everyday objects to represent abstract concepts,
[Spatial relationship]: Main subject on one side,
the thing it connects to on the other,
visual bridge or connection line between them,
[Storytelling details]: One or two tiny details that add charm,
(a small arrow, a dotted path, a gentle curve connecting elements),
[Optional label]: Single word or short phrase near each key element
""",
'negative_extra': "realistic scenes, complex environments, too many characters"
},
'diagram': {
'template': """
{style_base},
[Core visual metaphor]: {metaphor_description},
[Flow visualization]: Show {process_count} steps as connected nodes,
each node is a simple iconic symbol (circle, square, or small sketch),
hand-drawn arrows or dotted lines showing direction of flow,
[Layout]: Linear left-to-right or top-to-bottom arrangement,
equal spacing between steps, clear visual hierarchy,
numbered or lettered steps in small handwritten style,
[Connection style]: Curved hand-drawn arrows, not rigid straight lines,
small decorative elements at connection points (dots, small circles),
[Labels]: Minimal text labels (1-2 words) below or beside each node
""",
'negative_extra': "complex flowcharts, too many branches, corporate diagram style"
},
'comparison': {
'template': """
{style_base},
[Core visual metaphor]: {metaphor_description},
[Dual composition]: Split canvas into left and right zones,
each side contains one symbolic representation,
vertical dotted line or gentle divide in the middle,
[Contrast elements]: Left side shows concept A as simple icon,
right side shows concept B as contrasting icon,
visual weight and size indicate relative importance,
[Connecting element]: A small bridge, arrow, or "vs" symbol in center,
subtle visual cues showing the key difference,
[Optional labels]: Single word beneath each side identifying the concept
""",
'negative_extra': "complex charts, data visualizations, too many comparison points"
},
'mood': {
'template': """
{style_base},
[Core visual metaphor]: {metaphor_description},
[Atmosphere]: Single poetic visual element floating in space,
extremely minimal - just one or two strokes suggesting the theme,
80% negative space creating contemplative feeling,
[Emotional tone]: Gentle, reflective, like a pause for breath,
the illustration whispers rather than shouts,
[Element placement]: Off-center placement creating dynamic tension,
small scale relative to canvas size,
[Optional poetry]: One evocative word or very short phrase,
placed with intention, adding to the meditative quality
""",
'negative_extra': "busy scenes, strong emotions, action or movement"
}
}
def __init__(self, style_base: str = None, negative_base: str = None):
"""
初始化生成器
Args:
style_base: 自定义风格基底
negative_base: 自定义负面提示词
"""
self.style_base = style_base or self.DEFAULT_STYLE_BASE
self.negative_base = negative_base or self.DEFAULT_NEGATIVE
def _clean_text(self, text: str) -> str:
"""清理文本，移除多余空白"""
return ' '.join(text.split())
def _build_metaphor_description(self, slot: Dict, article_title: str = "") -> str:
"""
构建隐喻描述
优先使用 slot 中的 metaphor_description 字段（由 AI 生成的详细隐喻）
如果没有，则使用 context 和 anchor_text 作为后备
Args:
slot: 插图槽位信息，应包含：
- metaphor_description: AI 生成的详细隐喻描述（推荐）
例如："A friendly robot head in the center, with multiple glowing skill
modules floating around its brain area, connected by thin dotted
lines suggesting neural pathways"
- context: 段落上下文（后备）
- anchor_text: 锚定文本（后备）
article_title: 文章标题
Returns:
隐喻描述字符串
"""
# 优先使用 AI 生成的详细隐喻描述
if slot.get('metaphor_description'):
return slot['metaphor_description']
# 后备方案：从 context 和 anchor_text 构建
context = slot.get('context', '')
anchor = slot.get('anchor_text', '')
if anchor:
return f"Visual representation of: {anchor}"
elif context:
# 截取前100字符作为描述
return f"Visual representation of: {context[:100]}"
elif article_title:
return f"Visual representation of: {article_title}"
else:
return "Abstract conceptual illustration"
def _count_process_steps(self, context: str) -> int:
"""
从上下文中提取流程步骤数量
Args:
context: 上下文文本
Returns:
步骤数量，默认为 3
"""
import re
# 尝试匹配中文序号
chinese_patterns = [
r'第[一二三四五六七八九十]步',
r'[一二三四五六七八九十]、',
r'[①②③④⑤⑥⑦⑧⑨⑩]',
]
# 尝试匹配数字序号
digit_patterns = [
r'\d+\.',
r'\d+、',
r'\d+\)',
]
max_count = 0
for pattern in chinese_patterns + digit_patterns:
matches = re.findall(pattern, context)
if len(matches) > max_count:
max_count = len(matches)
return max_count if max_count >= 2 else 3 # 默认 3 步
def generate_prompt(
self,
slot: Dict,
article_title: str = "",
additional_context: str = ""
) -> ImagePrompt:
"""
为单个槽位生成 Prompt
slot 数据结构应包含：
{
'id': 'img_001',
'type': 'hero' | 'concept' | 'analogy' | 'diagram' | 'comparison' | 'mood',
'position': 'before_paragraph_1',
'context': '段落原文或摘要',
'anchor_text': '触发词或关键句',
'metaphor_description': '【推荐】AI 生成的详细隐喻描述，例如：
"A friendly robot head in the center, with multiple glowing skill
modules floating around its brain area, connected by thin dotted
lines suggesting neural pathways. The robot has simple dot eyes
showing curiosity."'
}
Args:
slot: 插图槽位信息
article_title: 文章标题
additional_context: 额外上下文
Returns:
生成的 Prompt
"""
slot_type = slot.get('type', 'concept')
template_config = self.TYPE_TEMPLATES.get(slot_type, self.TYPE_TEMPLATES['concept'])
context = slot.get('context', '')
# 构建隐喻描述
metaphor_description = self._build_metaphor_description(slot, article_title)
# 获取流程步骤数量（用于 diagram 类型）
process_count = self._count_process_steps(context)
# 根据类型填充模板
if slot_type == 'diagram':
filled_template = template_config['template'].format(
style_base=self.style_base,
metaphor_description=metaphor_description,
process_count=process_count
)
else:
filled_template = template_config['template'].format(
style_base=self.style_base,
metaphor_description=metaphor_description
)
# 组合负面提示词
negative = f"{self.negative_base}, {template_config.get('negative_extra', '')}"
return ImagePrompt(
id=slot.get('id', 'img_001'),
type=slot_type,
prompt=self._clean_text(filled_template),
negative_prompt=self._clean_text(negative),
position=slot.get('position', ''),
context=context
)
def generate_all(
self,
layout: Dict,
article_title: str = ""
) -> List[ImagePrompt]:
"""
为所有槽位生成 Prompt
Args:
layout: 编排方案
article_title: 文章标题
Returns:
Prompt 列表
"""
prompts = []
slots = layout.get('layout', [])
for slot in slots:
prompt = self.generate_prompt(slot, article_title)
prompts.append(prompt)
return prompts
def to_json(self, prompts: List[ImagePrompt]) -> str:
"""
转换为 JSON 格式
Args:
prompts: Prompt 列表
Returns:
JSON 字符串
"""
data = [asdict(p) for p in prompts]
return json.dumps(data, ensure_ascii=False, indent=2)
def save_to_file(self, prompts: List[ImagePrompt], filepath: str):
"""
保存到文件
Args:
prompts: Prompt 列表
filepath: 文件路径
"""
with open(filepath, 'w', encoding='utf-8') as f:
f.write(self.to_json(prompts))
# 便捷函数
def generate_image_prompts(
layout: Dict,
article_title: str = "",
style_base: str = None
) -> List[Dict]:
"""
生成图像 Prompt（便捷函数）
Args:
layout: 编排方案
article_title: 文章标题
style_base: 自定义风格
Returns:
Prompt 字典列表
"""
generator = PromptGenerator(style_base=style_base)
prompts = generator.generate_all(layout, article_title)
return [asdict(p) for p in prompts]
if __name__ == "__main__":
# 测试 - 展示新的 metaphor_description 字段用法
#
# 核心理念：
# 1. 先理解段落主旨
# 2. 抽象出隐喻和象征元素
# 3. 用极简手绘线条表现
#
# metaphor_description 字段应由 AI 在分析段落后生成，
# 描述具体的视觉元素和它们之间的空间关系
test_layout = {
'total_illustrations': 3,
'layout': [
{
'id': 'img_001',
'position': 'before_paragraph_1',
'type': 'hero',
'context': 'Claude Skills 就是给 AI 的外挂技能包',
'anchor_text': '',
# 【重要】metaphor_description 是核心字段
# 详细描述画面元素和空间关系
'metaphor_description': '''
A friendly robot head drawn with simple curved lines,
positioned at center. Around its brain area, 4-5 small
skill module icons (shaped like puzzle pieces or app icons)
float in an orbital pattern. Thin dotted lines connect
each module to the robot's head, suggesting neural pathways.
The robot has minimal dot eyes showing curiosity.
Small label "Skills" near one floating module.
'''
},
{
'id': 'img_002',
'position': 'after_paragraph_4',
'type': 'analogy',
'context': 'Token 就像乐高积木，AI 用这些积木来理解和组装语言',
'anchor_text': '就像乐高积木',
'metaphor_description': '''
Left side: A simple speech bubble or text block,
drawn with loose hand strokes.
Right side: The same shape broken into small LEGO-like
brick pieces, scattered but recognizable.
A curved arrow with hand-drawn wobble connects left to right,
showing the transformation.
Small handwritten label "Token" pointing to one brick.
'''
},
{
'id': 'img_003',
'position': 'after_paragraph_8',
'type': 'diagram',
'context': '''使用 Skills 的步骤：
第一步，找到合适的 Skill。
第二步，配置参数。
第三步，运行测试。
第四步，根据结果调整。''',
'anchor_text': '',
'metaphor_description': '''
4 simple circular nodes arranged horizontally,
each containing a minimal icon:
Node 1: magnifying glass (Find)
Node 2: slider controls (Configure)
Node 3: play button (Run)
Node 4: circular arrows (Adjust)
Hand-drawn curved arrows connect the nodes left to right.
Small handwritten numbers "1 2 3 4" above each node.
'''
}
]
}
prompts = generate_image_prompts(test_layout, "Claude Skills 入门指南")
print("=" * 60)
print("生成的 Prompt 示例（纯手绘线条风格）")
print("背景色: #F5F0E6 (高端米黄色)")
print("=" * 60)
print(json.dumps(prompts, ensure_ascii=False, indent=2))
research_analyzer.py
"""
Research 分析器
对采集到的多源内容进行深度分析，生成 Research Report
"""
import json
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass, field, asdict
from datetime import datetime
from enum import Enum
class ConsensusLevel(Enum):
"""共识度等级"""
HIGH = 5 # 3+ 篇文章提到
MEDIUM = 3 # 2 篇文章提到
LOW = 2 # 仅 1 篇文章提到
CONTROVERSIAL = 1 # 存在争议
@dataclass
class Insight:
"""洞察点"""
content: str
sources: List[int] # 来源文章索引
consensus_level: ConsensusLevel = ConsensusLevel.LOW
is_controversial: bool = False
conflicting_views: List[str] = field(default_factory=list)
@dataclass
class KnowledgeNode:
"""知识图谱节点"""
concept: str
definition: str = ""
components: List[str] = field(default_factory=list)
use_cases: List[str] = field(default_factory=list)
related_concepts: List[str] = field(default_factory=list)
common_misconceptions: List[str] = field(default_factory=list)
@dataclass
class SourceArticle:
"""源文章信息"""
index: int
title: str
source_type: str # 公众号 / 小红书 / Web
source_name: str
publish_time: str
url: str
citation_count: int = 0
cited_in: List[str] = field(default_factory=list) # 被哪些洞察引用
@dataclass
class ResearchReport:
"""Research 报告"""
topic: str
generated_at: str
# 信息源统计
source_stats: Dict[str, Dict[str, int]] = field(default_factory=dict)
# 核心定义
core_definition: str = ""
definition_consensus: ConsensusLevel = ConsensusLevel.LOW
# 关键洞察
insights: List[Insight] = field(default_factory=list)
# 知识图谱
knowledge_graph: Optional[KnowledgeNode] = None
# 争议点
controversies: List[Dict[str, str]] = field(default_factory=list)
# 源文章列表
sources: List[SourceArticle] = field(default_factory=list)
def to_dict(self) -> Dict:
"""转换为字典"""
data = asdict(self)
# 处理枚举
data['definition_consensus'] = self.definition_consensus.name
for insight in data['insights']:
insight['consensus_level'] = ConsensusLevel(insight['consensus_level']).name
return data
def to_markdown(self) -> str:
"""转换为 Markdown 格式"""
lines = []
# 标题
lines.append(f"# 📚 Deep Research Report")
lines.append(f"**主题**: {self.topic}")
lines.append(f"**生成时间**: {self.generated_at}")
lines.append("")
# 信息源概览
lines.append("## 一、信息源概览")
lines.append("")
lines.append("| 采集源 | 采集数 | 有效数 | 时效内 |")
lines.append("|--------|--------|--------|--------|")
total = {'collected': 0, 'valid': 0, 'in_time': 0}
for source, stats in self.source_stats.items():
lines.append(f"| {source} | {stats.get('collected', 0)} | {stats.get('valid', 0)} | {stats.get('in_time', 0)} |")
total['collected'] += stats.get('collected', 0)
total['valid'] += stats.get('valid', 0)
total['in_time'] += stats.get('in_time', 0)
lines.append(f"| **合计** | {total['collected']} | {total['valid']} | {total['in_time']} |")
lines.append("")
# 核心定义
lines.append("## 二、核心定义")
lines.append("")
consensus_stars = "★" * self.definition_consensus.value + "☆" * (5 - self.definition_consensus.value)
lines.append(f"> {self.core_definition}")
lines.append(f">")
lines.append(f"> **共识度**: {consensus_stars}")
lines.append("")
# 关键洞察
lines.append("## 三、关键洞察")
lines.append("")
for i, insight in enumerate(self.insights, 1):
consensus_stars = "★" * insight.consensus_level.value + "☆" * (5 - insight.consensus_level.value)
source_refs = ", ".join([f"[{s}]" for s in insight.sources])
lines.append(f"### {i}. {insight.content}")
lines.append(f"- **来源**: {source_refs}")
lines.append(f"- **共识度**: {consensus_stars}")
if insight.is_controversial:
lines.append(f"- **⚠️ 存在争议**")
for view in insight.conflicting_views:
lines.append(f" - {view}")
lines.append("")
# 知识图谱
if self.knowledge_graph:
lines.append("## 四、知识图谱")
lines.append("")
kg = self.knowledge_graph
lines.append(f"```")
lines.append(f"{kg.concept}")
if kg.definition:
lines.append(f"├── 定义: {kg.definition}")
if kg.components:
lines.append(f"├── 组成: {', '.join(kg.components)}")
if kg.use_cases:
lines.append(f"├── 用途: {', '.join(kg.use_cases)}")
if kg.related_concepts:
lines.append(f"├── 相关概念: {', '.join(kg.related_concepts)}")
if kg.common_misconceptions:
lines.append(f"└── 常见误解: {', '.join(kg.common_misconceptions)}")
lines.append(f"```")
lines.append("")
# 争议点
if self.controversies:
lines.append("## 五、争议/不确定点")
lines.append("")
for c in self.controversies:
lines.append(f"- **{c.get('question', '')}**")
lines.append(f" {c.get('detail', '')}")
lines.append("")
# 源文章列表
lines.append("## 六、源文章列表")
lines.append("")
for source in self.sources:
cited_info = f"(引用 {source.citation_count} 次: {', '.join(source.cited_in)})" if source.citation_count > 0 else ""
lines.append(f"**[{source.index}]** {source.title}")
lines.append(f"- 来源: {source.source_type}·{source.source_name}")
lines.append(f"- 日期: {source.publish_time}")
lines.append(f"- 链接: {source.url}")
if cited_info:
lines.append(f"- {cited_info}")
lines.append("")
return "\n".join(lines)
class ResearchAnalyzer:
"""Research 分析器"""
def __init__(self):
self.articles: List[Dict] = []
self.report: Optional[ResearchReport] = None
def add_articles(self, articles: List[Dict], source_type: str):
"""
添加文章到分析池
Args:
articles: 文章列表
source_type: 来源类型（公众号/小红书/Web）
"""
for article in articles:
article['_source_type'] = source_type
self.articles.append(article)
def analyze(self, topic: str) -> ResearchReport:
"""
执行分析，生成报告
Args:
topic: 研究主题
Returns:
Research 报告
Note:
这个方法提供了分析框架。实际的内容分析（提取洞察、
构建知识图谱等）需要 Claude 来完成。这里主要处理
数据结构和格式化。
"""
self.report = ResearchReport(
topic=topic,
generated_at=datetime.now().strftime("%Y-%m-%d %H:%M")
)
# 统计信息源
self._calculate_source_stats()
# 构建源文章列表
self._build_source_list()
return self.report
def _calculate_source_stats(self):
"""计算信息源统计"""
stats = {}
for article in self.articles:
source_type = article.get('_source_type', '未知')
if source_type not in stats:
stats[source_type] = {
'collected': 0,
'valid': 0,
'in_time': 0
}
stats[source_type]['collected'] += 1
# 有内容则视为有效
if article.get('content') or article.get('content_full'):
stats[source_type]['valid'] += 1
# 简化处理：有内容且有发布时间则视为时效内
if article.get('publish_time'):
stats[source_type]['in_time'] += 1
self.report.source_stats = stats
def _build_source_list(self):
"""构建源文章列表"""
sources = []
for i, article in enumerate(self.articles, 1):
source = SourceArticle(
index=i,
title=article.get('title', ''),
source_type=article.get('_source_type', ''),
source_name=article.get('source', article.get('author', '')),
publish_time=article.get('publish_time', ''),
url=article.get('url', ''),
citation_count=0,
cited_in=[]
)
sources.append(source)
self.report.sources = sources
def set_core_definition(self, definition: str, consensus: ConsensusLevel):
"""
设置核心定义
Args:
definition: 定义内容
consensus: 共识度
"""
if self.report:
self.report.core_definition = definition
self.report.definition_consensus = consensus
def add_insight(
self,
content: str,
sources: List[int],
is_controversial: bool = False,
conflicting_views: List[str] = None
):
"""
添加洞察
Args:
content: 洞察内容
sources: 来源文章索引列表
is_controversial: 是否有争议
conflicting_views: 冲突观点列表
"""
if not self.report:
return
# 根据来源数量确定共识度
if len(sources) >= 3:
consensus = ConsensusLevel.HIGH
elif len(sources) == 2:
consensus = ConsensusLevel.MEDIUM
else:
consensus = ConsensusLevel.LOW
if is_controversial:
consensus = ConsensusLevel.CONTROVERSIAL
insight = Insight(
content=content,
sources=sources,
consensus_level=consensus,
is_controversial=is_controversial,
conflicting_views=conflicting_views or []
)
self.report.insights.append(insight)
# 更新源文章的引用信息
for idx in sources:
for source in self.report.sources:
if source.index == idx:
source.citation_count += 1
source.cited_in.append(f"洞察{len(self.report.insights)}")
def set_knowledge_graph(
self,
concept: str,
definition: str = "",
components: List[str] = None,
use_cases: List[str] = None,
related_concepts: List[str] = None,
common_misconceptions: List[str] = None
):
"""
设置知识图谱
Args:
concept: 核心概念
definition: 定义
components: 组成部分
use_cases: 应用场景
related_concepts: 相关概念
common_misconceptions: 常见误解
"""
if self.report:
self.report.knowledge_graph = KnowledgeNode(
concept=concept,
definition=definition,
components=components or [],
use_cases=use_cases or [],
related_concepts=related_concepts or [],
common_misconceptions=common_misconceptions or []
)
def add_controversy(self, question: str, detail: str):
"""
添加争议点
Args:
question: 争议问题
detail: 详细说明
"""
if self.report:
self.report.controversies.append({
'question': question,
'detail': detail
})
def get_articles_for_analysis(self) -> List[Dict]:
"""
获取用于分析的文章内容
Returns:
文章列表（含完整内容）
"""
return [
{
'index': i,
'title': a.get('title', ''),
'source': a.get('_source_type', ''),
'content': a.get('content_full', a.get('content', ''))
}
for i, a in enumerate(self.articles, 1)
]
# 便捷函数
def create_research_report(
topic: str,
weixin_articles: List[Dict] = None,
redbook_notes: List[Dict] = None,
web_results: List[Dict] = None
) -> Tuple[ResearchAnalyzer, str]:
"""
创建 Research 报告（便捷函数）
Args:
topic: 研究主题
weixin_articles: 公众号文章
redbook_notes: 小红书笔记
web_results: Web 搜索结果
Returns:
(分析器实例, 文章内容摘要)
Note:
返回分析器实例，以便后续调用 Claude 进行内容分析后
补充洞察、知识图谱等信息
"""
analyzer = ResearchAnalyzer()
if weixin_articles:
analyzer.add_articles(weixin_articles, '公众号')
if redbook_notes:
analyzer.add_articles(redbook_notes, '小红书')
if web_results:
analyzer.add_articles(web_results, 'Web')
# 初始化报告框架
analyzer.analyze(topic)
# 获取文章摘要供 Claude 分析
articles_summary = analyzer.get_articles_for_analysis()
return analyzer, json.dumps(articles_summary, ensure_ascii=False, indent=2)
if __name__ == "__main__":
# 测试
analyzer = ResearchAnalyzer()
# 模拟添加文章
analyzer.add_articles([
{'title': '测试文章1', 'content': '内容1', 'url': 'http://example.com/1', 'publish_time': '2026-01-10'},
{'title': '测试文章2', 'content': '内容2', 'url': 'http://example.com/2', 'publish_time': '2026-01-08'},
], '公众号')
analyzer.add_articles([
{'title': '测试笔记1', 'content': '内容3', 'url': 'http://xiaohongshu.com/1', 'author': '作者1'},
], '小红书')
# 生成报告框架
report = analyzer.analyze("测试主题")
# 模拟 Claude 分析后补充信息
analyzer.set_core_definition(
"这是核心定义的描述...",
ConsensusLevel.HIGH
)
analyzer.add_insight(
"这是第一个洞察点",
sources=[1, 2],
is_controversial=False
)
analyzer.set_knowledge_graph(
concept="测试主题",
definition="定义",
components=["组件1", "组件2"],
use_cases=["用途1", "用途2"],
related_concepts=["相关1", "相关2"],
common_misconceptions=["误解1"]
)
# 输出 Markdown
print(report.to_markdown())
weixin_collector.py
"""
微信公众号文章采集工具
基于 weixin_search_mcp 项目
"""
import sys
import os
from typing import List, Dict, Optional
from datetime import datetime, timedelta
from dataclasses import dataclass
# 添加 weixin_search_mcp 到路径
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))))
sys.path.insert(0, os.path.join(PROJECT_ROOT, 'weixin_search_mcp'))
from weixin_search_mcp.tools.weixin_search import sogou_weixin_search, get_article_content
@dataclass
class WeixinArticle:
"""微信文章数据结构"""
title: str
url: str
source: str
publish_time: str
content: str
relevance_score: float = 0.0
class WeixinCollector:
"""微信公众号文章采集器"""
# 可信公众号列表
TRUSTED_SOURCES = [
"机器之心", "量子位", "新智元", "AI前线",
"硅星人", "极客公园", "少数派", "36氪",
"虎嗅", "爱范儿", "雷科技", "差评"
]
def __init__(self, months_limit: int = 3):
"""
初始化采集器
Args:
months_limit: 时效限制（月）
"""
self.months_limit = months_limit
self.cutoff_date = datetime.now() - timedelta(days=months_limit * 30)
def search(self, keyword: str, limit: int = 5) -> List[Dict]:
"""
搜索公众号文章
Args:
keyword: 搜索关键词
limit: 返回数量限制
Returns:
搜索结果列表
"""
try:
results = sogou_weixin_search(keyword)
return results[:limit] if results else []
except Exception as e:
print(f"搜索失败: {e}")
return []
def get_content(self, url: str, referer: str = None) -> str:
"""
获取文章内容
Args:
url: 文章真实链接
referer: 来源链接
Returns:
文章正文内容
"""
try:
return get_article_content(url, referer)
except Exception as e:
print(f"获取内容失败: {e}")
return ""
def _parse_publish_time(self, time_str: str) -> Optional[datetime]:
"""
解析发布时间
Args:
time_str: 时间字符串（可能包含 JS 代码）
Returns:
datetime 对象
"""
import re
# 尝试提取时间戳
match = re.search(r"timeConvert\('(\d+)'\)", time_str)
if match:
try:
timestamp = int(match.group(1))
return datetime.fromtimestamp(timestamp)
except:
pass
return None
def _calculate_relevance(self, article: Dict, core_keywords: List[str], refined_keywords: List[str]) -> float:
"""
计算文章相关性得分
Args:
article: 文章信息
core_keywords: 核心关键词
refined_keywords: 细化关键词
Returns:
相关性得分
"""
score = 0.0
title = article.get('title', '').lower()
# 标题命中核心词 +3
for kw in core_keywords:
if kw.lower() in title:
score += 3
# 标题命中细化词 +2
for kw in refined_keywords:
if kw.lower() in title:
score += 2
# 来源为可信源 +2
# 注：搜狗搜索结果中可能没有公众号名称，需要从内容中提取
return score
def collect(
self,
core_keywords: List[str],
refined_keywords: List[str] = None,
core_limit: int = 5,
refined_limit: int = 3
) -> List[WeixinArticle]:
"""
采集文章
Args:
core_keywords: 核心关键词列表
refined_keywords: 细化关键词列表
core_limit: 核心关键词每个取 top N
refined_limit: 细化关键词每个取 top N
Returns:
采集到的文章列表
"""
all_articles = []
seen_urls = set()
# 采集核心关键词结果
for kw in core_keywords:
results = self.search(kw, limit=core_limit)
for item in results:
real_url = item.get('real_url', '')
if real_url and real_url not in seen_urls:
seen_urls.add(real_url)
# 获取内容
content = self.get_content(real_url, item.get('link'))
# 计算相关性
score = self._calculate_relevance(
item,
core_keywords,
refined_keywords or []
)
article = WeixinArticle(
title=item.get('title', ''),
url=real_url,
source='公众号',
publish_time=item.get('publish_time', ''),
content=content,
relevance_score=score
)
all_articles.append(article)
# 采集细化关键词结果
if refined_keywords:
for kw in refined_keywords:
results = self.search(kw, limit=refined_limit)
for item in results:
real_url = item.get('real_url', '')
if real_url and real_url not in seen_urls:
seen_urls.add(real_url)
content = self.get_content(real_url, item.get('link'))
score = self._calculate_relevance(
item,
core_keywords,
refined_keywords
)
article = WeixinArticle(
title=item.get('title', ''),
url=real_url,
source='公众号',
publish_time=item.get('publish_time', ''),
content=content,
relevance_score=score
)
all_articles.append(article)
# 按相关性排序
all_articles.sort(key=lambda x: x.relevance_score, reverse=True)
return all_articles
def to_dict_list(self, articles: List[WeixinArticle]) -> List[Dict]:
"""
转换为字典列表
Args:
articles: 文章列表
Returns:
字典列表
"""
return [
{
'title': a.title,
'url': a.url,
'source': a.source,
'publish_time': a.publish_time,
'content': a.content[:500] + '...' if len(a.content) > 500 else a.content,
'content_full': a.content,
'relevance_score': a.relevance_score
}
for a in articles
]
# 便捷函数
def collect_weixin_articles(
core_keywords: List[str],
refined_keywords: List[str] = None,
core_limit: int = 5,
refined_limit: int = 3
) -> List[Dict]:
"""
采集微信公众号文章（便捷函数）
Args:
core_keywords: 核心关键词
refined_keywords: 细化关键词
core_limit: 核心词每个取 top N
refined_limit: 细化词每个取 top N
Returns:
文章列表
"""
collector = WeixinCollector()
articles = collector.collect(
core_keywords=core_keywords,
refined_keywords=refined_keywords,
core_limit=core_limit,
refined_limit=refined_limit
)
return collector.to_dict_list(articles)
if __name__ == "__main__":
# 测试
articles = collect_weixin_articles(
core_keywords=["Claude Skills"],
refined_keywords=["Claude Skills 教程", "Claude Skills 入门"],
core_limit=3,
refined_limit=2
)
print(f"采集到 {len(articles)} 篇文章:\n")
for i, a in enumerate(articles, 1):
print(f"{i}. {a['title']}")
print(f" 相关性: {a['relevance_score']}")
print(f" 链接: {a['url'][:60]}...")
print()
redbook_collector.py
"""
小红书笔记采集工具
基于 redbook_mcp 项目（Playwright 浏览器自动化）
"""
import asyncio
import os
import sys
from typing import List, Dict, Optional
from datetime import datetime, timedelta
from dataclasses import dataclass
# Playwright 相关
from playwright.async_api import async_playwright
# 项目根目录
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))))
@dataclass
class RedbookNote:
"""小红书笔记数据结构"""
title: str
url: str
author: str
content: str
publish_time: str
relevance_score: float = 0.0
class RedbookCollector:
"""小红书笔记采集器"""
BROWSER_DATA_DIR = os.path.join(PROJECT_ROOT, 'redbook_mcp', 'browser_data')
def __init__(self, months_limit: int = 3):
"""
初始化采集器
Args:
months_limit: 时效限制（月）
"""
self.months_limit = months_limit
self.browser_context = None
self.page = None
self.is_logged_in = False
# 确保目录存在
os.makedirs(self.BROWSER_DATA_DIR, exist_ok=True)
async def _ensure_browser(self) -> bool:
"""
确保浏览器已启动并登录
Returns:
是否已登录
"""
if self.browser_context is None:
playwright_instance = await async_playwright().start()
self.browser_context = await playwright_instance.chromium.launch_persistent_context(
user_data_dir=self.BROWSER_DATA_DIR,
headless=False,
viewport={"width": 1280, "height": 800},
timeout=60000
)
if self.browser_context.pages:
self.page = self.browser_context.pages[0]
else:
self.page = await self.browser_context.new_page()
self.page.set_default_timeout(60000)
# 检查登录状态
if not self.is_logged_in:
await self.page.goto("https://www.xiaohongshu.com", timeout=60000)
await asyncio.sleep(3)
login_elements = await self.page.query_selector_all('text="登录"')
if login_elements:
print("需要登录小红书，请在浏览器窗口扫码登录...")
print("等待登录完成（最多180秒）...")
max_wait = 180
waited = 0
while waited < max_wait:
still_login = await self.page.query_selector_all('text="登录"')
if not still_login:
print("登录成功！")
self.is_logged_in = True
break
await asyncio.sleep(5)
waited += 5
if not self.is_logged_in:
print("登录超时")
return False
else:
self.is_logged_in = True
return True
async def search(self, keyword: str, limit: int = 5) -> List[Dict]:
"""
搜索小红书笔记
Args:
keyword: 搜索关键词
limit: 返回数量限制
Returns:
搜索结果列表
"""
if not await self._ensure_browser():
return []
search_url = f"https://www.xiaohongshu.com/search_result?keyword={keyword}"
await self.page.goto(search_url, timeout=60000)
await asyncio.sleep(5)
# 获取帖子卡片
post_cards = await self.page.query_selector_all('section.note-item')
if not post_cards:
# 尝试备用选择器
post_cards = await self.page.query_selector_all('div[data-v-a264b01a]')
results = []
for card in post_cards[:limit]:
try:
# 获取标题
title_el = await card.query_selector('a.title span')
title = await title_el.text_content() if title_el else "未知标题"
# 获取链接
link_el = await card.query_selector('a[href*="/search_result/"]')
href = await link_el.get_attribute('href') if link_el else None
url = f"https://www.xiaohongshu.com{href}" if href and href.startswith('/') else href
# 获取作者
author_el = await card.query_selector('span.name')
author = await author_el.text_content() if author_el else "未知作者"
results.append({
'title': title.strip() if title else '',
'url': url,
'author': author.strip() if author else ''
})
except Exception as e:
print(f"解析笔记失败: {e}")
return results
async def get_note_content(self, url: str) -> Dict:
"""
获取笔记详细内容
Args:
url: 笔记链接
Returns:
笔记详情
"""
if not await self._ensure_browser():
return {}
try:
await self.page.goto(url, timeout=60000)
await asyncio.sleep(3)
# 获取标题
title_el = await self.page.query_selector('div.title')
title = await title_el.text_content() if title_el else ""
# 获取正文
content_el = await self.page.query_selector('div.desc')
content = await content_el.text_content() if content_el else ""
# 获取作者
author_el = await self.page.query_selector('span.username')
author = await author_el.text_content() if author_el else ""
# 获取发布时间
time_el = await self.page.query_selector('span.date')
publish_time = await time_el.text_content() if time_el else ""
return {
'title': title.strip() if title else '',
'content': content.strip() if content else '',
'author': author.strip() if author else '',
'publish_time': publish_time.strip() if publish_time else '',
'url': url
}
except Exception as e:
print(f"获取笔记内容失败: {e}")
return {}
def _calculate_relevance(self, note: Dict, core_keywords: List[str], refined_keywords: List[str]) -> float:
"""
计算笔记相关性得分
Args:
note: 笔记信息
core_keywords: 核心关键词
refined_keywords: 细化关键词
Returns:
相关性得分
"""
score = 0.0
title = note.get('title', '').lower()
content = note.get('content', '').lower()
# 标题命中核心词 +3
for kw in core_keywords:
if kw.lower() in title:
score += 3
# 标题命中细化词 +2
for kw in refined_keywords:
if kw.lower() in title:
score += 2
# 正文命中核心词 +1
for kw in core_keywords:
if kw.lower() in content:
score += 1
return score
async def collect(
self,
core_keywords: List[str],
refined_keywords: List[str] = None,
core_limit: int = 5,
refined_limit: int = 3,
fetch_content: bool = True
) -> List[RedbookNote]:
"""
采集笔记
Args:
core_keywords: 核心关键词列表
refined_keywords: 细化关键词列表
core_limit: 核心关键词每个取 top N
refined_limit: 细化关键词每个取 top N
fetch_content: 是否获取详细内容
Returns:
采集到的笔记列表
"""
all_notes = []
seen_urls = set()
# 采集核心关键词结果
for kw in core_keywords:
results = await self.search(kw, limit=core_limit)
for item in results:
url = item.get('url', '')
if url and url not in seen_urls:
seen_urls.add(url)
# 获取详细内容
if fetch_content:
detail = await self.get_note_content(url)
content = detail.get('content', '')
publish_time = detail.get('publish_time', '')
else:
content = ''
publish_time = ''
# 计算相关性
score = self._calculate_relevance(
{'title': item.get('title', ''), 'content': content},
core_keywords,
refined_keywords or []
)
note = RedbookNote(
title=item.get('title', ''),
url=url,
author=item.get('author', ''),
content=content,
publish_time=publish_time,
relevance_score=score
)
all_notes.append(note)
# 采集细化关键词结果
if refined_keywords:
for kw in refined_keywords:
results = await self.search(kw, limit=refined_limit)
for item in results:
url = item.get('url', '')
if url and url not in seen_urls:
seen_urls.add(url)
if fetch_content:
detail = await self.get_note_content(url)
content = detail.get('content', '')
publish_time = detail.get('publish_time', '')
else:
content = ''
publish_time = ''
score = self._calculate_relevance(
{'title': item.get('title', ''), 'content': content},
core_keywords,
refined_keywords
)
note = RedbookNote(
title=item.get('title', ''),
url=url,
author=item.get('author', ''),
content=content,
publish_time=publish_time,
relevance_score=score
)
all_notes.append(note)
# 按相关性排序
all_notes.sort(key=lambda x: x.relevance_score, reverse=True)
return all_notes
async def close(self):
"""关闭浏览器"""
if self.browser_context:
await self.browser_context.close()
self.browser_context = None
self.page = None
self.is_logged_in = False
def to_dict_list(self, notes: List[RedbookNote]) -> List[Dict]:
"""
转换为字典列表
Args:
notes: 笔记列表
Returns:
字典列表
"""
return [
{
'title': n.title,
'url': n.url,
'author': n.author,
'source': '小红书',
'publish_time': n.publish_time,
'content': n.content[:500] + '...' if len(n.content) > 500 else n.content,
'content_full': n.content,
'relevance_score': n.relevance_score
}
for n in notes
]
# 便捷函数（同步包装）
def collect_redbook_notes(
core_keywords: List[str],
refined_keywords: List[str] = None,
core_limit: int = 5,
refined_limit: int = 3,
fetch_content: bool = True
) -> List[Dict]:
"""
采集小红书笔记（便捷函数，同步）
Args:
core_keywords: 核心关键词
refined_keywords: 细化关键词
core_limit: 核心词每个取 top N
refined_limit: 细化词每个取 top N
fetch_content: 是否获取详细内容
Returns:
笔记列表
"""
async def _collect():
collector = RedbookCollector()
try:
notes = await collector.collect(
core_keywords=core_keywords,
refined_keywords=refined_keywords,
core_limit=core_limit,
refined_limit=refined_limit,
fetch_content=fetch_content
)
return collector.to_dict_list(notes)
finally:
await collector.close()
return asyncio.run(_collect())
if __name__ == "__main__":
# 测试
notes = collect_redbook_notes(
core_keywords=["Claude AI"],
refined_keywords=["Claude 教程"],
core_limit=3,
refined_limit=2,
fetch_content=False # 快速测试不获取详情
)
print(f"采集到 {len(notes)} 篇笔记:\n")
for i, n in enumerate(notes, 1):
print(f"{i}. {n['title']}")
print(f" 作者: {n['author']}")
print(f" 相关性: {n['relevance_score']}")
print()
layout_engine.py
"""
图文编排引擎
决定在文章的什么位置放置什么类型的插图
"""
import re
from typing import List, Dict, Optional
from dataclasses import dataclass, field, asdict
from enum import Enum
class IllustrationType(Enum):
"""插图类型"""
HERO = "hero" # 主视觉，抽象概念具象化
CONCEPT = "concept" # 概念可视化
ANALOGY = "analogy" # 类比场景图
DIAGRAM = "diagram" # 流程/结构图
COMPARISON = "comparison" # 对比图
MOOD = "mood" # 氛围图/留白图
class LengthTier(Enum):
"""字数档位"""
COMPACT = "compact" # 精炼版 1200-2000
STANDARD = "standard" # 标准版 2000-4000
DEEP = "deep" # 深度版 4000-7000
LONG = "long" # 长文版 7000-10000
@dataclass
class IllustrationSlot:
"""插图位置槽"""
id: str
position: str # before_paragraph_N / after_paragraph_N
paragraph_index: int
type: IllustrationType
priority: int # 1-5, 5 最高
context: str # 相关上下文
anchor_text: str = "" # 锚定文本（如类比的具体句子）
is_required: bool = False # 是否必需
def to_dict(self) -> Dict:
return {
**asdict(self),
'type': self.type.value
}
@dataclass
class LayoutPlan:
"""编排方案"""
article_title: str
length_tier: LengthTier
word_count: int
total_paragraphs: int
# 推荐的插图数量范围
min_illustrations: int
max_illustrations: int
# 插图槽位
slots: List[IllustrationSlot] = field(default_factory=list)
# 最终选中的槽位
selected_slots: List[IllustrationSlot] = field(default_factory=list)
def to_dict(self) -> Dict:
return {
'article_title': self.article_title,
'length_tier': self.length_tier.value,
'word_count': self.word_count,
'total_paragraphs': self.total_paragraphs,
'min_illustrations': self.min_illustrations,
'max_illustrations': self.max_illustrations,
'slots': [s.to_dict() for s in self.slots],
'selected_slots': [s.to_dict() for s in self.selected_slots]
}
class LayoutEngine:
"""图文编排引擎"""
# 字数档位配置
TIER_CONFIG = {
LengthTier.COMPACT: {
'min_words': 1200,
'max_words': 2000,
'min_illustrations': 1,
'max_illustrations': 2,
'required': [IllustrationType.HERO],
'optional': [IllustrationType.CONCEPT, IllustrationType.ANALOGY]
},
LengthTier.STANDARD: {
'min_words': 2000,
'max_words': 4000,
'min_illustrations': 2,
'max_illustrations': 3,
'required': [IllustrationType.HERO, IllustrationType.CONCEPT],
'optional': [IllustrationType.ANALOGY, IllustrationType.DIAGRAM]
},
LengthTier.DEEP: {
'min_words': 4000,
'max_words': 7000,
'min_illustrations': 3,
'max_illustrations': 5,
'required': [IllustrationType.HERO, IllustrationType.CONCEPT, IllustrationType.ANALOGY],
'optional': [IllustrationType.DIAGRAM, IllustrationType.COMPARISON]
},
LengthTier.LONG: {
'min_words': 7000,
'max_words': 10000,
'min_illustrations': 4,
'max_illustrations': 6,
'required': [IllustrationType.HERO, IllustrationType.CONCEPT, IllustrationType.ANALOGY, IllustrationType.DIAGRAM],
'optional': [IllustrationType.COMPARISON, IllustrationType.MOOD]
}
}
# 触发条件关键词
ANALOGY_TRIGGERS = [
"就像", "好比", "类似于", "想象一下", "打个比方",
"如同", "仿佛", "相当于", "可以理解为"
]
DIAGRAM_TRIGGERS = [
"首先", "其次", "然后", "最后", "第一步", "第二步",
"步骤", "流程", "分为", "包括以下"
]
COMPARISON_TRIGGERS = [
"相比", "对比", "区别", "不同", "vs", "VS",
"优势", "劣势", "差异"
]
def __init__(self):
self.plan: Optional[LayoutPlan] = None
def _detect_tier(self, word_count: int) -> LengthTier:
"""
根据字数检测档位
Args:
word_count: 字数
Returns:
档位
"""
if word_count < 2000:
return LengthTier.COMPACT
elif word_count < 4000:
return LengthTier.STANDARD
elif word_count < 7000:
return LengthTier.DEEP
else:
return LengthTier.LONG
def _split_paragraphs(self, content: str) -> List[str]:
"""
分割段落
Args:
content: 文章内容
Returns:
段落列表
"""
# 按换行分割，过滤空段落
paragraphs = [p.strip() for p in content.split('\n') if p.strip()]
return paragraphs
def _find_analogy_positions(self, paragraphs: List[str]) -> List[Dict]:
"""
找到类比段落位置
Args:
paragraphs: 段落列表
Returns:
类比位置信息列表
"""
positions = []
for i, para in enumerate(paragraphs):
for trigger in self.ANALOGY_TRIGGERS:
if trigger in para:
positions.append({
'index': i,
'trigger': trigger,
'text': para[:100]
})
break
return positions
def _find_diagram_positions(self, paragraphs: List[str]) -> List[Dict]:
"""
找到流程/步骤段落位置
Args:
paragraphs: 段落列表
Returns:
流程位置信息列表
"""
positions = []
# 检查连续的步骤
step_count = 0
step_start = -1
for i, para in enumerate(paragraphs):
has_trigger = any(t in para for t in self.DIAGRAM_TRIGGERS)
if has_trigger:
if step_start == -1:
step_start = i
step_count += 1
else:
if step_count >= 3: # 至少3步才触发
positions.append({
'index': step_start,
'step_count': step_count,
'text': paragraphs[step_start][:100]
})
step_count = 0
step_start = -1
# 处理结尾的步骤
if step_count >= 3:
positions.append({
'index': step_start,
'step_count': step_count,
'text': paragraphs[step_start][:100]
})
return positions
def _find_comparison_positions(self, paragraphs: List[str]) -> List[Dict]:
"""
找到对比段落位置
Args:
paragraphs: 段落列表
Returns:
对比位置信息列表
"""
positions = []
for i, para in enumerate(paragraphs):
for trigger in self.COMPARISON_TRIGGERS:
if trigger in para:
positions.append({
'index': i,
'trigger': trigger,
'text': para[:100]
})
break
return positions
def _find_mood_positions(self, paragraphs: List[str]) -> List[Dict]:
"""
找到适合放氛围图的位置（长段落后）
Args:
paragraphs: 段落列表
Returns:
氛围图位置信息列表
"""
positions = []
cumulative_length = 0
for i, para in enumerate(paragraphs):
cumulative_length += len(para)
# 每累积 1500 字左右考虑一个节奏点
if cumulative_length > 1500:
positions.append({
'index': i,
'cumulative_length': cumulative_length,
'text': '节奏调节点'
})
cumulative_length = 0
return positions
def analyze(self, content: str, title: str = "") -> LayoutPlan:
"""
分析文章并生成编排方案
Args:
content: 文章内容
title: 文章标题
Returns:
编排方案
"""
word_count = len(content)
tier = self._detect_tier(word_count)
config = self.TIER_CONFIG[tier]
paragraphs = self._split_paragraphs(content)
self.plan = LayoutPlan(
article_title=title,
length_tier=tier,
word_count=word_count,
total_paragraphs=len(paragraphs),
min_illustrations=config['min_illustrations'],
max_illustrations=config['max_illustrations']
)
slots = []
slot_id = 0
# 1. 头图（必需）
slot_id += 1
slots.append(IllustrationSlot(
id=f"img_{slot_id:03d}",
position="before_paragraph_1",
paragraph_index=0,
type=IllustrationType.HERO,
priority=5,
context=title or paragraphs[0][:50] if paragraphs else "",
is_required=IllustrationType.HERO in config['required']
))
# 2. 概念图位置（通常在第一个主要概念解释后）
if len(paragraphs) > 2:
slot_id += 1
slots.append(IllustrationSlot(
id=f"img_{slot_id:03d}",
position="after_paragraph_2",
paragraph_index=2,
type=IllustrationType.CONCEPT,
priority=4,
context=paragraphs[1][:100] if len(paragraphs) > 1 else "",
is_required=IllustrationType.CONCEPT in config['required']
))
# 3. 类比图位置
analogy_positions = self._find_analogy_positions(paragraphs)
for pos in analogy_positions[:2]: # 最多2个类比图
slot_id += 1
slots.append(IllustrationSlot(
id=f"img_{slot_id:03d}",
position=f"after_paragraph_{pos['index'] + 1}",
paragraph_index=pos['index'],
type=IllustrationType.ANALOGY,
priority=4,
context=pos['text'],
anchor_text=pos['trigger'],
is_required=IllustrationType.ANALOGY in config['required']
))
# 4. 流程图位置
diagram_positions = self._find_diagram_positions(paragraphs)
for pos in diagram_positions[:1]: # 最多1个流程图
slot_id += 1
slots.append(IllustrationSlot(
id=f"img_{slot_id:03d}",
position=f"after_paragraph_{pos['index'] + 1}",
paragraph_index=pos['index'],
type=IllustrationType.DIAGRAM,
priority=3,
context=f"{pos['step_count']}个步骤的流程",
is_required=IllustrationType.DIAGRAM in config['required']
))
# 5. 对比图位置
comparison_positions = self._find_comparison_positions(paragraphs)
for pos in comparison_positions[:1]: # 最多1个对比图
slot_id += 1
slots.append(IllustrationSlot(
id=f"img_{slot_id:03d}",
position=f"after_paragraph_{pos['index'] + 1}",
paragraph_index=pos['index'],
type=IllustrationType.COMPARISON,
priority=3,
context=pos['text'],
is_required=IllustrationType.COMPARISON in config['required']
))
# 6. 氛围图位置（仅长文版）
if tier == LengthTier.LONG:
mood_positions = self._find_mood_positions(paragraphs)
for pos in mood_positions[:2]: # 最多2个氛围图
slot_id += 1
slots.append(IllustrationSlot(
id=f"img_{slot_id:03d}",
position=f"after_paragraph_{pos['index'] + 1}",
paragraph_index=pos['index'],
type=IllustrationType.MOOD,
priority=2,
context=pos['text'],
is_required=False
))
self.plan.slots = slots
# 选择最终的插图槽位
self._select_slots()
return self.plan
def _select_slots(self):
"""选择最终的插图槽位"""
if not self.plan:
return
# 先选必需的
required = [s for s in self.plan.slots if s.is_required]
# 再按优先级选可选的
optional = [s for s in self.plan.slots if not s.is_required]
optional.sort(key=lambda x: x.priority, reverse=True)
# 计算还需要多少
needed = self.plan.max_illustrations - len(required)
selected_optional = optional[:needed] if needed > 0 else []
# 合并并按段落顺序排序
all_selected = required + selected_optional
all_selected.sort(key=lambda x: x.paragraph_index)
self.plan.selected_slots = all_selected
def get_layout_json(self) -> Dict:
"""
获取编排方案的 JSON 格式
Returns:
编排方案字典
"""
if not self.plan:
return {}
return {
'total_illustrations': len(self.plan.selected_slots),
'layout': [
{
'id': slot.id,
'position': slot.position,
'type': slot.type.value,
'context': slot.context,
'anchor_text': slot.anchor_text
}
for slot in self.plan.selected_slots
]
}
# 便捷函数
def generate_layout(content: str, title: str = "") -> Dict:
"""
生成图文编排方案（便捷函数）
Args:
content: 文章内容
title: 文章标题
Returns:
编排方案
"""
engine = LayoutEngine()
engine.analyze(content, title)
return engine.get_layout_json()
if __name__ == "__main__":
# 测试
test_content = """
你以为AI在思考，其实它在做填空题。
这不是在贬低它。恰恰相反，这个"填空题"做得实在太好了。
要理解这一点，我们需要先搞清楚什么是Token。Token就像乐高积木。一个汉字可能是一块积木，一个英文单词可能要两三块拼起来。AI就是用这些积木来理解和组装语言。
Claude Skills 是什么呢？想象一下，你有一个万能工匠，手边放着不同的工具箱。需要木工活就拿木工箱，需要电工活就拿电工箱。Skills 就是这些工具箱。
使用 Skills 的步骤很简单：
第一步，找到合适的 Skill。
第二步，配置参数。
第三步，运行测试。
第四步，根据结果调整。
相比传统的 System Prompt，Skills 更加模块化。相比 Agent，Skills 更加聚焦。它不是要替代什么，而是填补了一个空白。
这让我想起一句话：最好的工具不是替代你，而是让你变成更好的自己。
"""
layout = generate_layout(test_content, "Claude Skills 入门指南")
import json
print(json.dumps(layout, ensure_ascii=False, indent=2))