From 89d93e7e6f9866d5af2473982bb576e7295d52da Mon Sep 17 00:00:00 2001 From: CD Date: Sun, 15 Mar 2026 19:06:47 +0900 Subject: [PATCH] =?UTF-8?q?feat(tools):=20Wiki.js=20GraphQL=20=ED=81=B4?= =?UTF-8?q?=EB=9D=BC=EC=9D=B4=EC=96=B8=ED=8A=B8=20=EC=B6=94=EA=B0=80=20?= =?UTF-8?q?=E2=80=94=20=ED=8E=98=EC=9D=B4=EC=A7=80=20CRUD=20+=20=EB=A6=AC?= =?UTF-8?q?=EC=84=9C=EC=B9=98=20=EB=8C=80=EC=8B=9C=EB=B3=B4=EB=93=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tools/wiki_client.py | 330 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 330 insertions(+) create mode 100644 tools/wiki_client.py diff --git a/tools/wiki_client.py b/tools/wiki_client.py new file mode 100644 index 0000000..8cc263d --- /dev/null +++ b/tools/wiki_client.py @@ -0,0 +1,330 @@ +"""Wiki.js GraphQL 클라이언트 — 페이지 CRUD + 리서치 대시보드 관리. + +API: https://wiki.variet.net/graphql +""" + +import asyncio +import httpx +import logging +import re +import sys +from datetime import datetime +from typing import Optional +from dataclasses import dataclass + +import os +import sys +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) +import config + +logger = logging.getLogger("variet.tools.wiki") + +WIKI_URL = getattr(config, "WIKI_URL", "https://wiki.variet.net") +WIKI_API_KEY = getattr(config, "WIKI_API_KEY", "") +GRAPHQL_ENDPOINT = f"{WIKI_URL}/graphql" +DASHBOARD_PATH = "research/dashboard" + + +@dataclass +class WikiPage: + """위키 페이지 정보.""" + id: int + path: str + title: str + content: str = "" + updated_at: str = "" + description: str = "" + + +class WikiClient: + """Wiki.js GraphQL API 클라이언트.""" + + def __init__(self, timeout: float = 15.0): + self._timeout = timeout + self._headers = { + "Authorization": f"Bearer {WIKI_API_KEY}", + "Content-Type": "application/json", + } + + async def _query(self, query: str, variables: dict = None) -> dict: + """GraphQL 쿼리 실행.""" + payload = {"query": query} + if variables: + payload["variables"] = variables + + async with httpx.AsyncClient(timeout=self._timeout) as client: + resp = await client.post( + GRAPHQL_ENDPOINT, + json=payload, + headers=self._headers, + ) + resp.raise_for_status() + data = resp.json() + + if "errors" in data and data["errors"]: + err_msg = data["errors"][0].get("message", str(data["errors"])) + raise RuntimeError(f"Wiki.js API 오류: {err_msg}") + + return data.get("data", {}) + + # ────────────────────────────────────── + # 조회 + # ────────────────────────────────────── + + async def list_pages(self, prefix: str = "") -> list[WikiPage]: + """페이지 목록 조회. prefix로 경로 필터.""" + query = "{ pages { list { id, path, title, updatedAt } } }" + data = await self._query(query) + pages = [ + WikiPage( + id=p["id"], + path=p["path"], + title=p["title"], + updated_at=p.get("updatedAt", ""), + ) + for p in data.get("pages", {}).get("list", []) + ] + if prefix: + pages = [p for p in pages if p.path.startswith(prefix)] + return pages + + async def get_page(self, page_id: int) -> WikiPage: + """페이지 내용 조회 (ID 기반).""" + query = """ + query ($id: Int!) { + pages { single(id: $id) { + id, path, title, content, updatedAt, description + }} + } + """ + data = await self._query(query, {"id": page_id}) + p = data["pages"]["single"] + if not p: + raise RuntimeError(f"페이지 ID {page_id}를 찾을 수 없습니다.") + return WikiPage( + id=p["id"], path=p["path"], title=p["title"], + content=p.get("content", ""), updated_at=p.get("updatedAt", ""), + description=p.get("description", ""), + ) + + async def find_page(self, path: str) -> Optional[WikiPage]: + """경로로 페이지 찾기 (없으면 None).""" + pages = await self.list_pages() + for p in pages: + if p.path == path: + return await self.get_page(p.id) + return None + + # ────────────────────────────────────── + # 생성 / 수정 / 삭제 + # ────────────────────────────────────── + + async def create_page( + self, path: str, title: str, content: str, + description: str = "", tags: list[str] = None, + ) -> WikiPage: + """새 페이지 생성.""" + query = """ + mutation ($content: String!, $path: String!, $title: String!, + $description: String!, $tags: [String!]!) { + pages { create( + content: $content, description: $description, + editor: "markdown", isPublished: true, isPrivate: false, + locale: "ko", path: $path, tags: $tags, title: $title + ) { + responseResult { succeeded, message } + page { id, path } + }} + } + """ + variables = { + "content": content, "path": path, "title": title, + "description": description or title, + "tags": tags or [], + } + data = await self._query(query, variables) + result = data["pages"]["create"] + if not result["responseResult"]["succeeded"]: + raise RuntimeError(f"페이지 생성 실패: {result['responseResult']['message']}") + + page = result["page"] + logger.info(f"위키 페이지 생성: {path} (id: {page['id']})") + return WikiPage(id=page["id"], path=page["path"], title=title, content=content) + + async def update_page( + self, page_id: int, content: str, + title: str = None, description: str = None, tags: list[str] = None, + ) -> bool: + """기존 페이지 수정.""" + query = """ + mutation ($id: Int!, $content: String!, $title: String, + $description: String, $tags: [String!]) { + pages { update( + id: $id, content: $content, title: $title, + description: $description, tags: $tags + ) { + responseResult { succeeded, message } + }} + } + """ + variables = {"id": page_id, "content": content} + if title: + variables["title"] = title + if description: + variables["description"] = description + if tags is not None: + variables["tags"] = tags + + data = await self._query(query, variables) + result = data["pages"]["update"] + ok = result["responseResult"]["succeeded"] + if ok: + logger.info(f"위키 페이지 수정: id={page_id}") + else: + logger.error(f"수정 실패: {result['responseResult']['message']}") + return ok + + async def delete_page(self, page_id: int) -> bool: + """페이지 삭제.""" + query = """ + mutation ($id: Int!) { + pages { delete(id: $id) { + responseResult { succeeded, message } + }} + } + """ + data = await self._query(query, {"id": page_id}) + result = data["pages"]["delete"] + ok = result["responseResult"]["succeeded"] + if ok: + logger.info(f"위키 페이지 삭제: id={page_id}") + return ok + + # ────────────────────────────────────── + # 고수준 API + # ────────────────────────────────────── + + async def upsert_page( + self, path: str, title: str, content: str, + description: str = "", tags: list[str] = None, + ) -> WikiPage: + """있으면 수정, 없으면 생성.""" + existing = await self.find_page(path) + if existing: + await self.update_page( + existing.id, content, title=title, + description=description, tags=tags, + ) + existing.content = content + existing.title = title + return existing + else: + return await self.create_page( + path, title, content, description, tags, + ) + + async def update_dashboard(self, entries: list[dict] = None) -> WikiPage: + """리서치 대시보드 페이지 갱신. + + entries 형식: [{"topic": "주제", "status": "✅", "date": "2026-03-15", + "path": "research/topic-slug"}] + entries가 None이면 research/ 하위 페이지에서 자동 수집. + """ + if entries is None: + # research/ 하위 페이지 자동 수집 + pages = await self.list_pages("research/") + entries = [] + for p in pages: + if p.path == DASHBOARD_PATH: + continue + entries.append({ + "topic": p.title, + "status": "✅", + "date": p.updated_at[:10] if p.updated_at else "", + "path": p.path, + }) + + # 대시보드 마크다운 생성 + lines = [ + "# 📚 리서치 대시보드\n", + f"> 마지막 갱신: {datetime.now().strftime('%Y-%m-%d %H:%M')}\n", + "| 주제 | 상태 | 날짜 | 링크 |", + "|---|---|---|---|", + ] + for e in sorted(entries, key=lambda x: x.get("date", ""), reverse=True): + link = f"[보기](/{e['path']})" + lines.append(f"| {e['topic']} | {e['status']} | {e.get('date','')} | {link} |") + + if not entries: + lines.append("| *(등록된 리서치 없음)* | | | |") + + content = "\n".join(lines) + return await self.upsert_page( + DASHBOARD_PATH, "리서치 대시보드", content, + description="리서치 현황 요약", + tags=["dashboard", "research"], + ) + + @staticmethod + def slugify(text: str) -> str: + """한글/영문 제목을 URL-safe slug로 변환.""" + # 한글은 그대로, 특수문자 제거, 공백→하이픈 + slug = re.sub(r'[^\w\s가-힣-]', '', text.lower()) + slug = re.sub(r'\s+', '-', slug.strip()) + return slug or "untitled" + + +# ────────────────────────────────────── +# CLI +# ────────────────────────────────────── + +async def _cli(): + import io + if sys.stdout.encoding != "utf-8": + sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace") + + args = sys.argv[1:] + client = WikiClient() + + if not args or args[0] == "list": + prefix = args[1] if len(args) > 1 else "" + pages = await client.list_pages(prefix) + print(f"📄 페이지 {len(pages)}개:") + for p in pages: + print(f" [{p.id}] /{p.path} — {p.title}") + + elif args[0] == "get" and len(args) > 1: + page = await client.find_page(args[1]) + if page: + print(f"📄 {page.title} (/{page.path})") + print(f"{'─'*40}") + print(page.content[:2000]) + else: + print(f"❌ '{args[1]}' 페이지를 찾을 수 없습니다.") + + elif args[0] == "create" and len(args) > 2: + path, title = args[1], args[2] + content = args[3] if len(args) > 3 else f"# {title}\n\n(내용 없음)" + page = await client.create_page(path, title, content) + print(f"✅ 생성: /{page.path} (id: {page.id})") + + elif args[0] == "dashboard": + page = await client.update_dashboard() + print(f"✅ 대시보드 갱신: /{page.path} (id: {page.id})") + + elif args[0] == "delete" and len(args) > 1: + page_id = int(args[1]) + ok = await client.delete_page(page_id) + print(f"{'✅ 삭제 완료' if ok else '❌ 삭제 실패'}: id={page_id}") + + else: + print("사용법:") + print(" wiki_client.py list [prefix]") + print(" wiki_client.py get ") + print(" wiki_client.py create [content]") + print(" wiki_client.py dashboard") + print(" wiki_client.py delete <id>") + + +if __name__ == "__main__": + asyncio.run(_cli())