feat: 워크스페이스 이름충돌 감지 + 기본경로(VW_Proj) + 유령채널 자동정리
- workspace.py: find_by_name() 이름충돌, cleanup_orphans() 유령정리, 경로 자동생성 - config.py: WORKSPACE_BASE_DIR = VW_Proj - discord_bot.py: /workspace set path 선택적, 이름충돌 안내, on_ready 유령정리 - VW_Proj 폴더 생성
This commit is contained in:
@@ -119,11 +119,22 @@ async def on_ready():
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"슬래시 커맨드 동기화 실패: {e}")
|
logger.error(f"슬래시 커맨드 동기화 실패: {e}")
|
||||||
|
|
||||||
# 등록된 워크스페이스 채널 표시
|
# 유령 워크스페이스 정리
|
||||||
|
all_channel_ids = set()
|
||||||
|
for guild in bot.guilds:
|
||||||
|
for channel in guild.text_channels:
|
||||||
|
all_channel_ids.add(channel.id)
|
||||||
|
|
||||||
|
orphans = ws_manager.cleanup_orphans(all_channel_ids)
|
||||||
|
if orphans:
|
||||||
|
names = ", ".join(ws.name for ws in orphans)
|
||||||
|
logger.info(f"유령 워크스페이스 {len(orphans)}개 정리: {names}")
|
||||||
|
|
||||||
|
# 등록된 워크스페이스 표시
|
||||||
ws_list = ws_manager.list_all()
|
ws_list = ws_manager.list_all()
|
||||||
if ws_list:
|
if ws_list:
|
||||||
for ws in ws_list:
|
for ws in ws_list:
|
||||||
logger.info(f"워크스페이스 활성: #{ws.channel_id} → {ws.name} ({ws.path})")
|
logger.info(f"워크스페이스 활성: #{ws.channel_id} -> {ws.name} ({ws.path})")
|
||||||
else:
|
else:
|
||||||
logger.info("등록된 워크스페이스 없음 - /workspace set 으로 등록하세요")
|
logger.info("등록된 워크스페이스 없음 - /workspace set 으로 등록하세요")
|
||||||
|
|
||||||
@@ -370,13 +381,34 @@ workspace_group = app_commands.Group(name="workspace", description="워크스페
|
|||||||
|
|
||||||
|
|
||||||
@workspace_group.command(name="set", description="이 채널에 워크스페이스 등록")
|
@workspace_group.command(name="set", description="이 채널에 워크스페이스 등록")
|
||||||
@app_commands.describe(name="프로젝트 이름", path="로컬 프로젝트 경로")
|
@app_commands.describe(name="프로젝트 이름", path="로컬 경로 (미입력 시 VW_Proj/{name}에 자동 생성)")
|
||||||
async def workspace_set(interaction: discord.Interaction, name: str, path: str):
|
async def workspace_set(interaction: discord.Interaction, name: str, path: str = ""):
|
||||||
"""채널에 워크스페이스 등록."""
|
"""채널에 워크스페이스 등록."""
|
||||||
from pathlib import Path as P
|
from pathlib import Path as P
|
||||||
if not P(path).exists():
|
|
||||||
|
# 이름 충돌 검사
|
||||||
|
conflicts = ws_manager.find_by_name(name)
|
||||||
|
if conflicts:
|
||||||
|
old = conflicts[0]
|
||||||
|
embed = discord.Embed(
|
||||||
|
title="⚠️ 이름 충돌",
|
||||||
|
description=(
|
||||||
|
f"**{name}** 이름의 프로젝트가 이미 존재합니다.\n\n"
|
||||||
|
f"기존 등록: 채널 <#{old.channel_id}>\n"
|
||||||
|
f"경로: `{old.path}`\n\n"
|
||||||
|
f"**선택지:**\n"
|
||||||
|
f"1️⃣ 다른 이름으로 등록: `/workspace set name:새이름`\n"
|
||||||
|
f"2️⃣ 기존 프로젝트를 삭제 후 재등록: 기존 채널에서 `/workspace remove`"
|
||||||
|
),
|
||||||
|
color=0xF39C12,
|
||||||
|
)
|
||||||
|
await interaction.response.send_message(embed=embed)
|
||||||
|
return
|
||||||
|
|
||||||
|
# 부모 경로 검증 (명시적 경로 지정 시)
|
||||||
|
if path and not P(path).parent.exists():
|
||||||
await interaction.response.send_message(
|
await interaction.response.send_message(
|
||||||
f"❌ 경로가 존재하지 않습니다: `{path}`", ephemeral=True
|
f"❌ 부모 경로가 존재하지 않습니다: `{path}`", ephemeral=True
|
||||||
)
|
)
|
||||||
return
|
return
|
||||||
|
|
||||||
@@ -385,7 +417,7 @@ async def workspace_set(interaction: discord.Interaction, name: str, path: str):
|
|||||||
embed = discord.Embed(
|
embed = discord.Embed(
|
||||||
title="✅ 워크스페이스 등록 완료",
|
title="✅ 워크스페이스 등록 완료",
|
||||||
description=(
|
description=(
|
||||||
f"**{name}** → `{path}`\n\n"
|
f"**{name}** -> `{ws.path}`\n\n"
|
||||||
f"이 채널에서 봇과 대화할 수 있습니다.\n"
|
f"이 채널에서 봇과 대화할 수 있습니다.\n"
|
||||||
f"작업 실행을 위해 Git과 Vikunja를 설정해주세요:"
|
f"작업 실행을 위해 Git과 Vikunja를 설정해주세요:"
|
||||||
),
|
),
|
||||||
|
|||||||
@@ -44,3 +44,8 @@ GITEA_REPO: str = os.getenv("GITEA_REPO", "Variet/variet-agent")
|
|||||||
VIKUNJA_URL: str = os.getenv("VIKUNJA_URL", "https://plan.variet.net")
|
VIKUNJA_URL: str = os.getenv("VIKUNJA_URL", "https://plan.variet.net")
|
||||||
VIKUNJA_TOKEN: str = os.getenv("VIKUNJA_TOKEN", "")
|
VIKUNJA_TOKEN: str = os.getenv("VIKUNJA_TOKEN", "")
|
||||||
VIKUNJA_PROJECT_ID: int = int(os.getenv("VIKUNJA_PROJECT_ID", "7"))
|
VIKUNJA_PROJECT_ID: int = int(os.getenv("VIKUNJA_PROJECT_ID", "7"))
|
||||||
|
|
||||||
|
# === Workspace ===
|
||||||
|
WORKSPACE_BASE_DIR: str = os.getenv(
|
||||||
|
"WORKSPACE_BASE_DIR", r"c:\Users\Certes\Desktop\VW_Proj"
|
||||||
|
)
|
||||||
|
|||||||
@@ -7,6 +7,8 @@
|
|||||||
import json
|
import json
|
||||||
import logging
|
import logging
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
|
import config
|
||||||
from dataclasses import dataclass, field, asdict
|
from dataclasses import dataclass, field, asdict
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
@@ -125,12 +127,21 @@ class WorkspaceManager:
|
|||||||
encoding="utf-8",
|
encoding="utf-8",
|
||||||
)
|
)
|
||||||
|
|
||||||
def set_workspace(self, channel_id: int, name: str, path: str) -> Workspace:
|
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)
|
ws = Workspace(name=name, path=path, channel_id=channel_id)
|
||||||
self.workspaces[channel_id] = ws
|
self.workspaces[channel_id] = ws
|
||||||
self._save()
|
self._save()
|
||||||
logger.info(f"워크스페이스 설정: #{channel_id} → {name} ({path})")
|
logger.info(f"워크스페이스 설정: #{channel_id} -> {name} ({path})")
|
||||||
return ws
|
return ws
|
||||||
|
|
||||||
def get_workspace(self, channel_id: int) -> Optional[Workspace]:
|
def get_workspace(self, channel_id: int) -> Optional[Workspace]:
|
||||||
@@ -141,6 +152,30 @@ class WorkspaceManager:
|
|||||||
"""이 채널이 워크스페이스로 등록되어 있는지."""
|
"""이 채널이 워크스페이스로 등록되어 있는지."""
|
||||||
return channel_id in self.workspaces
|
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]:
|
||||||
|
"""존재하지 않는 채널의 워크스페이스 정리.
|
||||||
|
|
||||||
|
Returns: 제거된 워크스페이스 목록 (알림용)
|
||||||
|
"""
|
||||||
|
orphans = []
|
||||||
|
for ch_id, ws in list(self.workspaces.items()):
|
||||||
|
if ch_id not in valid_channel_ids:
|
||||||
|
orphans.append(ws)
|
||||||
|
del self.workspaces[ch_id]
|
||||||
|
logger.info(f"유령 워크스페이스 제거: {ws.name} (채널 {ch_id} 없음)")
|
||||||
|
|
||||||
|
if orphans:
|
||||||
|
self._save()
|
||||||
|
|
||||||
|
return orphans
|
||||||
|
|
||||||
def set_git(self, channel_id: int, url: str, token: str,
|
def set_git(self, channel_id: int, url: str, token: str,
|
||||||
repo: str = "", branch: str = "main") -> bool:
|
repo: str = "", branch: str = "main") -> bool:
|
||||||
"""워크스페이스에 Git 설정."""
|
"""워크스페이스에 Git 설정."""
|
||||||
|
|||||||
Reference in New Issue
Block a user