daec6a376d
- tools/kb-bridge.py: 原始物料→知识页核心脚本（384行完整实现） 功能：读取PDF/CSV/markdown → Claude API分析 → 生成frontmatter → 保存到projects/ - 完整落地链路.md: 端到端实施方案（Phase 0-3完整路径） - 完整项目现状报告.md: 真实状态验证（脚本真相+架构梳理） - docs/bot-comparison-analysis.md: kb-bot vs bot-v2 深度对比 - docs/kb-bot-usage-guide.md: kb-bot 使用指南 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
343 lines
10 KiB
Python
343 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
"""kb-bridge.py —— 原始物料 → 知识页（桥接脚本）
|
||
|
||
核心功能：
|
||
1. 读取原始文件（PDF/CSV/markdown/图片）
|
||
2. 调用 Claude API 分析内容
|
||
3. 判断 type（doc/decision/conversation）
|
||
4. 生成完整 frontmatter（符合 kb-contract.yaml）
|
||
5. 保存到 projects/<项目>/docs|decisions|conversations/
|
||
6. 调用 kb-lint-fm.py 校验
|
||
|
||
用法：
|
||
# 处理单个文件
|
||
python3 tools/kb-bridge.py --input 文件路径 [--project 项目代号]
|
||
|
||
# 批量处理目录
|
||
python3 tools/kb-bridge.py --input-dir 目录路径
|
||
|
||
# 示例
|
||
python3 tools/kb-bridge.py --input ../飞书同步/AI工作流.csv --project ai-workflow
|
||
python3 tools/kb-bridge.py --input ../会议记录/0713会议.pdf
|
||
|
||
依赖：
|
||
pip3 install anthropic pyyaml pdfplumber pandas
|
||
|
||
环境变量：
|
||
ANTHROPIC_API_KEY - Claude API Key（必需）
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import hashlib
|
||
import argparse
|
||
import subprocess
|
||
from pathlib import Path
|
||
from datetime import date
|
||
|
||
try:
|
||
import yaml
|
||
except ImportError:
|
||
print("ERROR: 需要 PyYAML。运行：pip3 install pyyaml", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
try:
|
||
import anthropic
|
||
except ImportError:
|
||
print("ERROR: 需要 anthropic。运行：pip3 install anthropic", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
# 定位库根
|
||
ROOT = Path(__file__).parent.parent.resolve()
|
||
|
||
# ========== 文件提取 ==========
|
||
|
||
def extract_content(file_path):
|
||
"""提取文件内容（支持 PDF/CSV/markdown/txt）"""
|
||
path = Path(file_path)
|
||
ext = path.suffix.lower()
|
||
|
||
if ext == '.pdf':
|
||
try:
|
||
import pdfplumber
|
||
except ImportError:
|
||
print("WARN: PDF 提取需要 pdfplumber。运行：pip3 install pdfplumber", file=sys.stderr)
|
||
return f"[PDF文件：{path.name}，无法提取文本]"
|
||
|
||
try:
|
||
with pdfplumber.open(file_path) as pdf:
|
||
text = "\n\n".join(page.extract_text() or "" for page in pdf.pages)
|
||
return text.strip() or "[PDF无文本内容]"
|
||
except Exception as e:
|
||
return f"[PDF提取失败：{e}]"
|
||
|
||
elif ext == '.csv':
|
||
try:
|
||
import pandas as pd
|
||
except ImportError:
|
||
print("WARN: CSV 处理需要 pandas。运行：pip3 install pandas", file=sys.stderr)
|
||
return f"[CSV文件：{path.name}，无法解析]"
|
||
|
||
try:
|
||
df = pd.read_csv(file_path)
|
||
# 限制行数，避免太长
|
||
if len(df) > 100:
|
||
preview = df.head(50).to_string()
|
||
return f"{preview}\n\n... (共{len(df)}行，仅显示前50行)"
|
||
return df.to_string()
|
||
except Exception as e:
|
||
return f"[CSV解析失败：{e}]"
|
||
|
||
elif ext in ['.md', '.txt']:
|
||
try:
|
||
return path.read_text(encoding='utf-8')
|
||
except Exception as e:
|
||
return f"[文件读取失败：{e}]"
|
||
|
||
else:
|
||
return f"[不支持的文件类型：{ext}]"
|
||
|
||
|
||
# ========== Claude API 分析 ==========
|
||
|
||
def analyze_with_claude(content, file_path):
|
||
"""调用 Claude API 分析文件内容"""
|
||
api_key = os.environ.get("ANTHROPIC_API_KEY")
|
||
if not api_key:
|
||
print("ERROR: 需要设置 ANTHROPIC_API_KEY 环境变量", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
client = anthropic.Anthropic(api_key=api_key)
|
||
|
||
# 限制内容长度（避免超token）
|
||
content_preview = content[:8000] if len(content) > 8000 else content
|
||
|
||
prompt = f"""分析这个文件，返回 JSON 格式的分析结果。
|
||
|
||
文件路径：{file_path}
|
||
内容（前8000字符）：
|
||
{content_preview}
|
||
|
||
请判断：
|
||
1. **type**（必须是以下之一）：
|
||
- "doc"：外部文档、参考资料、飞书文档
|
||
- "decision"：决策记录、方案选择、重要决定
|
||
- "conversation"：会议纪要、聊天记录、讨论总结
|
||
|
||
2. **title**：一句话标题（中文，20字内）
|
||
|
||
3. **description**：一句话描述（中文，50字内，说清楚这是什么内容）
|
||
|
||
4. **tags**：3-5个关键词（中文，用于分类和检索）
|
||
|
||
5. **key_points**：提炼3-5条核心要点（markdown格式，每条1-2句话）
|
||
|
||
6. **project**：建议的项目代号（英文/拼音，kebab-case，如 ai-workflow、meeting-2024q3）
|
||
- 如果是会议记录 → meeting-YYYY-qN
|
||
- 如果是某个主题的文档 → 主题-拼音
|
||
- 如果无法判断 → misc
|
||
|
||
返回格式（纯JSON，不要markdown代码块）：
|
||
{{
|
||
"type": "doc",
|
||
"title": "...",
|
||
"description": "...",
|
||
"tags": ["...", "...", "..."],
|
||
"key_points": "- 要点1\\n- 要点2\\n- 要点3",
|
||
"project": "..."
|
||
}}
|
||
"""
|
||
|
||
try:
|
||
response = client.messages.create(
|
||
model="claude-sonnet-3-5-20240620",
|
||
max_tokens=2000,
|
||
messages=[{"role": "user", "content": prompt}]
|
||
)
|
||
|
||
result_text = response.content[0].text.strip()
|
||
|
||
# 移除可能的 markdown 代码块标记
|
||
if result_text.startswith("```"):
|
||
lines = result_text.split("\n")
|
||
result_text = "\n".join(lines[1:-1])
|
||
|
||
return json.loads(result_text)
|
||
|
||
except Exception as e:
|
||
print(f"ERROR: Claude API 调用失败：{e}", file=sys.stderr)
|
||
# 返回默认值
|
||
return {
|
||
"type": "doc",
|
||
"title": Path(file_path).stem,
|
||
"description": "自动导入的文档",
|
||
"tags": ["待分类"],
|
||
"key_points": "[AI分析失败，需要手动补充]",
|
||
"project": "misc"
|
||
}
|
||
|
||
|
||
# ========== 生成 frontmatter ==========
|
||
|
||
def generate_frontmatter(analysis, source_file):
|
||
"""生成完整 frontmatter（符合 kb-contract.yaml）"""
|
||
|
||
# 判断 source（来源）
|
||
source_path = Path(source_file).resolve()
|
||
if "飞书同步" in str(source_path):
|
||
source = "飞书文档"
|
||
elif "会议记录" in str(source_path):
|
||
source = "会议"
|
||
else:
|
||
source = "NAS"
|
||
|
||
fm = {
|
||
"type": analysis["type"],
|
||
"title": analysis["title"],
|
||
"description": analysis["description"],
|
||
"tags": analysis["tags"],
|
||
"timestamp": str(date.today()),
|
||
"source": source,
|
||
"source_link": source_file if source_file.startswith("http") else f"file://{source_path}",
|
||
"status": "seed",
|
||
"created": str(date.today()),
|
||
"ingested": str(date.today()),
|
||
}
|
||
|
||
return "---\n" + yaml.dump(fm, allow_unicode=True, default_flow_style=False) + "---\n"
|
||
|
||
|
||
# ========== 保存知识页 ==========
|
||
|
||
def save_to_kb(content, analysis, source_file, project_override=None):
|
||
"""保存到知识库 projects/"""
|
||
|
||
project = project_override or analysis["project"]
|
||
|
||
# type → 子目录映射
|
||
type_dir = {
|
||
"doc": "docs",
|
||
"decision": "decisions",
|
||
"conversation": "conversations"
|
||
}.get(analysis["type"], "docs")
|
||
|
||
# 项目目录
|
||
project_dir = ROOT / "projects" / project
|
||
|
||
# 如果项目不存在，用 kb-init.sh 创建
|
||
if not project_dir.exists():
|
||
print(f"[创建项目] {project}")
|
||
result = subprocess.run(
|
||
["bash", str(ROOT / "tools" / "kb-init.sh"), project],
|
||
cwd=ROOT,
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
if result.returncode != 0:
|
||
print(f"ERROR: kb-init.sh 失败：{result.stderr}", file=sys.stderr)
|
||
return None
|
||
|
||
# 生成文件名（slug化）
|
||
def slugify(text):
|
||
import re
|
||
# 移除特殊字符，保留中文、字母、数字
|
||
text = re.sub(r'[^\w\s一-鿿-]', '', text)
|
||
# 空格替换为 -
|
||
text = re.sub(r'[\s]+', '-', text)
|
||
return text.strip('-')[:50] # 限制长度
|
||
|
||
filename = slugify(analysis["title"]) + ".md"
|
||
output_path = project_dir / type_dir / filename
|
||
|
||
# 生成完整内容
|
||
frontmatter = generate_frontmatter(analysis, source_file)
|
||
full_content = f"{frontmatter}\n{analysis['key_points']}\n\n## 原始文件\n\n文件：`{Path(source_file).name}`\n\n"
|
||
|
||
# 保存
|
||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||
output_path.write_text(full_content, encoding='utf-8')
|
||
|
||
return output_path
|
||
|
||
|
||
# ========== 主函数 ==========
|
||
|
||
def process_file(input_path, project=None):
|
||
"""处理单个文件"""
|
||
print(f"\n===== 处理文件：{input_path} =====")
|
||
|
||
# 1. 提取内容
|
||
print("[1/5] 提取文件内容...")
|
||
content = extract_content(input_path)
|
||
if not content or len(content) < 10:
|
||
print(f"WARN: 文件内容为空或太短，跳过")
|
||
return None
|
||
|
||
# 2. Claude 分析
|
||
print("[2/5] Claude API 分析...")
|
||
analysis = analyze_with_claude(content, input_path)
|
||
print(f" - type: {analysis['type']}")
|
||
print(f" - title: {analysis['title']}")
|
||
print(f" - project: {analysis['project']}")
|
||
|
||
# 3. 保存到知识库
|
||
print("[3/5] 保存到知识库...")
|
||
output_path = save_to_kb(content, analysis, input_path, project)
|
||
if not output_path:
|
||
return None
|
||
print(f" ✓ 已保存：{output_path.relative_to(ROOT)}")
|
||
|
||
# 4. Lint 校验
|
||
print("[4/5] frontmatter 校验...")
|
||
result = subprocess.run(
|
||
["python3", str(ROOT / "tools" / "kb-lint-fm.py"), str(output_path)],
|
||
cwd=ROOT,
|
||
capture_output=True,
|
||
text=True
|
||
)
|
||
if result.returncode == 0:
|
||
print(" ✓ 校验通过")
|
||
else:
|
||
print(f" ✗ 校验失败：\n{result.stdout}")
|
||
return None
|
||
|
||
# 5. 提示 Git 提交
|
||
print("[5/5] 提示：")
|
||
print(f" git add {output_path.relative_to(ROOT)}")
|
||
print(f" git commit -m 'Add {analysis['title']}'")
|
||
print(f" git push")
|
||
|
||
return output_path
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="原始物料 → 知识页")
|
||
parser.add_argument("--input", help="输入文件路径")
|
||
parser.add_argument("--input-dir", help="输入目录路径（批量处理）")
|
||
parser.add_argument("--project", help="指定项目代号（可选，否则 AI 自动判断）")
|
||
args = parser.parse_args()
|
||
|
||
if not args.input and not args.input_dir:
|
||
parser.print_help()
|
||
sys.exit(1)
|
||
|
||
# 单文件处理
|
||
if args.input:
|
||
process_file(args.input, args.project)
|
||
|
||
# 批量处理
|
||
elif args.input_dir:
|
||
input_dir = Path(args.input_dir)
|
||
files = list(input_dir.glob("*"))
|
||
print(f"发现 {len(files)} 个文件")
|
||
|
||
for i, file_path in enumerate(files, 1):
|
||
if file_path.is_file():
|
||
print(f"\n[{i}/{len(files)}] {file_path.name}")
|
||
process_file(str(file_path), args.project)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|