schema_adapted_search.py

#!/usr/bin/env python3
"""
Schema-adapted search function for the 83GB database
"""

import sqlite3
import logging

logger = logging.getLogger(__name__)

def adapted_search(db_path, query, limit=10):
    """
    Search function adapted for the actual database schema:
    - question_title (instead of title)
    - question_body, answer_body (instead of solution)
    - question_tags (instead of tags)
    - error_pattern
    """
    try:
        conn = sqlite3.connect(db_path, check_same_thread=False)
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()

        # Adapted query for actual schema
        search_query = """
        SELECT
            s.id,
            s.question_title as title,
            s.answer_body as solution,
            s.question_tags as tags,
            s.error_pattern,
            s.fix_command,
            s.answer_score,
            s.is_accepted,
            fts.rank,
            snippet(solutions_fts, 0, '<b>', '</b>', '...', 15) as title_snippet,
            snippet(solutions_fts, 2, '<b>', '</b>', '...', 20) as answer_snippet
        FROM solutions_fts fts
        JOIN solutions s ON s.id = fts.rowid
        WHERE solutions_fts MATCH ?
        ORDER BY rank
        LIMIT ?
        """

        cursor.execute(search_query, (query, limit))
        results = []

        for row in cursor.fetchall():
            result = {
                'id': row['id'],
                'title': row['title'] or 'No title',
                'solution': row['solution'] or 'No solution',
                'tags': row['tags'] or '',
                'error_pattern': row['error_pattern'] or '',
                'fix_command': row['fix_command'] or '',
                'answer_score': row['answer_score'] or 0,
                'is_accepted': row['is_accepted'] or False,
                'similarity': 1.0,  # Default similarity for now
                'confidence': row['answer_score'] / 100.0 if row['answer_score'] else 0.5,  # Convert score to confidence
                'avg_similarity': 0.8,  # Default avg similarity
                'search_strategy': 'fts_adapted',
                'title_snippet': row['title_snippet'] if 'title_snippet' in row.keys() else '',
                'answer_snippet': row['answer_snippet'] if 'answer_snippet' in row.keys() else '',
                'rank': row['rank'] if 'rank' in row.keys() else 0
            }
            results.append(result)

        conn.close()
        logger.info(f"Found {len(results)} results for query: {query}")
        return results

    except Exception as e:
        logger.error(f"Search failed: {e}")
        return []

if __name__ == "__main__":
    # Test the adapted search
    db_path = "/mnt/databases/SELF_HEALING_AGI.db"
    results = adapted_search(db_path, "python TypeError", 5)

    print(f"Found {len(results)} results:")
    for i, result in enumerate(results):
        print(f"{i+1}. {result['title'][:60]}...")
        if result['solution']:
            print(f"   Solution: {result['solution'][:100]}...")
        print()