MCP Localization Project

#!/usr/bin/env python3 """ Session Manager - 편집 세션 관리 시스템 파일 편집 세션의 생성, 관리, TTL 처리를 담당 """ import os import time import uuid import hashlib import threading import logging from datetime import datetime, timedelta from typing import Dict, List, Optional from .models import EditSession, EditOperation, EditStatus logger = logging.getLogger(__name__) class SessionManager: """세션 관리 시스템""" def __init__(self, default_ttl_minutes: int = 5): self.sessions: Dict[str, EditSession] = {} self.default_ttl = timedelta(minutes=default_ttl_minutes) self.lock = threading.RLock() self._start_cleanup_thread() logger.info(f"SessionManager 초기화 완료 (TTL: {default_ttl_minutes}분)") def create_session(self, file_path: str, operations: List[EditOperation]) -> str: """새 편집 세션 생성""" session_id = str(uuid.uuid4()) with self.lock: session = EditSession( session_id=session_id, file_path=os.path.abspath(file_path), file_hash=self._calculate_file_hash(file_path), operations=operations, status=EditStatus.PENDING, created_at=datetime.now(), expires_at=datetime.now() + self.default_ttl ) self.sessions[session_id] = session logger.info(f"편집 세션 생성: {session_id[:8]} -> {file_path}") return session_id def get_session(self, session_id: str) -> Optional[EditSession]: """세션 조회""" with self.lock: session = self.sessions.get(session_id) if session and datetime.now() > session.expires_at: self._cleanup_session(session_id) return None return session def extend_session_ttl(self, session_id: str, minutes: int = 5) -> bool: """세션 TTL 연장""" with self.lock: if session_id in self.sessions: session = self.sessions[session_id] session.expires_at = datetime.now() + timedelta(minutes=minutes) logger.debug(f"세션 TTL 연장: {session_id[:8]} (+{minutes}분)") return True return False def update_session(self, session_id: str, **kwargs): """세션 정보 업데이트""" with self.lock: if session_id in self.sessions: session = self.sessions[session_id] for key, value in kwargs.items(): if hasattr(session, key): setattr(session, key, value) def cleanup_session(self, session_id: str): """세션 정리 (외부 호출용)""" with self.lock: self._cleanup_session(session_id) def _cleanup_session(self, session_id: str): """개별 세션 정리 (내부용)""" if session_id in self.sessions: session = self.sessions.pop(session_id) logger.info(f"편집 세션 정리: {session_id[:8]}") def _calculate_file_hash(self, file_path: str) -> str: """파일 해시 계산""" try: hasher = hashlib.sha256() with open(file_path, 'rb') as f: for chunk in iter(lambda: f.read(4096), b""): hasher.update(chunk) return hasher.hexdigest() except Exception as e: logger.error(f"파일 해시 계산 실패 {file_path}: {e}") return "" def _cleanup_expired_sessions(self): """만료된 세션 자동 정리""" now = datetime.now() expired_sessions = [] with self.lock: for session_id, session in self.sessions.items(): if now > session.expires_at and session.status != EditStatus.APPLIED: expired_sessions.append(session_id) for session_id in expired_sessions: with self.lock: self._cleanup_session(session_id) if expired_sessions: logger.info(f"만료된 세션 정리: {len(expired_sessions)}개") def _start_cleanup_thread(self): """백그라운드 정리 스레드""" def cleanup_worker(): while True: try: self._cleanup_expired_sessions() time.sleep(60) # 1분마다 정리 except Exception as e: logger.error(f"정리 스레드 오류: {e}") time.sleep(60) cleanup_thread = threading.Thread(target=cleanup_worker, daemon=True) cleanup_thread.start() logger.debug("세션 정리 스레드 시작")