hoelee
86edd40c74
- docs/consultant-review/: review brief, per-batch post dumps (01-10), audit-tracker.json + AUDIT-TRACKER.md generated by scripts/manage_review.py - scripts/manage_review.py: list/approve/approve-batch/generate over the tracker - docs/consultant-review-2026-10-27-posts-and-casestudies-plan.md: locked decision register + 4 work orders for the /posts/ archive rework and the case-studies portfolio treatment
459 lines
20 KiB
Python
459 lines
20 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Blog Post Consultant Review Management Tool
|
||
- Tracks review status of all blog posts (pending, approved, needs_revision)
|
||
- Automatically partitions unapproved posts into batches (< 200k chars)
|
||
- Injects background, constraints, persona rules, and purpose into each batch
|
||
- Allows 1-command approval so approved posts are permanently skipped next time
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import glob
|
||
import re
|
||
import argparse
|
||
from datetime import datetime
|
||
|
||
REPO_ROOT = "D:/dev/hoelee-blog"
|
||
POSTS_DIR = os.path.join(REPO_ROOT, "src/content/posts")
|
||
ZH_POSTS_DIR = os.path.join(POSTS_DIR, "zh")
|
||
REVIEW_DIR = os.path.join(REPO_ROOT, "docs/consultant-review")
|
||
BATCHES_DIR = os.path.join(REVIEW_DIR, "batches")
|
||
STATUS_FILE = os.path.join(REVIEW_DIR, "audit-tracker.json")
|
||
TRACKER_MD = os.path.join(REVIEW_DIR, "AUDIT-TRACKER.md")
|
||
BRIEF_MD = os.path.join(REVIEW_DIR, "00-外部顾问审阅须知与背景规范.md")
|
||
|
||
BACKGROUND_HEADER = """# 外部顾问审阅前置须知与工程背景规范
|
||
|
||
> **本文档目标**：本批次包含 `blog.hoelee.com` 的待审阅技术文章全文。请外部顾问在审阅正文与技术主张时，以此规范为最高评估准则。
|
||
|
||
## 1. 站点定位与双重业务目的
|
||
- **受众 A（技术招聘官 / 资深技术主管）**：寻找具备 Senior 全栈 + DevOps + AI 智能体闭环能力的架构工程师。文章定位为**“不可伪造的工程实操交付证据（Proof of Expertise）”**，要求代码真实、日志可查、有架构权衡与技术深度。
|
||
- **受众 B（中小企业主 SME / 商业客户）**：寻找能提供系统研发、私有化部署、24/7 AI 获客系统、高可用基础设施的技术顾问。文章需清晰阐述业务痛点、商业 ROI、上线效率与交付保障。
|
||
|
||
## 2. 核心人设铁律与工程硬约束（必须严格遵守）
|
||
1. **彻底杜绝业余爱好者/玩具口吻**：
|
||
- ❌ **绝对禁用词**：`homelab`, `home lab`, `家用`, `小主机`, `玩具`, `折腾`。
|
||
- ✅ **标准替代词**：`the infrastructure I manage`（我管理的私有基础设施）、`self-hosted production node`（自托管生产节点）、`on-premise host`（私有化服务器）。
|
||
2. **严禁捏造虚假指标与性能数字**：
|
||
- 所有 benchmark 数据必须有明确的测试基准与单设备限制。
|
||
- 严格区分“耗时减少比例”与“吞吐加速倍数”（例如耗时从 352s 降至 218s 是耗时减少 38.0%，吞吐提升 1.61×，严禁偷换概念混淆）。
|
||
- 时间戳精度与物理帧对齐：视频帧率上限（如 30 fps 单帧 33.3ms）不可妄称“微秒级精确”。
|
||
3. **去模板化叙事与真实物理锚点**：
|
||
- 拒绝千篇一律的 AI 模板味。文章采用三大叙事架构：① 故障排查谜案（Symptom → False Lead → Kernel Evidence → Fix）；② 商业交付复盘（Client Bottleneck → Trade-offs → State Machine → ROI）；③ 架构演进对比（V1 Naive → Concurrency Crash → V2 Redesign）。
|
||
- 每篇文章必须包含至少一个物理真实世界锚点（主机真实硬件规格、生产维护窗口、真实业务场景）。
|
||
4. **代码与配置展示规范**：
|
||
- 超过 20 行的配置或日志采用 HTML `<details><summary>` 原生折叠，`<summary>` 必须精准标注文档名与行数。
|
||
|
||
## 3. 顾问重点开刀维度
|
||
- [ ] **事实与数字可核性**：技术判断与性能数据是否自洽？有无过度夸大？
|
||
- [ ] **人设与语调纯洁性**：有无遗漏的 hobbyist/homelab 口吻？
|
||
- [ ] **双轨转化引导有效性**：是否有清晰的业务价值提炼与行动召唤（CTA）？
|
||
- [ ] **代码与生产安全性**：有无未脱敏的凭据、绝对内部私密路径或不安全的权限配置？
|
||
"""
|
||
|
||
def load_or_init_tracker():
|
||
tracker = {}
|
||
if os.path.exists(STATUS_FILE):
|
||
with open(STATUS_FILE, "r", encoding="utf-8") as f:
|
||
try:
|
||
tracker = json.load(f)
|
||
except Exception:
|
||
tracker = {}
|
||
|
||
# Scan all posts in POSTS_DIR
|
||
en_files = sorted(glob.glob(os.path.join(POSTS_DIR, "*.md")))
|
||
for en_path in en_files:
|
||
slug = os.path.splitext(os.path.basename(en_path))[0]
|
||
zh_path = os.path.join(ZH_POSTS_DIR, f"{slug}.md")
|
||
|
||
with open(en_path, "r", encoding="utf-8") as f:
|
||
en_content = f.read()
|
||
zh_content = ""
|
||
if os.path.exists(zh_path):
|
||
with open(zh_path, "r", encoding="utf-8") as f:
|
||
zh_content = f.read()
|
||
|
||
# Extract category, title, pubDate safely without regex multiline newlines
|
||
category = "uncategorized"
|
||
title = slug
|
||
zh_title = slug
|
||
pub_date = "unknown"
|
||
for line in en_content.splitlines():
|
||
if line.startswith("category:"):
|
||
category = line.split(":", 1)[1].strip().strip('"\'')
|
||
elif line.startswith("title:"):
|
||
title = line.split(":", 1)[1].strip().strip('"\'')
|
||
elif line.startswith("pubDate:"):
|
||
pub_date = line.split(":", 1)[1].strip().strip('"\'')
|
||
for line in zh_content.splitlines():
|
||
if line.startswith("title:"):
|
||
zh_title = line.split(":", 1)[1].strip().strip('"\'')
|
||
|
||
if slug not in tracker:
|
||
tracker[slug] = {
|
||
"slug": slug,
|
||
"title": title,
|
||
"zh_title": zh_title,
|
||
"category": category,
|
||
"pub_date": pub_date,
|
||
"status": "pending", # pending | approved | rejected | needs_revision
|
||
"batch_id": None,
|
||
"approved_at": None,
|
||
"notes": "",
|
||
"en_chars": len(en_content),
|
||
"zh_chars": len(zh_content)
|
||
}
|
||
else:
|
||
# Update dynamic fields
|
||
tracker[slug]["title"] = title
|
||
tracker[slug]["zh_title"] = zh_title
|
||
tracker[slug]["category"] = category
|
||
tracker[slug]["pub_date"] = pub_date
|
||
tracker[slug]["en_chars"] = len(en_content)
|
||
tracker[slug]["zh_chars"] = len(zh_content)
|
||
|
||
save_tracker(tracker)
|
||
return tracker
|
||
|
||
def save_tracker(tracker):
|
||
os.makedirs(REVIEW_DIR, exist_ok=True)
|
||
with open(STATUS_FILE, "w", encoding="utf-8") as f:
|
||
json.dump(tracker, f, ensure_ascii=False, indent=2)
|
||
render_tracker_markdown(tracker)
|
||
|
||
def render_tracker_markdown(tracker):
|
||
total = len(tracker)
|
||
approved = sum(1 for p in tracker.values() if p.get("status") == "approved")
|
||
pending = total - approved
|
||
|
||
lines = [
|
||
"# 全站文章顾问审阅状态总控台 (Audit Tracker)",
|
||
"",
|
||
f"> **最后同步时间**：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ",
|
||
f"> **全站总文章数**：{total} 篇 | **已审批完成**：{approved} 篇 ({approved/total*100:.1f}%) | **待审阅**：{pending} 篇",
|
||
"",
|
||
"## 一、操作命令速查",
|
||
"- `python scripts/manage_review.py list` — 查看待审与已审概览",
|
||
"- `python scripts/manage_review.py approve <slug> [--notes '顾问通过']` — 标记单篇通过",
|
||
"- `python scripts/manage_review.py approve-batch <batch_num>` — 标记整个批次通过",
|
||
"- `python scripts/manage_review.py generate` — 重新分批（自动排除已审批文章，每批 < 200k 字）",
|
||
"",
|
||
"## 二、文章审核状态总表",
|
||
"| # | 分类 | 文章 Slug | 中文标题 | 状态 | 所在批次 | 审批时间 | 顾问备注 |",
|
||
"|---|---|---|---|:---:|:---:|:---:|---|"
|
||
]
|
||
|
||
sorted_posts = sorted(tracker.values(), key=lambda x: (x.get("category", ""), x.get("slug", "")))
|
||
for idx, p in enumerate(sorted_posts, 1):
|
||
status_icon = "✅ 已通过" if p.get("status") == "approved" else "⏳ 待审阅"
|
||
if p.get("status") == "needs_revision":
|
||
status_icon = "⚠️ 需修改"
|
||
batch = p.get("batch_id") or "未分配"
|
||
approved_time = p.get("approved_at") or "-"
|
||
notes = (p.get("notes") or "").replace("|", "/")
|
||
lines.append(f"| {idx} | `{p['category']}` | `{p['slug']}` | {p['zh_title']} | {status_icon} | {batch} | {approved_time} | {notes} |")
|
||
|
||
with open(TRACKER_MD, "w", encoding="utf-8") as f:
|
||
f.write("\n".join(lines) + "\n")
|
||
|
||
def write_master_brief():
|
||
with open(BRIEF_MD, "w", encoding="utf-8") as f:
|
||
f.write(BACKGROUND_HEADER + """
|
||
## 4. 博客 7 大分类与内容策略参考
|
||
- `case-studies` (商业案例)：求职与获客核心武器，以真实商业交付和解决瓶颈为核心。
|
||
- `devops` (私有基建)：高并发网关、Docker、Traefik、存储阵列与监控，最强差异化资产。
|
||
- `ai` (人工智能与智能体)：围绕自动化工作流、LiteLLM 网关、状态机防幻觉与本地私有大模型。
|
||
- `engineering` (系统架构与后端)：Java/Spring、PHP/CodeIgniter、TS/React 的硬核实战。
|
||
- `tutorials` (技术教程)：面向学习者的深度实战指南。
|
||
- `notes` (排障速记)：轻量级踩坑手记与高频高价值 gotcha。
|
||
- `web3` (智能合约)：Solidity/Foundry 严谨实践，诚实表达学习与验证阶段。
|
||
|
||
## 5. 既有核心参考文档清单（仓库内置）
|
||
- `docs/content-guide.md` — 博客内容定位、选题规范、反模式与 i18n 策略。
|
||
- `docs/post-publishing-sop-v2.md` — 5 步叙事模型、代码折叠与 Bento 卡片技术规范。
|
||
- `docs/consultant-review-2026-10-26-supplement.md` — 上一轮顾问诊断全景资料包。
|
||
""")
|
||
|
||
def generate_batches(mode="bilingual", max_chars=180000):
|
||
tracker = load_or_init_tracker()
|
||
write_master_brief()
|
||
os.makedirs(BATCHES_DIR, exist_ok=True)
|
||
|
||
# Clean previous batch files
|
||
for old_b in glob.glob(os.path.join(BATCHES_DIR, "batch-*.md")):
|
||
try:
|
||
os.remove(old_b)
|
||
except Exception:
|
||
pass
|
||
|
||
# Filter unapproved posts
|
||
pending_posts = [p for p in tracker.values() if p.get("status") != "approved"]
|
||
if not pending_posts:
|
||
print("All posts are approved! No new batches needed.")
|
||
return
|
||
|
||
# Sort deterministically: category, then slug
|
||
pending_posts.sort(key=lambda x: (x.get("category", ""), x.get("slug", "")))
|
||
|
||
batches = []
|
||
current_batch = []
|
||
current_chars = 0
|
||
|
||
for p in pending_posts:
|
||
slug = p['slug']
|
||
en_path = os.path.join(POSTS_DIR, f"{slug}.md")
|
||
zh_path = os.path.join(ZH_POSTS_DIR, f"{slug}.md")
|
||
|
||
with open(en_path, "r", encoding="utf-8") as f:
|
||
en_text = f.read()
|
||
zh_text = ""
|
||
if os.path.exists(zh_path):
|
||
with open(zh_path, "r", encoding="utf-8") as f:
|
||
zh_text = f.read()
|
||
|
||
if mode == "bilingual":
|
||
item_chars = len(en_text) + len(zh_text)
|
||
if item_chars > max_chars:
|
||
# Split giant post into ZH and EN individual sub-batches
|
||
if current_batch:
|
||
batches.append(current_batch)
|
||
current_batch = []
|
||
current_chars = 0
|
||
batches.append([{
|
||
"p": p, "mode": "zh_only", "slug": slug + "-zh",
|
||
"content_zh": zh_text, "content_en": "",
|
||
"chars": len(zh_text)
|
||
}])
|
||
batches.append([{
|
||
"p": p, "mode": "en_only", "slug": slug + "-en",
|
||
"content_zh": "", "content_en": en_text,
|
||
"chars": len(en_text)
|
||
}])
|
||
continue
|
||
item = {
|
||
"p": p, "mode": "bi", "slug": slug,
|
||
"content_zh": zh_text, "content_en": en_text,
|
||
"chars": item_chars
|
||
}
|
||
elif mode == "zh":
|
||
item_chars = len(zh_text)
|
||
item = {
|
||
"p": p, "mode": "zh_only", "slug": slug,
|
||
"content_zh": zh_text, "content_en": "",
|
||
"chars": item_chars
|
||
}
|
||
else: # en
|
||
item_chars = len(en_text)
|
||
item = {
|
||
"p": p, "mode": "en_only", "slug": slug,
|
||
"content_zh": "", "content_en": en_text,
|
||
"chars": item_chars
|
||
}
|
||
|
||
if current_chars + item_chars > max_chars and current_batch:
|
||
batches.append(current_batch)
|
||
current_batch = [item]
|
||
current_chars = item_chars
|
||
else:
|
||
current_batch.append(item)
|
||
current_chars += item_chars
|
||
|
||
if current_batch:
|
||
batches.append(current_batch)
|
||
|
||
# Write each batch file
|
||
for b_idx, batch_items in enumerate(batches, 1):
|
||
batch_id = f"Batch-{b_idx:02d}"
|
||
batch_char_sum = sum(x['chars'] for x in batch_items)
|
||
batch_filename = f"batch-{b_idx:02d}.md"
|
||
batch_filepath = os.path.join(BATCHES_DIR, batch_filename)
|
||
|
||
lines = [
|
||
f"# 博客文章顾问审阅资料包 — {batch_id}",
|
||
"",
|
||
f"> **批次编号**：`{batch_id}` ",
|
||
f"> **包含文章数**：{len(batch_items)} 篇 ",
|
||
f"> **本批次总字符量**：{batch_char_sum:,} 字（严格限制 < 200,000 字） ",
|
||
f"> **模式**：{mode.upper()} ",
|
||
f"> **生成时间**：{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ",
|
||
"",
|
||
BACKGROUND_HEADER,
|
||
"",
|
||
"---",
|
||
"",
|
||
"## 本批次文章目录索引",
|
||
""
|
||
]
|
||
|
||
for i, item in enumerate(batch_items, 1):
|
||
p = item['p']
|
||
lines.append(f"{i}. [{p['category']}] **{p['zh_title']}** (`{item['slug']}`) — ~{item['chars']:,} 字符")
|
||
|
||
lines.append("")
|
||
lines.append("---")
|
||
lines.append("")
|
||
|
||
for i, item in enumerate(batch_items, 1):
|
||
p = item['p']
|
||
slug = p['slug']
|
||
tracker[slug]["batch_id"] = batch_id
|
||
|
||
lines.append(f"## 文章 {i:02d} / {len(batch_items):02d}：{p['zh_title']}")
|
||
lines.append(f"- **Slug**: `{slug}`")
|
||
lines.append(f"- **英文标题**: {p['title']}")
|
||
lines.append(f"- **分类**: `{p['category']}` | **发布日期**: {p['pub_date']}")
|
||
lines.append(f"- **线上地址**: https://blog.hoelee.com/posts/{slug}/ (EN) | https://blog.hoelee.com/posts/zh/{slug}/ (ZH)")
|
||
lines.append(f"- **本篇审核状态**: ⏳ 待审核 (Pending)")
|
||
lines.append("")
|
||
lines.append("### ✍️ 顾问审阅意见区 (Review Feedback)")
|
||
lines.append("```text")
|
||
lines.append("【审阅结论】：[ ] 通过 [ ] 需微调 [ ] 需重构")
|
||
lines.append("【事实与数字核对】：")
|
||
lines.append("【人设与用词卫生】：")
|
||
lines.append("【CRO与转化建议】：")
|
||
lines.append("```")
|
||
lines.append("")
|
||
|
||
if item["content_zh"]:
|
||
lines.append("### 🇨🇳 中文版原文 (Chinese Markdown Twin)")
|
||
lines.append("````markdown")
|
||
lines.append(item["content_zh"].strip())
|
||
lines.append("````")
|
||
lines.append("")
|
||
|
||
if item["content_en"]:
|
||
lines.append("### 🇬🇧 英文版原文 (English Markdown Primary)")
|
||
lines.append("````markdown")
|
||
lines.append(item["content_en"].strip())
|
||
lines.append("````")
|
||
lines.append("")
|
||
|
||
lines.append("---")
|
||
lines.append("")
|
||
|
||
with open(batch_filepath, "w", encoding="utf-8") as bf:
|
||
bf.write("\n".join(lines) + "\n")
|
||
|
||
print(f"Generated {batch_id}: {len(batch_items)} posts, {batch_char_sum:,} chars -> {batch_filepath}")
|
||
|
||
save_tracker(tracker)
|
||
print(f"\n[OK] Generated {len(batches)} batches in {BATCHES_DIR}. Total pending posts processed: {len(pending_posts)}")
|
||
|
||
def approve_posts(slugs, notes=None):
|
||
tracker = load_or_init_tracker()
|
||
now_str = datetime.now().strftime("%Y-%m-%d %H:%M")
|
||
count = 0
|
||
for slug in slugs:
|
||
if slug in tracker:
|
||
tracker[slug]["status"] = "approved"
|
||
tracker[slug]["approved_at"] = now_str
|
||
if notes:
|
||
tracker[slug]["notes"] = notes
|
||
count += 1
|
||
print(f"Approved: {slug}")
|
||
else:
|
||
print(f"Warning: Slug '{slug}' not found in tracker.")
|
||
save_tracker(tracker)
|
||
print(f"Successfully marked {count} post(s) as approved.")
|
||
|
||
def approve_batch(batch_num, notes=None):
|
||
tracker = load_or_init_tracker()
|
||
target_batch = f"Batch-{int(batch_num):02d}"
|
||
matched_slugs = [slug for slug, p in tracker.items() if p.get("batch_id") == target_batch]
|
||
if not matched_slugs:
|
||
print(f"No posts found belonging to {target_batch}.")
|
||
return
|
||
approve_posts(matched_slugs, notes=notes or f"Approved via {target_batch}")
|
||
|
||
def reset_posts(slugs):
|
||
tracker = load_or_init_tracker()
|
||
count = 0
|
||
for slug in slugs:
|
||
if slug in tracker:
|
||
tracker[slug]["status"] = "pending"
|
||
tracker[slug]["approved_at"] = None
|
||
tracker[slug]["notes"] = ""
|
||
count += 1
|
||
print(f"Reset to pending: {slug}")
|
||
save_tracker(tracker)
|
||
print(f"Successfully reset {count} post(s).")
|
||
|
||
def reset_all():
|
||
tracker = load_or_init_tracker()
|
||
for slug in tracker:
|
||
tracker[slug]["status"] = "pending"
|
||
tracker[slug]["approved_at"] = None
|
||
tracker[slug]["notes"] = ""
|
||
save_tracker(tracker)
|
||
print(f"Successfully reset all {len(tracker)} posts to pending.")
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="Blog Post Consultant Review Management")
|
||
subparsers = parser.add_subparsers(dest="command")
|
||
|
||
# init
|
||
subparsers.add_parser("init", help="Initialize or update tracker from markdown files")
|
||
|
||
# generate
|
||
gen_p = subparsers.add_parser("generate", help="Generate review batches (< 200k chars each)")
|
||
gen_p.add_argument("--mode", choices=["bilingual", "zh", "en"], default="bilingual", help="Content mode")
|
||
gen_p.add_argument("--max-chars", type=int, default=180000, help="Max characters per batch")
|
||
|
||
# approve
|
||
app_p = subparsers.add_parser("approve", help="Mark slug(s) as approved")
|
||
app_p.add_argument("slugs", nargs="+", help="Post slug(s) to approve")
|
||
app_p.add_argument("--notes", default="Approved by consultant", help="Consultant review note")
|
||
|
||
# approve-batch
|
||
app_b = subparsers.add_parser("approve-batch", help="Mark all posts in a batch as approved")
|
||
app_b.add_argument("batch_num", type=int, help="Batch number (e.g. 1)")
|
||
app_b.add_argument("--notes", default="Approved by consultant", help="Consultant review note")
|
||
|
||
# reset
|
||
rst_p = subparsers.add_parser("reset", help="Reset slug(s) to pending")
|
||
rst_p.add_argument("slugs", nargs="+", help="Post slug(s) to reset")
|
||
|
||
# reset-all
|
||
subparsers.add_parser("reset-all", help="Reset all posts to pending")
|
||
|
||
# list
|
||
subparsers.add_parser("list", help="List summary of review status")
|
||
|
||
args = parser.parse_args()
|
||
|
||
if args.command == "init":
|
||
t = load_or_init_tracker()
|
||
print(f"Tracker initialized with {len(t)} posts.")
|
||
elif args.command == "generate":
|
||
generate_batches(mode=args.mode, max_chars=args.max_chars)
|
||
elif args.command == "approve":
|
||
approve_posts(args.slugs, notes=args.notes)
|
||
elif args.command == "approve-batch":
|
||
approve_batch(args.batch_num, notes=args.notes)
|
||
elif args.command == "reset":
|
||
reset_posts(args.slugs)
|
||
elif args.command == "reset-all":
|
||
reset_all()
|
||
elif args.command == "list" or not args.command:
|
||
tracker = load_or_init_tracker()
|
||
total = len(tracker)
|
||
approved = sum(1 for p in tracker.values() if p.get("status") == "approved")
|
||
pending = total - approved
|
||
print(f"=== Blog Post Consultant Review Status ===")
|
||
print(f"Total Posts: {total} | Approved: {approved} | Pending: {pending}")
|
||
by_cat = {}
|
||
for p in tracker.values():
|
||
cat = p['category']
|
||
by_cat.setdefault(cat, [0, 0])
|
||
if p.get('status') == 'approved':
|
||
by_cat[cat][0] += 1
|
||
else:
|
||
by_cat[cat][1] += 1
|
||
for cat, (app, pend) in sorted(by_cat.items()):
|
||
print(f" {cat:15s}: {app:2d} approved / {pend:2d} pending")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|