MCP Localization Project

#!/usr/bin/env python3 """ UniGUI-MCP 통합 GUI 자동화 시스템 - 기본 구조 모든 GUI 프로그램을 제어할 수 있는 확장 가능한 통합 자동화 시스템 주요 기능: - 확장 가능한 모듈 아키텍처 - 웹, 데스크톱, AI 비전 모듈 통합 - 통합 세션 관리 - 스텔스 자동화 지원 """ import asyncio import sqlite3 import json import time import uuid import logging from abc import ABC, abstractmethod from typing import Dict, List, Optional, Any, Union, Tuple from dataclasses import dataclass, asdict from pathlib import Path from datetime import datetime, timedelta from enum import Enum # 로깅 설정 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class ModuleType(Enum): """모듈 타입 정의""" WEB = "web" # 웹 브라우저 모듈 (Playwright) DESKTOP = "desktop" # 데스크톱 앱 모듈 (PyAutoGUI, OpenCV) VISION = "vision" # AI 비전 모듈 (화면 인식) HYBRID = "hybrid" # 하이브리드 모듈 class ActionType(Enum): """액션 타입 정의""" NAVIGATE = "navigate" CLICK = "click" FILL = "fill" SELECT = "select" HOVER = "hover" DRAG = "drag" KEY_PRESS = "key_press" SCREENSHOT = "screenshot" EVALUATE = "evaluate" HTTP_REQUEST = "http_request" WINDOW_CONTROL = "window_control" CUSTOM = "custom" @dataclass class ActionResult: """액션 실행 결과""" success: bool data: Any message: str action_type: ActionType module_type: ModuleType execution_time: float metadata: Dict[str, Any] @dataclass class CodegenSession: """코드 생성 세션 정보""" session_id: str output_path: str test_name_prefix: str include_comments: bool actions: List[Dict[str, Any]] created_at: datetime last_updated: datetime active: bool @dataclass class ElementInfo: """GUI 요소 정보""" selector: str element_type: str text: str attributes: Dict[str, str] position: Tuple[int, int] size: Tuple[int, int] module_type: ModuleType class UniGUIModule(ABC): """모든 GUI 모듈의 기본 추상 클래스""" def __init__(self, module_type: ModuleType): self.module_type = module_type self.active = False self.session_data = {} @abstractmethod async def initialize(self) -> bool: """모듈 초기화""" pass @abstractmethod async def cleanup(self): """모듈 정리""" pass @abstractmethod async def is_available(self) -> bool: """모듈 사용 가능 여부 확인""" pass @abstractmethod async def execute_action(self, action_type: ActionType, **kwargs) -> ActionResult: """액션 실행""" pass class UniGUIDatabase: """UniGUI 데이터베이스 관리자""" def __init__(self, db_path: str = "unigui_mcp.db"): """ 데이터베이스 초기화 Args: db_path: 데이터베이스 파일 경로 """ self.db_path = db_path self.init_database() def init_database(self): """데이터베이스 초기화 및 테이블 생성""" try: with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() # 코드 생성 세션 테이블 cursor.execute(""" CREATE TABLE IF NOT EXISTS codegen_sessions ( session_id TEXT PRIMARY KEY, output_path TEXT NOT NULL, test_name_prefix TEXT DEFAULT 'GeneratedTest', include_comments BOOLEAN DEFAULT TRUE, actions TEXT DEFAULT '[]', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP, active BOOLEAN DEFAULT TRUE ) """) # 액션 히스토리 테이블 cursor.execute(""" CREATE TABLE IF NOT EXISTS action_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT, action_type TEXT NOT NULL, module_type TEXT NOT NULL, parameters TEXT, result TEXT, success BOOLEAN, execution_time REAL, timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (session_id) REFERENCES codegen_sessions (session_id) ) """) # 요소 정보 캐시 테이블 cursor.execute(""" CREATE TABLE IF NOT EXISTS element_cache ( id INTEGER PRIMARY KEY AUTOINCREMENT, selector TEXT NOT NULL, element_type TEXT, text TEXT, attributes TEXT, position_x INTEGER, position_y INTEGER, width INTEGER, height INTEGER, module_type TEXT, page_url TEXT, timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) # 스크린샷 기록 테이블 cursor.execute(""" CREATE TABLE IF NOT EXISTS screenshots ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT, file_path TEXT NOT NULL, description TEXT, width INTEGER, height INTEGER, timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (session_id) REFERENCES codegen_sessions (session_id) ) """) # 인덱스 생성 cursor.execute("CREATE INDEX IF NOT EXISTS idx_session_id ON action_history(session_id)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_action_type ON action_history(action_type)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_element_selector ON element_cache(selector)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_screenshots_session ON screenshots(session_id)") conn.commit() logger.info("📁 UniGUI 데이터베이스 초기화 완료") except Exception as e: logger.error(f"❌ 데이터베이스 초기화 실패: {e}") raise async def create_codegen_session( self, output_path: str, test_name_prefix: str = "GeneratedTest", include_comments: bool = True ) -> str: """코드 생성 세션 생성""" try: session_id = str(uuid.uuid4()) with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(""" INSERT INTO codegen_sessions (session_id, output_path, test_name_prefix, include_comments, actions) VALUES (?, ?, ?, ?, ?) """, (session_id, output_path, test_name_prefix, include_comments, "[]")) conn.commit() logger.info(f"✅ 코드 생성 세션 생성: {session_id}") return session_id except Exception as e: logger.error(f"❌ 코드 생성 세션 생성 실패: {e}") raise async def get_codegen_session(self, session_id: str) -> Optional[CodegenSession]: """코드 생성 세션 조회""" try: with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(""" SELECT session_id, output_path, test_name_prefix, include_comments, actions, created_at, last_updated, active FROM codegen_sessions WHERE session_id = ? """, (session_id,)) row = cursor.fetchone() if row: return CodegenSession( session_id=row[0], output_path=row[1], test_name_prefix=row[2], include_comments=bool(row[3]), actions=json.loads(row[4]), created_at=datetime.fromisoformat(row[5]), last_updated=datetime.fromisoformat(row[6]), active=bool(row[7]) ) except Exception as e: logger.error(f"❌ 코드 생성 세션 조회 실패: {e}") return None async def update_session_actions(self, session_id: str, actions: List[Dict[str, Any]]) -> bool: """세션 액션 업데이트""" try: with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(""" UPDATE codegen_sessions SET actions = ?, last_updated = CURRENT_TIMESTAMP WHERE session_id = ? """, (json.dumps(actions), session_id)) conn.commit() return cursor.rowcount > 0 except Exception as e: logger.error(f"❌ 세션 액션 업데이트 실패: {e}") return False async def close_session(self, session_id: str) -> bool: """세션 종료""" try: with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(""" UPDATE codegen_sessions SET active = FALSE, last_updated = CURRENT_TIMESTAMP WHERE session_id = ? """, (session_id,)) conn.commit() return cursor.rowcount > 0 except Exception as e: logger.error(f"❌ 세션 종료 실패: {e}") return False async def clear_session(self, session_id: str) -> bool: """세션 데이터 정리""" try: with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() # 액션 히스토리 삭제 cursor.execute("DELETE FROM action_history WHERE session_id = ?", (session_id,)) # 스크린샷 기록 삭제 (실제 파일은 별도 정리 필요) cursor.execute("DELETE FROM screenshots WHERE session_id = ?", (session_id,)) # 세션 삭제 cursor.execute("DELETE FROM codegen_sessions WHERE session_id = ?", (session_id,)) conn.commit() return cursor.rowcount > 0 except Exception as e: logger.error(f"❌ 세션 정리 실패: {e}") return False async def log_action( self, session_id: str, action_type: ActionType, module_type: ModuleType, parameters: Dict[str, Any], result: ActionResult ) -> bool: """액션 히스토리 기록""" try: with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(""" INSERT INTO action_history (session_id, action_type, module_type, parameters, result, success, execution_time) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( session_id, action_type.value, module_type.value, json.dumps(parameters), json.dumps(asdict(result)), result.success, result.execution_time )) conn.commit() return True except Exception as e: logger.error(f"❌ 액션 히스토리 기록 실패: {e}") return False class UniGUIManager: """UniGUI 통합 관리자 클래스""" def __init__(self, db_path: str = "unigui_mcp.db"): """ UniGUI 매니저 초기화 Args: db_path: 데이터베이스 파일 경로 """ self.db = UniGUIDatabase(db_path) self.modules: Dict[str, UniGUIModule] = {} self.active_sessions: Dict[str, str] = {} # session_id -> module_name self.initialized = False async def initialize(self) -> bool: """매니저 초기화""" try: # 데이터베이스는 이미 생성자에서 초기화됨 self.initialized = True logger.info("✅ UniGUI 매니저 초기화 완료") return True except Exception as e: logger.error(f"❌ UniGUI 매니저 초기화 실패: {e}") return False async def register_module(self, name: str, module: UniGUIModule) -> bool: """ 모듈 등록 Args: name: 모듈 이름 module: 모듈 인스턴스 Returns: 등록 성공 여부 """ try: if await module.initialize(): self.modules[name] = module logger.info(f"✅ 모듈 '{name}' 등록 완료") return True else: logger.error(f"❌ 모듈 '{name}' 초기화 실패") return False except Exception as e: logger.error(f"❌ 모듈 '{name}' 등록 실패: {e}") return False async def unregister_module(self, name: str) -> bool: """ 모듈 등록 해제 Args: name: 모듈 이름 Returns: 해제 성공 여부 """ try: if name in self.modules: await self.modules[name].cleanup() del self.modules[name] logger.info(f"✅ 모듈 '{name}' 등록 해제 완료") return True return False except Exception as e: logger.error(f"❌ 모듈 '{name}' 등록 해제 실패: {e}") return False async def get_module(self, name: str) -> Optional[UniGUIModule]: """ 모듈 조회 Args: name: 모듈 이름 Returns: 모듈 인스턴스 또는 None """ return self.modules.get(name) async def list_modules(self) -> List[str]: """등록된 모듈 목록 조회""" return list(self.modules.keys()) async def execute_action( self, module_name: str, action_type: ActionType, session_id: Optional[str] = None, **kwargs ) -> ActionResult: """ 액션 실행 Args: module_name: 모듈 이름 action_type: 액션 타입 session_id: 세션 ID (선택사항) **kwargs: 액션 매개변수 Returns: 액션 실행 결과 """ try: if module_name not in self.modules: return ActionResult( success=False, data=None, message=f"모듈 '{module_name}'을 찾을 수 없습니다.", action_type=action_type, module_type=ModuleType.WEB, # 기본값 execution_time=0.0, metadata={} ) module = self.modules[module_name] # 액션 실행 start_time = time.time() result = await module.execute_action(action_type, **kwargs) result.execution_time = time.time() - start_time # 세션이 있으면 액션 기록 if session_id: await self.db.log_action( session_id, action_type, module.module_type, kwargs, result ) return result except Exception as e: logger.error(f"❌ 액션 실행 실패: {e}") return ActionResult( success=False, data=None, message=f"액션 실행 중 오류 발생: {e}", action_type=action_type, module_type=ModuleType.WEB, # 기본값 execution_time=0.0, metadata={"error": str(e)} ) async def create_session( self, output_path: str, test_name_prefix: str = "GeneratedTest", include_comments: bool = True ) -> str: """코드 생성 세션 생성""" return await self.db.create_codegen_session( output_path, test_name_prefix, include_comments ) async def get_session(self, session_id: str) -> Optional[CodegenSession]: """세션 정보 조회""" return await self.db.get_codegen_session(session_id) async def close_session(self, session_id: str) -> bool: """세션 종료""" result = await self.db.close_session(session_id) if session_id in self.active_sessions: del self.active_sessions[session_id] return result async def cleanup(self): """매니저 정리""" try: # 모든 모듈 정리 for name in list(self.modules.keys()): await self.unregister_module(name) # 활성 세션 정리 self.active_sessions.clear() logger.info("✅ UniGUI 매니저 정리 완료") except Exception as e: logger.error(f"❌ UniGUI 매니저 정리 실패: {e}") class UniGUIUtils: """UniGUI 유틸리티 함수들""" @staticmethod def generate_session_id() -> str: """고유한 세션 ID 생성""" return str(uuid.uuid4()) @staticmethod def validate_selector(selector: str, module_type: ModuleType) -> bool: """선택자 유효성 검증""" if not selector or not selector.strip(): return False if module_type == ModuleType.WEB: # CSS 선택자 기본 검증 return bool(selector.strip()) elif module_type == ModuleType.DESKTOP: # 데스크톱 요소 선택자 검증 (향후 구현) return bool(selector.strip()) return True @staticmethod def sanitize_filename(filename: str) -> str: """파일명 정리""" import re # 위험한 문자 제거 sanitized = re.sub(r'[<>:"/\\|?*]', '_', filename) return sanitized.strip() @staticmethod def calculate_element_similarity(elem1: ElementInfo, elem2: ElementInfo) -> float: """요소 유사도 계산""" similarity = 0.0 # 선택자 유사도 if elem1.selector == elem2.selector: similarity += 0.4 # 텍스트 유사도 if elem1.text and elem2.text: if elem1.text.lower() == elem2.text.lower(): similarity += 0.3 elif elem1.text.lower() in elem2.text.lower() or elem2.text.lower() in elem1.text.lower(): similarity += 0.15 # 위치 유사도 if elem1.position and elem2.position: distance = ((elem1.position[0] - elem2.position[0]) ** 2 + (elem1.position[1] - elem2.position[1]) ** 2) ** 0.5 if distance < 50: # 50픽셀 이내 similarity += 0.3 elif distance < 100: # 100픽셀 이내 similarity += 0.15 return min(similarity, 1.0) @staticmethod def format_action_code(action_type: ActionType, parameters: Dict[str, Any], include_comments: bool = True) -> str: """액션을 코드로 변환""" code_lines = [] if include_comments: code_lines.append(f"// {action_type.value.title()} action") if action_type == ActionType.NAVIGATE: url = parameters.get('url', '') code_lines.append(f"await page.goto('{url}');") elif action_type == ActionType.CLICK: selector = parameters.get('selector', '') code_lines.append(f"await page.click('{selector}');") elif action_type == ActionType.FILL: selector = parameters.get('selector', '') value = parameters.get('value', '') code_lines.append(f"await page.fill('{selector}', '{value}');") elif action_type == ActionType.SELECT: selector = parameters.get('selector', '') value = parameters.get('value', '') code_lines.append(f"await page.selectOption('{selector}', '{value}');") elif action_type == ActionType.HOVER: selector = parameters.get('selector', '') code_lines.append(f"await page.hover('{selector}');") elif action_type == ActionType.KEY_PRESS: key = parameters.get('key', '') selector = parameters.get('selector') if selector: code_lines.append(f"await page.press('{selector}', '{key}');") else: code_lines.append(f"await page.keyboard.press('{key}');") elif action_type == ActionType.SCREENSHOT: path = parameters.get('path', 'screenshot.png') code_lines.append(f"await page.screenshot({{ path: '{path}' }});") else: code_lines.append(f"// Custom action: {action_type.value}") code_lines.append(f"// Parameters: {json.dumps(parameters)}") return "\n".join(code_lines) # 테스트 함수 async def test_unigui_base(): """UniGUI 기본 구조 테스트""" print("🧪 UniGUI 기본 구조 테스트 시작...") # 데이터베이스 초기화 db = UniGUIDatabase("test_unigui.db") # 코드 생성 세션 생성 session_id = await db.create_codegen_session( output_path="/tmp/test_output", test_name_prefix="TestDemo", include_comments=True ) print(f"✅ 세션 생성: {session_id}") # 세션 조회 session = await db.get_codegen_session(session_id) print(f"📄 세션 정보: {session.test_name_prefix if session else 'None'}") # 액션 로깅 테스트 test_result = ActionResult( success=True, data={"url": "https://example.com"}, message="Navigation successful", action_type=ActionType.NAVIGATE, module_type=ModuleType.WEB, execution_time=1.5, metadata={} ) logged = await db.log_action( session_id, ActionType.NAVIGATE, ModuleType.WEB, {"url": "https://example.com"}, test_result ) print(f"📝 액션 로깅: {'성공' if logged else '실패'}") # 코드 생성 테스트 code = UniGUIUtils.format_action_code( ActionType.NAVIGATE, {"url": "https://example.com"}, include_comments=True ) print(f"💻 생성된 코드:\n{code}") # 세션 종료 closed = await db.close_session(session_id) print(f"🔚 세션 종료: {'성공' if closed else '실패'}") print("🎯 UniGUI 기본 구조 테스트 완료!") if __name__ == "__main__": asyncio.run(test_unigui_base())