feat(tools): Wiki.js GraphQL 클라이언트 추가 — 페이지 CRUD + 리서치 대시보드
This commit is contained in:
330
tools/wiki_client.py
Normal file
330
tools/wiki_client.py
Normal file
@@ -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 <path>")
|
||||
print(" wiki_client.py create <path> <title> [content]")
|
||||
print(" wiki_client.py dashboard")
|
||||
print(" wiki_client.py delete <id>")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(_cli())
|
||||
Reference in New Issue
Block a user