"""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 import shutil 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: # 이미 orphan 접미사가 있으면 스킵 if "_orphan_" in ws.name: continue old_name = ws.name old_path = Path(ws.path) new_name = f"{ws.name}{suffix}" # 실제 폴더 이름 변경 if old_path.exists(): new_path = old_path.parent / f"{old_path.name}{suffix}" # 이미 같은 이름 폴더가 있으면 숫자 추가 counter = 1 while new_path.exists(): new_path = old_path.parent / f"{old_path.name}{suffix}_{counter}" counter += 1 try: old_path.rename(new_path) ws.path = str(new_path) logger.info(f"폴더 이름 변경: {old_path} -> {new_path}") except OSError as e: logger.warning(f"폴더 이름 변경 실패: {e} (JSON만 업데이트)") ws.name = new_name ws.channel_id = 0 renamed.append(ws) # 키를 orphan 키로 변경 del self.workspaces[ch_id] orphan_key = -abs(hash(new_name)) % (10**10) self.workspaces[orphan_key] = ws logger.info( f"유령 워크스페이스 보존: {old_name} -> {new_name} " f"(채널 {ch_id} 없음, 설정+폴더 유지)" ) if renamed: 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())