""" Baseline keyword parser for Text-to-SQL. Rule-based model (no ML) that maps natural language questions to SQL using keyword matching and fuzzy string matching. Exists solely to provide a comparison point for the fine-tuned T5 model. Expected accuracy: ~10-20% exact match — by design. """ import json import os import re from difflib import get_close_matches # ── Aggregation keyword map ────────────────────────────────────────────────── # Multi-word phrases are checked first (sorted by length, descending). AGG_KEYWORDS = { "average": "AVG", "mean": "AVG", "how many": "COUNT", "count": "COUNT", "number of": "COUNT", "maximum": "MAX", "highest": "MAX", "most": "MAX", "largest": "MAX", "greatest": "MAX", "biggest": "MAX", "minimum": "MIN", "lowest": "MIN", "least": "MIN", "smallest": "MIN", "fewest": "MIN", "total": "SUM", "sum": "SUM", } # Pre-sort so multi-word phrases are checked before single words _AGG_PHRASES_SORTED = sorted(AGG_KEYWORDS, key=len, reverse=True) # ── Internal helpers ───────────────────────────────────────────────────────── def _detect_aggregation(question: str) -> str | None: """Return SQL aggregation keyword (AVG, COUNT, …) or None for plain SELECT.""" q_lower = question.lower() for phrase in _AGG_PHRASES_SORTED: if phrase in q_lower: return AGG_KEYWORDS[phrase] return None def _fuzzy_match_column(text: str, columns: list[str], cutoff: float = 0.4) -> str | None: """Find the best-matching column name for a piece of text. Strategy: 1. Exact substring — if a full column name appears inside the text, pick the longest match (avoids "No." matching before "No. of goals"). 2. Fuzzy word-level — for each column, check if any of its words fuzzy-match any word in the text. """ text_lower = text.lower() # --- exact substring pass --- best_exact = None best_exact_len = 0 for col in columns: cl = col.lower() if cl in text_lower and len(cl) > best_exact_len: best_exact = col best_exact_len = len(cl) if best_exact: return best_exact # --- fuzzy word-level pass --- text_words = re.findall(r"[a-z0-9]+", text_lower) best_score = 0.0 best_col = None for col in columns: col_words = re.findall(r"[a-z0-9]+", col.lower()) for cw in col_words: if len(cw) <= 2: # skip tiny words like "no", "id" continue matches = get_close_matches(cw, text_words, n=1, cutoff=cutoff) if matches: # use ratio as a rough score from difflib import SequenceMatcher score = SequenceMatcher(None, cw, matches[0]).ratio() if score > best_score: best_score = score best_col = col return best_col def _extract_where_condition(question: str, columns: list[str], sel_column: str) -> tuple[str, str] | None: """Try to extract a (column, value) pair for a WHERE clause. Very crude — scans for column names other than the SELECT column appearing in the question, then grabs the surrounding text as the value. Falls back to fuzzy matching if no exact column substring is found. """ q_lower = question.lower() remaining = [c for c in columns if c != sel_column] # Pass 1: exact column-name substring match in the question for col in remaining: cl = col.lower() pos = q_lower.find(cl) if pos == -1: continue # Grab everything *after* the column name as candidate value after = question[pos + len(col):].strip() after = re.sub(r"^[\s,=:]+", "", after).rstrip("?.!,").strip() # Grab everything *before* the column name as alternate candidate before = question[:pos].strip() before = re.sub(r"[\s,=:]+$", "", before).rstrip("?.!,").strip() # Prefer "after" if non-empty, else try "before" tail if after: return (col, after) if before: words = before.split() value = " ".join(words[-4:]) if len(words) > 4 else before return (col, value) # Pass 2: pattern-based extraction # Look for common question structures that imply a WHERE condition: # "What is the X where/when/for/of/with Y" # "... that were/was/is/are Y" signal_patterns = [ r"(?:where|when|if|that)\s+(?:the\s+)?(.+?)(?:\?|$)", r"\bfor\s+(?:the\s+|a\s+)?(.+?)(?:\?|$)", r"\bwith\s+(?:the\s+|a\s+)?(.+?)(?:\?|$)", r"\bof\s+(?:the\s+|a\s+)?(.+?)(?:\?|$)", r"\bhad\s+(.+?)(?:\?|$)", ] for pattern in signal_patterns: m = re.search(pattern, q_lower) if not m: continue fragment = m.group(1).strip().rstrip("?.!,") if not fragment or len(fragment) < 2: continue # Try to fuzzy-match a column name within this fragment cond_col = _fuzzy_match_column(fragment, remaining, cutoff=0.5) if cond_col: # Value = everything in the fragment that isn't the column name cl = cond_col.lower() frag_lower = fragment.lower() col_pos = frag_lower.find(cl) if col_pos != -1: value = fragment[col_pos + len(cl):].strip() value = re.sub(r"^[\s,=:]+", "", value).rstrip("?.!,").strip() if value: return (cond_col, value) # Column matched but couldn't isolate value — skip continue # No column match in fragment — if there's a "=" or "is" pattern, # try to split on it is_split = re.split(r"\s+(?:is|are|was|were|=)\s+", fragment, maxsplit=1) if len(is_split) == 2: left, right = is_split left_col = _fuzzy_match_column(left, remaining, cutoff=0.5) if left_col and right.strip(): return (left_col, right.strip().rstrip("?.!,")) return None # ── Public API ─────────────────────────────────────────────────────────────── def baseline_predict(question: str, columns: list[str], types: list[str]) -> str: """Generate SQL from a question using keyword + fuzzy matching. Parameters ---------- question : str Natural-language question. columns : list[str] Column names from the table header. types : list[str] Column types (``"real"`` or ``"text"``). Currently unused by the baseline but kept in the signature so it mirrors the T5 prompt builder interface. Returns ------- str SQL string in WikiSQL ``human_readable`` format. """ # 1 — aggregation agg = _detect_aggregation(question) # 2 — SELECT column sel_col = _fuzzy_match_column(question, columns) or columns[0] # 3 — WHERE condition (best-effort) where = _extract_where_condition(question, columns, sel_col) # 4 — assemble if agg: sql = f"SELECT {agg} {sel_col} FROM table" else: sql = f"SELECT {sel_col} FROM table" if where: cond_col, cond_val = where sql += f" WHERE {cond_col} = {cond_val}" return sql # ── Manual sanity test ─────────────────────────────────────────────────────── if __name__ == "__main__": DATA_DIR = os.path.join(os.path.dirname(__file__), "..", "data", "wikisql_export") TEST_FILE = os.path.join(DATA_DIR, "test.json") # Load first 20 test samples samples = [] with open(TEST_FILE, "r", encoding="utf-8") as f: for i, line in enumerate(f): if i >= 20: break samples.append(json.loads(line.strip())) exact_matches = 0 total = len(samples) print("=" * 90) print("BASELINE KEYWORD PARSER — Manual Sanity Check (first 20 test samples)") print("=" * 90) for idx, sample in enumerate(samples): question = sample["question"] columns = sample["table"]["header"] types = sample["table"]["types"] expected = sample["sql"]["human_readable"] predicted = baseline_predict(question, columns, types) match = predicted.strip() == expected.strip() if match: exact_matches += 1 icon = "✅" if match else "❌" print(f"\n--- Sample {idx + 1} ---") print(f" Question: {question}") print(f" Expected: {expected}") print(f" Got: {predicted}") print(f" Match: {icon}") print("\n" + "=" * 90) print(f"Exact matches: {exact_matches}/{total} ({exact_matches/total:.0%})") print("=" * 90) print("\nNote: Low accuracy is EXPECTED. This is a dumb baseline.") print("The real model (T5) will be evaluated against these same metrics in Phase 4.")