video downloads: quality evaluation seam (owned-copy-vs-profile verdict)
The shared judge both the Download modal and the later-phase engine will use:
core/video/quality_eval.py (pure, isolated) — resolution_rank/resolution_label,
meets_cutoff (loose resolution target), and evaluate_owned(file, profile) →
{meets, resolution_label, reasons[]}. A copy is 'below target' when its resolution
is under the loose cutoff, or its codec is on the reject list.
Exposed as POST /api/video/downloads/evaluate (loads the stored profile, judges the
posted file). 9 tests green, isolation guard green, ruff clean.
This commit is contained in:
parent
ddde6cba68
commit
e880be9910
4 changed files with 199 additions and 0 deletions
|
|
@ -80,6 +80,18 @@ def register_routes(bp):
|
|||
body = request.get_json(silent=True) or {}
|
||||
return jsonify(save(get_video_db(), body))
|
||||
|
||||
@bp.route("/downloads/evaluate", methods=["POST"])
|
||||
def video_quality_evaluate():
|
||||
"""Judge a video file the user already owns against their quality profile —
|
||||
powers the Download modal's 'In your library · … (below your target)' line.
|
||||
Body: {"file": {resolution, video_codec, …}}."""
|
||||
from . import get_video_db
|
||||
from core.video.quality_eval import evaluate_owned
|
||||
from core.video.quality_profile import load
|
||||
body = request.get_json(silent=True) or {}
|
||||
profile = load(get_video_db())
|
||||
return jsonify(evaluate_owned(body.get("file"), profile))
|
||||
|
||||
@bp.route("/downloads/youtube-quality", methods=["GET"])
|
||||
def video_youtube_quality():
|
||||
# Separate, smaller profile — YouTube is yt-dlp, not scene/p2p releases.
|
||||
|
|
|
|||
99
core/video/quality_eval.py
Normal file
99
core/video/quality_eval.py
Normal file
|
|
@ -0,0 +1,99 @@
|
|||
"""Evaluate a video file/release against the quality profile.
|
||||
|
||||
Two consumers, one source of truth:
|
||||
- the Download modal — to tell the user whether the copy they already own meets
|
||||
their quality target (or is eligible for an upgrade), and
|
||||
- the (later-phase) download engine — to filter/score search results.
|
||||
|
||||
Pure functions (no DB, no network) so they're unit-tested in isolation. Isolated —
|
||||
imports only the sibling video ``quality_profile`` constants; nothing from music.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
# Resolution ranking (higher = better). The loose cutoff and the owned-vs-target
|
||||
# check both compare on this rank, so "1920x1080", "1080p" and "1080" all agree.
|
||||
_RES_RANK = (("2160", 4), ("4k", 4), ("1440", 3), ("1080", 3),
|
||||
("720", 2), ("576", 1), ("480", 1), ("sd", 1))
|
||||
_RES_LABEL = {4: "4K", 3: "1080p", 2: "720p", 1: "SD", 0: ""}
|
||||
|
||||
|
||||
def resolution_rank(res: Any) -> int:
|
||||
"""Map a raw resolution token to a rank int (4=4K … 1=SD, 0=unknown)."""
|
||||
s = str(res or "").strip().lower()
|
||||
for token, rank in _RES_RANK:
|
||||
if token in s:
|
||||
return rank
|
||||
return 0
|
||||
|
||||
|
||||
def resolution_label(res: Any) -> str:
|
||||
"""A friendly resolution label ('4K' / '1080p' / '720p' / 'SD' / '')."""
|
||||
return _RES_LABEL.get(resolution_rank(res), "")
|
||||
|
||||
|
||||
def _cutoff_label(cutoff: str) -> str:
|
||||
return _RES_LABEL.get(resolution_rank(cutoff), "best")
|
||||
|
||||
|
||||
def _codec_family(codec: Any) -> str:
|
||||
"""Normalise a stored video codec to a reject-list key ('x264'/'hevc'/'av1')."""
|
||||
s = str(codec or "").strip().lower()
|
||||
if not s:
|
||||
return ""
|
||||
if "av1" in s:
|
||||
return "av1"
|
||||
if "265" in s or "hevc" in s:
|
||||
return "hevc"
|
||||
if "264" in s or "avc" in s:
|
||||
return "x264"
|
||||
return ""
|
||||
|
||||
|
||||
def meets_cutoff(resolution: Any, profile: dict) -> bool:
|
||||
"""Does an owned item's resolution already satisfy the loose cutoff target?
|
||||
An empty cutoff ('always upgrade') is never 'good enough'."""
|
||||
cut = (profile or {}).get("cutoff_resolution", "")
|
||||
if not cut:
|
||||
return False
|
||||
return resolution_rank(resolution) >= resolution_rank(cut)
|
||||
|
||||
|
||||
def evaluate_owned(file: Any, profile: Any) -> dict:
|
||||
"""Verdict for a copy the user already owns, vs their quality profile.
|
||||
|
||||
Returns ``{"meets": bool, "resolution_label": str, "reasons": [{ok, text}]}``
|
||||
— ``meets`` False means it's eligible for an upgrade. ``reasons`` is an ordered,
|
||||
render-ready list of the checks (ok=True is reassuring, ok=False explains why
|
||||
an upgrade would help)."""
|
||||
file = file if isinstance(file, dict) else {}
|
||||
profile = profile if isinstance(profile, dict) else {}
|
||||
reasons: list = []
|
||||
meets = True
|
||||
|
||||
res = file.get("resolution")
|
||||
cut = profile.get("cutoff_resolution", "")
|
||||
if not cut:
|
||||
meets = False
|
||||
reasons.append({"ok": False, "text": "You're set to always chase the best — eligible for an upgrade."})
|
||||
elif resolution_rank(res) >= resolution_rank(cut):
|
||||
reasons.append({"ok": True, "text": "Meets your " + _cutoff_label(cut) + " target."})
|
||||
else:
|
||||
meets = False
|
||||
reasons.append({"ok": False, "text": "Below your " + _cutoff_label(cut) + " target — eligible for an upgrade."})
|
||||
|
||||
fam = _codec_family(file.get("video_codec"))
|
||||
if fam and fam in (profile.get("rejects") or []):
|
||||
meets = False
|
||||
reasons.append({"ok": False, "text": "Its " + fam + " codec is on your reject list."})
|
||||
|
||||
return {
|
||||
"meets": meets,
|
||||
"resolution_label": resolution_label(res),
|
||||
"reasons": reasons,
|
||||
}
|
||||
|
||||
|
||||
__all__ = ["resolution_rank", "resolution_label", "meets_cutoff", "evaluate_owned"]
|
||||
|
|
@ -431,6 +431,28 @@ def test_youtube_quality_profile_endpoint_roundtrips(tmp_path):
|
|||
videoapi._video_db = None
|
||||
|
||||
|
||||
def test_quality_evaluate_endpoint_judges_owned_copy(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:
|
||||
# Default profile cuts off at 1080p → a 720p copy is below target.
|
||||
below = client.post("/api/video/downloads/evaluate",
|
||||
json={"file": {"resolution": "720p", "video_codec": "x265"}}).get_json()
|
||||
assert below["meets"] is False and below["resolution_label"] == "720p"
|
||||
# A 1080p copy meets it.
|
||||
ok = client.post("/api/video/downloads/evaluate",
|
||||
json={"file": {"resolution": "1920x1080", "video_codec": "x265"}}).get_json()
|
||||
assert ok["meets"] is True
|
||||
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
|
||||
|
|
|
|||
66
tests/test_video_quality_eval.py
Normal file
66
tests/test_video_quality_eval.py
Normal file
|
|
@ -0,0 +1,66 @@
|
|||
"""Quality evaluation seam — owned-copy-vs-profile verdict (resolution rank +
|
||||
loose cutoff + codec reject), isolated from music. Shared by the Download modal
|
||||
and the later-phase download engine."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from core.video.quality_eval import (
|
||||
evaluate_owned,
|
||||
meets_cutoff,
|
||||
resolution_label,
|
||||
resolution_rank,
|
||||
)
|
||||
|
||||
|
||||
def test_resolution_rank_agrees_across_formats():
|
||||
assert resolution_rank("2160p") == resolution_rank("4K") == resolution_rank("3840x2160")
|
||||
assert resolution_rank("1080p") == resolution_rank("1920x1080") == 3
|
||||
assert resolution_rank("720p") == 2
|
||||
assert resolution_rank("480p") == resolution_rank("576p") == 1
|
||||
assert resolution_rank("garbage") == 0
|
||||
|
||||
|
||||
def test_resolution_label():
|
||||
assert resolution_label("1920x1080") == "1080p"
|
||||
assert resolution_label("2160p") == "4K"
|
||||
assert resolution_label(None) == ""
|
||||
|
||||
|
||||
def test_meets_cutoff_loose_target():
|
||||
assert meets_cutoff("1080p", {"cutoff_resolution": "1080p"}) is True
|
||||
assert meets_cutoff("4K", {"cutoff_resolution": "1080p"}) is True # better than target
|
||||
assert meets_cutoff("720p", {"cutoff_resolution": "1080p"}) is False # below target
|
||||
assert meets_cutoff("4K", {"cutoff_resolution": ""}) is False # always-upgrade
|
||||
|
||||
|
||||
def test_evaluate_owned_meets_target():
|
||||
out = evaluate_owned({"resolution": "1080p", "video_codec": "x265"},
|
||||
{"cutoff_resolution": "1080p", "rejects": ["cam"]})
|
||||
assert out["meets"] is True
|
||||
assert out["resolution_label"] == "1080p"
|
||||
assert out["reasons"][0]["ok"] is True
|
||||
|
||||
|
||||
def test_evaluate_owned_below_target_is_upgradeable():
|
||||
out = evaluate_owned({"resolution": "720p", "video_codec": "x265"},
|
||||
{"cutoff_resolution": "1080p", "rejects": []})
|
||||
assert out["meets"] is False
|
||||
assert any(not r["ok"] and "Below your 1080p target" in r["text"] for r in out["reasons"])
|
||||
|
||||
|
||||
def test_evaluate_owned_flags_rejected_codec_even_if_resolution_ok():
|
||||
out = evaluate_owned({"resolution": "1080p", "video_codec": "AVC (H.264)"},
|
||||
{"cutoff_resolution": "1080p", "rejects": ["x264"]})
|
||||
assert out["meets"] is False # resolution fine, but codec is rejected
|
||||
assert any("x264 codec is on your reject list" in r["text"] for r in out["reasons"])
|
||||
|
||||
|
||||
def test_evaluate_owned_always_upgrade_when_cutoff_empty():
|
||||
out = evaluate_owned({"resolution": "4K"}, {"cutoff_resolution": ""})
|
||||
assert out["meets"] is False
|
||||
assert "always chase the best" in out["reasons"][0]["text"]
|
||||
|
||||
|
||||
def test_evaluate_owned_handles_garbage_inputs():
|
||||
assert evaluate_owned(None, None)["meets"] in (True, False) # never raises
|
||||
assert evaluate_owned("nope", 42)["resolution_label"] == ""
|
||||
Loading…
Reference in a new issue