a050f2e7d5
- kb_writer: uuid4 short hash로 파일명 경합 방지, counter 기반 중복 방어 제거 - kb_writer: qdrant_id 외부 수신 지원 (body.qdrant_id) - n8n: Set pid 노드 추가 — 분기 전 pid 한 번 생성, Handle Note/Embed & Save Memory에 전달 - Handle Note/Embed & Save Memory: 동일 pid를 kb_writer(qdrant_id)와 Qdrant point ID에 사용 - restore_kb.sh: DS1525+ → 맥미니 knowledge-base 복구 스크립트 추가 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
102 lines
3.0 KiB
Python
102 lines
3.0 KiB
Python
"""Knowledge Base Writer — 마크다운 파일 저장 마이크로서비스 (port 8095)
|
|
|
|
DEVONthink AppleScript 브릿지 대체. 순수 파일 I/O로 knowledge-base/ 에 마크다운 저장.
|
|
DEVONthink에서는 인덱스 그룹으로 읽기만 하면 됨.
|
|
"""
|
|
|
|
import logging
|
|
import re
|
|
import unicodedata
|
|
from datetime import datetime, timezone, timedelta
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.responses import JSONResponse
|
|
|
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
|
|
logger = logging.getLogger("kb_writer")
|
|
|
|
KST = timezone(timedelta(hours=9))
|
|
BASE_DIR = Path(__file__).resolve().parent / "knowledge-base"
|
|
|
|
app = FastAPI()
|
|
|
|
|
|
def _slugify(text: str, max_len: int = 50) -> str:
|
|
"""한글 + 영문 친화적 슬러그 생성."""
|
|
text = unicodedata.normalize("NFC", text)
|
|
text = re.sub(r"[^\w가-힣\s-]", "", text)
|
|
text = re.sub(r"[\s_]+", "-", text).strip("-")
|
|
return text[:max_len] or "untitled"
|
|
|
|
|
|
@app.post("/save")
|
|
async def save(request: Request):
|
|
body = await request.json()
|
|
title = body.get("title", "Untitled")
|
|
content = body.get("content", "")
|
|
doc_type = body.get("type", "note") # note | chat-memory | news
|
|
tags = body.get("tags", [])
|
|
username = body.get("username", "unknown")
|
|
topic = body.get("topic", "general")
|
|
source = body.get("source", "synology-chat")
|
|
|
|
now = datetime.now(KST)
|
|
month_dir = now.strftime("%Y-%m")
|
|
date_str = now.strftime("%Y-%m-%d")
|
|
iso_str = now.isoformat()
|
|
|
|
# 파일명 생성 (uuid4 short hash로 경합 방지)
|
|
slug = _slugify(title)
|
|
uid = uuid4().hex[:6]
|
|
if doc_type == "chat-memory":
|
|
time_str = now.strftime("%H%M")
|
|
filename = f"{date_str}T{time_str}-{uid}-{_slugify(topic)}.md"
|
|
elif doc_type == "news":
|
|
filename = f"{date_str}-{uid}-{_slugify(source)}-{slug}.md"
|
|
else:
|
|
filename = f"{date_str}-{uid}-{slug}.md"
|
|
|
|
# 디렉토리 생성
|
|
type_dir = BASE_DIR / doc_type / month_dir
|
|
type_dir.mkdir(parents=True, exist_ok=True)
|
|
filepath = type_dir / filename
|
|
|
|
# YAML frontmatter + 본문
|
|
tags_yaml = ", ".join(f'"{t}"' for t in tags)
|
|
qdrant_id = body.get("qdrant_id") or int(now.timestamp() * 1000)
|
|
|
|
md_content = f"""---
|
|
title: "{title}"
|
|
date: "{iso_str}"
|
|
source: {source}
|
|
type: {doc_type}
|
|
tags: [{tags_yaml}]
|
|
username: {username}
|
|
topic: {topic}
|
|
qdrant_id: {qdrant_id}
|
|
---
|
|
|
|
{content}
|
|
"""
|
|
|
|
try:
|
|
filepath.write_text(md_content, encoding="utf-8")
|
|
logger.info(f"Saved: {filepath.relative_to(BASE_DIR)}")
|
|
return JSONResponse({
|
|
"success": True,
|
|
"path": str(filepath.relative_to(BASE_DIR)),
|
|
"filename": filepath.name,
|
|
"qdrant_id": qdrant_id,
|
|
})
|
|
except Exception as e:
|
|
logger.error(f"Save failed: {e}")
|
|
return JSONResponse({"success": False, "error": str(e)}, status_code=500)
|
|
|
|
|
|
@app.get("/health")
|
|
async def health():
|
|
kb_exists = BASE_DIR.is_dir()
|
|
return {"status": "ok", "knowledge_base_exists": kb_exists}
|