From 1ce1e7d48ac5c3b9544d9af37a4b43d4710aaa5d Mon Sep 17 00:00:00 2001 From: Ethanfel Date: Sun, 15 Mar 2026 14:36:55 +0100 Subject: [PATCH] feat: add stats-based sorting (downloads, rating, thumbsup) Co-Authored-By: Claude Sonnet 4.6 --- proxy.py | 109 +++++++++++++++++++++++++++++++++++++++ tests/test_stats_sort.py | 61 ++++++++++++++++++++++ 2 files changed, 170 insertions(+) create mode 100644 tests/test_stats_sort.py diff --git a/proxy.py b/proxy.py index f95b773..5f0317f 100644 --- a/proxy.py +++ b/proxy.py @@ -233,6 +233,106 @@ def _enrich_list_response(response_data: dict, db=None) -> dict: return response_data +# --------------------------------------------------------------------------- +# Stats-based sorting +# --------------------------------------------------------------------------- + +_STATS_SORT_KEYS = {"downloads", "rating", "thumbsup"} +_STATS_SORT_COLUMN = { + "downloads": "download_count", + "rating": "rating", + "thumbsup": "thumbs_up_count", +} + + +def _is_stats_sort(sort_key: str) -> bool: + """Return True if sort_key is a stats-based sort.""" + return sort_key in _STATS_SORT_KEYS + + +def _sort_items_by_stats(items, sort_key, order, db=None): + """Sort items by a stats column. Items without stats go last.""" + if db is None: + db = _get_stats_db() + + column = _STATS_SORT_COLUMN.get(sort_key, "download_count") + hashes = [item["sha256"] for item in items if item.get("sha256")] + stats_map = db.get_by_hashes(hashes) + reverse = order == "desc" + + def get_value(item): + sha = item.get("sha256") + if sha and sha in stats_map: + return stats_map[sha].get(column, 0) + return None + + # Split items into those with stats and those without, then sort only + # the former so that items without stats always end up last. + with_stats = [i for i in items if get_value(i) is not None] + without_stats = [i for i in items if get_value(i) is None] + + with_stats.sort(key=get_value, reverse=reverse) + return with_stats + without_stats + + +async def _proxy_list_with_stats_sort( + request: web.Request, sort_by: str +) -> web.Response: + """Fetch all items from remote, sort by stats locally, paginate.""" + from urllib.parse import urlencode + + page = int(request.query.get("page", "1")) + page_size = int(request.query.get("page_size", "20")) + sort_key, _, order = sort_by.partition(":") + order = order or "desc" + + # Build query to fetch all items from remote with default sort + query = dict(request.query) + query["page"] = "1" + query["page_size"] = "9999" + query["sort_by"] = "name:asc" + remote_url = f"{remote_config.remote_url}{request.path}?{urlencode(query)}" + + # Forward relevant headers + headers = {} + skip = {"host", "transfer-encoding", "connection", "keep-alive", "upgrade"} + for k, v in request.headers.items(): + if k.lower() not in skip: + headers[k] = v + + session = await _get_proxy_session() + try: + async with session.get(remote_url, headers=headers) as resp: + if resp.status != 200: + return await _proxy_http(request) + data = await resp.json() + except Exception: + return await _proxy_http(request) + + items = data.get("items", []) + db = _get_stats_db() + + # Enrich all items with stats + _enrich_list_response(data, db) + + # Sort by stats + sorted_items = _sort_items_by_stats(items, sort_key, order, db) + + # Paginate locally + total = len(sorted_items) + start = (page - 1) * page_size + end = start + page_size + page_items = sorted_items[start:end] + + return web.json_response({ + "items": page_items, + "total": total, + "page": page, + "page_size": page_size, + "total_pages": (total + page_size - 1) // page_size, + }) + + # Shared HTTP session for proxied requests (connection pooling) _proxy_session: aiohttp.ClientSession | None = None @@ -385,6 +485,15 @@ async def lm_remote_proxy_middleware(request: web.Request, handler): if _is_ws_route(path): return await _proxy_ws(request) + # Stats-based sorting: intercept list requests, fetch all from remote, + # sort locally, paginate locally + if (_should_proxy(path) and path.endswith("/list") + and path.startswith("/api/lm/") and request.method == "GET"): + sort_by = request.query.get("sort_by", "name") + sort_key = sort_by.split(":")[0] if ":" in sort_by else sort_by + if _is_stats_sort(sort_key): + return await _proxy_list_with_stats_sort(request, sort_by) + # Regular proxy routes if _should_proxy(path): response = await _proxy_http(request) diff --git a/tests/test_stats_sort.py b/tests/test_stats_sort.py new file mode 100644 index 0000000..df58fad --- /dev/null +++ b/tests/test_stats_sort.py @@ -0,0 +1,61 @@ +"""Tests for stats-based sorting.""" +import pytest + + +@pytest.fixture +def db(tmp_path): + from stats_db import StatsDB + db = StatsDB(tmp_path / "test.db") + db.init() + return db + + +def test_sort_by_downloads_desc(db): + """Items sorted by download_count descending.""" + from proxy import _sort_items_by_stats + + db.upsert("h1", {"download_count": 100, "rating": 0, "rating_count": 0, "thumbs_up_count": 0}) + db.upsert("h2", {"download_count": 5000, "rating": 0, "rating_count": 0, "thumbs_up_count": 0}) + db.upsert("h3", {"download_count": 500, "rating": 0, "rating_count": 0, "thumbs_up_count": 0}) + + items = [ + {"sha256": "h1", "file_name": "a"}, + {"sha256": "h2", "file_name": "b"}, + {"sha256": "h3", "file_name": "c"}, + ] + sorted_items = _sort_items_by_stats(items, "downloads", "desc", db) + assert [i["sha256"] for i in sorted_items] == ["h2", "h3", "h1"] + + +def test_sort_by_rating_desc(db): + """Items sorted by rating descending.""" + from proxy import _sort_items_by_stats + + db.upsert("h1", {"download_count": 0, "rating": 3.0, "rating_count": 0, "thumbs_up_count": 0}) + db.upsert("h2", {"download_count": 0, "rating": 4.8, "rating_count": 0, "thumbs_up_count": 0}) + + items = [{"sha256": "h1"}, {"sha256": "h2"}] + sorted_items = _sort_items_by_stats(items, "rating", "desc", db) + assert sorted_items[0]["sha256"] == "h2" + + +def test_sort_items_without_stats_go_last(db): + """Items without stats sort to the end.""" + from proxy import _sort_items_by_stats + + db.upsert("h1", {"download_count": 100, "rating": 0, "rating_count": 0, "thumbs_up_count": 0}) + + items = [{"sha256": "no_stats"}, {"sha256": "h1"}] + sorted_items = _sort_items_by_stats(items, "downloads", "desc", db) + assert sorted_items[0]["sha256"] == "h1" + assert sorted_items[1]["sha256"] == "no_stats" + + +def test_is_stats_sort(): + """Identify stats-based sort keys.""" + from proxy import _is_stats_sort + assert _is_stats_sort("downloads") is True + assert _is_stats_sort("rating") is True + assert _is_stats_sort("thumbsup") is True + assert _is_stats_sort("name") is False + assert _is_stats_sort("date") is False