MCP Localization Project

#!/usr/bin/env python3 """ Shrimp Task Manager - 기본 데이터 구조 및 핵심 클래스 로컬 AI의 사고 구조를 형성하는 핵심 데이터 모델들 이 모듈은 Shrimp Task Manager의 기반이 되는 모든 데이터 구조를 정의합니다. - Task: 개별 작업 단위 - TaskDependency: 작업 간 의존성 - ThoughtProcess: 사고 과정 추적 - ProjectRules: 프로젝트 규칙 - Evidence: 증거 기반 의사결정을 위한 데이터 """ from enum import Enum from dataclasses import dataclass, field from typing import Dict, List, Optional, Any, Set, Tuple, Union from datetime import datetime, timedelta import uuid import json import hashlib # ============================================================================= # 핵심 열거형 (Core Enums) # ============================================================================= class TaskStatus(Enum): """작업 상태 정의""" PENDING = "pending" # 대기 중 IN_PROGRESS = "in_progress" # 진행 중 COMPLETED = "completed" # 완료 BLOCKED = "blocked" # 차단됨 CANCELLED = "cancelled" # 취소됨 FAILED = "failed" # 실패 REVIEWING = "reviewing" # 검토 중 class TaskPriority(Enum): """작업 우선순위""" LOW = "low" MEDIUM = "medium" HIGH = "high" CRITICAL = "critical" URGENT = "urgent" class TaskCategory(Enum): """작업 카테고리""" DEVELOPMENT = "development" # 개발 ANALYSIS = "analysis" # 분석 TESTING = "testing" # 테스트 DOCUMENTATION = "documentation" # 문서화 BUGFIX = "bugfix" # 버그 수정 REFACTORING = "refactoring" # 리팩토링 RESEARCH = "research" # 연구 INFRASTRUCTURE = "infrastructure" # 인프라 PLANNING = "planning" # 계획 VALIDATION = "validation" # 검증 class FileRelationType(Enum): """파일 관련 타입""" TO_MODIFY = "TO_MODIFY" # 수정할 파일 REFERENCE = "REFERENCE" # 참고 파일 CREATE = "CREATE" # 생성할 파일 DEPENDENCY = "DEPENDENCY" # 의존성 파일 OTHER = "OTHER" # 기타 class ThoughtStage(Enum): """사고 단계""" PROBLEM_DEFINITION = "Problem Definition" INFORMATION_GATHERING = "Information Gathering" RESEARCH = "Research" ANALYSIS = "Analysis" SYNTHESIS = "Synthesis" CONCLUSION = "Conclusion" CRITICAL_QUESTIONING = "Critical Questioning" PLANNING = "Planning" class QualityScore(Enum): """품질 점수""" EXCELLENT = "excellent" # 90-100점 GOOD = "good" # 70-89점 ACCEPTABLE = "acceptable" # 50-69점 POOR = "poor" # 30-49점 CRITICAL = "critical" # 0-29점 # ============================================================================= # 핵심 데이터 클래스 (Core Data Classes) # ============================================================================= @dataclass class RelatedFile: """작업과 관련된 파일 정보""" path: str # 파일 경로 type: FileRelationType # 관련 타입 description: Optional[str] = None # 파일 설명 line_start: Optional[int] = None # 시작 라인 line_end: Optional[int] = None # 끝 라인 def to_dict(self) -> Dict[str, Any]: """딕셔너리로 변환""" return { 'path': self.path, 'type': self.type.value, 'description': self.description, 'line_start': self.line_start, 'line_end': self.line_end } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'RelatedFile': """딕셔너리에서 생성""" return cls( path=data['path'], type=FileRelationType(data['type']), description=data.get('description'), line_start=data.get('line_start'), line_end=data.get('line_end') ) @dataclass class Evidence: """증거 기반 의사결정을 위한 증거 데이터""" source: str # 증거 출처 content: str # 증거 내용 reliability: float # 신뢰도 (0.0-1.0) timestamp: datetime # 생성 시간 evidence_type: str # 증거 타입 (data, analysis, measurement 등) metadata: Dict[str, Any] = field(default_factory=dict) def __post_init__(self): """초기화 후 처리""" if not 0.0 <= self.reliability <= 1.0: raise ValueError("Reliability must be between 0.0 and 1.0") def to_dict(self) -> Dict[str, Any]: """딕셔너리로 변환""" return { 'source': self.source, 'content': self.content, 'reliability': self.reliability, 'timestamp': self.timestamp.isoformat(), 'evidence_type': self.evidence_type, 'metadata': self.metadata } @dataclass class TaskDependency: """작업 의존성 정보""" task_id: str # 의존하는 작업 ID dependency_type: str = "requires" # 의존성 타입 (requires, blocks, enables) description: Optional[str] = None # 의존성 설명 def to_dict(self) -> Dict[str, Any]: """딕셔너리로 변환""" return { 'task_id': self.task_id, 'dependency_type': self.dependency_type, 'description': self.description } @dataclass class Task: """핵심 작업 클래스 - 로컬 AI 사고의 기본 단위""" # 필수 필드 id: str # 고유 ID name: str # 작업 이름 description: str # 작업 설명 status: TaskStatus # 작업 상태 priority: TaskPriority # 우선순위 category: TaskCategory # 카테고리 # 시간 관련 created_at: datetime # 생성 시간 updated_at: datetime # 수정 시간 due_date: Optional[datetime] = None # 마감일 estimated_hours: Optional[float] = None # 예상 소요 시간 actual_hours: Optional[float] = None # 실제 소요 시간 # 구현 관련 implementation_guide: str = "" # 구현 가이드 verification_criteria: str = "" # 검증 기준 notes: str = "" # 추가 노트 # 관계 및 의존성 dependencies: List[TaskDependency] = field(default_factory=list) # 의존성 목록 related_files: List[RelatedFile] = field(default_factory=list) # 관련 파일 목록 subtasks: List[str] = field(default_factory=list) # 하위 작업 ID 목록 parent_task: Optional[str] = None # 상위 작업 ID # 실행 결과 및 검증 completion_percentage: float = 0.0 # 완료율 (0.0-100.0) quality_score: Optional[float] = None # 품질 점수 (0.0-100.0) verification_result: Optional[str] = None # 검증 결과 execution_log: List[str] = field(default_factory=list) # 실행 로그 # 증거 기반 의사결정 evidence: List[Evidence] = field(default_factory=list) # 관련 증거들 decision_rationale: str = "" # 결정 근거 # 메타데이터 tags: Set[str] = field(default_factory=set) # 태그 metadata: Dict[str, Any] = field(default_factory=dict) # 추가 메타데이터 def __post_init__(self): """초기화 후 검증""" if not 0.0 <= self.completion_percentage <= 100.0: raise ValueError("Completion percentage must be between 0.0 and 100.0") if self.quality_score is not None and not 0.0 <= self.quality_score <= 100.0: raise ValueError("Quality score must be between 0.0 and 100.0") def add_dependency(self, task_id: str, dependency_type: str = "requires", description: str = None): """의존성 추가""" dependency = TaskDependency(task_id, dependency_type, description) self.dependencies.append(dependency) self.updated_at = datetime.now() def add_related_file(self, path: str, file_type: FileRelationType, description: str = None, line_start: int = None, line_end: int = None): """관련 파일 추가""" related_file = RelatedFile(path, file_type, description, line_start, line_end) self.related_files.append(related_file) self.updated_at = datetime.now() def add_evidence(self, source: str, content: str, reliability: float, evidence_type: str = "data"): """증거 추가""" evidence = Evidence(source, content, reliability, datetime.now(), evidence_type) self.evidence.append(evidence) self.updated_at = datetime.now() def add_execution_log(self, message: str): """실행 로그 추가""" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") log_entry = f"[{timestamp}] {message}" self.execution_log.append(log_entry) self.updated_at = datetime.now() def update_status(self, new_status: TaskStatus, reason: str = ""): """상태 업데이트""" old_status = self.status self.status = new_status self.updated_at = datetime.now() log_message = f"Status changed from {old_status.value} to {new_status.value}" if reason: log_message += f" - {reason}" self.add_execution_log(log_message) def calculate_quality_score(self) -> float: """품질 점수 계산 (0-100)""" score = 0.0 # 기본 완성도 (40%) score += self.completion_percentage * 0.4 # 문서화 품질 (20%) doc_score = 0.0 if self.description.strip(): doc_score += 25.0 if self.implementation_guide.strip(): doc_score += 25.0 if self.verification_criteria.strip(): doc_score += 25.0 if self.related_files: doc_score += 25.0 score += doc_score * 0.2 # 의존성 관리 (15%) if self.dependencies: score += 15.0 # 증거 기반 의사결정 (15%) if self.evidence: avg_reliability = sum(e.reliability for e in self.evidence) / len(self.evidence) score += avg_reliability * 15.0 # 실행 로그 품질 (10%) if self.execution_log: score += min(len(self.execution_log) * 2.0, 10.0) return min(score, 100.0) def get_cache_key(self) -> str: """캐시 키 생성""" content = f"{self.id}:{self.name}:{self.updated_at.isoformat()}" return hashlib.md5(content.encode()).hexdigest() def to_dict(self) -> Dict[str, Any]: """딕셔너리로 변환""" return { 'id': self.id, 'name': self.name, 'description': self.description, 'status': self.status.value, 'priority': self.priority.value, 'category': self.category.value, 'created_at': self.created_at.isoformat(), 'updated_at': self.updated_at.isoformat(), 'due_date': self.due_date.isoformat() if self.due_date else None, 'estimated_hours': self.estimated_hours, 'actual_hours': self.actual_hours, 'implementation_guide': self.implementation_guide, 'verification_criteria': self.verification_criteria, 'notes': self.notes, 'dependencies': [dep.to_dict() for dep in self.dependencies], 'related_files': [file.to_dict() for file in self.related_files], 'subtasks': self.subtasks, 'parent_task': self.parent_task, 'completion_percentage': self.completion_percentage, 'quality_score': self.quality_score, 'verification_result': self.verification_result, 'execution_log': self.execution_log, 'evidence': [ev.to_dict() for ev in self.evidence], 'decision_rationale': self.decision_rationale, 'tags': list(self.tags), 'metadata': self.metadata } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'Task': """딕셔너리에서 생성""" # 의존성 복원 dependencies = [TaskDependency(**dep) for dep in data.get('dependencies', [])] # 관련 파일 복원 related_files = [RelatedFile.from_dict(file) for file in data.get('related_files', [])] # 증거 복원 evidence = [] for ev_data in data.get('evidence', []): ev_data['timestamp'] = datetime.fromisoformat(ev_data['timestamp']) evidence.append(Evidence(**ev_data)) return cls( id=data['id'], name=data['name'], description=data['description'], status=TaskStatus(data['status']), priority=TaskPriority(data['priority']), category=TaskCategory(data['category']), created_at=datetime.fromisoformat(data['created_at']), updated_at=datetime.fromisoformat(data['updated_at']), due_date=datetime.fromisoformat(data['due_date']) if data.get('due_date') else None, estimated_hours=data.get('estimated_hours'), actual_hours=data.get('actual_hours'), implementation_guide=data.get('implementation_guide', ''), verification_criteria=data.get('verification_criteria', ''), notes=data.get('notes', ''), dependencies=dependencies, related_files=related_files, subtasks=data.get('subtasks', []), parent_task=data.get('parent_task'), completion_percentage=data.get('completion_percentage', 0.0), quality_score=data.get('quality_score'), verification_result=data.get('verification_result'), execution_log=data.get('execution_log', []), evidence=evidence, decision_rationale=data.get('decision_rationale', ''), tags=set(data.get('tags', [])), metadata=data.get('metadata', {}) ) @dataclass class ThoughtProcess: """사고 과정 추적 클래스""" id: str # 고유 ID task_id: Optional[str] # 관련 작업 ID thought: str # 사고 내용 stage: ThoughtStage # 사고 단계 thought_number: int # 사고 번호 total_thoughts: int # 총 사고 수 next_thought_needed: bool # 다음 사고 필요 여부 # 메타인지 관련 assumptions_challenged: List[str] = field(default_factory=list) # 도전받은 가정들 axioms_used: List[str] = field(default_factory=list) # 사용된 공리들 tags: List[str] = field(default_factory=list) # 태그들 # 시간 관련 created_at: datetime = field(default_factory=datetime.now) def to_dict(self) -> Dict[str, Any]: """딕셔너리로 변환""" return { 'id': self.id, 'task_id': self.task_id, 'thought': self.thought, 'stage': self.stage.value, 'thought_number': self.thought_number, 'total_thoughts': self.total_thoughts, 'next_thought_needed': self.next_thought_needed, 'assumptions_challenged': self.assumptions_challenged, 'axioms_used': self.axioms_used, 'tags': self.tags, 'created_at': self.created_at.isoformat() } @dataclass class ProjectRules: """프로젝트 규칙 정의""" project_id: str # 프로젝트 ID coding_standards: Dict[str, Any] # 코딩 표준 architecture_patterns: List[str] # 아키텍처 패턴 quality_gates: Dict[str, float] # 품질 게이트 (최소 점수) file_naming_conventions: Dict[str, str] # 파일 명명 규칙 dependency_rules: List[str] # 의존성 규칙 testing_requirements: Dict[str, Any] # 테스트 요구사항 documentation_standards: Dict[str, str] # 문서화 표준 created_at: datetime = field(default_factory=datetime.now) updated_at: datetime = field(default_factory=datetime.now) def to_dict(self) -> Dict[str, Any]: """딕셔너리로 변환""" return { 'project_id': self.project_id, 'coding_standards': self.coding_standards, 'architecture_patterns': self.architecture_patterns, 'quality_gates': self.quality_gates, 'file_naming_conventions': self.file_naming_conventions, 'dependency_rules': self.dependency_rules, 'testing_requirements': self.testing_requirements, 'documentation_standards': self.documentation_standards, 'created_at': self.created_at.isoformat(), 'updated_at': self.updated_at.isoformat() } # ============================================================================= # 유틸리티 함수들 (Utility Functions) # ============================================================================= def generate_task_id() -> str: """새로운 작업 ID 생성""" return f"TASK-{datetime.now().strftime('%Y%m%d')}-{str(uuid.uuid4())[:8].upper()}" def generate_thought_id() -> str: """새로운 사고 과정 ID 생성""" return f"THOUGHT-{datetime.now().strftime('%Y%m%d')}-{str(uuid.uuid4())[:8].upper()}" def calculate_dependency_depth(task: Task, all_tasks: Dict[str, Task]) -> int: """의존성 깊이 계산 (순환 의존성 탐지 포함)""" visited = set() def _calculate_depth(current_task_id: str, path: Set[str]) -> int: if current_task_id in path: raise ValueError(f"Circular dependency detected: {' -> '.join(path)} -> {current_task_id}") if current_task_id in visited: return 0 visited.add(current_task_id) current_task = all_tasks.get(current_task_id) if not current_task or not current_task.dependencies: return 0 max_depth = 0 new_path = path | {current_task_id} for dep in current_task.dependencies: dep_depth = _calculate_depth(dep.task_id, new_path) max_depth = max(max_depth, dep_depth + 1) return max_depth return _calculate_depth(task.id, set()) def validate_task_data(task_data: Dict[str, Any]) -> List[str]: """작업 데이터 유효성 검증""" errors = [] # 필수 필드 검증 required_fields = ['name', 'description', 'status', 'priority', 'category'] for field in required_fields: if not task_data.get(field): errors.append(f"Required field '{field}' is missing or empty") # 상태 검증 if task_data.get('status') and task_data['status'] not in [s.value for s in TaskStatus]: errors.append(f"Invalid status: {task_data['status']}") # 우선순위 검증 if task_data.get('priority') and task_data['priority'] not in [p.value for p in TaskPriority]: errors.append(f"Invalid priority: {task_data['priority']}") # 완료율 검증 completion = task_data.get('completion_percentage', 0.0) if not 0.0 <= completion <= 100.0: errors.append("Completion percentage must be between 0.0 and 100.0") return errors if __name__ == "__main__": # 기본 테스트 print("🚀 Shrimp Task Manager 기본 데이터 구조 테스트") # 샘플 작업 생성 task = Task( id=generate_task_id(), name="테스트 작업", description="테스트용 작업입니다", status=TaskStatus.PENDING, priority=TaskPriority.HIGH, category=TaskCategory.DEVELOPMENT, created_at=datetime.now(), updated_at=datetime.now() ) # 증거 추가 task.add_evidence("test_source", "테스트 증거", 0.9, "analysis") # 관련 파일 추가 task.add_related_file("/test/file.py", FileRelationType.TO_MODIFY, "테스트 파일") # 품질 점수 계산 quality_score = task.calculate_quality_score() print(f"✅ 작업 생성 성공: {task.name}") print(f"✅ 작업 ID: {task.id}") print(f"✅ 품질 점수: {quality_score:.2f}") print(f"✅ 캐시 키: {task.get_cache_key()}") # 딕셔너리 변환 테스트 task_dict = task.to_dict() restored_task = Task.from_dict(task_dict) print(f"✅ 직렬화/역직렬화 성공: {restored_task.name == task.name}") # 사고 과정 테스트 thought = ThoughtProcess( id=generate_thought_id(), task_id=task.id, thought="이것은 테스트 사고입니다", stage=ThoughtStage.ANALYSIS, thought_number=1, total_thoughts=3, next_thought_needed=True ) print(f"✅ 사고 과정 생성 성공: {thought.id}") print("🎯 모든 기본 테스트 완료!")