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

f

Browse Source
This commit is contained in:
Nose
2026-01-17 02:36:06 -08:00
parent 3a7c443004
commit c6fd6b4224
9 changed files with 440 additions and 226 deletions
Download Patch File Download Diff File Expand all files Collapse all files

281
cmdlet/download_file.py
View File

@@ -81,23 +81,6 @@ class Download_File(Cmdlet):
alias="o",
description="(deprecated) Output directory (use -path instead)",
),
CmdletArg(
name="audio",
type="flag",
alias="a",
description="Download audio only (yt-dlp)",
),
CmdletArg(
name="-magnet-id",
type="string",
description="(internal) AllDebrid magnet id used by provider selection hooks",
),
CmdletArg(
name="format",
type="string",
alias="fmt",
description="Explicit yt-dlp format selector",
),
QueryArg(
"clip",
key="clip",
@@ -183,6 +166,42 @@ class Download_File(Cmdlet):
path_value: Optional[Any] = path
if isinstance(path, dict):
provider_action = str(
path.get("action")
or path.get("provider_action")
or ""
).strip().lower()
if provider_action == "download_items" or bool(path.get("download_items")):
request_metadata = path.get("metadata") or path.get("full_metadata") or {}
if not isinstance(request_metadata, dict):
request_metadata = {}
magnet_id = path.get("magnet_id") or request_metadata.get("magnet_id")
if magnet_id is not None:
request_metadata.setdefault("magnet_id", magnet_id)
if SearchResult is None:
debug("Provider download_items requested but SearchResult unavailable")
continue
sr = SearchResult(
table=str(provider_name),
title=str(path.get("title") or path.get("name") or f"{provider_name} item"),
path=str(path.get("path") or path.get("url") or url),
full_metadata=request_metadata,
)
downloaded_extra = self._download_provider_items(
provider=provider,
provider_name=str(provider_name),
search_result=sr,
output_dir=final_output_dir,
progress=progress,
quiet_mode=quiet_mode,
config=config,
)
if downloaded_extra:
downloaded_count += int(downloaded_extra)
continue
path_value = path.get("path") or path.get("file_path")
extra_meta = path.get("metadata") or path.get("full_metadata")
title_hint = path.get("title") or path.get("name")
@@ -451,6 +470,24 @@ class Download_File(Cmdlet):
provider_sr = sr
debug(f"[download-file] Provider download result: {downloaded_path}")
if downloaded_path is None:
try:
downloaded_extra = self._download_provider_items(
provider=provider_obj,
provider_name=str(provider_key),
search_result=sr,
output_dir=output_dir,
progress=progress,
quiet_mode=quiet_mode,
config=config,
)
except Exception:
downloaded_extra = 0
if downloaded_extra:
downloaded_count += int(downloaded_extra)
continue
# Fallback: if we have a direct HTTP URL and no provider successfully handled it
if (downloaded_path is None and not attempted_provider_download
and isinstance(target, str) and target.startswith("http")):
@@ -539,6 +576,68 @@ class Download_File(Cmdlet):
return downloaded_count, queued_magnet_submissions
def _download_provider_items(
self,
*,
provider: Any,
provider_name: str,
search_result: Any,
output_dir: Path,
progress: PipelineProgress,
quiet_mode: bool,
config: Dict[str, Any],
) -> int:
if provider is None or not hasattr(provider, "download_items"):
return 0
def _on_emit(path: Path, file_url: str, relpath: str, metadata: Dict[str, Any]) -> None:
title_hint = None
try:
title_hint = metadata.get("name") or relpath
except Exception:
title_hint = relpath
title_hint = title_hint or (Path(path).name if path else "download")
self._emit_local_file(
downloaded_path=path,
source=file_url,
title_hint=title_hint,
tags_hint=None,
media_kind_hint="file",
full_metadata=metadata if isinstance(metadata, dict) else None,
progress=progress,
config=config,
provider_hint=provider_name,
)
try:
downloaded_count = provider.download_items(
search_result,
output_dir,
emit=_on_emit,
progress=progress,
quiet_mode=quiet_mode,
path_from_result=coerce_to_path,
config=config,
)
except TypeError:
downloaded_count = provider.download_items(
search_result,
output_dir,
emit=_on_emit,
progress=progress,
quiet_mode=quiet_mode,
path_from_result=coerce_to_path,
)
except Exception as exc:
log(f"Provider {provider_name} download_items error: {exc}", file=sys.stderr)
return 0
try:
return int(downloaded_count or 0)
except Exception:
return 0
def _emit_local_file(
self,
*,
@@ -1221,11 +1320,27 @@ class Download_File(Cmdlet):
# Add base command for display
format_dict["cmd"] = base_cmd
def _merge_query_args(selection_args: List[str], query_value: str) -> List[str]:
if not query_value:
return selection_args
merged = list(selection_args or [])
if "-query" in merged:
idx_query = merged.index("-query")
if idx_query + 1 < len(merged):
existing = str(merged[idx_query + 1] or "").strip()
merged[idx_query + 1] = f"{existing},{query_value}" if existing else query_value
else:
merged.append(query_value)
else:
merged.extend(["-query", query_value])
return merged
# Append clip values to selection args if needed
selection_args: List[str] = format_dict["_selection_args"].copy()
selection_args: List[str] = list(format_dict.get("_selection_args") or [])
try:
if (not clip_spec) and clip_values:
selection_args.extend(["-query", f"clip:{','.join([v for v in clip_values if v])}"])
clip_query = f"clip:{','.join([v for v in clip_values if v])}"
selection_args = _merge_query_args(selection_args, clip_query)
except Exception:
pass
format_dict["_selection_args"] = selection_args
@@ -1253,7 +1368,9 @@ class Download_File(Cmdlet):
pipeline_context.set_last_result_table(table, results_list)
debug(f"[ytdlp.formatlist] Format table registered with {len(results_list)} formats")
debug(f"[ytdlp.formatlist] When user selects @N, will invoke: download-file {url} -format <format_id>")
debug(
f"[ytdlp.formatlist] When user selects @N, will invoke: download-file {url} -query 'format:<format_id>'"
)
log(f"", file=sys.stderr)
return 0
@@ -1518,7 +1635,7 @@ class Download_File(Cmdlet):
"url": url,
"item_selector": selection_format_id,
},
"_selection_args": ["-format", selection_format_id],
"_selection_args": ["-query", f"format:{selection_format_id}"],
}
results_list.append(format_dict)
@@ -1748,12 +1865,10 @@ class Download_File(Cmdlet):
except Exception:
query_wants_audio = False
audio_flag = bool(parsed.get("audio") is True)
wants_audio = audio_flag
if query_audio is not None:
wants_audio = wants_audio or bool(query_audio)
wants_audio = bool(query_audio)
else:
wants_audio = wants_audio or bool(query_wants_audio)
wants_audio = bool(query_wants_audio)
mode = "audio" if wants_audio else "video"
clip_ranges, clip_invalid, clip_values = self._parse_clip_ranges_and_apply_items(
@@ -1777,8 +1892,8 @@ class Download_File(Cmdlet):
formats_cache: Dict[str, Optional[List[Dict[str, Any]]]] = {}
playlist_items = str(parsed.get("item")) if parsed.get("item") else None
ytdl_format = parsed.get("format")
if not ytdl_format and query_format and not query_wants_audio:
ytdl_format = None
if query_format and not query_wants_audio:
try:
height_selector = self._format_selector_for_query_height(query_format)
except ValueError as e:
@@ -1825,7 +1940,7 @@ class Download_File(Cmdlet):
sample_pipeline = f'download-file "{candidate_url}"'
hint = (
"To select non-interactively, re-run with an explicit format: "
"e.g. mm \"{pipeline} -format {fmt} | add-file -store <store>\" or "
"e.g. mm \"{pipeline} -query 'format:{fmt}' | add-file -store <store>\" or "
"mm \"{pipeline} -query 'format:{index}' | add-file -store <store>\""
).format(
pipeline=sample_pipeline,
@@ -2735,18 +2850,6 @@ class Download_File(Cmdlet):
downloaded_count = 0
# Special-case: support selection-inserted magnet-id arg to drive provider downloads
magnet_ret = self._process_magnet_id(
parsed=parsed,
registry=registry,
config=config,
final_output_dir=final_output_dir,
progress=progress,
quiet_mode=quiet_mode
)
if magnet_ret is not None:
return magnet_ret
urls_downloaded, early_exit = self._process_explicit_urls(
raw_urls=raw_url,
final_output_dir=final_output_dir,
@@ -2800,104 +2903,6 @@ class Download_File(Cmdlet):
pass
progress.close_local_ui(force_complete=True)
def _process_magnet_id(
self,
*,
parsed: Dict[str, Any],
registry: Dict[str, Any],
config: Dict[str, Any],
final_output_dir: Path,
progress: PipelineProgress,
quiet_mode: bool
) -> Optional[int]:
magnet_id_raw = parsed.get("magnet-id")
if not magnet_id_raw:
return None
try:
magnet_id = int(str(magnet_id_raw).strip())
except Exception:
log(f"[download-file] invalid magnet-id: {magnet_id_raw}", file=sys.stderr)
return 1
get_provider = registry.get("get_provider")
provider_name = str(parsed.get("provider") or "alldebrid").strip().lower()
provider_obj = None
if get_provider is not None:
try:
provider_obj = get_provider(provider_name, config)
except Exception:
provider_obj = None
if provider_obj is None:
log(f"[download-file] provider '{provider_name}' not available", file=sys.stderr)
return 1
SearchResult = registry.get("SearchResult")
try:
if SearchResult is not None:
sr = SearchResult(
table=provider_name,
title=f"magnet-{magnet_id}",
path=f"alldebrid:magnet:{magnet_id}",
full_metadata={
"magnet_id": magnet_id,
"provider": provider_name,
"provider_view": "files",
},
)
else:
sr = None
except Exception:
sr = None
def _on_emit(path: Path, file_url: str, relpath: str, metadata: Dict[str, Any]) -> None:
title_hint = metadata.get("name") or relpath or f"magnet-{magnet_id}"
self._emit_local_file(
downloaded_path=path,
source=file_url or f"alldebrid:magnet:{magnet_id}",
title_hint=title_hint,
tags_hint=None,
media_kind_hint="file",
full_metadata=metadata,
progress=progress,
config=config,
provider_hint=provider_name,
)
try:
downloaded_extra = provider_obj.download_items(
sr,
final_output_dir,
emit=_on_emit,
progress=progress,
quiet_mode=quiet_mode,
path_from_result=coerce_to_path,
config=config,
)
except TypeError:
downloaded_extra = provider_obj.download_items(
sr,
final_output_dir,
emit=_on_emit,
progress=progress,
quiet_mode=quiet_mode,
path_from_result=coerce_to_path,
)
except Exception as exc:
log(f"[download-file] failed to download magnet {magnet_id}: {exc}", file=sys.stderr)
return 1
if downloaded_extra:
debug(f"[download-file] AllDebrid magnet {magnet_id} emitted {downloaded_extra} files")
return 0
log(
f"[download-file] AllDebrid magnet {magnet_id} produced no downloads",
file=sys.stderr,
)
return 1
def _maybe_show_provider_picker(
self,
*,