25969302440

"""Lightweight CLI argument parsing for Vector Inspector.

Handles --version, --help, and runtime-only flags before any Qt/GUI modules
are imported.  Runtime flags are propagated via environment variables so that
main.py and lower-level modules pick them up without any signature change.

Environment variables set (never persisted to disk):
  VI_NO_TELEMETRY=1   — disables telemetry for this process
  VI_NO_SPLASH=1      — skips the loading splash screen
  VI_CONFIG_PATH=PATH — alternate settings file path for this run
  LOG_LEVEL=LEVEL     — logging level (read by core/logging.py at import time)
"""

import argparse
import json
import os
import platform
import sys
from datetime import UTC, datetime
from pathlib import Path

from vector_inspector import get_version

GITHUB_URL = "https://github.com/anthonypdawson/vector-inspector"
_FIRST_RUN_KEY = "cli.first_run_done"
_LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR")


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="vector-inspector",
        description=(
            "Vector Inspector — a desktop GUI for visualizing and managing vector databases.\n"
            "\n"
            "Running 'vector-inspector' or 'python -m vector_inspector' without arguments\n"
            "starts the GUI application."
        ),
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=f"For problems, visit: {GITHUB_URL}",
    )
    parser.add_argument(
        "--version",
        action="version",
        version=f"%(prog)s {get_version()}",
    )
    parser.add_argument(
        "--no-telemetry",
        action="store_true",
        help="Disable telemetry for this run only (does not change saved settings).",
    )
    parser.add_argument(
        "--log-level",
        metavar="LEVEL",
        choices=_LOG_LEVELS,
        type=str.upper,
        default=None,
        help=f"Set logging verbosity for this run only. Choices: {', '.join(_LOG_LEVELS)}.",
    )
    parser.add_argument(
        "--no-splash",
        action="store_true",
        help="Skip the loading splash screen (does not change saved settings).",
    )
    parser.add_argument(
        "--config",
        metavar="PATH",
        default=None,
        help="Use an alternate settings file for this run only (your default settings are not modified).",
    )
    parser.add_argument(
        "--dump-settings",
        action="store_true",
        help="Print current settings as JSON and exit.",
    )
    parser.add_argument(
        "--llm-console",
        action="store_true",
        help=argparse.SUPPRESS,  # Hidden debug/dev flag — not shown in --help
    )
    parser.add_argument(
        "--install",
        metavar="PROVIDER",
        nargs="?",
        const="_wizard_",
        default=None,
        help=(
            "Install a database provider package without launching the GUI. "
            "Pass a provider ID (e.g. chromadb, qdrant, pinecone, lancedb, pgvector, "
            "weaviate, milvus) or omit the value to run an interactive wizard that lists "
            "all unavailable providers."
        ),
    )
    return parser


def _maybe_send_first_run_telemetry(command: str) -> None:
    """Send a one-time cli_first_use event on the first --version or --help invocation.

    Respects VI_NO_TELEMETRY env var and the telemetry.enabled setting.
    Completely best-effort — any failure is silently swallowed.
    """
    if os.environ.get("VI_NO_TELEMETRY"):
        return
    try:
        from vector_inspector.services.settings_service import SettingsService
        from vector_inspector.services.telemetry_service import TelemetryService

        settings = SettingsService()
        if settings.get(_FIRST_RUN_KEY, False):
            return

        telemetry = TelemetryService(settings_service=settings)
        if not telemetry.is_enabled():
            return

        telemetry.queue_event(
            {
                "event_name": "cli_first_use",
                "metadata": {
                    "command": command,
                    "platform": platform.system(),
                    "ts": datetime.now(UTC).isoformat(),
                },
            }
        )
        telemetry.send_batch()
        settings.set(_FIRST_RUN_KEY, True)
    except Exception:
        pass  # Best-effort: never crash the CLI for a telemetry failure


def _handle_dump_settings(config_path: str | None) -> None:
    """Print settings JSON to stdout and exit 0.  Never imports Qt."""
    settings_file = Path(config_path) if config_path else Path.home() / ".vector-inspector" / "settings.json"
    try:
        data = json.loads(settings_file.read_text(encoding="utf-8")) if settings_file.exists() else {}
        print(json.dumps(data, indent=2))  # noqa: T201
    except Exception as exc:
        print(f"Error reading settings: {exc}", file=sys.stderr)  # noqa: T201
        sys.exit(1)
    sys.exit(0)


def _handle_install(provider_arg: str) -> None:
    """Interactive or direct provider install wizard.  Never imports Qt.

    When ``provider_arg`` is ``"_wizard_"`` (the const from ``--install``
    with no value), an interactive numbered menu is shown.  Otherwise
    ``provider_arg`` is treated as a provider ID and installed directly.
    """
    from vector_inspector.core.provider_detection import get_all_providers
    from vector_inspector.services.install_service import (
        get_install_command,
        get_valid_provider_ids,
        install,
    )

    _DIVIDER = "=" * 54

    print("\nVector Inspector — Provider Installer")  # noqa: T201
    print(_DIVIDER)  # noqa: T201
    print("Checking installed providers…")  # noqa: T201

    all_providers = get_all_providers()
    unavailable = [p for p in all_providers if not p.available]
    available = [p for p in all_providers if p.available]

    if available:
        print(f"Already installed: {', '.join(p.name for p in available)}")  # noqa: T201

    if provider_arg == "_wizard_":
        # Interactive wizard mode — list unavailable providers and ask.
        if not unavailable:
            print("\n✓ All providers are already installed!")  # noqa: T201
            sys.exit(0)

        print("\nAvailable to install:")  # noqa: T201
        for idx, p in enumerate(unavailable, start=1):
            print(f"  {idx}. {p.name:<30} ({p.install_command})")  # noqa: T201

        print()  # noqa: T201
        try:
            raw = input("Enter a number or provider ID (or press Enter to cancel): ").strip()
        except (EOFError, KeyboardInterrupt):
            print("\nCancelled.")  # noqa: T201
            sys.exit(0)

        if not raw:
            print("Cancelled.")  # noqa: T201
            sys.exit(0)

        # Accept a number from the menu or a literal provider ID.
        selected_provider = None
        if raw.isdigit():
            idx = int(raw) - 1
            if 0 <= idx < len(unavailable):
                selected_provider = unavailable[idx]
        if selected_provider is None:
            # Try as a direct provider ID.
            matches = [p for p in unavailable if p.id == raw]
            if matches:
                selected_provider = matches[0]

        if selected_provider is None:
            print(f"Unknown selection: {raw!r}. Aborting.", file=sys.stderr)  # noqa: T201
            sys.exit(1)
    else:
        # Direct mode — the user passed a provider ID on the command line.
        if provider_arg not in get_valid_provider_ids():
            print(  # noqa: T201
                f"Unknown provider: {provider_arg!r}\nValid providers: {', '.join(sorted(get_valid_provider_ids()))}",
                file=sys.stderr,
            )
            sys.exit(1)

        matches = [p for p in all_providers if p.id == provider_arg]
        selected_provider = matches[0] if matches else None
        if selected_provider is None:
            print(f"Provider info not found for {provider_arg!r}.", file=sys.stderr)  # noqa: T201
            sys.exit(1)

        if selected_provider.available:
            print(f"\n✓ {selected_provider.name} is already installed.")  # noqa: T201
            sys.exit(0)

    # Run the install.
    cmd = get_install_command(selected_provider.id)
    print(f"\nInstalling {selected_provider.name}…")  # noqa: T201
    print(f"Running: {' '.join(cmd)}")  # noqa: T201
    print(_DIVIDER)  # noqa: T201

    returncode, _combined = install(
        selected_provider.id,
        on_output=lambda line: print(line, end="", flush=True),  # noqa: T201
    )

    print(_DIVIDER)  # noqa: T201
    if returncode == 0:
        print(f"\n✓ {selected_provider.name} installed successfully!")  # noqa: T201
        print("Restart Vector Inspector (or use the 🔄 Refresh button) to use it.")  # noqa: T201
        sys.exit(0)
    else:
        print(f"\n✗ Installation failed (exit code {returncode}).", file=sys.stderr)  # noqa: T201
        sys.exit(returncode)


def parse_cli_args(argv: list[str] | None = None) -> None:
    """Parse CLI arguments, apply runtime-only env vars, and exit where appropriate.

    Must be called before importing Qt/GUI modules.  Returns normally
    (without exiting) when no early-exit flag is supplied.
    """
    argv = sys.argv[1:] if argv is None else argv

    # Step 1: Pre-parse non-exit flags so env vars are set before telemetry
    # fires or --dump-settings reads settings (VI_CONFIG_PATH must be ready).
    pre = argparse.ArgumentParser(add_help=False)
    pre.add_argument("--no-telemetry", action="store_true")
    pre.add_argument("--log-level", type=str.upper, default=None)
    pre.add_argument("--config", default=None)
    pre_ns, _ = pre.parse_known_args(argv)

    if pre_ns.no_telemetry:
        os.environ["VI_NO_TELEMETRY"] = "1"
    if pre_ns.log_level:
        os.environ["LOG_LEVEL"] = pre_ns.log_level
        # Also update the already-imported logger level so the setting takes
        # effect even on code paths that imported logging before this call.
        try:
            import logging

            logging.getLogger("vector_inspector").setLevel(getattr(logging, pre_ns.log_level))
        except Exception:
            pass
    if pre_ns.config:
        os.environ["VI_CONFIG_PATH"] = pre_ns.config

    # Step 2: Detect command for first-run telemetry before argparse exits.
    command: str | None = None
    if "--version" in argv:
        command = "--version"
    elif "--help" in argv or "-h" in argv:
        command = "--help"
    elif "--dump-settings" in argv:
        command = "--dump-settings"

    if command:
        _maybe_send_first_run_telemetry(command)

    # Step 3: Full parse — exits here for --version / --help.
    parser = _build_parser()
    args = parser.parse_args(argv)

    # Step 4: --dump-settings is an early exit (no Qt needed).
    if args.dump_settings:
        _handle_dump_settings(args.config)

    # Step 4b: --install is an early exit (no Qt needed).
    if args.install is not None:
        _handle_install(args.install)

    # Step 5: Set remaining runtime env vars.
    if args.no_splash:
        os.environ["VI_NO_SPLASH"] = "1"
    if args.llm_console:
        os.environ["VI_LLM_CONSOLE"] = "1"


def console_entry() -> None:
    """Console script entry point for the ``vector-inspector`` command."""
    parse_cli_args()
    # Only reached when no early-exit flag was given; launch the GUI.
    # --llm-console (VI_LLM_CONSOLE=1) causes main.py to also open the
    # LLM debug window alongside the main application window.
    from vector_inspector.main import main

    main()

1	"""Lightweight CLI argument parsing for Vector Inspector.
2
3	Handles --version, --help, and runtime-only flags before any Qt/GUI modules
4	are imported. Runtime flags are propagated via environment variables so that
5	main.py and lower-level modules pick them up without any signature change.
6
7	Environment variables set (never persisted to disk):
8	VI_NO_TELEMETRY=1 — disables telemetry for this process
9	VI_NO_SPLASH=1 — skips the loading splash screen
10	VI_CONFIG_PATH=PATH — alternate settings file path for this run
11	LOG_LEVEL=LEVEL — logging level (read by core/logging.py at import time)
12	"""
13
14	import argparse	1✔
15	import json	1✔
16	import os	1✔
17	import platform	1✔
18	import sys	1✔
19	from datetime import UTC, datetime	1✔
20	from pathlib import Path	1✔
21
22	from vector_inspector import get_version	1✔
23
24	GITHUB_URL = "https://github.com/anthonypdawson/vector-inspector"	1✔
25	_FIRST_RUN_KEY = "cli.first_run_done"	1✔
26	_LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR")	1✔
27
28
29	def _build_parser() -> argparse.ArgumentParser:	1✔
30	parser = argparse.ArgumentParser(	1✔
31	prog="vector-inspector",
32	description=(
33	"Vector Inspector — a desktop GUI for visualizing and managing vector databases.\n"
34	"\n"
35	"Running 'vector-inspector' or 'python -m vector_inspector' without arguments\n"
36	"starts the GUI application."
37	),
38	formatter_class=argparse.RawDescriptionHelpFormatter,
39	epilog=f"For problems, visit: {GITHUB_URL}",
40	)
41	parser.add_argument(	1✔
42	"--version",
43	action="version",
44	version=f"%(prog)s {get_version()}",
45	)
46	parser.add_argument(	1✔
47	"--no-telemetry",
48	action="store_true",
49	help="Disable telemetry for this run only (does not change saved settings).",
50	)
51	parser.add_argument(	1✔
52	"--log-level",
53	metavar="LEVEL",
54	choices=_LOG_LEVELS,
55	type=str.upper,
56	default=None,
57	help=f"Set logging verbosity for this run only. Choices: {', '.join(_LOG_LEVELS)}.",
58	)
59	parser.add_argument(	1✔
60	"--no-splash",
61	action="store_true",
62	help="Skip the loading splash screen (does not change saved settings).",
63	)
64	parser.add_argument(	1✔
65	"--config",
66	metavar="PATH",
67	default=None,
68	help="Use an alternate settings file for this run only (your default settings are not modified).",
69	)
70	parser.add_argument(	1✔
71	"--dump-settings",
72	action="store_true",
73	help="Print current settings as JSON and exit.",
74	)
75	parser.add_argument(	1✔
76	"--llm-console",
77	action="store_true",
78	help=argparse.SUPPRESS, # Hidden debug/dev flag — not shown in --help
79	)
80	parser.add_argument(	1✔
81	"--install",
82	metavar="PROVIDER",
83	nargs="?",
84	const="_wizard_",
85	default=None,
86	help=(
87	"Install a database provider package without launching the GUI. "
88	"Pass a provider ID (e.g. chromadb, qdrant, pinecone, lancedb, pgvector, "
89	"weaviate, milvus) or omit the value to run an interactive wizard that lists "
90	"all unavailable providers."
91	),
92	)
93	return parser	1✔
94
95
96	def _maybe_send_first_run_telemetry(command: str) -> None:	1✔
97	"""Send a one-time cli_first_use event on the first --version or --help invocation.
98
99	Respects VI_NO_TELEMETRY env var and the telemetry.enabled setting.
100	Completely best-effort — any failure is silently swallowed.
101	"""
102	if os.environ.get("VI_NO_TELEMETRY"):	1✔
103	return	1✔
104	try:	1✔
105	from vector_inspector.services.settings_service import SettingsService	1✔
106	from vector_inspector.services.telemetry_service import TelemetryService	1✔
107
108	settings = SettingsService()	1✔
109	if settings.get(_FIRST_RUN_KEY, False):	1✔
110	return	1✔
111
112	telemetry = TelemetryService(settings_service=settings)	1✔
113	if not telemetry.is_enabled():	1✔
114	return	1✔
115
116	telemetry.queue_event(	1✔
117	{
118	"event_name": "cli_first_use",
119	"metadata": {
120	"command": command,
121	"platform": platform.system(),
122	"ts": datetime.now(UTC).isoformat(),
123	},
124	}
125	)
126	telemetry.send_batch()	1✔
127	settings.set(_FIRST_RUN_KEY, True)	1✔
128	except Exception:	×
129	pass # Best-effort: never crash the CLI for a telemetry failure	×
130
131
132	def _handle_dump_settings(config_path: str \| None) -> None:	1✔
133	"""Print settings JSON to stdout and exit 0. Never imports Qt."""
134	settings_file = Path(config_path) if config_path else Path.home() / ".vector-inspector" / "settings.json"	1✔
135	try:	1✔
136	data = json.loads(settings_file.read_text(encoding="utf-8")) if settings_file.exists() else {}	1✔
137	print(json.dumps(data, indent=2)) # noqa: T201	1✔
138	except Exception as exc:	×
139	print(f"Error reading settings: {exc}", file=sys.stderr) # noqa: T201	×
140	sys.exit(1)	×
141	sys.exit(0)	1✔
142
143
144	def _handle_install(provider_arg: str) -> None:	1✔
145	"""Interactive or direct provider install wizard. Never imports Qt.
146
147	When ``provider_arg`` is ``"_wizard_"`` (the const from ``--install``
148	with no value), an interactive numbered menu is shown. Otherwise
149	``provider_arg`` is treated as a provider ID and installed directly.
150	"""
151	from vector_inspector.core.provider_detection import get_all_providers	1✔
152	from vector_inspector.services.install_service import (	1✔
153	get_install_command,
154	get_valid_provider_ids,
155	install,
156	)
157
158	_DIVIDER = "=" * 54	1✔
159
160	print("\nVector Inspector — Provider Installer") # noqa: T201	1✔
161	print(_DIVIDER) # noqa: T201	1✔
162	print("Checking installed providers…") # noqa: T201	1✔
163
164	all_providers = get_all_providers()	1✔
165	unavailable = [p for p in all_providers if not p.available]	1✔
166	available = [p for p in all_providers if p.available]	1✔
167
168	if available:	1✔
169	print(f"Already installed: {', '.join(p.name for p in available)}") # noqa: T201	1✔
170
171	if provider_arg == "_wizard_":	1✔
172	# Interactive wizard mode — list unavailable providers and ask.
173	if not unavailable:	1✔
174	print("\n✓ All providers are already installed!") # noqa: T201	1✔
175	sys.exit(0)	1✔
176
177	print("\nAvailable to install:") # noqa: T201	1✔
178	for idx, p in enumerate(unavailable, start=1):	1✔
179	print(f" {idx}. {p.name:<30} ({p.install_command})") # noqa: T201	1✔
180
181	print() # noqa: T201	1✔
182	try:	1✔
183	raw = input("Enter a number or provider ID (or press Enter to cancel): ").strip()	1✔
NEW 184	except (EOFError, KeyboardInterrupt):	×
NEW 185	print("\nCancelled.") # noqa: T201	×
NEW 186	sys.exit(0)	×
187
188	if not raw:	1✔
189	print("Cancelled.") # noqa: T201	1✔
190	sys.exit(0)	1✔
191
192	# Accept a number from the menu or a literal provider ID.
193	selected_provider = None	1✔
194	if raw.isdigit():	1✔
195	idx = int(raw) - 1	1✔
196	if 0 <= idx < len(unavailable):	1✔
197	selected_provider = unavailable[idx]	1✔
198	if selected_provider is None:	1✔
199	# Try as a direct provider ID.
200	matches = [p for p in unavailable if p.id == raw]	1✔
201	if matches:	1✔
NEW 202	selected_provider = matches[0]	×
203
204	if selected_provider is None:	1✔
205	print(f"Unknown selection: {raw!r}. Aborting.", file=sys.stderr) # noqa: T201	1✔
206	sys.exit(1)	1✔
207	else:
208	# Direct mode — the user passed a provider ID on the command line.
209	if provider_arg not in get_valid_provider_ids():	1✔
210	print( # noqa: T201	1✔
211	f"Unknown provider: {provider_arg!r}\nValid providers: {', '.join(sorted(get_valid_provider_ids()))}",
212	file=sys.stderr,
213	)
214	sys.exit(1)	1✔
215
216	matches = [p for p in all_providers if p.id == provider_arg]	1✔
217	selected_provider = matches[0] if matches else None	1✔
218	if selected_provider is None:	1✔
NEW 219	print(f"Provider info not found for {provider_arg!r}.", file=sys.stderr) # noqa: T201	×
NEW 220	sys.exit(1)	×
221
222	if selected_provider.available:	1✔
223	print(f"\n✓ {selected_provider.name} is already installed.") # noqa: T201	1✔
224	sys.exit(0)	1✔
225
226	# Run the install.
227	cmd = get_install_command(selected_provider.id)	1✔
228	print(f"\nInstalling {selected_provider.name}…") # noqa: T201	1✔
229	print(f"Running: {' '.join(cmd)}") # noqa: T201	1✔
230	print(_DIVIDER) # noqa: T201	1✔
231
232	returncode, _combined = install(	1✔
233	selected_provider.id,
234	on_output=lambda line: print(line, end="", flush=True), # noqa: T201
235	)
236
237	print(_DIVIDER) # noqa: T201	1✔
238	if returncode == 0:	1✔
239	print(f"\n✓ {selected_provider.name} installed successfully!") # noqa: T201	1✔
240	print("Restart Vector Inspector (or use the 🔄 Refresh button) to use it.") # noqa: T201	1✔
241	sys.exit(0)	1✔
242	else:
243	print(f"\n✗ Installation failed (exit code {returncode}).", file=sys.stderr) # noqa: T201	1✔
244	sys.exit(returncode)	1✔
245
246
247	def parse_cli_args(argv: list[str] \| None = None) -> None:	1✔
248	"""Parse CLI arguments, apply runtime-only env vars, and exit where appropriate.
249
250	Must be called before importing Qt/GUI modules. Returns normally
251	(without exiting) when no early-exit flag is supplied.
252	"""
253	argv = sys.argv[1:] if argv is None else argv	1✔
254
255	# Step 1: Pre-parse non-exit flags so env vars are set before telemetry
256	# fires or --dump-settings reads settings (VI_CONFIG_PATH must be ready).
257	pre = argparse.ArgumentParser(add_help=False)	1✔
258	pre.add_argument("--no-telemetry", action="store_true")	1✔
259	pre.add_argument("--log-level", type=str.upper, default=None)	1✔
260	pre.add_argument("--config", default=None)	1✔
261	pre_ns, _ = pre.parse_known_args(argv)	1✔
262
263	if pre_ns.no_telemetry:	1✔
264	os.environ["VI_NO_TELEMETRY"] = "1"	1✔
265	if pre_ns.log_level:	1✔
266	os.environ["LOG_LEVEL"] = pre_ns.log_level	1✔
267	# Also update the already-imported logger level so the setting takes
268	# effect even on code paths that imported logging before this call.
269	try:	1✔
270	import logging	1✔
271
272	logging.getLogger("vector_inspector").setLevel(getattr(logging, pre_ns.log_level))	1✔
273	except Exception:	×
274	pass	×
275	if pre_ns.config:	1✔
276	os.environ["VI_CONFIG_PATH"] = pre_ns.config	1✔
277
278	# Step 2: Detect command for first-run telemetry before argparse exits.
279	command: str \| None = None	1✔
280	if "--version" in argv:	1✔
281	command = "--version"	1✔
282	elif "--help" in argv or "-h" in argv:	1✔
283	command = "--help"	1✔
284	elif "--dump-settings" in argv:	1✔
285	command = "--dump-settings"	1✔
286
287	if command:	1✔
288	_maybe_send_first_run_telemetry(command)	1✔
289
290	# Step 3: Full parse — exits here for --version / --help.
291	parser = _build_parser()	1✔
292	args = parser.parse_args(argv)	1✔
293
294	# Step 4: --dump-settings is an early exit (no Qt needed).
295	if args.dump_settings:	1✔
296	_handle_dump_settings(args.config)	1✔
297
298	# Step 4b: --install is an early exit (no Qt needed).
299	if args.install is not None:	1✔
300	_handle_install(args.install)	1✔
301
302	# Step 5: Set remaining runtime env vars.
303	if args.no_splash:	1✔
304	os.environ["VI_NO_SPLASH"] = "1"	1✔
305	if args.llm_console:	1✔
306	os.environ["VI_LLM_CONSOLE"] = "1"	×
307
308
309	def console_entry() -> None:	1✔
310	"""Console script entry point for the ``vector-inspector`` command."""
311	parse_cli_args()	×
312	# Only reached when no early-exit flag was given; launch the GUI.
313	# --llm-console (VI_LLM_CONSOLE=1) causes main.py to also open the
314	# LLM debug window alongside the main application window.
315	from vector_inspector.main import main	×
316
317	main()	×

anthonypdawson / vector-inspector / 25969302440

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous