syn-chat-bot/devonthink_bridge.py

"""DEVONthink Bridge — AppleScript REST API 래퍼 (port 8093)"""

import json
import logging
import os
import subprocess

from dotenv import load_dotenv
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

load_dotenv()

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
logger = logging.getLogger("devonthink_bridge")

DT_APP = os.getenv("DEVONTHINK_APP_NAME", "DEVONthink")

app = FastAPI()


def _run_applescript(script: str, timeout: int = 15) -> str:
    """AppleScript 실행."""
    result = subprocess.run(
        ["osascript", "-e", script],
        capture_output=True, text=True, timeout=timeout,
    )
    if result.returncode != 0:
        raise RuntimeError(f"AppleScript error: {result.stderr.strip()}")
    return result.stdout.strip()


def _escape_as(s: str) -> str:
    """AppleScript 문자열 이스케이프."""
    return s.replace("\\", "\\\\").replace('"', '\\"')


@app.post("/save")
async def save_record(request: Request):
    body = await request.json()
    title = body.get("title", "Untitled")
    content = body.get("content", "")
    record_type = body.get("type", "markdown")
    database = body.get("database")
    group = body.get("group")
    tags = body.get("tags", [])

    # Map type to DEVONthink record type
    type_map = {"markdown": "markdown", "text": "txt", "html": "html"}
    dt_type = type_map.get(record_type, "markdown")

    # Build AppleScript
    esc_title = _escape_as(title)
    esc_content = _escape_as(content)
    tags_str = ", ".join(f'"{_escape_as(t)}"' for t in tags) if tags else ""

    if database:
        db_line = f'set theDB to open database "{_escape_as(database)}"'
    else:
        db_line = "set theDB to first database"

    if group:
        group_line = f'set theGroup to get record at "/{_escape_as(group)}" in theDB'
    else:
        group_line = "set theGroup to incoming group of theDB"

    script = f'''tell application "{DT_APP}"
    {db_line}
    {group_line}
    set theRecord to create record with {{name:"{esc_title}", type:{dt_type}, plain text:"{esc_content}"}} in theGroup
    {f'set tags of theRecord to {{{tags_str}}}' if tags_str else ''}
    set theUUID to uuid of theRecord
    set theName to name of theRecord
    return theUUID & "|" & theName
end tell'''

    try:
        result = _run_applescript(script)
        parts = result.split("|", 1)
        uuid_val = parts[0] if parts else result
        name_val = parts[1] if len(parts) > 1 else title
        logger.info(f"Record saved: {uuid_val} '{name_val}'")
        return JSONResponse({"success": True, "uuid": uuid_val, "name": name_val})
    except Exception as e:
        logger.error(f"Save failed: {e}")
        return JSONResponse({"success": False, "error": str(e)}, status_code=500)


@app.get("/databases")
async def list_databases():
    script = f'''tell application "{DT_APP}"
    set dbList to {{}}
    repeat with theDB in databases
        set end of dbList to (name of theDB) & "|" & (uuid of theDB)
    end repeat
    set AppleScript's text item delimiters to "\\n"
    return dbList as text
end tell'''

    try:
        result = _run_applescript(script)
        databases = []
        for line in result.strip().split("\n"):
            if "|" in line:
                parts = line.split("|", 1)
                databases.append({"name": parts[0], "uuid": parts[1]})
        return JSONResponse({"databases": databases})
    except Exception as e:
        logger.error(f"List databases failed: {e}")
        return JSONResponse({"databases": [], "error": str(e)}, status_code=500)


@app.get("/health")
async def health():
    devonthink_running = False
    try:
        result = _run_applescript(
            f'tell application "System Events" to return (name of processes) contains "{DT_APP}"',
            timeout=5,
        )
        devonthink_running = result.lower() == "true"
    except Exception:
        pass

    return {"status": "ok", "devonthink_running": devonthink_running}