Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 34 additions & 8 deletions app/api/torznab/api.py
Original file line number Diff line number Diff line change
Expand Up @@ -65,9 +65,13 @@ def _default_languages_for_site(site: str) -> List[str]:

def _coerce_positive_int(value: object) -> Optional[int]:
"""
Parse an input value into a positive integer.
Coerce an arbitrary value into a positive integer.

Returns None when parsing fails or value is <= 0.
Parameters:
value (object): Value to convert to an integer.

Returns:
The parsed positive integer if conversion succeeds and is greater than zero, `None` otherwise.
"""
try:
parsed = int(value) # type: ignore[arg-type]
Expand All @@ -83,11 +87,15 @@ def _resolve_tvsearch_query_from_ids(
imdbid: Optional[str],
) -> Optional[str]:
"""
Resolve a canonical series title from Torznab identifier parameters.
Resolve a canonical TV series title from provided Torznab identifiers.

Uses SkyHook in this order:
1) direct show lookup by tvdbid
2) tvdb resolution via tmdb/imdb id search, then show lookup
If a positive `tvdbid` is supplied, the function looks up the show title for that ID.
If not, it attempts to resolve a `tvdbid` by querying SkyHook using `tmdbid` and/or `imdbid`,
then looks up the show title for the resolved `tvdbid`.

Returns:
title (str): The resolved show title when found.
None: If no title could be resolved.
"""
tvdb_id = _coerce_positive_int(tvdbid)
if tvdb_id is None:
Expand Down Expand Up @@ -150,9 +158,27 @@ def _try_mapped_special_probe(
special_map,
) -> tuple[bool, Optional[int], Optional[str], Optional[str], int, int, int, int]:
"""
Probe mapped AniWorld special coordinates using cache first, then live probe.
Probe availability and quality for an AniWorld special that maps to a different source episode, using cached availability when possible.

Returns availability tuple together with resolved source/alias coordinates.
Parameters:
tn_module: Provider module exposing `get_availability` and `probe_episode_quality` used to fetch cached availability or probe live quality.
session (Session): Database/session object used by `get_availability`.
slug (str): Show identifier used for probing.
lang (str): Language to probe (e.g., "German Dub").
site_found (str): Catalogue site name where the source episode is hosted.
special_map: Mapping object containing `source_season`, `source_episode`, `alias_season`, and `alias_episode` that describe the source coordinates and their alias.

Returns:
tuple: (
available (bool): `True` if the source episode is available, `False` otherwise,
height (Optional[int]): video height in pixels if known, otherwise `None`,
vcodec (Optional[str]): video codec identifier if known, otherwise `None`,
provider (Optional[str]): provider name that supplied the quality info if known, otherwise `None`,
source_season (int): season number of the mapped source episode,
source_episode (int): episode number of the mapped source episode,
alias_season (int): alias season number requested,
alias_episode (int): alias episode number requested
)
"""
source_season = special_map.source_season
source_episode = special_map.source_episode
Expand Down
50 changes: 35 additions & 15 deletions app/utils/title_resolver.py
Original file line number Diff line number Diff line change
Expand Up @@ -531,16 +531,23 @@ def _normalize_tokens(s: str) -> Set[str]:


def _normalize_alnum(s: str) -> str:
"""Lowercase and filter to alphanumeric characters."""
"""
Produce a lowercase string containing only the alphanumeric characters from the input.

Returns:
str: The input lowercased with all non-alphanumeric characters removed.
"""
return "".join(ch.lower() for ch in s if ch.isalnum())


def _match_tokens(s: str) -> Set[str]:
"""
Build query/title tokens for scoring while ignoring common stop words.
Produce a set of query/title tokens suitable for matching by removing common stopwords.

Falls back to the unfiltered token set when filtering would remove every
token, so very short titles/queries still remain matchable.
If removing stopwords would remove every token, returns the original normalized token set.

Returns:
Set[str]: Tokens lowercased and split on non-alphanumeric boundaries with common stopwords removed; if that yields an empty set, the unfiltered normalized tokens are returned.
"""
tokens = _normalize_tokens(s)
if not tokens:
Expand All @@ -553,10 +560,12 @@ def _score_title_candidate(
query_tokens: Set[str], query_norm: str, candidate_title: str
) -> float:
"""
Score how well a candidate title matches the query.
Assigns a numeric relevance score indicating how well `candidate_title` matches the query.

The score increases with token overlap and balance between precision and recall (F1), and is boosted for exact or substring matches and for higher normalized string similarity.

Uses token overlap, precision/recall, normalized string similarity and
exact/substring checks. Higher is better.
Returns:
float: Relevance score (higher is better). Returns 0.0 when there is no meaningful match.
"""
title_tokens = _match_tokens(candidate_title)
title_norm = _normalize_alnum(candidate_title)
Expand Down Expand Up @@ -599,11 +608,11 @@ def _score_title_candidate(


def _build_sto_search_terms(query: str) -> List[str]:
"""Build ordered S.to search variants from a raw query.
"""
Builds ordered search variants for S.to from a raw query.

Returns the raw query, a compact alphanumeric-only variant, and a dashed
variant when the compact form is numeric with length >= 3. Empty values are
filtered and the list is de-duplicated while preserving order.
Returns:
terms (List[str]): Ordered, de-duplicated list of non-empty search variants including the original trimmed query, a compact alphanumeric-only variant when different, and a dashed numeric variant when the compact form is all digits of length >= 3.
"""
raw = (query or "").strip()
if not raw:
Expand Down Expand Up @@ -675,19 +684,30 @@ def _search_sto_slug(query: str) -> Optional[str]:

def slug_from_query(q: str, site: Optional[str] = None) -> Optional[Tuple[str, str]]:
"""
Find the best-matching site and slug for a free-text query by comparing token overlap with titles and alternative titles.
Determine the best matching catalog site and slug for a free-text series query.

Parameters:
q (str): Free-text query used to match against series titles.
site (Optional[str]): If provided, restricts the search to this site; otherwise searches all configured sites.
q (str): Free-text query to match against site indexes and alternative titles.
site (Optional[str]): If provided, restrict search to this site; otherwise searches configured catalog sites and applies site-specific fallbacks.

Returns:
Optional[Tuple[str, str]]: `(site, slug)` of the best match, `None` if the query is empty or no match is found.
Optional[Tuple[str, str]]: Tuple `(site, slug)` for the best match, or `None` if the query is empty or no acceptable match is found.
"""
if not q:
return None

def _search_sites(sites: List[str]) -> Optional[Tuple[str, str]]:
"""
Finds the best matching (site, slug) for the current free-text query across the given sites.

Evaluates each site's cached index and alternative titles using the module's title-scoring logic, returning the site and slug with the highest score that meets the minimum match threshold. For sites without an index, attempts a search-only slug derivation; if no indexed match is found and "s.to" is among the sites, queries the S.to suggest API as a fallback.

Parameters:
sites (List[str]): Ordered list of site identifiers to search.

Returns:
Optional[Tuple[str, str]]: `(site, slug)` of the best match if a candidate meets the minimum score, otherwise `None`.
"""
q_tokens = _match_tokens(q)
q_norm = _normalize_alnum(q)
best_slug: Optional[str] = None
Expand Down
13 changes: 13 additions & 0 deletions tests/test_torznab.py
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,19 @@ class Rec:
seen = {"query": None}

def _slug_from_query(query, site=None):
"""
Record the provided query in the shared `seen` mapping and return a fixed (site, slug) pair.

Parameters:
query (str): The query string to record.
site (str | None): Optional site hint (unused by this stub).

Returns:
tuple: A two-element tuple (site, slug) where `site` is `"aniworld.to"` and `slug` is `"slug"`.

Side effects:
Mutates the `seen` mapping by setting `seen["query"] = query`.
"""
seen["query"] = query
return ("aniworld.to", "slug")

Expand Down