syn-chat-bot/note_bridge.py

"""Note Bridge — Synology Note Station REST API 래퍼 (port 8096)"""

import logging
import os

import httpx
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("note_bridge")

# Note Station 앱 포털 포트 (9350)
NOTE_STATION_URL = os.getenv("NOTE_STATION_URL", "http://192.168.1.227:9350")
NOTE_ACCOUNT = os.getenv("NOTE_ACCOUNT", "chatbot-api")
NOTE_PASSWORD = os.getenv("NOTE_PASSWORD", "U1Ob2L3a")

# 메모장 ID
NOTEBOOK_MEMO = os.getenv("NOTEBOOK_MEMO", "1030_3CLANHDEQH4T9DRONETDNOG2U4")      # chatbot (메모용)

# 세션
_sid: str = ""


async def _login() -> str:
    """Note Station 앱 포털에 로그인하여 SID를 반환."""
    global _sid
    async with httpx.AsyncClient(verify=False, timeout=15) as c:
        resp = await c.post(f"{NOTE_STATION_URL}/webapi/entry.cgi", data={
            "api": "SYNO.API.Auth",
            "version": "7",
            "method": "login",
            "account": NOTE_ACCOUNT,
            "passwd": NOTE_PASSWORD,
        })
        data = resp.json()
        if data.get("success"):
            _sid = data["data"]["sid"]
            logger.info("Note Station login successful")
            return _sid
        else:
            logger.error(f"Note Station login failed: {data}")
            raise Exception(f"Login failed: {data}")


async def _ensure_sid() -> str:
    """SID가 유효한지 확인하고, 필요하면 재로그인."""
    global _sid
    if not _sid:
        return await _login()
    # SID 유효성 간단 체크
    async with httpx.AsyncClient(verify=False, timeout=10) as c:
        resp = await c.get(f"{NOTE_STATION_URL}/webapi/entry.cgi", params={
            "api": "SYNO.NoteStation.Info",
            "version": "1",
            "method": "get",
            "_sid": _sid,
        })
        data = resp.json()
        if data.get("success"):
            return _sid
    return await _login()


async def _api_call(api: str, method: str, version: int = 3, **kwargs) -> dict:
    """Note Station API 호출 래퍼."""
    sid = await _ensure_sid()
    params = {
        "api": api,
        "version": str(version),
        "method": method,
        "_sid": sid,
        **kwargs,
    }
    async with httpx.AsyncClient(verify=False, timeout=30) as c:
        resp = await c.post(f"{NOTE_STATION_URL}/webapi/entry.cgi", data=params)
        data = resp.json()
        if not data.get("success"):
            # SID 만료일 수 있으므로 재로그인 후 재시도
            sid = await _login()
            params["_sid"] = sid
            resp = await c.post(f"{NOTE_STATION_URL}/webapi/entry.cgi", data=params)
            data = resp.json()
        return data


# ==================== FastAPI App ====================

app = FastAPI(title="Note Bridge")


@app.post("/note/create")
async def create_note(request: Request):
    """노트 생성. body: {title, content}"""
    body = await request.json()
    title = body.get("title", "제목 없음")
    content = body.get("content", "")

    # HTML 래핑 (이미 HTML이 아닌 경우)
    if content and not content.strip().startswith("<"):
        content = f"<p>{content}</p>"

    result = await _api_call(
        "SYNO.NoteStation.Note",
        "create",
        parent_id=NOTEBOOK_MEMO,
        title=title,
        commit_msg='{"device":"api","listable":false}',
        encrypt="false",
    )

    if result.get("success"):
        note_id = result["data"].get("object_id", "")
        # 내용 업데이트 (create는 빈 노트만 생성)
        if content:
            await _api_call(
                "SYNO.NoteStation.Note",
                "update",
                object_id=note_id,
                content=content,
                commit_msg='{"device":"api","listable":false}',
            )
        logger.info(f"Note created: {note_id} '{title}'")
        return {"success": True, "note_id": note_id, "title": title,
                "message": f"메모 저장 완료: {title}"}
    else:
        logger.error(f"Note creation failed: {result}")
        return JSONResponse(status_code=500, content={
            "success": False, "error": str(result),
            "message": f"메모 저장 실패: {title}"})


@app.post("/note/append")
async def append_to_note(request: Request):
    """기존 노트에 내용 추가. body: {note_id, content}"""
    body = await request.json()

    note_id = body.get("note_id")
    content = body.get("content", "")

    if not content:
        return JSONResponse(status_code=400, content={"success": False, "error": "content required"})
    if not note_id:
        return JSONResponse(status_code=400, content={"success": False, "error": "note_id required"})

    # 기존 내용 가져오기
    existing = await _api_call("SYNO.NoteStation.Note", "get", object_id=note_id)
    if not existing.get("success"):
        return JSONResponse(status_code=404, content={"success": False, "error": "Note not found"})

    old_content = existing["data"].get("content", "")

    # HTML 래핑
    if not content.strip().startswith("<"):
        content = f"<p>{content}</p>"

    updated_content = old_content + content

    result = await _api_call(
        "SYNO.NoteStation.Note",
        "update",
        object_id=note_id,
        content=updated_content,
        commit_msg='{"device":"api","listable":false}',
    )

    if result.get("success"):
        logger.info(f"Note appended: {note_id}")
        return {"success": True, "note_id": note_id, "message": f"메모 추가 완료"}
    else:
        return JSONResponse(status_code=500, content={"success": False, "error": str(result)})


@app.get("/health")
async def health():
    """헬스체크."""
    try:
        sid = await _ensure_sid()
        return {"status": "ok", "sid_active": bool(sid)}
    except Exception as e:
        return {"status": "error", "error": str(e)}