Add kb-bridge.py + 文档整理
- 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>
This commit is contained in:
yangqianqian
342
tools/kb-bridge.py
Normal file
342
tools/kb-bridge.py
Normal file
@@ -0,0 +1,342 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user