852f5cb648
- kb_writer.py: DEVONthink AppleScript 브릿지 → 마크다운 파일 기반 전환 (포트 8095) - knowledge-base/ 디렉토리 구조 (note, chat-memory, news) - Handle Note: kb_writer 파일 저장 + Qdrant 임베딩 추가 - Embed & Save Memory: DEVONthink → kb_writer 교체 - mail_bridge.py: IMAP 날짜 기반 메일 조회 (포트 8094) - mail-processing-pipeline: IMAP Trigger → Schedule + mail_bridge + dedup - docker-compose, manage_services, LaunchAgent plist 업데이트 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
107 lines
3.1 KiB
Python
107 lines
3.1 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 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()
|
|
|
|
# 파일명 생성
|
|
slug = _slugify(title)
|
|
if doc_type == "chat-memory":
|
|
time_str = now.strftime("%H%M")
|
|
filename = f"{date_str}T{time_str}-{_slugify(topic)}.md"
|
|
elif doc_type == "news":
|
|
filename = f"{date_str}-{_slugify(source)}-{slug}.md"
|
|
else:
|
|
filename = f"{date_str}-{slug}.md"
|
|
|
|
# 디렉토리 생성
|
|
type_dir = BASE_DIR / doc_type / month_dir
|
|
type_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 중복 파일명 처리
|
|
filepath = type_dir / filename
|
|
counter = 1
|
|
while filepath.exists():
|
|
stem = filename.rsplit(".", 1)[0]
|
|
filepath = type_dir / f"{stem}-{counter}.md"
|
|
counter += 1
|
|
|
|
# YAML frontmatter + 본문
|
|
tags_yaml = ", ".join(f'"{t}"' for t in tags)
|
|
qdrant_id = 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}
|