MCP Localization Project

#!/usr/bin/env python3 """ Context7 MCP 로컬화 - 기본 데이터 구조 및 유틸리티 라이브러리 문서 검색과 캐시를 위한 기반 클래스들 주요 기능: - 라이브러리 ID 해결 시스템 - 문서 검색 및 캐시 - 로컬 데이터베이스 관리 """ import asyncio import sqlite3 import hashlib import json import time import aiohttp import aiofiles from datetime import datetime, timedelta from typing import Dict, List, Optional, Any, Union, Tuple from dataclasses import dataclass from pathlib import Path import logging import re # 로깅 설정 logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @dataclass class LibraryInfo: """라이브러리 정보 데이터 클래스""" library_id: str # Context7 호환 ID (예: "/axios/axios") name: str # 라이브러리 이름 (예: "axios") description: str # 설명 version: str # 버전 trust_score: float = 7.0 # 신뢰도 점수 (0-10) code_snippets: int = 0 # 코드 스니펫 수 last_updated: str = "" # 마지막 업데이트 시간 @dataclass class LibraryDocs: """라이브러리 문서 데이터 클래스""" library_id: str content: str # 문서 내용 topic: str # 주제 tokens: int # 토큰 수 timestamp: float # 캐시 타임스탬프 source: str # 데이터 소스 (local/web) @dataclass class SearchResult: """검색 결과 데이터 클래스""" library_id: str relevance_score: float name_similarity: float description_match: float trust_score: float code_snippets: int class Context7Database: """Context7 로컬 데이터베이스 관리자""" def __init__(self, db_path: str = "context7_local.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 libraries ( library_id TEXT PRIMARY KEY, name TEXT NOT NULL, description TEXT, version TEXT, trust_score REAL DEFAULT 7.0, code_snippets INTEGER DEFAULT 0, last_updated TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) # 문서 캐시 테이블 cursor.execute(""" CREATE TABLE IF NOT EXISTS docs_cache ( id TEXT PRIMARY KEY, library_id TEXT, content TEXT, topic TEXT, tokens INTEGER, timestamp REAL, source TEXT, expires_at TIMESTAMP, FOREIGN KEY (library_id) REFERENCES libraries (library_id) ) """) # 검색 키워드 매핑 테이블 cursor.execute(""" CREATE TABLE IF NOT EXISTS keyword_mapping ( keyword TEXT, library_id TEXT, relevance_score REAL, PRIMARY KEY (keyword, library_id), FOREIGN KEY (library_id) REFERENCES libraries (library_id) ) """) # 인덱스 생성 cursor.execute("CREATE INDEX IF NOT EXISTS idx_library_name ON libraries(name)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_docs_library_id ON docs_cache(library_id)") cursor.execute("CREATE INDEX IF NOT EXISTS idx_keyword_mapping ON keyword_mapping(keyword)") conn.commit() logger.info("📁 Context7 데이터베이스 초기화 완료") except Exception as e: logger.error(f"❌ 데이터베이스 초기화 실패: {e}") raise async def add_library(self, library_info: LibraryInfo) -> bool: """라이브러리 정보 추가""" try: with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(""" INSERT OR REPLACE INTO libraries (library_id, name, description, version, trust_score, code_snippets, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?) """, ( library_info.library_id, library_info.name, library_info.description, library_info.version, library_info.trust_score, library_info.code_snippets, library_info.last_updated )) conn.commit() logger.info(f"✅ 라이브러리 추가: {library_info.library_id}") return True except Exception as e: logger.error(f"❌ 라이브러리 추가 실패: {e}") return False async def search_libraries(self, query: str, limit: int = 10) -> List[LibraryInfo]: """라이브러리 검색""" try: with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() # 다양한 검색 조건으로 검색 search_queries = [ # 이름 정확 매칭 ("SELECT * FROM libraries WHERE name = ? ORDER BY trust_score DESC", [query]), # 이름 부분 매칭 ("SELECT * FROM libraries WHERE name LIKE ? ORDER BY trust_score DESC", [f"%{query}%"]), # 설명 매칭 ("SELECT * FROM libraries WHERE description LIKE ? ORDER BY trust_score DESC", [f"%{query}%"]), # 키워드 매핑을 통한 검색 ("""SELECT l.* FROM libraries l JOIN keyword_mapping k ON l.library_id = k.library_id WHERE k.keyword LIKE ? ORDER BY k.relevance_score DESC, l.trust_score DESC""", [f"%{query}%"]) ] results = [] seen_ids = set() for sql_query, params in search_queries: cursor.execute(sql_query, params) rows = cursor.fetchall() for row in rows: if row[0] not in seen_ids and len(results) < limit: library_info = LibraryInfo( library_id=row[0], name=row[1], description=row[2] or "", version=row[3] or "", trust_score=row[4] or 7.0, code_snippets=row[5] or 0, last_updated=row[6] or "" ) results.append(library_info) seen_ids.add(row[0]) logger.info(f"🔍 라이브러리 검색 완료: '{query}' -> {len(results)}개 결과") return results except Exception as e: logger.error(f"❌ 라이브러리 검색 실패: {e}") return [] async def cache_docs(self, docs: LibraryDocs, ttl_hours: int = 24) -> bool: """문서 캐시 저장""" try: # 캐시 ID 생성 (라이브러리 ID + 주제 해시) cache_id = hashlib.md5(f"{docs.library_id}:{docs.topic}".encode()).hexdigest() expires_at = datetime.now() + timedelta(hours=ttl_hours) with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(""" INSERT OR REPLACE INTO docs_cache (id, library_id, content, topic, tokens, timestamp, source, expires_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( cache_id, docs.library_id, docs.content, docs.topic, docs.tokens, docs.timestamp, docs.source, expires_at )) conn.commit() logger.info(f"💾 문서 캐시 저장: {docs.library_id} (topic: {docs.topic})") return True except Exception as e: logger.error(f"❌ 문서 캐시 저장 실패: {e}") return False async def get_cached_docs(self, library_id: str, topic: str = "") -> Optional[LibraryDocs]: """캐시된 문서 조회""" try: cache_id = hashlib.md5(f"{library_id}:{topic}".encode()).hexdigest() with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(""" SELECT * FROM docs_cache WHERE id = ? AND expires_at > CURRENT_TIMESTAMP """, (cache_id,)) row = cursor.fetchone() if row: docs = LibraryDocs( library_id=row[1], content=row[2], topic=row[3], tokens=row[4], timestamp=row[5], source=row[6] ) logger.info(f"📄 캐시된 문서 조회: {library_id} (topic: {topic})") return docs except Exception as e: logger.error(f"❌ 캐시된 문서 조회 실패: {e}") return None class Context7Utils: """Context7 유틸리티 함수들""" @staticmethod def calculate_name_similarity(query: str, library_name: str) -> float: """이름 유사도 계산""" query_lower = query.lower() name_lower = library_name.lower() # 정확 매칭 if query_lower == name_lower: return 1.0 # 부분 매칭 if query_lower in name_lower or name_lower in query_lower: return 0.8 # 단어 매칭 query_words = set(query_lower.split()) name_words = set(name_lower.split()) if query_words and name_words: intersection = query_words.intersection(name_words) union = query_words.union(name_words) return len(intersection) / len(union) if union else 0.0 return 0.0 @staticmethod def calculate_description_match(query: str, description: str) -> float: """설명 매칭 점수 계산""" if not description: return 0.0 query_lower = query.lower() desc_lower = description.lower() # 키워드 매칭 query_words = set(query_lower.split()) desc_words = set(desc_lower.split()) if query_words and desc_words: intersection = query_words.intersection(desc_words) return len(intersection) / len(query_words) if query_words else 0.0 return 0.0 @staticmethod def calculate_relevance_score( name_similarity: float, description_match: float, trust_score: float, code_snippets: int ) -> float: """전체 관련성 점수 계산""" # 가중치 적용 weights = { 'name': 0.4, 'description': 0.3, 'trust': 0.2, 'snippets': 0.1 } # 코드 스니펫 점수 정규화 (0-1) snippet_score = min(code_snippets / 100.0, 1.0) trust_score_normalized = trust_score / 10.0 total_score = ( name_similarity * weights['name'] + description_match * weights['description'] + trust_score_normalized * weights['trust'] + snippet_score * weights['snippets'] ) return round(total_score, 3) @staticmethod def validate_library_id(library_id: str) -> bool: """라이브러리 ID 형식 검증""" # Context7 호환 ID 형식: /org/project 또는 /org/project/version pattern = r'^/[a-zA-Z0-9\-_]+/[a-zA-Z0-9\-_.]+(?:/[a-zA-Z0-9\-_.]+)?$' return bool(re.match(pattern, library_id)) @staticmethod def sanitize_content(content: str, max_tokens: int = 10000) -> str: """문서 내용 정리 및 토큰 제한""" # 기본적인 정리 content = content.strip() # 토큰 근사치 계산 (단어 수 * 1.3) estimated_tokens = len(content.split()) * 1.3 if estimated_tokens > max_tokens: # 내용을 줄임 words = content.split() target_words = int(max_tokens / 1.3) content = ' '.join(words[:target_words]) + "..." return content # 테스트 함수 async def test_context7_base(): """Context7 기본 기능 테스트""" print("🧪 Context7 기본 기능 테스트 시작...") # 데이터베이스 초기화 db = Context7Database("test_context7.db") # 테스트 라이브러리 추가 test_library = LibraryInfo( library_id="/axios/axios", name="axios", description="Promise based HTTP client for the browser and node.js", version="1.6.2", trust_score=9.5, code_snippets=150, last_updated="2025-01-01" ) success = await db.add_library(test_library) print(f"✅ 라이브러리 추가: {success}") # 검색 테스트 results = await db.search_libraries("axios") print(f"🔍 검색 결과: {len(results)}개") for result in results: print(f" - {result.library_id}: {result.name} (trust: {result.trust_score})") # 문서 캐시 테스트 test_docs = LibraryDocs( library_id="/axios/axios", content="# Axios Documentation\n\nAxios is a promise-based HTTP client...", topic="usage", tokens=50, timestamp=time.time(), source="local" ) cache_success = await db.cache_docs(test_docs) print(f"💾 문서 캐시: {cache_success}") # 캐시 조회 테스트 cached = await db.get_cached_docs("/axios/axios", "usage") print(f"📄 캐시 조회: {'성공' if cached else '실패'}") # 유틸리티 함수 테스트 similarity = Context7Utils.calculate_name_similarity("axios", "axios") print(f"🔢 이름 유사도: {similarity}") relevance = Context7Utils.calculate_relevance_score(1.0, 0.8, 9.5, 150) print(f"📊 관련성 점수: {relevance}") print("🎯 Context7 기본 기능 테스트 완료!") if __name__ == "__main__": asyncio.run(test_context7_base())