feat: 워크스페이스 시스템 + 통합 프롬프트 + Docs 기록 관리
- workspace.py: 채널별 워크스페이스 모델 + JSON 영속 저장 - discord_bot.py: /workspace 슬래시 커맨드 (set/git/vikunja/info/remove/list) - 등록 채널만 자동 응답, 미등록 채널 무시 - Git/Vikunja 미설정 시 작업 차단 + 안내 - 통합 프롬프트 1회 호출 (router+planner+chat 통합) - docs_manager.py: Wiki 인덱스, 세션 기록, Changelog 자동 업데이트 - task_pipeline.py: 모든 Gemini 호출에 docs 컨텍스트 주입, 완료 시 기록 - unified.md: 분류+즉답/계획 통합 프롬프트
This commit is contained in:
132
core/docs_manager.py
Normal file
132
core/docs_manager.py
Normal file
@@ -0,0 +1,132 @@
|
||||
"""Docs Manager — 작업 기록 + Wiki 문서 관리.
|
||||
|
||||
모든 작업은 docs/에 기록되며, Gemini 호출 시
|
||||
문서 경로와 기존 문서 목록을 프롬프트에 주입합니다.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
logger = logging.getLogger("variet.docs")
|
||||
|
||||
|
||||
class DocsManager:
|
||||
"""프로젝트 문서 관리자."""
|
||||
|
||||
def __init__(self, project_path: str, docs_subpath: str = "docs/wiki"):
|
||||
self.project_path = Path(project_path)
|
||||
self.docs_path = self.project_path / docs_subpath
|
||||
self.sessions_path = self.project_path / "docs" / "sessions"
|
||||
self.changelog_path = self.project_path / "docs" / "changelog.md"
|
||||
|
||||
# 디렉토리 생성
|
||||
self.docs_path.mkdir(parents=True, exist_ok=True)
|
||||
self.sessions_path.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
def get_docs_index(self) -> str:
|
||||
"""docs/wiki 내 문서 목록을 문자열로 반환 (프롬프트 주입용)."""
|
||||
if not self.docs_path.exists():
|
||||
return "문서 없음"
|
||||
|
||||
files = sorted(self.docs_path.glob("*.md"))
|
||||
if not files:
|
||||
return "문서 없음"
|
||||
|
||||
lines = ["=== PROJECT DOCS ==="]
|
||||
for f in files:
|
||||
size = f.stat().st_size
|
||||
lines.append(f" - {f.name} ({size}B)")
|
||||
lines.append(f"경로: {self.docs_path}")
|
||||
lines.append("=== END DOCS ===")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
def get_doc_content(self, filename: str) -> str:
|
||||
"""특정 문서 내용 반환."""
|
||||
path = self.docs_path / filename
|
||||
if path.exists():
|
||||
return path.read_text(encoding="utf-8", errors="ignore")
|
||||
return ""
|
||||
|
||||
def get_all_docs_content(self, max_total_bytes: int = 30000) -> str:
|
||||
"""전체 문서 내용 반환 (예산 내에서)."""
|
||||
files = sorted(self.docs_path.glob("*.md"))
|
||||
parts = ["=== PROJECT KNOWLEDGE BASE ===\n"]
|
||||
total = 0
|
||||
|
||||
for f in files:
|
||||
content = f.read_text(encoding="utf-8", errors="ignore")
|
||||
if total + len(content.encode("utf-8")) > max_total_bytes:
|
||||
parts.append(f"\n--- {f.name} (예산 초과, 생략) ---")
|
||||
continue
|
||||
parts.append(f"\n--- {f.name} ---\n{content}")
|
||||
total += len(content.encode("utf-8"))
|
||||
|
||||
parts.append("\n=== END KNOWLEDGE BASE ===")
|
||||
return "\n".join(parts)
|
||||
|
||||
def record_session(self, request: str, summary: dict, plan: dict = None) -> str:
|
||||
"""작업 세션을 기록."""
|
||||
now = datetime.now()
|
||||
filename = f"{now.strftime('%Y-%m-%d_%H%M%S')}.md"
|
||||
filepath = self.sessions_path / filename
|
||||
|
||||
lines = [
|
||||
f"# 작업 기록 — {now.strftime('%Y-%m-%d %H:%M')}",
|
||||
"",
|
||||
f"## 요청",
|
||||
f"{request}",
|
||||
"",
|
||||
]
|
||||
|
||||
if plan:
|
||||
lines.extend([
|
||||
f"## 계획",
|
||||
f"{plan.get('summary', str(plan)[:300])}",
|
||||
"",
|
||||
])
|
||||
|
||||
if isinstance(summary, dict):
|
||||
lines.extend([
|
||||
f"## 결과",
|
||||
f"{summary.get('summary', str(summary)[:300])}",
|
||||
"",
|
||||
])
|
||||
|
||||
changes = summary.get("changes", [])
|
||||
if changes:
|
||||
lines.append("## 변경 파일")
|
||||
for c in changes:
|
||||
lines.append(f"- `{c.get('file', '?')}` — {c.get('description', '')}")
|
||||
lines.append("")
|
||||
|
||||
warnings = summary.get("warnings", [])
|
||||
if warnings:
|
||||
lines.append("## 주의사항")
|
||||
for w in warnings:
|
||||
lines.append(f"- {w}")
|
||||
lines.append("")
|
||||
|
||||
filepath.write_text("\n".join(lines), encoding="utf-8")
|
||||
logger.info(f"세션 기록: {filepath}")
|
||||
return str(filepath)
|
||||
|
||||
def append_changelog(self, entry: str):
|
||||
"""Changelog에 항목 추가."""
|
||||
now = datetime.now().strftime("%Y-%m-%d %H:%M")
|
||||
line = f"\n- [{now}] {entry}"
|
||||
|
||||
if self.changelog_path.exists():
|
||||
content = self.changelog_path.read_text(encoding="utf-8")
|
||||
else:
|
||||
content = "# Changelog\n"
|
||||
|
||||
content += line
|
||||
self.changelog_path.write_text(content, encoding="utf-8")
|
||||
|
||||
def update_wiki(self, filename: str, content: str):
|
||||
"""Wiki 문서 생성/수정."""
|
||||
path = self.docs_path / filename
|
||||
path.write_text(content, encoding="utf-8")
|
||||
logger.info(f"Wiki 업데이트: {path}")
|
||||
Reference in New Issue
Block a user