feat: Add text normalization for improved search functionality in CalibreOPDSClient

This commit is contained in:
JB
2025-11-29 06:02:04 -08:00
parent bce1419d92
commit 252be6d4b7
+31 -13
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import dataclasses import dataclasses
import html import html
import re import re
import unicodedata
from collections import deque from collections import deque
from dataclasses import dataclass, field from dataclasses import dataclass, field
from pathlib import PurePosixPath from pathlib import PurePosixPath
@@ -922,6 +923,14 @@ class CalibreOPDSClient:
return link.title.strip() return link.title.strip()
return "" return ""
@staticmethod
def _normalize_text(text: str) -> str:
if not text:
return ""
# Normalize unicode characters to their base form (e.g. é -> e)
normalized = unicodedata.normalize('NFKD', text).encode('ASCII', 'ignore').decode('utf-8')
return normalized.lower().strip()
@staticmethod @staticmethod
def _alphabet_letter_for_entry(entry: OPDSEntry, mode: str) -> Optional[str]: def _alphabet_letter_for_entry(entry: OPDSEntry, mode: str) -> Optional[str]:
source = CalibreOPDSClient._alphabet_source(entry, mode) source = CalibreOPDSClient._alphabet_source(entry, mode)
@@ -929,45 +938,54 @@ class CalibreOPDSClient:
return None return None
if mode == "title": if mode == "title":
source = CalibreOPDSClient._strip_leading_article(source) source = CalibreOPDSClient._strip_leading_article(source)
source = re.sub(r"^[^0-9A-Za-z]+", "", source)
if not source: # Normalize to handle accents (É -> E)
normalized_source = unicodedata.normalize('NFKD', source).encode('ASCII', 'ignore').decode('utf-8')
cleaned = re.sub(r"^[^0-9A-Za-z]+", "", normalized_source)
if not cleaned:
return "#" return "#"
initial = source[0] initial = cleaned[0]
if initial.isalpha(): if initial.isalpha():
return initial.upper() return initial.upper()
if initial.isdigit(): if initial.isdigit():
return "#" return "#"
normalized = initial.upper() return "#"
return normalized if normalized.isalpha() else "#"
@staticmethod @staticmethod
def _entry_matches_query(entry: OPDSEntry, tokens: List[str]) -> bool: def _entry_matches_query(entry: OPDSEntry, tokens: List[str]) -> bool:
if not tokens: if not tokens:
return True return True
search_fragments: List[str] = [] search_fragments: List[str] = []
if entry.title: if entry.title:
search_fragments.append(entry.title.strip().lower()) search_fragments.append(CalibreOPDSClient._normalize_text(entry.title))
if entry.series: if entry.series:
search_fragments.append(entry.series.strip().lower()) search_fragments.append(CalibreOPDSClient._normalize_text(entry.series))
for author in entry.authors: for author in entry.authors:
cleaned = (author or "").strip() cleaned = (author or "").strip()
if not cleaned: if not cleaned:
continue continue
lower_value = cleaned.lower() normalized_author = CalibreOPDSClient._normalize_text(cleaned)
search_fragments.append(lower_value) search_fragments.append(normalized_author)
for part in re.split(r"[\s,]+", lower_value): for part in re.split(r"[\s,]+", normalized_author):
part = part.strip() part = part.strip()
if part: if part:
search_fragments.append(part) search_fragments.append(part)
if not search_fragments: if not search_fragments:
return False return False
# Check if all tokens match at least one fragment
# Tokens are already normalized in _filter_feed_entries
return all(any(token in fragment for fragment in search_fragments) for token in tokens) return all(any(token in fragment for fragment in search_fragments) for token in tokens)
def _filter_feed_entries(self, feed: OPDSFeed, query: str) -> OPDSFeed: def _filter_feed_entries(self, feed: OPDSFeed, query: str) -> OPDSFeed:
normalized = (query or "").strip().lower() normalized_query = CalibreOPDSClient._normalize_text(query)
if not normalized: if not normalized_query:
return feed return feed
tokens = [token for token in re.split(r"\s+", normalized) if token] tokens = [token for token in re.split(r"\s+", normalized_query) if token]
if not tokens: if not tokens:
return feed return feed
filtered = [entry for entry in feed.entries if self._entry_matches_query(entry, tokens)] filtered = [entry for entry in feed.entries if self._entry_matches_query(entry, tokens)]