CD
1e8e8c324f
- cleanup_orphans: 삭제 -> _orphan_YYYYMMDD 접미사 추가 - Git/Vikunja 설정 등 모든 데이터 보존됨 - 이름 충돌 없이 같은 이름으로 새 채널 등록 가능
243 lines
8.1 KiB
Python
243 lines
8.1 KiB
Python
"""Workspace Manager — 채널별 프로젝트 설정 관리.
|
|
|
|
워크스페이스는 JSON 파일로 영속 저장됩니다.
|
|
각 디스코드 채널은 하나의 워크스페이스에 바인딩됩니다.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
|
|
import config
|
|
from dataclasses import dataclass, field, asdict
|
|
from typing import Optional
|
|
|
|
logger = logging.getLogger("variet.workspace")
|
|
|
|
# 워크스페이스 설정 파일 경로
|
|
WORKSPACES_FILE = Path(__file__).parent.parent / "workspaces.json"
|
|
|
|
|
|
@dataclass
|
|
class GitConfig:
|
|
url: str = ""
|
|
token: str = ""
|
|
repo: str = "" # "Owner/RepoName"
|
|
branch: str = "main"
|
|
|
|
@property
|
|
def is_configured(self) -> bool:
|
|
return bool(self.url and self.token)
|
|
|
|
|
|
@dataclass
|
|
class VikunjaConfig:
|
|
url: str = ""
|
|
token: str = ""
|
|
project_id: int = 0
|
|
|
|
@property
|
|
def is_configured(self) -> bool:
|
|
return bool(self.url and self.token and self.project_id)
|
|
|
|
|
|
@dataclass
|
|
class Workspace:
|
|
name: str
|
|
path: str
|
|
channel_id: int
|
|
git: GitConfig = field(default_factory=GitConfig)
|
|
vikunja: VikunjaConfig = field(default_factory=VikunjaConfig)
|
|
docs_path: str = "docs/wiki"
|
|
|
|
@property
|
|
def is_ready(self) -> bool:
|
|
"""Git + Vikunja 모두 설정되었는지."""
|
|
return self.git.is_configured and self.vikunja.is_configured
|
|
|
|
@property
|
|
def missing_configs(self) -> list[str]:
|
|
"""미설정 항목 목록."""
|
|
missing = []
|
|
if not self.git.is_configured:
|
|
missing.append("Git")
|
|
if not self.vikunja.is_configured:
|
|
missing.append("Vikunja")
|
|
return missing
|
|
|
|
@property
|
|
def abs_docs_path(self) -> Path:
|
|
return Path(self.path) / self.docs_path
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"name": self.name,
|
|
"path": self.path,
|
|
"channel_id": self.channel_id,
|
|
"git": asdict(self.git),
|
|
"vikunja": asdict(self.vikunja),
|
|
"docs_path": self.docs_path,
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: dict) -> "Workspace":
|
|
return cls(
|
|
name=data["name"],
|
|
path=data["path"],
|
|
channel_id=data["channel_id"],
|
|
git=GitConfig(**data.get("git", {})),
|
|
vikunja=VikunjaConfig(**data.get("vikunja", {})),
|
|
docs_path=data.get("docs_path", "docs/wiki"),
|
|
)
|
|
|
|
|
|
class WorkspaceManager:
|
|
"""워크스페이스 CRUD + 영속 저장."""
|
|
|
|
def __init__(self, config_path: Path = WORKSPACES_FILE):
|
|
self.config_path = config_path
|
|
self.workspaces: dict[int, Workspace] = {} # channel_id → Workspace
|
|
self._load()
|
|
|
|
def _load(self):
|
|
"""JSON에서 워크스페이스 로드."""
|
|
if not self.config_path.exists():
|
|
self.workspaces = {}
|
|
return
|
|
|
|
try:
|
|
data = json.loads(self.config_path.read_text(encoding="utf-8"))
|
|
self.workspaces = {
|
|
int(ch_id): Workspace.from_dict(ws)
|
|
for ch_id, ws in data.items()
|
|
}
|
|
logger.info(f"워크스페이스 {len(self.workspaces)}개 로드됨")
|
|
except Exception as e:
|
|
logger.error(f"워크스페이스 로드 실패: {e}")
|
|
self.workspaces = {}
|
|
|
|
def _save(self):
|
|
"""JSON으로 워크스페이스 저장."""
|
|
data = {
|
|
str(ch_id): ws.to_dict()
|
|
for ch_id, ws in self.workspaces.items()
|
|
}
|
|
self.config_path.write_text(
|
|
json.dumps(data, ensure_ascii=False, indent=2),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def set_workspace(self, channel_id: int, name: str, path: str = "") -> Workspace:
|
|
"""채널에 워크스페이스 등록.
|
|
|
|
path가 비어있으면 WORKSPACE_BASE_DIR/{name} 으로 자동 생성.
|
|
"""
|
|
if not path:
|
|
path = str(Path(config.WORKSPACE_BASE_DIR) / name)
|
|
|
|
# 경로 디렉토리 자동 생성
|
|
Path(path).mkdir(parents=True, exist_ok=True)
|
|
|
|
ws = Workspace(name=name, path=path, channel_id=channel_id)
|
|
self.workspaces[channel_id] = ws
|
|
self._save()
|
|
logger.info(f"워크스페이스 설정: #{channel_id} -> {name} ({path})")
|
|
return ws
|
|
|
|
def get_workspace(self, channel_id: int) -> Optional[Workspace]:
|
|
"""채널의 워크스페이스 조회."""
|
|
return self.workspaces.get(channel_id)
|
|
|
|
def is_workspace_channel(self, channel_id: int) -> bool:
|
|
"""이 채널이 워크스페이스로 등록되어 있는지."""
|
|
return channel_id in self.workspaces
|
|
|
|
def find_by_name(self, name: str) -> list[Workspace]:
|
|
"""이름으로 워크스페이스 검색 (채널 삭제 후 재생성 시 충돌 감지용)."""
|
|
return [
|
|
ws for ws in self.workspaces.values()
|
|
if ws.name.lower() == name.lower()
|
|
]
|
|
|
|
def cleanup_orphans(self, valid_channel_ids: set[int]) -> list[Workspace]:
|
|
"""존재하지 않는 채널의 워크스페이스를 이름 변경하여 보존.
|
|
|
|
삭제하지 않고 이름에 _orphan_날짜 접미사를 붙여 보존합니다.
|
|
기존 설정(Git, Vikunja)은 그대로 유지됩니다.
|
|
|
|
Returns: 이름 변경된 워크스페이스 목록 (알림용)
|
|
"""
|
|
from datetime import datetime
|
|
|
|
renamed = []
|
|
suffix = f"_orphan_{datetime.now().strftime('%Y%m%d')}"
|
|
|
|
for ch_id, ws in list(self.workspaces.items()):
|
|
if ch_id not in valid_channel_ids:
|
|
old_name = ws.name
|
|
# 이미 orphan 접미사가 있으면 스킵
|
|
if "_orphan_" in ws.name:
|
|
continue
|
|
ws.name = f"{ws.name}{suffix}"
|
|
ws.channel_id = 0 # 채널 바인딩 해제
|
|
renamed.append(ws)
|
|
logger.info(
|
|
f"유령 워크스페이스 보존: {old_name} -> {ws.name} "
|
|
f"(채널 {ch_id} 없음, 설정 유지)"
|
|
)
|
|
|
|
if renamed:
|
|
# 유령 항목은 channel_id=0 으로 이동
|
|
to_remove = [
|
|
ch_id for ch_id in self.workspaces
|
|
if ch_id not in valid_channel_ids and "_orphan_" not in self.workspaces[ch_id].name
|
|
]
|
|
# 이미 이름 변경됐으므로 키만 정리
|
|
for ch_id in list(self.workspaces.keys()):
|
|
if self.workspaces[ch_id].channel_id == 0 and ch_id != 0:
|
|
ws = self.workspaces.pop(ch_id)
|
|
# orphan은 이름을 키로 저장 (음수 해시)
|
|
orphan_key = -abs(hash(ws.name)) % (10**10)
|
|
self.workspaces[orphan_key] = ws
|
|
self._save()
|
|
|
|
return renamed
|
|
|
|
def set_git(self, channel_id: int, url: str, token: str,
|
|
repo: str = "", branch: str = "main") -> bool:
|
|
"""워크스페이스에 Git 설정."""
|
|
ws = self.workspaces.get(channel_id)
|
|
if not ws:
|
|
return False
|
|
|
|
ws.git = GitConfig(url=url, token=token, repo=repo, branch=branch)
|
|
self._save()
|
|
logger.info(f"Git 설정: {ws.name} → {url}")
|
|
return True
|
|
|
|
def set_vikunja(self, channel_id: int, url: str, token: str,
|
|
project_id: int) -> bool:
|
|
"""워크스페이스에 Vikunja 설정."""
|
|
ws = self.workspaces.get(channel_id)
|
|
if not ws:
|
|
return False
|
|
|
|
ws.vikunja = VikunjaConfig(url=url, token=token, project_id=project_id)
|
|
self._save()
|
|
logger.info(f"Vikunja 설정: {ws.name} → {url} (project {project_id})")
|
|
return True
|
|
|
|
def remove_workspace(self, channel_id: int) -> bool:
|
|
"""워크스페이스 제거."""
|
|
if channel_id in self.workspaces:
|
|
name = self.workspaces[channel_id].name
|
|
del self.workspaces[channel_id]
|
|
self._save()
|
|
logger.info(f"워크스페이스 제거: {name}")
|
|
return True
|
|
return False
|
|
|
|
def list_all(self) -> list[Workspace]:
|
|
"""전체 워크스페이스 목록."""
|
|
return list(self.workspaces.values())
|