search.py•3.5 kB
"""
Core search functionality - Thread-safe wrapper for scsold.py
"""
import sys
import sqlite3
import numpy as np
from pathlib import Path
# Import the original SmartCodeSearch
sys.path.insert(0, str(Path(__file__).parent.parent.parent))
from scsold import SmartCodeSearch as OriginalSmartCodeSearch
from src.core.db_wrapper import ThreadSafeDB
class SmartCodeSearch(OriginalSmartCodeSearch):
"""Thread-safe version of SmartCodeSearch"""
def __init__(self, project_root=".", quiet=False):
# Initialize path and model from parent
self.root = Path(project_root)
self.db_path = self.root / ".claude-symbols" / "search.db"
self.db_path.parent.mkdir(exist_ok=True)
self.debug = "--debug" in sys.argv
# Initialize model using parent's logic
if not quiet:
print("🤖 Initializing AI model...")
# Import and load model
from sentence_transformers import SentenceTransformer
self.model = SentenceTransformer('all-MiniLM-L6-v2')
# Create thread-safe database wrapper
self.db = ThreadSafeDB(self.db_path)
# Initialize database schema
self._init_db()
# Copy language patterns from parent
super().__init__(project_root, quiet=True)
# Close the parent's connection if it exists
if hasattr(self, 'conn') and self.conn:
try:
self.conn.close()
except:
pass
@property
def conn(self):
"""Compatibility property that returns current thread's connection"""
return self.db.get_connection()
@conn.setter
def conn(self, value):
"""Ignore connection setting attempts from parent class"""
pass
def _init_db(self):
"""Initialize database schema using thread-safe wrapper"""
# Create table first
self.db.execute('''
CREATE TABLE IF NOT EXISTS symbols (
id INTEGER PRIMARY KEY AUTOINCREMENT,
file_path TEXT,
symbol_name TEXT,
symbol_type TEXT,
content TEXT,
embedding BLOB,
file_hash TEXT,
indexed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
''')
# Create indexes separately to ensure table exists
try:
self.db.execute('CREATE INDEX IF NOT EXISTS idx_file_path ON symbols(file_path)')
self.db.execute('CREATE INDEX IF NOT EXISTS idx_symbol_name ON symbols(symbol_name)')
self.db.execute('CREATE INDEX IF NOT EXISTS idx_symbol_type ON symbols(symbol_type)')
self.db.execute('CREATE INDEX IF NOT EXISTS idx_file_hash ON symbols(file_hash)')
except sqlite3.OperationalError:
# Indexes might already exist, that's fine
pass
def search(self, query: str, limit: int = 10):
"""Thread-safe search - delegates to parent with thread-safe connection"""
# Use parent's search but with our thread-safe connection
return super().search(query, limit)
def index_project(self, force: bool = False):
"""Thread-safe project indexing - delegates to parent"""
return super().index_project(force)
def __del__(self):
"""Cleanup database connections"""
if hasattr(self, 'db'):
self.db.close_all()
__all__ = ['SmartCodeSearch']