src/validator.py

import difflib
import re
from typing import Tuple, List

_WORD_RE = re.compile(r'\b[a-zA-Z]+\b')
_STOP_WORDS = {'what', 'is', 'the', 'how', 'many', 'show', 'me', 'list', 'of', 'for', 'in', 'and', 'or', 'to', 'a'}
_AGG_KEYWORDS = {'total', 'count', 'average', 'max', 'min', 'highest', 'lowest', 'sum'}

def extract_keywords(text: str) -> List[str]:
    """
    Extracts words from a question, ignoring common stop words
    and punctuation to prepare for fuzzy matching.
    """
    words = _WORD_RE.findall(text.lower())
    return [w for w in words if w not in _STOP_WORDS]

def validate_question(question: str, columns: List[str], threshold: float = 0.3) -> Tuple[bool, float]:
    """
    Validates if a question is broadly related to the CSV columns.
    Returns (bool, float) -> (is_valid, confidence_score)
    """
    q_words = extract_keywords(question)
    if not q_words:
        # If it's a completely empty or stop-word only question, reject it.
        return False, 0.0
        
    # Build a vocabulary of column words
    col_words = []
    for col in columns:
        col_words.extend(extract_keywords(col))
    col_words_set = set(col_words)
    
    if not col_words_set:
        return True, 1.0 # Edge case: no text columns, allow it
        
    # Check overlap
    matches = 0
    for qw in q_words:
        # Check if this question word matches or is very similar to any column word
        close = difflib.get_close_matches(qw, col_words_set, n=1, cutoff=0.7)
        if close:
            matches += 1
            
    # Also check if there are aggregation keywords which are universally valid
    for qw in q_words:
        if qw in _AGG_KEYWORDS:
            matches += 1
            
    score = matches / len(q_words)
    
    # We are lenient. A score above threshold means it's vaguely related.
    return score >= threshold, score