276 lines
9.6 KiB
Python
276 lines
9.6 KiB
Python
"""Packaged CLI entrypoint used by installers and console scripts.
|
||
|
||
This module provides the `main` entrypoint for `mm`/`medeia` and supports
|
||
running from a development checkout (by importing the top-level
|
||
`CLI.MedeiaCLI`) or when running tests that inject a legacy
|
||
`medeia_entry` shim into `sys.modules`.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Optional, List, Tuple
|
||
import sys
|
||
import importlib
|
||
from pathlib import Path
|
||
import shlex
|
||
|
||
|
||
def _ensure_repo_root_on_sys_path(pkg_file: Optional[Path] = None) -> None:
|
||
"""Ensure the repository root (where top-level modules live) is importable.
|
||
|
||
The project currently keeps key modules like `CLI.py` at the repo root.
|
||
When `mm` is invoked from a different working directory, that repo root is
|
||
not necessarily on `sys.path`, which breaks `import CLI`.
|
||
|
||
We infer the repo root by walking up from this package location and looking
|
||
for a sibling `CLI.py`.
|
||
|
||
`pkg_file` exists for unit tests; production uses this module's `__file__`.
|
||
"""
|
||
try:
|
||
pkg_dir = (pkg_file or Path(__file__)).resolve().parent
|
||
except Exception:
|
||
return
|
||
|
||
for parent in pkg_dir.parents:
|
||
try:
|
||
if (parent / "CLI.py").exists():
|
||
parent_str = str(parent)
|
||
if parent_str not in sys.path:
|
||
sys.path.insert(0, parent_str)
|
||
return
|
||
except Exception:
|
||
continue
|
||
|
||
|
||
def _parse_mode_and_strip_args(args: List[str]) -> Tuple[Optional[str], List[str]]:
|
||
"""Parse --gui/--cli/--mode flags and return (mode, cleaned_args).
|
||
|
||
The function removes any mode flags from the argument list so the selected
|
||
runner can receive the remaining arguments untouched.
|
||
|
||
Supported forms:
|
||
--gui, -g, --gui=true
|
||
--cli, -c, --cli=true
|
||
--mode=gui|cli
|
||
--mode gui|cli
|
||
|
||
Raises ValueError on conflicting or invalid flags.
|
||
"""
|
||
mode: Optional[str] = None
|
||
out: List[str] = []
|
||
i = 0
|
||
while i < len(args):
|
||
a = args[i]
|
||
la = a.lower()
|
||
|
||
# --gui / -g
|
||
if la in ("--gui", "-g"):
|
||
if mode and mode != "gui":
|
||
raise ValueError("Conflicting mode flags: found both 'gui' and 'cli'")
|
||
mode = "gui"
|
||
i += 1
|
||
continue
|
||
if la.startswith("--gui="):
|
||
val = la.split("=", 1)[1]
|
||
if val and val not in ("0", "false", "no", "off"):
|
||
if mode and mode != "gui":
|
||
raise ValueError(
|
||
"Conflicting mode flags: found both 'gui' and 'cli'"
|
||
)
|
||
mode = "gui"
|
||
i += 1
|
||
continue
|
||
|
||
# --cli / -c
|
||
if la in ("--cli", "-c"):
|
||
if mode and mode != "cli":
|
||
raise ValueError("Conflicting mode flags: found both 'gui' and 'cli'")
|
||
mode = "cli"
|
||
i += 1
|
||
continue
|
||
if la.startswith("--cli="):
|
||
val = la.split("=", 1)[1]
|
||
if val and val not in ("0", "false", "no", "off"):
|
||
if mode and mode != "cli":
|
||
raise ValueError(
|
||
"Conflicting mode flags: found both 'gui' and 'cli'"
|
||
)
|
||
mode = "cli"
|
||
i += 1
|
||
continue
|
||
|
||
# --mode
|
||
if la.startswith("--mode="):
|
||
val = la.split("=", 1)[1]
|
||
val = val.lower()
|
||
if val not in ("gui", "cli"):
|
||
raise ValueError("--mode must be 'gui' or 'cli'")
|
||
if mode and mode != val:
|
||
raise ValueError("Conflicting mode flags: found both 'gui' and 'cli'")
|
||
mode = val
|
||
i += 1
|
||
continue
|
||
if la == "--mode":
|
||
if i + 1 >= len(args):
|
||
raise ValueError("--mode requires a value ('gui' or 'cli')")
|
||
val = args[i + 1].lower()
|
||
if val not in ("gui", "cli"):
|
||
raise ValueError("--mode must be 'gui' or 'cli'")
|
||
if mode and mode != val:
|
||
raise ValueError("Conflicting mode flags: found both 'gui' and 'cli'")
|
||
mode = val
|
||
i += 2
|
||
continue
|
||
|
||
# Not a mode flag; keep it
|
||
out.append(a)
|
||
i += 1
|
||
|
||
return mode, out
|
||
|
||
|
||
def _run_cli(clean_args: List[str]) -> int:
|
||
"""Run the CLI runner (MedeiaCLI) with cleaned argv list.
|
||
|
||
The function supports three modes (in order):
|
||
1) If a `medeia_entry` module is present in sys.modules (for tests/legacy
|
||
scenarios), use it as the source of `MedeiaCLI`.
|
||
2) If running from a development checkout, import the top-level `CLI` and
|
||
use `MedeiaCLI` from there.
|
||
3) Otherwise fail with a helpful message suggesting an editable install.
|
||
"""
|
||
try:
|
||
sys.argv[1:] = list(clean_args)
|
||
except Exception:
|
||
pass
|
||
|
||
# 1) Check for an in-memory 'medeia_entry' module (tests/legacy installs)
|
||
MedeiaCLI = None
|
||
if "medeia_entry" in sys.modules:
|
||
mod = sys.modules.get("medeia_entry")
|
||
if hasattr(mod, "MedeiaCLI"):
|
||
MedeiaCLI = getattr(mod, "MedeiaCLI")
|
||
else:
|
||
# Preserve the existing error message used by tests that inject a
|
||
# dummy 'medeia_entry' without a `MedeiaCLI` attribute.
|
||
raise ImportError(
|
||
"Imported module 'medeia_entry' does not define 'MedeiaCLI' and direct import of top-level 'CLI' failed.\n"
|
||
"Remedy: ensure the top-level 'medeia_entry' module exports 'MedeiaCLI' or run from the project root/debug the checkout."
|
||
)
|
||
|
||
# 2) If no in-memory module provided the class, try importing the repo-root CLI
|
||
if MedeiaCLI is None:
|
||
try:
|
||
_ensure_repo_root_on_sys_path()
|
||
from CLI import MedeiaCLI as _M # type: ignore
|
||
MedeiaCLI = _M
|
||
except Exception:
|
||
raise ImportError(
|
||
"Could not import 'MedeiaCLI'. This often means the project is not available on sys.path (run 'pip install -e scripts' or re-run the bootstrap script)."
|
||
)
|
||
|
||
try:
|
||
app = MedeiaCLI()
|
||
app.run()
|
||
return 0
|
||
except SystemExit as exc:
|
||
return int(getattr(exc, "code", 0) or 0)
|
||
|
||
|
||
def _run_gui(clean_args: List[str]) -> int:
|
||
"""Run the TUI runner (PipelineHubApp).
|
||
|
||
The TUI is imported lazily; if Textual or the TUI code is unavailable we
|
||
give a helpful error message and exit non‑zero.
|
||
"""
|
||
try:
|
||
tui_mod = importlib.import_module("TUI.tui")
|
||
except Exception as exc:
|
||
print(
|
||
"Error: Unable to import TUI (Textual may not be installed):",
|
||
exc,
|
||
file=sys.stderr,
|
||
)
|
||
return 2
|
||
|
||
try:
|
||
PipelineHubApp = getattr(tui_mod, "PipelineHubApp")
|
||
except AttributeError:
|
||
print("Error: 'TUI.tui' does not expose 'PipelineHubApp'", file=sys.stderr)
|
||
return 2
|
||
|
||
try:
|
||
app = PipelineHubApp()
|
||
app.run()
|
||
return 0
|
||
except SystemExit as exc:
|
||
return int(getattr(exc, "code", 0) or 0)
|
||
|
||
|
||
def main(argv: Optional[List[str]] = None) -> int:
|
||
"""Entry point for console_scripts.
|
||
|
||
Accepts an optional argv list (useful for testing). Mode flags are parsed
|
||
and removed before dispatching to the selected runner.
|
||
"""
|
||
args = list(argv) if argv is not None else list(sys.argv[1:])
|
||
|
||
try:
|
||
mode, clean_args = _parse_mode_and_strip_args(args)
|
||
except ValueError as exc:
|
||
print(f"Error parsing mode flags: {exc}", file=sys.stderr)
|
||
return 2
|
||
|
||
# Early environment sanity check to detect urllib3/urllib3-future conflicts.
|
||
# When a broken urllib3 is detected we print an actionable message and
|
||
# exit early to avoid confusing import-time errors later during startup.
|
||
try:
|
||
from SYS.env_check import ensure_urllib3_ok
|
||
|
||
try:
|
||
ensure_urllib3_ok(exit_on_error=True)
|
||
except SystemExit as exc:
|
||
# Bubble out the exit code as the CLI return value for clearer
|
||
# behavior in shell sessions and scripts.
|
||
return int(getattr(exc, "code", 2) or 2)
|
||
except Exception:
|
||
# If the sanity check itself cannot be imported or run, don't block
|
||
# startup; we'll continue and let normal import errors surface.
|
||
pass
|
||
|
||
# If GUI requested, delegate directly (GUI may decide to honor any args itself)
|
||
if mode == "gui":
|
||
return _run_gui(clean_args)
|
||
|
||
# Support quoting a pipeline (or even a single full command) on the command line.
|
||
#
|
||
# - If the user provides a single argument that contains a pipe character,
|
||
# treat it as a pipeline and rewrite the args to call the internal `pipeline`
|
||
# subcommand so existing CLI pipeline handling is used.
|
||
#
|
||
# - If the user provides a single argument that contains whitespace but no pipe,
|
||
# expand it into argv tokens (PowerShell commonly encourages quoting strings).
|
||
#
|
||
# Examples:
|
||
# mm "download-media <url> | add-tag 'x' | add-file -store local"
|
||
# mm "download-media '<url>' -query 'format:720p' -path 'C:\\out'"
|
||
if len(clean_args) == 1:
|
||
single = clean_args[0]
|
||
if "|" in single and not single.startswith("-"):
|
||
clean_args = ["pipeline", "--pipeline", single]
|
||
elif (not single.startswith("-")) and any(ch.isspace() for ch in single):
|
||
try:
|
||
expanded = shlex.split(single, posix=True)
|
||
if expanded:
|
||
clean_args = list(expanded)
|
||
except Exception:
|
||
pass
|
||
|
||
# Default to CLI if --cli is requested or no explicit mode provided.
|
||
return _run_cli(clean_args)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|