Spaces:
Sleeping
Sleeping
File size: 4,062 Bytes
b336134 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 | """
Spelling correction module using SymSpell (Symmetric Delete Spelling Correction).
Optimized for O(1) delete lookups to match column names and operation keywords.
"""
from __future__ import annotations
import re
def _get_deletes(word: str, max_edit_distance: int = 2) -> set[str]:
"""Generate deletes for a word up to a maximum edit distance."""
deletes = set()
queue = {word}
for _ in range(max_edit_distance):
next_queue = set()
for w in queue:
if len(w) > 1:
for i in range(len(w)):
del_w = w[:i] + w[i+1:]
deletes.add(del_w)
next_queue.add(del_w)
queue = next_queue
return deletes
def _lev_dist(s1: str, s2: str) -> int:
"""Calculate the Levenshtein distance between two strings."""
if len(s1) < len(s2):
return _lev_dist(s2, s1)
if len(s2) == 0:
return len(s1)
previous_row = list(range(len(s2) + 1))
for i, c1 in enumerate(s1):
current_row = [i + 1]
for j, c2 in enumerate(s2):
insertions = previous_row[j + 1] + 1
deletions = current_row[j] + 1
substitutions = previous_row[j] + (c1 != c2)
current_row.append(min(insertions, deletions, substitutions))
previous_row = current_row
return previous_row[-1]
class SymSpell:
"""A lightweight symmetric delete spelling corrector."""
def __init__(self, max_edit_distance: int = 2):
self.max_edit_distance = max_edit_distance
# Maps delete_item -> set of original_words
self.deletes: dict[str, set[str]] = {}
self.words: set[str] = set()
def add_word(self, word: str) -> None:
"""Index a word and its deletes for O(1) spelling lookup."""
word = word.lower().strip()
if not word or word in self.words:
return
self.words.add(word)
# Index word itself
if word not in self.deletes:
self.deletes[word] = set()
self.deletes[word].add(word)
# Index deletions
for delete in _get_deletes(word, self.max_edit_distance):
if delete not in self.deletes:
self.deletes[delete] = set()
self.deletes[delete].add(word)
def lookup(self, word: str) -> list[str]:
"""Find candidate words matching the spelling of input word."""
word = word.lower().strip()
if not word:
return []
if word in self.words:
return [word]
candidates: set[str] = set()
# 1. Direct delete match
if word in self.deletes:
candidates.update(self.deletes[word])
# 2. Deletes of word match
for delete in _get_deletes(word, self.max_edit_distance):
if delete in self.deletes:
candidates.update(self.deletes[delete])
if delete in self.words:
candidates.add(delete)
# Score and rank candidates by Levenshtein distance
scored = []
for cand in candidates:
dist = _lev_dist(word, cand)
if dist <= self.max_edit_distance:
scored.append((cand, dist))
# Sort by distance first, then length (longer words first for ties)
scored.sort(key=lambda x: (x[1], -len(x[0])))
return [c for c, _ in scored]
def correct_query(self, query: str) -> str:
"""Correct typos in words within the query string."""
# Find all alphabet-only words
words = re.findall(r'[a-zA-Z]+', query)
corrected = query
for w in words:
if len(w) > 2: # Only correct words longer than 2 characters
suggestions = self.lookup(w)
if suggestions:
# Match exact word boundary to prevent partial replacements
corrected = re.sub(rf'\b{re.escape(w)}\b', suggestions[0], corrected, flags=re.IGNORECASE)
return corrected
|