#!/usr/bin/env python3 """todo_sync.py task-637.1 TDD GREEN Phase — utils/todo_sync.py 구현 (불칸) 기능: - sync_task_completion(task_id): todo.json의 sub_item을 완료 처리 - retroactive_sync(): task-timers.json에서 완료된 태스크를 일괄 동기화 - link_task_to_issue(task_id, issue_id, sub_item_title): 태스크↔이슈 연결 안전장치: - 수정 전 .bak 백업 - fcntl 파일 락 (동시 수정 방지) - done=True → done=False 불가 (단방향) - 변경 시 last_synced 필드 기록 """ import argparse import fcntl import json import sys from datetime import datetime, timezone from pathlib import Path # --------------------------------------------------------------------------- # 모듈 상수 (monkeypatch 대상) # --------------------------------------------------------------------------- WORKSPACE_ROOT = Path("/home/jay/workspace") TODO_FILE = WORKSPACE_ROOT / "memory" / "todo.json" TIMERS_FILE = WORKSPACE_ROOT / "memory" / "task-timers.json" # --------------------------------------------------------------------------- # 내부 헬퍼 # --------------------------------------------------------------------------- def _load_json(path: Path) -> dict: """JSON 파일을 로드한다. 파일이 없으면 빈 dict를 반환한다.""" if not path.exists(): return {} return json.loads(path.read_text(encoding="utf-8")) def _save_todo(data: dict, todo_file: Path) -> None: """todo.json을 .bak 백업 후 저장한다. fcntl 락으로 동시 수정을 방지한다.""" bak_file = todo_file.parent / (todo_file.name + ".bak") # 수정 전 백업 (기존 파일이 있을 때만) if todo_file.exists(): bak_file.write_bytes(todo_file.read_bytes()) # 파일 락 후 저장 todo_file.parent.mkdir(parents=True, exist_ok=True) with open(str(todo_file), "w", encoding="utf-8") as f: fcntl.flock(f, fcntl.LOCK_EX) try: json.dump(data, f, ensure_ascii=False, indent=2) finally: fcntl.flock(f, fcntl.LOCK_UN) def _now_iso() -> str: """현재 시각을 ISO 8601 형식으로 반환한다.""" return datetime.now(timezone.utc).astimezone().isoformat() # --------------------------------------------------------------------------- # 공개 API # --------------------------------------------------------------------------- def sync_task_completion(task_id: str) -> dict: """todo.json에서 task_id에 해당하는 sub_item을 완료 처리한다. 매칭 전략: 1. 직접 매칭: sub_item.task_id == task_id → done=True 2. linked_tasks 매칭: issue.linked_tasks에 task_id 포함 → 해당 issue의 미완료 sub_items 중 task_id=None인 첫 번째 항목을 done=True + task_id 설정 Returns: {"updated": int, "details": list[str]} """ data = _load_json(TODO_FILE) if not data: return {"updated": 0, "details": []} updated = 0 details: list[str] = [] changed = False issues: list[dict] = data.get("issues", []) for issue in issues: issue_id: str = issue.get("id", "") sub_items: list[dict] = issue.get("sub_items", []) # 전략 1: 직접 매칭 for sub in sub_items: if sub.get("task_id") == task_id: if sub.get("done") is True: # 이미 done=True → 스킵 (단방향) continue sub["done"] = True updated += 1 changed = True details.append(f"[직접매칭] issue={issue_id}, title={sub.get('title')}, task_id={task_id} → done=True") break # 같은 issue에서 중복 처리 방지 # 전략 2: linked_tasks 매칭 linked_tasks: list[str] = issue.get("linked_tasks", []) if task_id in linked_tasks: # 직접 매칭이 이미 처리한 경우는 건너뜀 direct_match_done = any(s.get("task_id") == task_id and s.get("done") is True for s in sub_items) direct_match_updated = any(s.get("task_id") == task_id for s in sub_items) if not direct_match_updated: # task_id=None이고 done=False인 첫 번째 sub_item을 처리 for sub in sub_items: if sub.get("task_id") is None and sub.get("done") is False: sub["done"] = True sub["task_id"] = task_id updated += 1 changed = True details.append( f"[linked_tasks매칭] issue={issue_id}, title={sub.get('title')}, task_id={task_id} → done=True" ) break # 모든 sub_items가 done이면 issue.status = "done" if sub_items and all(s.get("done") is True for s in sub_items): if issue.get("status") != "done": issue["status"] = "done" issue["completed_at"] = _now_iso() changed = True details.append(f"[이슈완료] issue={issue_id} → status=done, completed_at 기록") if changed: data["last_synced"] = _now_iso() _save_todo(data, TODO_FILE) return {"updated": updated, "details": details} def retroactive_sync() -> dict: """task-timers.json에서 status=completed인 모든 task_id를 일괄 동기화한다. Returns: {"total_synced": int, "details": list[str]} """ timers_data = _load_json(TIMERS_FILE) tasks: dict = timers_data.get("tasks", {}) total_synced = 0 all_details: list[str] = [] for task_id, task_info in tasks.items(): if task_info.get("status") == "completed": result = sync_task_completion(task_id) if result["updated"] > 0: total_synced += result["updated"] all_details.extend(result["details"]) return {"total_synced": total_synced, "details": all_details} def link_task_to_issue( task_id: str, issue_id: str, sub_item_title: str | None = None, ) -> dict: """issue의 linked_tasks에 task_id를 추가하고, 선택적으로 sub_item.task_id도 설정한다. Args: task_id: 연결할 태스크 ID issue_id: 연결 대상 이슈 ID sub_item_title: 지정 시 해당 title의 sub_item.task_id에도 task_id 설정 Returns: {"linked": bool, "message": str} """ data = _load_json(TODO_FILE) if not data: return {"linked": False, "message": f"todo.json을 로드할 수 없습니다."} issues: list[dict] = data.get("issues", []) target_issue: dict | None = None for issue in issues: if issue.get("id") == issue_id: target_issue = issue break if target_issue is None: return {"linked": False, "message": f"issue_id={issue_id}를 찾을 수 없습니다."} changed = False messages: list[str] = [] # linked_tasks에 task_id 추가 (중복 방지) linked_tasks: list[str] = target_issue.setdefault("linked_tasks", []) if task_id not in linked_tasks: linked_tasks.append(task_id) changed = True messages.append(f"linked_tasks에 {task_id} 추가") else: messages.append(f"linked_tasks에 {task_id} 이미 존재") # sub_item_title 지정 시 해당 sub_item.task_id 설정 if sub_item_title is not None: sub_items: list[dict] = target_issue.get("sub_items", []) for sub in sub_items: if sub.get("title") == sub_item_title: sub["task_id"] = task_id changed = True messages.append(f"sub_item '{sub_item_title}'.task_id = {task_id} 설정") break else: messages.append(f"WARNING: sub_item_title='{sub_item_title}'을 찾을 수 없습니다.") if changed: data["last_synced"] = _now_iso() _save_todo(data, TODO_FILE) return {"linked": True, "message": " / ".join(messages)} # --------------------------------------------------------------------------- # CLI 인터페이스 # --------------------------------------------------------------------------- def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="todo_sync — todo.json 동기화 유틸리티") subparsers = parser.add_subparsers(dest="command") # sync sync_parser = subparsers.add_parser("sync", help="특정 태스크를 완료 처리") sync_parser.add_argument("--task-id", required=True, help="태스크 ID (예: task-635.1)") # retroactive subparsers.add_parser("retroactive", help="completed 태스크 일괄 동기화") # link link_parser = subparsers.add_parser("link", help="태스크를 이슈에 연결") link_parser.add_argument("--task-id", required=True, help="태스크 ID") link_parser.add_argument("--issue-id", required=True, help="이슈 ID") link_parser.add_argument("--sub-item", default=None, help="연결할 sub_item title") return parser def main() -> None: parser = _build_parser() args = parser.parse_args() if args.command == "sync": result = sync_task_completion(args.task_id) print(json.dumps(result, ensure_ascii=False, indent=2)) elif args.command == "retroactive": result = retroactive_sync() print(json.dumps(result, ensure_ascii=False, indent=2)) elif args.command == "link": result = link_task_to_issue( task_id=args.task_id, issue_id=args.issue_id, sub_item_title=args.sub_item, ) print(json.dumps(result, ensure_ascii=False, indent=2)) else: parser.print_help() sys.exit(1) if __name__ == "__main__": main()