fix(search): typed text is a literal in every LIKE pattern

The feed's channel-name match and the live search's term fold built their
LIKE pattern from the typed text as it was, so `_` and `%` acted as
wildcards: `build_log` also matched `buildXlog`. The term fold also tested
a substring, so a term already present inside a longer word (`cat` in
`category`) was never folded into search_terms. It now compares whole
terms, escaped.

`like_escape` moves from the Plex query module to `app.utils`, beside
`to_tsquery_str`, and every pattern built from typed text goes through it.
This commit is contained in:
2026-09-19 01:33:32 +02:00
parent 59fd068711
commit 66dbd0af81
6 changed files with 116 additions and 18 deletions
+4 -3
View File
@@ -30,7 +30,7 @@ from app.models import (
Video,
VideoState,
)
from app.utils import iso, to_tsquery_str
from app.utils import iso, like_escape, to_tsquery_str
def _channel_url(channel_id: str, handle: str | None) -> str:
@@ -272,8 +272,9 @@ def filtered_query(
# `unaccent(channels.title) ILIKE '%q%'` straight into the per-video predicate defeated the
# search_vector GIN — the planner detoasted every non-matching tsvector to test the `@@`
# (~2.6s/search on the prod-sized set); channel_id.in_(...) keeps the GIN as a BitmapOr branch
# (PERF-12). Empty match -> false() so the OR collapses to the text match alone.
like = func.unaccent(f"%{q}%")
# (PERF-12). Empty match -> false() so the OR collapses to the text match alone. The query
# is a literal inside the pattern: `_` in `my_channel` is not a wildcard.
like = func.unaccent(f"%{like_escape(q)}%")
matching_channel_ids = (
db.execute(select(Channel.id).where(func.unaccent(Channel.title).ilike(like)))
.scalars()
-4
View File
@@ -298,10 +298,6 @@ class LibraryFilters:
}
def like_escape(s: str) -> str:
return s.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
# --- Watch-state scope (shared by the /library grid AND the /facets aggregation) ---------------
#
# The `show` scope (all | watched | in_progress | unwatched) filters the browse result by per-user
+2 -1
View File
@@ -24,7 +24,8 @@ from app.plex import sync as plex_sync
from app.plex.cards import movie_card, show_card
from app.plex.client import PlexClient, PlexError, PlexNotConfigured
from app.plex.lookups import by_rating_key
from app.plex.query import content_libs, like_escape, runtime_by_key, top_art_keys
from app.plex.query import content_libs, runtime_by_key, top_art_keys
from app.utils import like_escape
# The most-restrictive → least ordering used to pick a collection's effective content rating from its
# members (Plex shows one cert on the collection; we don't store it, so derive it).
+21 -10
View File
@@ -38,6 +38,7 @@ from app.models import (
VideoState,
)
from app.feedquery import feed_columns, serialize
from app.utils import like_escape
from app.sync.subscriptions import apply_channel_details
from app.sync.videos import insert_stubs, apply_video_details, parse_dt
from app.youtube import search_scrape
@@ -224,6 +225,25 @@ def _new_first(db: Session, user: User, ids: list[str]) -> list[str]:
return [i for i in ids if i not in have] + [i for i in ids if i in have]
def _fold_term(db: Session, video_ids: list[str], term: str) -> None:
"""Append `term` to each video's space-joined `search_terms`, unless it is already one of them.
"Already one of them" is a whole-term match, as a literal: a substring test skipped `cat` on a
video already found for `category`, and an unescaped one read the `_` in `my_channel` as a
wildcard — either way the term was never folded in, and local search lost that relevance."""
padded = func.concat(" ", func.coalesce(Video.search_terms, ""), " ")
db.execute(
update(Video)
.where(Video.id.in_(video_ids))
.where(padded.notilike(f"% {like_escape(term)} %"))
.values(
search_terms=func.trim(
func.coalesce(Video.search_terms, "").op("||")(" ").op("||")(term)
)
)
)
@router.get("/youtube")
def search_youtube(
q: str | None = None,
@@ -364,16 +384,7 @@ def search_youtube(
.values([{"user_id": user.id, "video_id": vid} for vid in ordered])
.on_conflict_do_nothing(index_elements=["user_id", "video_id"])
)
db.execute(
update(Video)
.where(Video.id.in_(ordered))
.where(func.coalesce(Video.search_terms, "").notilike(f"%{term}%"))
.values(
search_terms=func.trim(
func.coalesce(Video.search_terms, "").op("||")(" ").op("||")(term)
)
)
)
_fold_term(db, ordered, term)
db.commit()
ordered = _new_first(db, user, ordered)
+8
View File
@@ -73,6 +73,14 @@ def to_tsquery_str(q: str) -> str | None:
return " & ".join(tokens)
def like_escape(s: str) -> str:
"""`s` as a LITERAL inside a LIKE/ILIKE pattern: `%` and `_` are wildcards there, and `\\` is
Postgres's default escape character, so all three are escaped. Every pattern built from text a
person typed goes through this — unescaped, a search for `100%` or `my_channel` matches rows it
should not."""
return s.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
def normalize_description(text: str | None) -> str | None:
"""Strip whitespace-only lines from a CHANNEL description before storing. YouTube channel
descriptions pad their contact/link blocks with runs of blank lines; keeping them wastes storage,
+81
View File
@@ -0,0 +1,81 @@
"""What a person types is a literal inside a LIKE pattern, never a wildcard (DB-lane).
`%` and `_` are LIKE wildcards. The feed's channel-name match and the live search's term fold both
built their pattern from the typed text as it was, so `my_channel` also matched `myXchannel`, and a
term already present as part of a longer word was never folded in.
"""
from datetime import datetime, timezone
from sqlalchemy import select
from app.models import Channel, Subscription, Video
from app.routes.search import _fold_term
def _video(vid: str, ch: str, title: str, search_terms: str | None = None) -> Video:
return Video(
id=vid,
channel_id=ch,
published_at=datetime(2024, 1, 1, tzinfo=timezone.utc),
title=title,
is_short=False,
live_status="none",
via_search=False,
via_explore=False,
search_terms=search_terms,
)
def test_an_underscore_in_a_feed_search_matches_only_itself(db, as_user, make_user):
user = make_user()
db.add_all(
[
Channel(id="UCunder", title="build_log daily", subscriber_count=1),
Channel(id="UClookalike", title="buildXlog daily", subscriber_count=1),
]
)
db.add_all(
[
Subscription(user_id=user.id, channel_id=ch, priority=0, hidden=False)
for ch in ("UCunder", "UClookalike")
]
)
db.add_all(
[
_video("vidUnder", "UCunder", "Nothing in common"),
_video("vidLook", "UClookalike", "Nothing in common either"),
]
)
db.commit()
resp = as_user(user).get("/api/feed?q=build_log&show=all&limit=50")
assert resp.status_code == 200, resp.text
assert {it["id"] for it in resp.json()["items"]} == {"vidUnder"}
def test_a_term_is_folded_in_unless_it_is_already_a_whole_term(db):
db.add(Channel(id="UCfold", title="Fold", subscriber_count=1))
db.add_all(
[
_video("vidWord", "UCfold", "a", search_terms="category"),
_video("vidWild", "UCfold", "b", search_terms="catXdog"),
_video("vidHas", "UCfold", "c", search_terms="old cat_dog new"),
_video("vidNone", "UCfold", "d"),
]
)
db.commit()
_fold_term(db, ["vidWord", "vidWild", "vidHas", "vidNone"], "cat")
_fold_term(db, ["vidWild", "vidHas"], "cat_dog")
db.commit()
db.expire_all()
terms = dict(db.execute(select(Video.id, Video.search_terms)).all())
assert terms == {
"vidWord": "category cat",
"vidWild": "catXdog cat cat_dog",
"vidHas": "old cat_dog new cat",
"vidNone": "cat",
}