"""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 | add-tag 'x' | add-file -store local" # mm "download-media '' -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())