video search: evaluate/rank pipeline + mock indexer + /downloads/search
Phase 2 of live search (indexer mocked, pipeline real):
- quality_eval.evaluate_release(parsed, profile, scope, want_season/episode, size_gb)
→ {accepted, score, rejected, tier, quality_label}. Sonarr-style: rejects junk
sources/codecs/3D, requires an ENABLED ladder tier, honours HDR-require + size caps,
and VALIDATES scope (episode wants SxxExx, season wants the whole-season PACK, show
wants a complete-series pack, with season/episode-number checks). Scores keepers by
resolution/source/codec-pref/HDR/audio/repack for ranking.
- mock_search.py: deterministic stand-in indexer returning scope-shaped raw hits
(the single swap-point for real slskd/Prowlarr).
- POST /downloads/search: mock → parse → evaluate → rank (accepted, score, seeders).
16 tests, isolation guard + ruff clean. UI wiring next.
This commit is contained in:
BoulderBadgeDad
parent
958a5cd0a4
commit
44b074772b
5 changed files with 351 additions and 1 deletions
|
|
@ -92,6 +92,53 @@ def register_routes(bp):
|
|||
profile = load(get_video_db())
|
||||
return jsonify(evaluate_owned(body.get("file"), profile))
|
||||
|
||||
@bp.route("/downloads/search", methods=["POST"])
|
||||
def video_downloads_search():
|
||||
"""Search a scope (movie / episode / season / series) and return candidates
|
||||
ranked + filtered against the stored quality profile. The indexer is mocked
|
||||
for now (core.video.mock_search) — the parse→evaluate→rank pipeline is real,
|
||||
so swapping in slskd/Prowlarr later needs no change here.
|
||||
Body: {scope, title, year?, season?, episode?, season_end?}."""
|
||||
from . import get_video_db
|
||||
from core.video.mock_search import mock_search
|
||||
from core.video.quality_eval import evaluate_release
|
||||
from core.video.quality_profile import load as load_profile
|
||||
from core.video.release_parse import parse_release
|
||||
|
||||
body = request.get_json(silent=True) or {}
|
||||
scope = str(body.get("scope") or "movie").lower()
|
||||
title = body.get("title") or ""
|
||||
|
||||
def _int(v):
|
||||
try:
|
||||
return int(v)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
want_season, want_episode = _int(body.get("season")), _int(body.get("episode"))
|
||||
profile = load_profile(get_video_db())
|
||||
raw = mock_search(scope, title, year=body.get("year"), season=want_season,
|
||||
episode=want_episode, season_end=_int(body.get("season_end")))
|
||||
|
||||
results = []
|
||||
for hit in raw:
|
||||
parsed = parse_release(hit.get("title"))
|
||||
size_gb = round((hit.get("size_bytes") or 0) / (1024 ** 3), 1)
|
||||
verdict = evaluate_release(parsed, profile, scope=scope, want_season=want_season,
|
||||
want_episode=want_episode, size_gb=size_gb)
|
||||
results.append({
|
||||
"title": hit.get("title"), "size_gb": size_gb, "seeders": hit.get("seeders", 0),
|
||||
"quality_label": verdict["quality_label"], "accepted": verdict["accepted"],
|
||||
"rejected": verdict["rejected"], "score": verdict["score"],
|
||||
"resolution": parsed.get("resolution"), "source": parsed.get("source"),
|
||||
"codec": parsed.get("codec"), "hdr": parsed.get("hdr"),
|
||||
"audio": parsed.get("audio"), "group": parsed.get("group"),
|
||||
"repack": parsed.get("repack") or parsed.get("proper"),
|
||||
})
|
||||
# accepted first, then best score, then most seeders.
|
||||
results.sort(key=lambda r: (r["accepted"], r["score"], r["seeders"]), reverse=True)
|
||||
return jsonify({"scope": scope, "results": results})
|
||||
|
||||
@bp.route("/downloads/youtube-quality", methods=["GET"])
|
||||
def video_youtube_quality():
|
||||
# Separate, smaller profile — YouTube is yt-dlp, not scene/p2p releases.
|
||||
|
|
|
|||
100
core/video/mock_search.py
Normal file
100
core/video/mock_search.py
Normal file
|
|
@ -0,0 +1,100 @@
|
|||
"""Mock indexer — a stand-in for a real slskd/Prowlarr search while the download
|
||||
engine is being built.
|
||||
|
||||
Given a scope (movie / episode / season / series) and a title, it returns a list of
|
||||
plausible raw 'indexer hits' ({title, size_bytes, seeders}). The real parse → evaluate
|
||||
→ rank pipeline (release_parse + quality_eval) runs on these exactly as it will on real
|
||||
hits, so the search UI is fully exercised. Deterministic (no RNG) so results are stable
|
||||
for tests and reloads. THIS is the single swap-point: replace ``mock_search`` with a
|
||||
real indexer client and nothing downstream changes.
|
||||
|
||||
Pure (no DB, no network). Isolated — imports only typing; the music side never imports it.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
_GB = 1024 ** 3
|
||||
|
||||
# (quality suffix, size in GB). Ordered best→worst-ish; the evaluator re-ranks anyway.
|
||||
_MOVIE = [
|
||||
("2160p.UHD.BluRay.REMUX.HDR.DV.TrueHD.Atmos-FraMeSToR", 58),
|
||||
("2160p.WEB-DL.DDP5.1.HDR.HEVC-FLUX", 19),
|
||||
("1080p.BluRay.x265.10bit.DTS-HD.MA.5.1-GROUP", 11),
|
||||
("1080p.WEB-DL.DDP5.1.H264-NTb", 7),
|
||||
("1080p.WEBRip.x264.AAC-RARBG", 4),
|
||||
("720p.HDTV.x264-GROUP", 2),
|
||||
("HDCAM.x264-CRUDE", 2),
|
||||
]
|
||||
_EPISODE = [
|
||||
("2160p.WEB-DL.DDP5.1.HDR.HEVC-FLUX", 5),
|
||||
("1080p.BluRay.x265-GROUP", 3),
|
||||
("1080p.WEB-DL.H264-NTb", 2),
|
||||
("720p.HDTV.x264-GROUP", 1),
|
||||
]
|
||||
_SEASON = [
|
||||
("2160p.WEB-DL.DDP5.1.HDR.HEVC-NTb", 48),
|
||||
("1080p.BluRay.x265.10bit-GROUP", 26),
|
||||
("1080p.WEB-DL.H264-FLUX", 18),
|
||||
("720p.HDTV.x264-GROUP", 9),
|
||||
]
|
||||
_SERIES = [
|
||||
("COMPLETE.1080p.BluRay.x265.10bit-GROUP", 120),
|
||||
("COMPLETE.1080p.WEB-DL.H264-NTb", 88),
|
||||
("COMPLETE.720p.WEB-DL.x264-GROUP", 40),
|
||||
]
|
||||
|
||||
|
||||
def _slug(title: Any) -> str:
|
||||
s = "".join(c if (c.isalnum() or c == " ") else " " for c in str(title or "").strip())
|
||||
return ".".join(p for p in s.split() if p) or "Unknown"
|
||||
|
||||
|
||||
def _seeders(slug: str, i: int) -> int:
|
||||
# Deterministic spread (no RNG): a stable hash of the title + index.
|
||||
h = 0
|
||||
for ch in slug:
|
||||
h = (h * 31 + ord(ch)) & 0xFFFFFFFF
|
||||
return ((h >> (i % 7)) % 240) + 3
|
||||
|
||||
|
||||
def _ss(n) -> str:
|
||||
try:
|
||||
return "S%02d" % int(n)
|
||||
except (TypeError, ValueError):
|
||||
return "S01"
|
||||
|
||||
|
||||
def mock_search(scope: str, title: Any, *, year: Any = None, season: Any = None,
|
||||
episode: Any = None, season_end: Any = None) -> list:
|
||||
"""Return plausible raw hits for a scope. Replace with a real indexer client later."""
|
||||
slug = _slug(title)
|
||||
scope = (scope or "movie").lower()
|
||||
if scope == "movie":
|
||||
prefix = slug + ("." + str(year) if year else "")
|
||||
rows = _MOVIE
|
||||
elif scope == "episode":
|
||||
prefix = slug + "." + _ss(season) + ("E%02d" % int(episode) if episode is not None else "E01")
|
||||
rows = _EPISODE
|
||||
elif scope == "season":
|
||||
prefix = slug + "." + _ss(season)
|
||||
rows = _SEASON
|
||||
elif scope == "series":
|
||||
end = season_end or 5
|
||||
prefix = slug + ".S01-" + _ss(end)
|
||||
rows = _SERIES
|
||||
else:
|
||||
return []
|
||||
|
||||
hits = []
|
||||
for i, (suffix, gb) in enumerate(rows):
|
||||
hits.append({
|
||||
"title": prefix + "." + suffix,
|
||||
"size_bytes": int(gb * _GB),
|
||||
"seeders": _seeders(slug, i),
|
||||
})
|
||||
return hits
|
||||
|
||||
|
||||
__all__ = ["mock_search"]
|
||||
|
|
@ -96,4 +96,113 @@ def evaluate_owned(file: Any, profile: Any) -> dict:
|
|||
}
|
||||
|
||||
|
||||
__all__ = ["resolution_rank", "resolution_label", "meets_cutoff", "evaluate_owned"]
|
||||
# ── release (search hit) evaluation ───────────────────────────────────────────
|
||||
# Map a parsed source → the tier-key prefix used in the quality profile's ladder.
|
||||
_SRC_TIER = {"remux": "remux", "bluray": "bluray", "web-dl": "web",
|
||||
"webrip": "webrip", "hdtv": "hdtv", "dvd": "dvd"}
|
||||
_RES_SCORE = {"2160p": 400, "1080p": 300, "720p": 200, "480p": 100}
|
||||
_SRC_SCORE = {"remux": 90, "bluray": 70, "web-dl": 55, "webrip": 40, "hdtv": 25, "dvd": 10}
|
||||
|
||||
|
||||
def tier_key(source, resolution) -> str:
|
||||
"""The quality-ladder key for a parsed (source, resolution), or '' if it isn't a
|
||||
ladder tier (junk sources like cam/screener have no tier)."""
|
||||
pre = _SRC_TIER.get(source)
|
||||
if not pre:
|
||||
return ""
|
||||
if pre == "dvd":
|
||||
return "dvd"
|
||||
return (pre + "-" + resolution) if resolution else ""
|
||||
|
||||
|
||||
def _scope_ok(parsed, scope, want_season, want_episode):
|
||||
"""Validate a hit actually matches what was searched (Sonarr-style): an episode
|
||||
search wants SxxExx, a season search wants the whole season PACK, a show search
|
||||
wants a complete-series pack."""
|
||||
season, episode = parsed.get("season"), parsed.get("episode")
|
||||
if scope == "movie":
|
||||
return (None, None) if season is None else (None, "This is a TV release, not the movie")
|
||||
if scope == "episode":
|
||||
if episode is None:
|
||||
return None, "Not a single episode"
|
||||
if want_season is not None and season != want_season:
|
||||
return None, "Wrong season"
|
||||
if want_episode is not None and episode != want_episode:
|
||||
return None, "Wrong episode"
|
||||
return None, None
|
||||
if scope == "season":
|
||||
if not parsed.get("is_season_pack"):
|
||||
return None, "Not a full-season pack"
|
||||
if want_season is not None and season != want_season:
|
||||
return None, "Wrong season"
|
||||
return None, None
|
||||
if scope == "series":
|
||||
return (None, None) if parsed.get("is_series_pack") else (None, "Not a complete-series pack")
|
||||
return None, None
|
||||
|
||||
|
||||
def evaluate_release(parsed, profile, *, scope="movie", want_season=None,
|
||||
want_episode=None, size_gb=None) -> dict:
|
||||
"""Judge a parsed search hit against the quality profile + the search scope.
|
||||
|
||||
Returns ``{accepted, score, rejected, tier, quality_label}`` — ``accepted`` False
|
||||
means it's filtered out (``rejected`` says why); ``score`` ranks the keepers."""
|
||||
parsed = parsed if isinstance(parsed, dict) else {}
|
||||
profile = profile if isinstance(profile, dict) else {}
|
||||
res, source = parsed.get("resolution"), parsed.get("source")
|
||||
rejects = profile.get("rejects") or []
|
||||
rejected = None
|
||||
|
||||
# 1) hard rejects — junk source / 3D / rejected codec
|
||||
if source in ("cam", "screener", "workprint") and source in rejects:
|
||||
rejected = source + " is on your reject list"
|
||||
fam = _codec_family(parsed.get("codec"))
|
||||
if not rejected and fam and fam in rejects:
|
||||
rejected = fam + " codec is on your reject list"
|
||||
if not rejected and parsed.get("three_d") and "3d" in rejects:
|
||||
rejected = "3D is on your reject list"
|
||||
|
||||
# 2) must be an enabled ladder tier
|
||||
tier = tier_key(source, res)
|
||||
if not rejected:
|
||||
enabled = {t.get("key") for t in (profile.get("tiers") or []) if t.get("enabled")}
|
||||
if not tier:
|
||||
rejected = "Unknown / unsupported quality"
|
||||
elif tier not in enabled:
|
||||
rejected = (resolution_label(res) or "This quality") + " " + (source or "") + " isn't in your enabled tiers"
|
||||
|
||||
# 3) HDR required (a real filter when set)
|
||||
if not rejected and profile.get("prefer_hdr") == "require" and not parsed.get("hdr"):
|
||||
rejected = "HDR required but this is SDR"
|
||||
|
||||
# 4) scope validation (episode vs season pack vs series pack)
|
||||
if not rejected:
|
||||
_, scope_reason = _scope_ok(parsed, scope, want_season, want_episode)
|
||||
if scope_reason:
|
||||
rejected = scope_reason
|
||||
|
||||
# 5) size guard (movie/episode only — packs are legitimately large)
|
||||
if not rejected and size_gb:
|
||||
cap = profile.get("max_movie_gb") if scope == "movie" else (profile.get("max_episode_gb") if scope == "episode" else 0)
|
||||
if cap and size_gb > cap:
|
||||
rejected = "Over your " + str(cap) + " GB size cap"
|
||||
|
||||
# score the keepers (higher = better)
|
||||
score = _RES_SCORE.get(res, 0) + _SRC_SCORE.get(source, 0)
|
||||
if profile.get("prefer_codec") not in (None, "any") and fam == profile.get("prefer_codec"):
|
||||
score += 40
|
||||
if parsed.get("hdr") and profile.get("prefer_hdr") in ("prefer", "require"):
|
||||
score += 30
|
||||
if parsed.get("audio") in ("atmos", "truehd", "dts-hd"):
|
||||
score += 15
|
||||
if profile.get("prefer_repack") and (parsed.get("repack") or parsed.get("proper")):
|
||||
score += 10
|
||||
|
||||
label = " · ".join([x for x in [resolution_label(res),
|
||||
(source or "").upper() if source else "", fam.upper() if fam else ""] if x])
|
||||
return {"accepted": rejected is None, "score": score, "rejected": rejected,
|
||||
"tier": tier, "quality_label": label}
|
||||
|
||||
|
||||
__all__ = ["resolution_rank", "resolution_label", "meets_cutoff", "evaluate_owned",
|
||||
"tier_key", "evaluate_release"]
|
||||
|
|
|
|||
|
|
@ -453,6 +453,31 @@ def test_quality_evaluate_endpoint_judges_owned_copy(tmp_path):
|
|||
videoapi._video_db = None
|
||||
|
||||
|
||||
def test_downloads_search_endpoint_ranks_and_filters(tmp_path):
|
||||
import api.video as videoapi
|
||||
from database.video_database import VideoDatabase
|
||||
|
||||
db = VideoDatabase(database_path=str(tmp_path / "video_library.db"))
|
||||
videoapi._video_db = db
|
||||
app = Flask(__name__)
|
||||
app.register_blueprint(videoapi.create_video_blueprint(), url_prefix="/api/video")
|
||||
client = app.test_client()
|
||||
try:
|
||||
d = client.post("/api/video/downloads/search",
|
||||
json={"scope": "movie", "title": "The Matrix", "year": 1999}).get_json()
|
||||
assert d["scope"] == "movie" and d["results"]
|
||||
# accepted hits sort ahead of rejected ones; the cam hit is rejected.
|
||||
accepted = [r for r in d["results"] if r["accepted"]]
|
||||
assert accepted and d["results"][0]["accepted"] is True
|
||||
assert any(r["rejected"] and "reject" in r["rejected"] for r in d["results"]) # the HDCAM
|
||||
# a season search returns season packs (validated against the profile/scope).
|
||||
s = client.post("/api/video/downloads/search",
|
||||
json={"scope": "season", "title": "The Wire", "season": 2}).get_json()
|
||||
assert s["results"] and all(".S02" in r["title"] for r in s["results"])
|
||||
finally:
|
||||
videoapi._video_db = None
|
||||
|
||||
|
||||
def test_slskd_config_shared_via_config_manager(tmp_path, monkeypatch):
|
||||
import api.video as videoapi
|
||||
import config.settings as cfg
|
||||
|
|
|
|||
69
tests/test_video_search_pipeline.py
Normal file
69
tests/test_video_search_pipeline.py
Normal file
|
|
@ -0,0 +1,69 @@
|
|||
"""Search pipeline — evaluate_release (accept/reject/score vs profile + scope) and
|
||||
the mock indexer. Isolated from music."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from core.video.mock_search import mock_search
|
||||
from core.video.quality_eval import evaluate_release, tier_key
|
||||
from core.video.quality_profile import default_profile
|
||||
from core.video.release_parse import parse_release
|
||||
|
||||
|
||||
def test_tier_key_mapping():
|
||||
assert tier_key("web-dl", "1080p") == "web-1080p"
|
||||
assert tier_key("remux", "2160p") == "remux-2160p"
|
||||
assert tier_key("dvd", None) == "dvd"
|
||||
assert tier_key("cam", "1080p") == "" # junk has no ladder tier
|
||||
|
||||
|
||||
def test_evaluate_accepts_an_enabled_tier():
|
||||
p = default_profile() # bluray-1080p etc. enabled, cutoff 1080p
|
||||
v = evaluate_release(parse_release("Movie 2020 1080p BluRay x265-GRP"), p, scope="movie")
|
||||
assert v["accepted"] is True and v["score"] > 0
|
||||
assert "1080p" in v["quality_label"]
|
||||
|
||||
|
||||
def test_evaluate_rejects_cam_and_disabled_tier():
|
||||
p = default_profile()
|
||||
cam = evaluate_release(parse_release("Movie 2020 HDCAM x264-CRUDE"), p, scope="movie")
|
||||
assert cam["accepted"] is False and "reject list" in cam["rejected"]
|
||||
# 2160p tiers are OFF by default → a 4K hit is filtered out (not enabled)
|
||||
uhd = evaluate_release(parse_release("Movie 2020 2160p WEB-DL HEVC"), p, scope="movie")
|
||||
assert uhd["accepted"] is False and "enabled tiers" in uhd["rejected"]
|
||||
|
||||
|
||||
def test_evaluate_scope_validation():
|
||||
p = default_profile()
|
||||
# an episode hit fails a SEASON search (not a full-season pack)
|
||||
ep = parse_release("Show S02E03 1080p WEB-DL x265-GRP")
|
||||
assert evaluate_release(ep, p, scope="season", want_season=2)["rejected"] == "Not a full-season pack"
|
||||
# the season PACK passes the season search
|
||||
pack = parse_release("Show S02 1080p WEB-DL x265-GRP")
|
||||
assert evaluate_release(pack, p, scope="season", want_season=2)["accepted"] is True
|
||||
# wrong season is rejected
|
||||
assert evaluate_release(pack, p, scope="season", want_season=5)["rejected"] == "Wrong season"
|
||||
|
||||
|
||||
def test_evaluate_size_cap_movie_only():
|
||||
p = default_profile(); p["max_movie_gb"] = 10
|
||||
big = evaluate_release(parse_release("Movie 2020 1080p BluRay x265-GRP"), p, scope="movie", size_gb=14)
|
||||
assert big["accepted"] is False and "size cap" in big["rejected"]
|
||||
ok = evaluate_release(parse_release("Movie 2020 1080p BluRay x265-GRP"), p, scope="movie", size_gb=8)
|
||||
assert ok["accepted"] is True
|
||||
|
||||
|
||||
def test_mock_search_scopes_are_shaped_right():
|
||||
movie = mock_search("movie", "The Matrix", year=1999)
|
||||
assert movie and all("size_bytes" in h and "seeders" in h for h in movie)
|
||||
assert "1999" in movie[0]["title"]
|
||||
season = mock_search("season", "The Wire", season=2)
|
||||
assert all(".S02" in h["title"] for h in season)
|
||||
series = mock_search("series", "The Wire", season_end=5)
|
||||
assert all("S01-S05" in h["title"] for h in series)
|
||||
assert mock_search("bogus", "x") == []
|
||||
|
||||
|
||||
def test_mock_search_is_deterministic():
|
||||
a = mock_search("episode", "Severance", season=1, episode=4)
|
||||
b = mock_search("episode", "Severance", season=1, episode=4)
|
||||
assert a == b # no RNG → stable across calls/reloads
|
||||
Loading…
Reference in a new issue