| name | blog-rss-wiki-graph |
| description | 从博客 RSS 生成 Wiki 并分析页面间语义关联关系 — 自动生成概念页、提取 wikilinks、构建链接图、输出 CSV 报告 |
| version | 1.1.1 |
| author | Hermes Agent |
| tags | ["wiki","blog","rss","graph-analysis","semantic-links","knowledge-base"] |
| metadata | {"hermes":{"tags":["wiki","blog","rss","graph-analysis","semantic-links"],"category":"research","related_skills":["llm-wiki"]}} |
Blog RSS Wiki Graph Analyzer
从博客 RSS feed 批量 ingest 文章，为每篇文章生成语义关联的 wiki 概念页（含 [[wikilinks]]），构建页面间链接图，输出 CSV 格式的关联分析报告。
核心流程
RSS Feed → prepare_ingest.py（拉取+索引）→ LLM 语义分析 → 创建 concept 页（含 wikilinks）→ execute_ingest.py（提取链接+生成报告）
每篇文章经历 4 步：
- Fetch — 从 RSS 拉取文章，保存 raw，更新 page_index
- Analyze — LLM 分析文章内容，识别与已有页面的语义关联
- Write — 创建 concept 页，写入 YAML frontmatter + 摘要 +
[[wikilinks]]
- Update — 提取 wikilinks，追加到 link.csv，重新生成报告
目录结构
wiki/
├── page_index.csv # 索引：index,page,pub_date,raw_url
├── link.csv # 边集：source,target（无向图）
├── link_inbound.csv # 报告：按度数排名
├── link_outbound_top3.csv # 报告：每页 outbound 前3高度数页面
├── raw/articles/ # 原始文章（不可变）
├── concepts/ # 概念页（LLM 生成，含 wikilinks）
└── scripts/
├── prepare_ingest.py # 拉取 RSS + 更新索引 + slug 去重
└── execute_ingest.py # 提取 wikilinks + 更新链接图 + 重新生成报告
CSV 文件格式
page_index.csv
index,page,pub_date,raw_url
1,hermes-python,2026-04-21,https://zelikk.blogspot.com/2026/04/hermes-python.html
link.csv（无向图，source < target 排序去重）
source,target
1,4
2,7
link_inbound.csv（按 degree 降序）
rank,index,page,degree,neighbors
1,42,develop-with-gpt,15,"13|27|30|36|..."
link_outbound_top3.csv
index,page,top1_index,top1_page,top2_index,top2_page,top3_index,top3_page
1,hermes-python,13,hermes-agent-oracle-vps-ubuntu-root,4,hermes-agent-telegram-group-topic-agent,...
概念页模板
---
title: "<文章标题>"
created: YYYY-MM-DD
updated: YYYY-MM-DD
type: article
tags: [relevant, tags]
source_url: <原始链接>
raw_source: raw/articles/<slug>.md
---
# <文章标题>
<1-3 句中文摘要>
## 语义关联
- [[related-page-slug-1]]
- [[related-page-slug-2]]
- [[related-page-slug-3]]
关键脚本
prepare_ingest.py
拉取 RSS 最新文章，通过 URL 比对检测未索引的新文章，保存 raw，更新 page_index。
核心逻辑（URL 比对）：Blogger RSS 的 start-index=1 永远返回最新文章，新文章插入顶部会导致旧文章 index 后移。因此不能按顺序索引号 fetch，必须：
fetch_rss(start_index=1, max_results=50) 拉取最新文章列表- 与
page_index.csv 中的 raw_url 比对，过滤出未索引的文章
- 按
pub_date 升序排序，每次处理最早的一篇（保持时间序）
- 连续运行可清空积压（一次一篇，直到无新文章返回
{"error": "No new article"}）
- 冷启动兜底：当最新 50 篇都已索引但 RSS 总数 > 已索引数时，自动分页遍历全部 RSS（每次 50 篇），找到剩余未索引的旧文章
自动初始化：如果 page_index.csv 不存在，自动创建带 header 的空文件。
Slug 去重逻辑（必须）：遇到重复 slug 时自动加 -index 后缀。
def deduplicate_slug(slug, index, existing_slugs):
if slug not in existing_slugs:
return slug
import random
chars = 'abcdefghjkmnpqrstuvwxyz23456789'
while True:
rand = ''.join(random.choices(chars, k=4))
candidate = f"{slug}-{index}-{rand}"
if candidate not in existing_slugs:
return candidate
此逻辑同时覆盖两种场景：
- Blogger 默认 slug
blog-post：作者未设标题时，多篇文章会共享同一 slug，自动加后缀避免覆盖
- 同 slug 不同日期：如果 RSS 中出现同一 slug 但不同日期的文章（如更新版），
-index 后缀同样确保唯一性
execute_ingest.py
从 concept 页提取 [[wikilinks]]，查找 page_index 获取 index，追加边到 link.csv（跳过已存在的边），重新生成报告。
批量处理策略
冷启动 vs 增量更新
冷启动（从零构建 wiki）
当 page_index.csv 不存在时，prepare_ingest.py 会自动创建空索引文件。运行流程：
- 前 50 次运行：每次拉取 RSS 最新 50 篇，取
pub_date 最早的一篇处理
- 第 51 次及之后：脚本检测到最新 50 篇都已索引，但 RSS 总数 > 已索引数，自动分页遍历全部 RSS（每次 50 篇），找到剩余未索引的旧文章继续处理
- 直到全部文章处理完毕，返回
{"error": "No new article"}
增量更新（已有 wiki）
每次运行只检查 RSS 最新 50 篇。新文章总是在 RSS 顶部，因此不需要全量遍历，速度快、开销小。
使用 delegate_task
每批 5 篇文章，通过 delegate_task 并行处理：
delegate_task:
goal: "Process 5 blog articles (next 5 RSS indices) into the wiki at ~/wiki/"
toolsets: ["terminal", "file"]
max_iterations: 60
LLM 语义分析要点
- 读取
existing_pages 列表（格式 "index: slug"）
- 识别语义相关页面（同主题、同工具、同问题域）
- 每个 concept 页添加 2-6 个
[[wikilinks]]
- 使用中文摘要
验收检查清单
完成后运行以下检查：
cd wiki
echo "index:$(tail -n +2 page_index.csv | wc -l) raw:$(ls raw/articles/*.md | wc -l) concepts:$(ls concepts/*.md | wc -l)"
python3 -c "
import csv
from collections import Counter
slugs = [r['page'] for r in csv.DictReader(open('page_index.csv'))]
dupes = {k:v for k,v in Counter(slugs).items() if v > 1}
print(f'Duplicate slugs: {dupes if dupes else \"NONE\"}')
"
python3 -c "
import csv
edges = set()
dupes = 0
for row in csv.DictReader(open('link.csv')):
e = tuple(sorted([row['source'], row['target']]))
if e in edges: dupes += 1
edges.add(e)
print(f'Edges: {len(edges)}, Duplicates: {dupes}')
"
python3 -c "
import csv, os, re
valid = {r['page'] for r in csv.DictReader(open('page_index.csv'))}
broken = []
for f in os.listdir('concepts'):
if not f.endswith('.md'): continue
with open(f'concepts/{f}') as fh:
for l in re.findall(r'\[\[([^\]|]+?)(?:\|[^\]]+)?\]\]', fh.read()):
if l not in valid: broken.append(f'{f[:-3]} → [[{l}]]')
print(f'Broken: {len(broken)}')
for b in broken: print(f' {b}')
"
