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

This commit is contained in:
2026-07-12 17:49:52 -04:00
parent 77692b1414
commit 8d8c8cd3dd
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 json
import logging import logging
import re import re
import tomllib
from collections import defaultdict from collections import defaultdict
from dataclasses import dataclass from dataclasses import dataclass
from functools import cache
from pathlib import Path
from time import perf_counter from time import perf_counter
from typing import TYPE_CHECKING, Protocol from typing import TYPE_CHECKING, Protocol
from sqlalchemy import and_, delete, func, or_, select from sqlalchemy import and_, delete, func, or_, select
from python.ebook_search.llm_interface import request_chat_completion 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 ( from python.orm.richie import (
EbookCandidatePhrase, EbookCandidatePhrase,
EbookChunk, EbookChunk,
@@ -36,7 +45,7 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
PHRASE_CONFIG_DIR = Path(__file__).resolve().parent
BAD_START_SCORE_PENALTY = 3.0 BAD_START_SCORE_PENALTY = 3.0
BAD_END_SCORE_PENALTY = 3.0 BAD_END_SCORE_PENALTY = 3.0
SOURCE_FIELDS = ( SOURCE_FIELDS = (
@@ -91,15 +100,6 @@ class YakeExtractorFactory(Protocol):
"""Create a YAKE keyword extractor.""" """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) @dataclass(slots=True)
class PhraseCandidate: class PhraseCandidate:
"""A phrase candidate with merged extraction-source metadata.""" """A phrase candidate with merged extraction-source metadata."""
@@ -225,88 +225,6 @@ class PhraseRecalculationResult:
candidate_phrases: int 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: def strip_leading_articles(phrase_norm: str) -> str:
"""Remove one leading English article from a normalized phrase.""" """Remove one leading English article from a normalized phrase."""
tokens_ = phrase_norm.split() 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 " "