refactor(protected-phrases): extract config and text normalization helpers

This commit is contained in:
2026-07-24 11:38:50 -04:00
parent 8e2ca365c2
commit 11b5d5db3c
8 changed files with 145 additions and 95 deletions
@@ -0,0 +1,15 @@
"""Protected phrase extraction, storage, and runtime matching."""
from python.ebook_search.protected_phrases.config.lib import (
get_bad_ends,
get_bad_starts,
get_ignored_phrases,
get_most_common_words,
)
__all__ = [
"get_bad_ends",
"get_bad_starts",
"get_ignored_phrases",
"get_most_common_words",
]
@@ -0,0 +1,54 @@
"""Protected phrase extraction, storage, and runtime matching."""
from __future__ import annotations
import logging
import tomllib
from functools import cache
from pathlib import Path
from python.ebook_search.protected_phrases.text_normalization import normalize_text
logger = logging.getLogger(__name__)
def _load_toml_string_set(path: Path, key: str) -> frozenset[str]:
"""Load and validate a TOML string list as a normalized immutable set."""
with path.open("rb") as file:
body = tomllib.load(file)
values = body.get(key)
if not isinstance(values, list) or not all(isinstance(item, str) for item in values):
msg = f"{path} must contain a {key!r} string list"
raise ValueError(msg)
return frozenset(normalize_text(value) for value in values if normalize_text(value))
@cache
def _get_phrase_config_dir() -> Path:
"""Return the directory containing phrase configuration files."""
return Path(__file__).resolve().parent
@cache
def get_ignored_phrases() -> frozenset[str]:
"""Return ignored phrase strings loaded from TOML."""
return _load_toml_string_set(_get_phrase_config_dir() / "ignored_phrases.toml", "phrases")
@cache
def get_bad_ends() -> frozenset[str]:
"""Return bad phrase-ending tokens loaded from TOML."""
return _load_toml_string_set(_get_phrase_config_dir() / "bad_ends.toml", "tokens")
@cache
def get_bad_starts() -> frozenset[str]:
"""Return bad phrase-starting tokens loaded from TOML."""
return _load_toml_string_set(_get_phrase_config_dir() / "bad_starts.toml", "tokens")
@cache
def get_most_common_words() -> frozenset[str]:
"""Return the most common English words loaded from TOML."""
return _load_toml_string_set(_get_phrase_config_dir() / "most_common_words.toml", "words")
+13 -95
View File
@@ -6,17 +6,26 @@ import importlib
import json
import logging
import re
import tomllib
from collections import defaultdict
from dataclasses import dataclass
from functools import cache
from pathlib import Path
from time import perf_counter
from typing import TYPE_CHECKING, Protocol
from sqlalchemy import and_, delete, func, or_, select
from python.ebook_search.llm_interface import request_chat_completion
from python.ebook_search.protected_phrases.config import (
get_bad_ends,
get_bad_starts,
get_ignored_phrases,
get_most_common_words,
)
from python.ebook_search.protected_phrases.text_normalization import (
NormalizedToken,
normalize_text,
tokenize,
tokenize_with_offsets,
)
from python.orm.richie import (
EbookCandidatePhrase,
EbookChunk,
@@ -36,7 +45,7 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
PHRASE_CONFIG_DIR = Path(__file__).resolve().parent
BAD_START_SCORE_PENALTY = 3.0
BAD_END_SCORE_PENALTY = 3.0
SOURCE_FIELDS = (
@@ -91,15 +100,6 @@ class YakeExtractorFactory(Protocol):
"""Create a YAKE keyword extractor."""
@dataclass(frozen=True, slots=True)
class NormalizedToken:
"""A normalized token plus its source character span."""
text: str
start_char: int
end_char: int
@dataclass(slots=True)
class PhraseCandidate:
"""A phrase candidate with merged extraction-source metadata."""
@@ -225,88 +225,6 @@ class PhraseRecalculationResult:
candidate_phrases: int
@cache
def get_ignored_phrases() -> frozenset[str]:
"""Return ignored phrase strings loaded from TOML."""
return load_toml_string_set(PHRASE_CONFIG_DIR / "ignored_phrases.toml", "phrases")
@cache
def get_bad_ends() -> frozenset[str]:
"""Return bad phrase-ending tokens loaded from TOML."""
return load_toml_string_set(PHRASE_CONFIG_DIR / "bad_ends.toml", "tokens")
@cache
def get_bad_starts() -> frozenset[str]:
"""Return bad phrase-starting tokens loaded from TOML."""
return load_toml_string_set(PHRASE_CONFIG_DIR / "bad_starts.toml", "tokens")
@cache
def get_most_common_words() -> frozenset[str]:
"""Return the most common English words loaded from TOML."""
return load_toml_string_set(PHRASE_CONFIG_DIR / "most_common_words.toml", "words")
def load_toml_string_set(path: Path, key: str) -> frozenset[str]:
"""Load and validate a TOML string list as a normalized immutable set."""
with path.open("rb") as file:
body = tomllib.load(file)
values = body.get(key)
if not isinstance(values, list) or not all(isinstance(item, str) for item in values):
msg = f"{path} must contain a {key!r} string list"
raise ValueError(msg)
return frozenset(normalize_text(value) for value in values if normalize_text(value))
def normalize_text(text: str) -> str:
"""Normalize text for phrase storage and lookup."""
return " ".join(token.text for token in tokenize_with_offsets(text))
def tokenize(text: str) -> list[str]:
"""Normalize and split text into phrase-detection tokens."""
return [token.text for token in tokenize_with_offsets(text)]
def tokenize_with_offsets(text: str) -> list[NormalizedToken]:
"""Normalize text into tokens while preserving original character offsets."""
tokens: list[NormalizedToken] = []
current: list[str] = []
start_char: int | None = None
for index, char in enumerate(text):
normalized = normalize_char(char)
if normalized == " ":
if current and start_char is not None:
tokens.append(NormalizedToken(text="".join(current), start_char=start_char, end_char=index))
current = []
start_char = None
continue
if start_char is None:
start_char = index
current.append(normalized)
if current and start_char is not None:
tokens.append(NormalizedToken(text="".join(current), start_char=start_char, end_char=len(text)))
return tokens
def normalize_char(char: str) -> str:
"""Normalize one character into a token character or a separator."""
if char in {"\u2019", "\u2018"}:
return "'"
if char in {"-", "\u2013", "\u2014"}:
return " "
lowered = char.lower()
if lowered in "abcdefghijklmnopqrstuvwxyz0123456789'":
return lowered
return " "
def strip_leading_articles(phrase_norm: str) -> str:
"""Remove one leading English article from a normalized phrase."""
tokens_ = phrase_norm.split()
@@ -0,0 +1,63 @@
"""Protected phrase extraction, storage, and runtime matching."""
from __future__ import annotations
import logging
from dataclasses import dataclass
logger = logging.getLogger(__name__)
@dataclass(frozen=True, slots=True)
class NormalizedToken:
"""A normalized token plus its source character span."""
text: str
start_char: int
end_char: int
def normalize_text(text: str) -> str:
"""Normalize text for phrase storage and lookup."""
return " ".join(token.text for token in tokenize_with_offsets(text))
def tokenize(text: str) -> list[str]:
"""Normalize and split text into phrase-detection tokens."""
return [token.text for token in tokenize_with_offsets(text)]
def tokenize_with_offsets(text: str) -> list[NormalizedToken]:
"""Normalize text into tokens while preserving original character offsets."""
tokens: list[NormalizedToken] = []
current: list[str] = []
start_char: int | None = None
for index, char in enumerate(text):
normalized = normalize_char(char)
if normalized == " ":
if current and start_char is not None:
tokens.append(NormalizedToken(text="".join(current), start_char=start_char, end_char=index))
current = []
start_char = None
continue
if start_char is None:
start_char = index
current.append(normalized)
if current and start_char is not None:
tokens.append(NormalizedToken(text="".join(current), start_char=start_char, end_char=len(text)))
return tokens
def normalize_char(char: str) -> str:
"""Normalize one character into a token character or a separator."""
if char in {"\u2019", "\u2018"}:
return "'"
if char in {"-", "\u2013", "\u2014"}:
return " "
lowered = char.lower()
if lowered in "abcdefghijklmnopqrstuvwxyz0123456789'":
return lowered
return " "