#!/usr/bin/env python3 """cleanup-workspace.py — 워크스페이스 파일 자동 정리 스크립트. 사용법: python3 cleanup-workspace.py --dry-run # 삭제 후보 출력 (기본) python3 cleanup-workspace.py --execute # 실제 삭제 python3 cleanup-workspace.py --report # 디스크 사용량 보고 """ import argparse, os, shutil from datetime import datetime from pathlib import Path from typing import Any PROTECTED_NAMES: list[str] = [ "CLAUDE.md", "MEMORY.md", ".env", ".env.keys", "organization-structure.json", "task-timers.json", "token-ledger.json", ] PROTECTED_DIRS: list[str] = [ "memory/reports", "memory/research", "memory/specs", "memory/meetings", "memory/plans", ] def _age(p: Path) -> float: return (datetime.now().timestamp() - p.stat().st_mtime) / 86400 def is_protected(path: Path, workspace: Path) -> bool: if path.name in PROTECTED_NAMES: return True try: rel = path.relative_to(workspace).as_posix() except ValueError: return False return any(rel == d or rel.startswith(d + "/") for d in PROTECTED_DIRS) def _size(p: Path) -> int: if p.is_dir(): return sum(f.stat().st_size for f in p.rglob("*") if f.is_file()) try: return p.stat().st_size except OSError: return 0 def find_tmp_candidates(workspace: Path) -> list[Path]: d = workspace / "tmp" return [f for f in d.rglob("*") if d.exists() and f.is_file() and _age(f) > 7] if d.exists() else [] def find_events_candidates(workspace: Path) -> list[Path]: d = workspace / "memory" / "events" if not d.exists(): return [] return [f for f in d.iterdir() if f.is_file() and (f.name.endswith(".done.clear") or f.name.endswith(".done.acked")) and _age(f) > 30] def find_dispatch_candidates(workspace: Path) -> list[Path]: d = workspace / "memory" / "tasks" if not d.exists(): return [] return [f for f in d.iterdir() if f.is_file() and f.name.startswith("dispatch-") and f.name.endswith(".md") and _age(f) > 90] def find_logs_candidates(workspace: Path) -> list[Path]: d = workspace / "logs" if not d.exists(): return [] return [f for f in d.rglob("*") if f.is_file() and not f.name.startswith("cleanup-") and _age(f) > 60] def find_backups_candidates(workspace: Path) -> list[Path]: d = workspace / "memory" / "backups" return [f for f in d.rglob("*") if d.exists() and f.is_file() and _age(f) > 90] if d.exists() else [] def find_team_task_candidates(workspace: Path) -> list[Path]: teams_dir = workspace / "teams" reports_dir = workspace / "memory" / "reports" if not teams_dir.exists(): return [] result: list[Path] = [] for team in teams_dir.glob("dev*"): for task_dir in team.glob("task-*"): if not task_dir.is_dir() or _age(task_dir) <= 30: continue if reports_dir.exists() and any(reports_dir.glob(f"*{task_dir.name}*")): result.append(task_dir) return result def find_cokacdir_candidates(cokacdir_ws: Path) -> list[Path]: if not cokacdir_ws.exists(): return [] return [f for proj in cokacdir_ws.iterdir() if proj.is_dir() for f in proj.rglob("*") if f.is_file() and _age(f) > 30] def _all_candidates(workspace: Path, cokacdir_ws: Path) -> dict[str, list[Path]]: return { "tmp": find_tmp_candidates(workspace), "events": find_events_candidates(workspace), "dispatch": find_dispatch_candidates(workspace), "logs": find_logs_candidates(workspace), "backups": find_backups_candidates(workspace), "team_tasks": find_team_task_candidates(workspace), "cokacdir": find_cokacdir_candidates(cokacdir_ws), } def run_cleanup(workspace: Path, dry_run: bool, cokacdir_ws: Path) -> None: cats = _all_candidates(workspace, cokacdir_ws) total = [p for items in cats.values() for p in items if not is_protected(p, workspace)] tag = "[DRY-RUN]" if dry_run else "[DELETE]" log_lines: list[str] = [] deleted = 0 for p in total: line = f"{tag} {p} ({_size(p):,} bytes)" print(line) log_lines.append(line) if not dry_run: try: shutil.rmtree(p) if p.is_dir() else p.unlink() deleted += 1 except OSError as e: err = f"[ERROR] {p}: {e}" print(err); log_lines.append(err) if not dry_run: _write_log(workspace, log_lines) print(f"\n삭제 완료: {deleted}/{len(total)} 항목") else: print(f"\n[DRY-RUN] 삭제 후보: {len(total)} 항목 (실제 삭제 없음)") def _write_log(workspace: Path, lines: list[str]) -> None: logs_dir = workspace / "logs" logs_dir.mkdir(parents=True, exist_ok=True) log_file = logs_dir / f"cleanup-{datetime.now().strftime('%Y-%m-%d')}.log" with open(log_file, "a", encoding="utf-8") as fh: fh.write(f"\n=== cleanup-workspace {datetime.now().isoformat()} ===\n") fh.write("\n".join(lines) + "\n") def build_report(workspace: Path, cokacdir_ws: Path) -> dict[str, Any]: cats = _all_candidates(workspace, cokacdir_ws) cat_info: dict[str, dict[str, Any]] = {} total_n, total_sz = 0, 0 for cat, items in cats.items(): filtered = [p for p in items if not is_protected(p, workspace)] sz = sum(_size(p) for p in filtered) cat_info[cat] = {"count": len(filtered), "size_bytes": sz} total_n += len(filtered); total_sz += sz return {"total_candidates": total_n, "total_size_bytes": total_sz, "categories": cat_info} def print_report(workspace: Path, cokacdir_ws: Path) -> None: data = build_report(workspace, cokacdir_ws) print("=" * 60) print(f" 워크스페이스 정리 보고서 | 기준: {workspace}") print("=" * 60) for cat, info in data["categories"].items(): print(f" {cat:<15} {info['count']:>5}개 {info['size_bytes']/1048576:>8.2f} MB") print("-" * 60) print(f" {'합계':<15} {data['total_candidates']:>5}개 {data['total_size_bytes']/1048576:>8.2f} MB 정리 가능") print("=" * 60) def main() -> None: parser = argparse.ArgumentParser(description="워크스페이스 파일 자동 정리") g = parser.add_mutually_exclusive_group() g.add_argument("--dry-run", action="store_true", default=True, help="삭제 후보 출력 (기본값)") g.add_argument("--execute", action="store_true", help="실제 삭제") g.add_argument("--report", action="store_true", help="디스크 사용량 보고") parser.add_argument("--workspace", type=Path, default=Path(os.environ.get("WORKSPACE_ROOT", "/home/jay/workspace"))) args = parser.parse_args() ws: Path = args.workspace cokac = Path.home() / ".cokacdir" / "workspace" if args.report: print_report(ws, cokac) elif args.execute: run_cleanup(workspace=ws, dry_run=False, cokacdir_ws=cokac) else: run_cleanup(workspace=ws, dry_run=True, cokacdir_ws=cokac) if __name__ == "__main__": main()