131 lines
3.8 KiB
Python
131 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
批量生成 session 对话摘要。
|
||
|
||
使用 Claude API 为每个对话文件生成 50-400 字的中文摘要。
|
||
摘要规则：
|
||
- 列出本次对话完成的所有功能点/任务
|
||
- 包含关键技术细节：文件路径、模块名、数据库表、API 端点等
|
||
- bug 修复要说明原因和方案
|
||
- 不写过程性描述，只写结果
|
||
- 内容太短或无实质内容的，写"无实质内容"
|
||
"""
|
||
|
||
import os
|
||
import json
|
||
import sys
|
||
from pathlib import Path
|
||
from dotenv import load_dotenv
|
||
import anthropic
|
||
|
||
# 加载环境变量
|
||
load_dotenv()
|
||
|
||
from _env_paths import ensure_repo_root
|
||
ensure_repo_root()
|
||
|
||
# 初始化 Anthropic 客户端
|
||
client = anthropic.Anthropic(api_key=os.environ.get("ANTHROPIC_API_KEY"))
|
||
|
||
# 路径配置
|
||
SESSION_DIR = Path("export/session_summaries")
|
||
OUTPUT_DIR = SESSION_DIR / "out"
|
||
OUTPUT_DIR.mkdir(exist_ok=True)
|
||
|
||
SYSTEM_PROMPT = """你是一个专业的技术对话分析师。你的任务是为 AI 编程助手与开发者的对话生成简洁的中文摘要。
|
||
|
||
摘要规则：
|
||
1. 列出本次对话完成的所有功能点/任务（一个对话可能完成多个）
|
||
2. 包含关键技术细节：文件路径、模块名、数据库表、API 端点等
|
||
3. bug 修复要说明原因和方案
|
||
4. 不写过程性描述（"用户说..."），只写结果
|
||
5. 内容太短或无实质内容的，写"无实质内容"
|
||
6. 字数控制在 50-400 字之间
|
||
|
||
示例摘要：
|
||
"修复 ETL DWD_LOAD_FROM_ODS 的 SCD2 时区问题（Asia/Shanghai → UTC 遗漏），影响 dwd.dim_member。新增 quality/row_count_check.py 质检规则。涉及：tasks/dwd_load.py、scd/merge.py。"
|
||
|
||
请直接输出摘要，不要添加任何前缀或解释。"""
|
||
|
||
|
||
def generate_summary(content: str) -> str:
|
||
"""使用 Claude 生成摘要"""
|
||
try:
|
||
message = client.messages.create(
|
||
model="claude-3-5-sonnet-20241022",
|
||
max_tokens=500,
|
||
system=SYSTEM_PROMPT,
|
||
messages=[
|
||
{
|
||
"role": "user",
|
||
"content": f"请为以下对话生成摘要：\n\n{content}",
|
||
}
|
||
],
|
||
)
|
||
return message.content[0].text.strip()
|
||
except Exception as e:
|
||
print(f"生成摘要失败: {e}", file=sys.stderr)
|
||
return f"生成失败: {str(e)}"
|
||
|
||
|
||
def process_file(input_path: Path) -> bool:
|
||
"""处理单个文件，返回是否成功"""
|
||
output_path = OUTPUT_DIR / input_path.name
|
||
|
||
# 如果输出文件已存在，跳过
|
||
if output_path.exists():
|
||
print(f"✓ 已存在: {input_path.name}")
|
||
return True
|
||
|
||
try:
|
||
# 读取输入文件
|
||
with open(input_path, "r", encoding="utf-8") as f:
|
||
content = f.read()
|
||
|
||
# 生成摘要
|
||
summary = generate_summary(content)
|
||
|
||
# 写入输出文件
|
||
with open(output_path, "w", encoding="utf-8") as f:
|
||
f.write(summary)
|
||
|
||
print(f"✓ 完成: {input_path.name}")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"✗ 错误 {input_path.name}: {e}", file=sys.stderr)
|
||
return False
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
# 获取所有 .txt 文件（跳过 _manifest.json 和以 _ 开头的文件）
|
||
files = sorted(
|
||
[
|
||
f
|
||
for f in SESSION_DIR.glob("*.txt")
|
||
if not f.name.startswith("_") and f.name != "_manifest.json"
|
||
]
|
||
)
|
||
|
||
if not files:
|
||
print("未找到要处理的文件")
|
||
return
|
||
|
||
print(f"找到 {len(files)} 个文件")
|
||
|
||
# 处理文件
|
||
success_count = 0
|
||
for i, file_path in enumerate(files, 1):
|
||
if process_file(file_path):
|
||
success_count += 1
|
||
# 每处理 10 个文件打印进度
|
||
if i % 10 == 0:
|
||
print(f"进度: {i}/{len(files)}")
|
||
|
||
print(f"\n完成: {success_count}/{len(files)} 个文件")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|