Source code for search_names.engines
"""Correct, deterministic exact and fuzzy name matching."""
from collections import defaultdict
from typing import Any
import regex
MAX_RESULTS = 20
RESULT_FIELDS = ("uniqid", "n", "match", "start", "end")
FuzzyRule = tuple[int, int]
[docs]
class SearchEngine:
"""Compile name patterns once and search many documents."""
def __init__(
self,
names: list[tuple[str, str]],
fuzzy_rules: list[FuzzyRule] | None = None,
) -> None:
"""Validate rules and compile grouped name patterns.
Args:
names: ``(identifier, name)`` pairs.
fuzzy_rules: ``(minimum name length, maximum edit distance)`` pairs.
Raises:
ValueError: If a name or fuzzy rule is invalid.
"""
self.fuzzy_rules = self._validate_fuzzy_rules(fuzzy_rules or [])
grouped_names: dict[str, list[str]] = defaultdict(list)
for identifier, name in names:
normalized_name = name.strip()
if not normalized_name:
raise ValueError("search names cannot be blank")
grouped_names[str(identifier)].append(normalized_name)
self.patterns = [
(identifier, self._compile(names_for_identifier))
for identifier, names_for_identifier in grouped_names.items()
]
@staticmethod
def _validate_fuzzy_rules(rules: list[FuzzyRule]) -> list[FuzzyRule]:
validated: list[FuzzyRule] = []
for minimum_length, edit_distance in rules:
if minimum_length < 1:
raise ValueError("fuzzy-rule minimum lengths must be positive")
if edit_distance < 0:
raise ValueError("fuzzy-rule edit distances cannot be negative")
validated.append((minimum_length, edit_distance))
return sorted(validated)
def _edit_distance(self, name: str) -> int:
distance = 0
for minimum_length, candidate_distance in self.fuzzy_rules:
if len(name) >= minimum_length:
distance = candidate_distance
return distance
def _compile(self, names: list[str]) -> regex.Pattern[str]:
alternatives = []
for name in names:
escaped_name = regex.escape(name)
edit_distance = self._edit_distance(name)
if edit_distance:
alternatives.append(f"(?:{escaped_name}){{e<={edit_distance}}}")
else:
alternatives.append(escaped_name)
return regex.compile(
rf"(?<!\w)(?:{'|'.join(alternatives)})(?!\w)", flags=regex.IGNORECASE
)
[docs]
def search(self, text: str, max_results: int = MAX_RESULTS) -> list[dict[str, Any]]:
"""Return up to ``max_results`` matches in stable pattern order."""
if max_results < 1:
raise ValueError("max_results must be positive")
matches: list[dict[str, Any]] = []
for identifier, pattern in self.patterns:
for match in pattern.finditer(text):
matches.append(
{
"uniqid": identifier,
"n": 1,
"match": match.group(),
"start": match.start(),
"end": match.end(),
}
)
if len(matches) == max_results:
return matches
return matches