Files
Medios-Macina/medeia_macina/cli_entry.py

279 lines
10 KiB
Python
Raw Normal View History

2025-12-23 16:36:39 -08:00
"""CLI entrypoint module compatible with console scripts.
This wraps the existing `medeia_entry.py` runner so installers can set
entry points to `medeia_macina.cli_entry:main`.
"""
from __future__ import annotations
from typing import Optional, List, Tuple
import sys
import importlib
from pathlib import Path
import shlex
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 _import_medeia_entry_module():
"""Import and return the top-level 'medeia_entry' module.
This attempts a regular import first. If that fails with ImportError it will
try a few fallbacks useful for editable installs and running directly from
the repository (searching for .egg-link, walking parents, or checking CWD).
"""
try:
return importlib.import_module("medeia_entry")
except ImportError:
# Try to find the project root next to this installed package
pkg_dir = Path(__file__).resolve().parent
# 1) Look for an .egg-link that points to the project root
try:
for egg in pkg_dir.glob("*.egg-link"):
try:
project_root = egg.read_text().splitlines()[0].strip()
if project_root:
candidate = Path(project_root) / "medeia_entry.py"
if candidate.exists():
if str(Path(project_root)) not in sys.path:
sys.path.insert(0, str(Path(project_root)))
return importlib.import_module("medeia_entry")
except Exception:
continue
except Exception:
pass
# 2) Walk upwards looking for a top-level 'medeia_entry.py'
for parent in pkg_dir.parents:
candidate = parent / "medeia_entry.py"
if candidate.exists():
if str(parent) not in sys.path:
sys.path.insert(0, str(parent))
return importlib.import_module("medeia_entry")
# 3) Check current working directory
candidate = Path.cwd() / "medeia_entry.py"
if candidate.exists():
if str(Path.cwd()) not in sys.path:
sys.path.insert(0, str(Path.cwd()))
return importlib.import_module("medeia_entry")
raise ImportError(
2025-12-24 02:13:21 -08:00
"Could not import 'medeia_entry'. This often means the package is not installed into the active virtualenv or is an outdated install.\n"
"Remedy: activate your venv and run: pip install -e . (or re-run the bootstrap script).\n"
"If problems persist, recreate the venv and reinstall the project."
2025-12-23 16:36:39 -08:00
)
def _run_cli(clean_args: List[str]) -> int:
"""Run the CLI runner (MedeiaCLI) with cleaned argv list."""
try:
sys.argv[1:] = list(clean_args)
except Exception:
pass
mod = _import_medeia_entry_module()
2025-12-24 02:13:21 -08:00
# Backwards compatibility: the imported module may not expose `MedeiaCLI` as
# an attribute (for example, the installed `medeia_entry` delegates to the
# packaged entrypoint instead of importing the top-level `CLI` module at
# import-time). Try a few strategies to obtain or invoke the CLI:
MedeiaCLI = None
if hasattr(mod, "MedeiaCLI"):
2025-12-23 16:36:39 -08:00
MedeiaCLI = getattr(mod, "MedeiaCLI")
2025-12-24 02:13:21 -08:00
else:
# Try importing the top-level `CLI` module directly (editable/repo mode)
try:
from CLI import MedeiaCLI as _M # type: ignore
MedeiaCLI = _M
except Exception:
raise ImportError(
"Imported module 'medeia_entry' does not define 'MedeiaCLI' and direct import of top-level 'CLI' failed.\n"
"Remedy: activate your venv and run: pip install -e . (or re-run the bootstrap script).\n"
"If problems persist, recreate the venv and reinstall the project."
)
2025-12-23 16:36:39 -08:00
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 nonzero.
"""
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
2025-12-24 02:13:21 -08:00
# 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
2025-12-23 16:36:39 -08:00
# 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())