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