goyimnose/Medios-Macina
1
0
Fork 0
Code Issues Pull Requests Packages Releases 1 Wiki Activity

d

Browse Source
This commit is contained in:
Nose
2026-01-31 23:41:47 -08:00
parent 753cdfb196
commit 95748698fa
15 changed files with 218 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
cmdlet/search_file.py
View File

@@ -30,9 +30,6 @@ from ._shared import (
)
from SYS import pipeline as ctx
STORAGE_ORIGINS = {"local",
"hydrus"}
class _WorkerLogger:
def __init__(self, worker_id: str) -> None:
@@ -94,7 +91,6 @@ class search_file(Cmdlet):
"provider",
type="string",
description="External provider name (e.g., tidal, youtube, soulseek, etc)",
choices=["bandcamp", "libgen", "soulseek", "youtube", "alldebrid", "loc", "internetarchive", "tidal", "tidal"],
),
CmdletArg(
"open",
@@ -142,22 +138,8 @@ class search_file(Cmdlet):
ext = "".join(ch for ch in ext if ch.isalnum())
return ext[:5]
@staticmethod
def _get_tidal_view_from_query(query: str) -> str:
text = str(query or "").strip()
if not text:
return "track"
if re.search(r"\balbum\s*:", text, flags=re.IGNORECASE):
return "album"
if re.search(r"\bartist\s*:", text, flags=re.IGNORECASE):
return "artist"
return "track"
def _ensure_storage_columns(self, payload: Dict[str, Any]) -> Dict[str, Any]:
"""Ensure storage results have the necessary fields for result_table display."""
store_value = str(payload.get("store") or "").lower()
if store_value not in STORAGE_ORIGINS:
return payload
# Ensure we have title field
if "title" not in payload:
@@ -265,78 +247,37 @@ class search_file(Cmdlet):
provider_text = str(provider_name or "").strip()
provider_lower = provider_text.lower()
id_match = re.search(r"\bid\s*[=:]\s*(\d+)", query, flags=re.IGNORECASE)
parsed_open_id = open_id
if id_match and parsed_open_id is None:
try:
parsed_open_id = int(id_match.group(1))
except Exception:
parsed_open_id = None
query = re.sub(r"\bid\s*[=:]\s*\d+", "", query, flags=re.IGNORECASE).strip()
if not query:
query = "*"
effective_open_id = parsed_open_id if parsed_open_id is not None else open_id
if provider_lower == "youtube":
provider_label = "Youtube"
elif provider_lower == "openlibrary":
provider_label = "OpenLibrary"
elif provider_lower == "loc":
provider_label = "LoC"
else:
provider_label = provider_text[:1].upper() + provider_text[1:] if provider_text else "Provider"
# Dynamic query/filter extraction via provider
normalized_query = str(query or "").strip()
provider_filters: Dict[str, Any] = {}
try:
normalized_query, provider_filters = provider.extract_query_arguments(query)
except Exception:
provider_filters = {}
normalized_query = (normalized_query or "").strip()
query = normalized_query or "*"
provider_filters = dict(provider_filters or {})
search_filters = dict(provider_filters or {})
if provider_lower == "alldebrid" and effective_open_id is not None:
table_title = f"{provider_label} Files: {effective_open_id}".strip().rstrip(":")
else:
table_title = f"{provider_label}: {query}".strip().rstrip(":")
# Dynamic table generation via provider
table_title = provider.get_table_title(query, search_filters).strip().rstrip(":")
table_type = provider.get_table_type(query, search_filters)
table_meta = provider.get_table_metadata(query, search_filters)
preserve_order = provider.preserve_order
preserve_order = provider_lower in {"youtube", "openlibrary", "loc", "torrent"}
table_type = provider_name
table_meta: Dict[str, Any] = {"provider": provider_name}
if provider_lower == "tidal":
view = self._get_tidal_view_from_query(query)
table_meta["view"] = view
table_type = f"tidal.{view}"
elif provider_lower == "internetarchive":
# Internet Archive search results are effectively folders (items); selecting @N
# should open a list of downloadable files for the chosen item.
table_type = "internetarchive.folder"
table = Table(table_title)._perseverance(preserve_order)
table.set_table(table_type)
if provider_lower == "alldebrid":
table_meta["view"] = "files" if effective_open_id is not None else "folders"
if effective_open_id is not None:
table_meta["magnet_id"] = effective_open_id
try:
table.set_table_metadata(table_meta)
except Exception:
pass
if provider_lower == "vimm":
# Keep auto-staged download-file from inheriting raw query tokens;
# only propagate provider hint so @N expands to a clean downloader call.
table.set_source_command("search-file", ["-provider", provider_name])
else:
table.set_source_command("search-file", list(args_list))
# Dynamic source command via provider
source_cmd, source_args = provider.get_source_command(args_list)
table.set_source_command(source_cmd, source_args)
search_filters = dict(provider_filters)
debug(f"[search-file] Calling {provider_name}.search(filters={search_filters})")
if provider_lower == "alldebrid":
search_open_id = parsed_open_id if parsed_open_id is not None else open_id
view_value = "files" if search_open_id is not None else "folders"
search_filters["view"] = view_value
if search_open_id is not None:
search_filters["magnet_id"] = search_open_id
results = provider.search(query, limit=limit, filters=search_filters or None)
debug(f"[search-file] {provider_name} -> {len(results or [])} result(s)")