Variet Agent
cbc9db0439
- wiki_client.py: find_page()를 singleByPath 직접 조회로 교체 (O(1)) - debate-agent/: gemini/opus AG 프로젝트 생성 (GEMINI.md + wiki_debate.py + /start) - debate_handler.py: 로컬 파일 I/O → Wiki.js API 전환, _sync_wiki() 삭제 - response 페이지 매 턴 초기화 (누적 방지)
353 lines
13 KiB
Python
353 lines
13 KiB
Python
"""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 = ""
|
|
tags: list[str] = None
|
|
|
|
|
|
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, tags { tag }
|
|
}}
|
|
}
|
|
"""
|
|
data = await self._query(query, {"id": page_id})
|
|
p = data["pages"]["single"]
|
|
if not p:
|
|
raise RuntimeError(f"페이지 ID {page_id}를 찾을 수 없습니다.")
|
|
tags = [t["tag"] for t in p.get("tags", [])]
|
|
return WikiPage(
|
|
id=p["id"], path=p["path"], title=p["title"],
|
|
content=p.get("content", ""), updated_at=p.get("updatedAt", ""),
|
|
description=p.get("description", ""), tags=tags,
|
|
)
|
|
|
|
async def find_page(self, path: str) -> Optional[WikiPage]:
|
|
"""경로로 페이지 찾기 (없으면 None). singleByPath 직접 조회."""
|
|
query = """
|
|
query ($path: String!) {
|
|
pages { singleByPath(path: $path, locale: "ko") {
|
|
id, path, title, content, updatedAt, description,
|
|
tags { tag }
|
|
}}
|
|
}
|
|
"""
|
|
data = await self._query(query, {"path": path})
|
|
p = data.get("pages", {}).get("singleByPath")
|
|
if not p:
|
|
return None
|
|
tags = [t["tag"] for t in p.get("tags", [])]
|
|
return WikiPage(
|
|
id=p["id"], path=p["path"], title=p["title"],
|
|
content=p.get("content", ""), updated_at=p.get("updatedAt", ""),
|
|
description=p.get("description", ""), tags=tags,
|
|
)
|
|
|
|
# ──────────────────────────────────────
|
|
# 생성 / 수정 / 삭제
|
|
# ──────────────────────────────────────
|
|
|
|
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:
|
|
"""기존 페이지 수정."""
|
|
if tags is None:
|
|
existing = await self.get_page(page_id)
|
|
tags = existing.tags
|
|
|
|
args_def = ["$id: Int!", "$content: String!", "$tags: [String!]"]
|
|
args_pass = ["id: $id", "content: $content", "tags: $tags"]
|
|
variables = {"id": page_id, "content": content, "tags": tags}
|
|
|
|
if title:
|
|
args_def.append("$title: String")
|
|
args_pass.append("title: $title")
|
|
variables["title"] = title
|
|
if description:
|
|
args_def.append("$description: String")
|
|
args_pass.append("description: $description")
|
|
variables["description"] = description
|
|
|
|
query = f"""
|
|
mutation ({", ".join(args_def)}) {{
|
|
pages {{ update(
|
|
{", ".join(args_pass)}
|
|
) {{
|
|
responseResult {{ succeeded, message }}
|
|
}}}}
|
|
}}
|
|
"""
|
|
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())
|