Something went wrong. Try again.
Command line interface for accessing the Open Web Index shards and slices.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511"""Admin Commands - Typer CLI for administrative operations."""from typing import Optionalfrom pathlib import Pathimport typerfrom rich.table import Table
from ._common.context import CLIContextfrom ._common.ui import ask_yes_nofrom owilix.core.utils import split_query_access
# Create sub-app for admin commandsapp = typer.Typer( name="admin", help="Administrative operations", no_args_is_help=True,)
def _all_repo_names(cli_ctx: CLIContext) -> list[str]: try: return sorted(cli_ctx.owi.remote_data.get_repo_names()) except Exception: return []
def _get_repo(cli_ctx: CLIContext, repository: str): repo = cli_ctx.owi.remote_data.get_single_repo(repository) if repo is None: raise ValueError(f"Repository '{repository}' not found") return repo
@app.command("repos")def repos(ctx: typer.Context): """ List configured repositories and status. """ cli_ctx: CLIContext = ctx.obj names = _all_repo_names(cli_ctx) if not names: cli_ctx.console.print("[yellow]No repositories configured[/yellow]") return
table = Table(title="Configured Repositories") table.add_column("Name", style="cyan") table.add_column("Backend", style="magenta") table.add_column("Status", style="green") table.add_column("Details", style="dim")
for name in names: try: repo = _get_repo(cli_ctx, name) backend = getattr(repo, "_backend_name", "unknown") st = repo.status() if hasattr(repo, "status") else {"status": None, "message": "n/a"} ok = bool(st.get("status")) status_text = "ok" if ok else "error" details = str(st.get("message") or "").strip() table.add_row(name, str(backend), status_text, details) except Exception as exc: table.add_row(name, "unknown", "error", str(exc))
cli_ctx.console.print(table)
@app.command("path")def path_check( ctx: typer.Context, path: str = typer.Argument(..., help="Remote path to check (e.g., /IT4ILexisV2/public/<dataset-id>)"), repository: Optional[str] = typer.Option(None, "--repository", "-r", help="Repository name (default: check all)"),): """ Check whether a path exists in one repository or all configured repositories. """ cli_ctx: CLIContext = ctx.obj names = [repository] if repository else _all_repo_names(cli_ctx) if not names: cli_ctx.console.print("[yellow]No repositories configured[/yellow]") return
table = Table(title=f"Path Check: {path}") table.add_column("Repository", style="cyan") table.add_column("Exists", style="green") table.add_column("Type", style="yellow") table.add_column("Info", style="dim")
for name in names: try: repo = _get_repo(cli_ctx, name) fs = getattr(repo, "fs", None) if fs is None: table.add_row(name, "no", "n/a", "repository has no filesystem") continue exists = fs.exists(path) if not exists: table.add_row(name, "no", "-", "") continue is_dir = fs.isdir(path) info = fs.info(path) kind = "directory" if is_dir else "file" info_bits = [] if isinstance(info, dict): size = info.get("size") if size is not None: info_bits.append(f"size={size}") table.add_row(name, "yes", kind, ", ".join(info_bits)) except Exception as exc: table.add_row(name, "error", "-", str(exc))
cli_ctx.console.print(table)
@app.command("fs")def filesystem( ctx: typer.Context, operation: str = typer.Argument(..., help="Operation: ls, exists, info, find, glob, cat, mkdir, cp, mv, rm"), path: str = typer.Argument(..., help="Path (source path for cp/mv)"), repository: str = typer.Option("lexis", "--repository", "-r", help="Repository name"), target_path: Optional[str] = typer.Option(None, "--target-path", "-t", help="Target path for cp/mv"), pattern: Optional[str] = typer.Option(None, "--pattern", "-p", help="Pattern for glob/find"), recursive: bool = typer.Option(False, "--recursive", help="Recursive remove or listing behavior where applicable"), max_items: int = typer.Option(50, "--max-items", help="Maximum items to display for ls/find/glob"), yes: bool = typer.Option(False, "--yes", "-y", help="Skip confirmation for modifying operations"),): """ Run low-level filesystem operations on repository backends. """ cli_ctx: CLIContext = ctx.obj op = operation.strip().lower() mutable_ops = {"mkdir", "cp", "mv", "rm"} read_ops = {"ls", "exists", "info", "find", "glob", "cat"} allowed = mutable_ops | read_ops if op not in allowed: cli_ctx.console.print(f"[red]Unsupported operation '{operation}'. Allowed: {', '.join(sorted(allowed))}[/red]") raise typer.Exit(code=1)
repo = _get_repo(cli_ctx, repository) fs = getattr(repo, "fs", None) if fs is None: cli_ctx.console.print(f"[red]Repository '{repository}' has no filesystem[/red]") raise typer.Exit(code=1)
if op in mutable_ops and not yes: action = f"{op} on repository={repository} path={path}" if target_path: action += f" target={target_path}" if not ask_yes_no(cli_ctx.console, f"Proceed with {action}?"): cli_ctx.console.print("[yellow]Aborted[/yellow]") raise typer.Exit(code=1)
try: if op == "exists": cli_ctx.console.print(str(fs.exists(path))) return if op == "info": cli_ctx.console.print(fs.info(path)) return if op == "ls": items = fs.ls(path, detail=False) for item in items[:max_items]: cli_ctx.console.print(item) if len(items) > max_items: cli_ctx.console.print(f"[dim]... {len(items) - max_items} more[/dim]") return if op == "find": found = fs.find(path, withdirs=recursive) if pattern: import fnmatch found = [f for f in found if fnmatch.fnmatch(str(f), pattern)] for item in found[:max_items]: cli_ctx.console.print(item) if len(found) > max_items: cli_ctx.console.print(f"[dim]... {len(found) - max_items} more[/dim]") return if op == "glob": pat = pattern or path items = fs.glob(pat) for item in items[:max_items]: cli_ctx.console.print(item) if len(items) > max_items: cli_ctx.console.print(f"[dim]... {len(items) - max_items} more[/dim]") return if op == "cat": with fs.open(path, "rb") as handle: data = handle.read(4096) try: cli_ctx.console.print(data.decode("utf-8", errors="replace")) except Exception: cli_ctx.console.print(data) return
if op == "mkdir": fs.makedirs(path, exist_ok=True) cli_ctx.console.print(f"[green]created:[/green] {path}") return if op == "cp": if not target_path: raise ValueError("--target-path is required for cp") if not hasattr(fs, "cp"): raise ValueError("Filesystem does not support cp") fs.cp(path, target_path, recursive=recursive) cli_ctx.console.print(f"[green]copied:[/green] {path} -> {target_path}") return if op == "mv": if not target_path: raise ValueError("--target-path is required for mv") if not hasattr(fs, "mv"): raise ValueError("Filesystem does not support mv") fs.mv(path, target_path, recursive=recursive) cli_ctx.console.print(f"[green]moved:[/green] {path} -> {target_path}") return if op == "rm": fs.rm(path, recursive=recursive) cli_ctx.console.print(f"[green]removed:[/green] {path}") return except Exception as exc: cli_ctx.console.print(f"[red]filesystem operation failed:[/red] {exc}") raise typer.Exit(code=1)
@app.command()def logs( ctx: typer.Context, log_type: str = typer.Argument("latest", help="Log selector: latest, location, events, sql, or a concrete filename"), lines: int = typer.Option(50, "--lines", "-n", help="Number of lines to show"), follow: bool = typer.Option(False, "--follow", "-f", help="Follow log output"),): """ View application logs. Shows recent log entries from the selected log target.
Examples: owi admin logs latest owi admin logs sql --lines 100 owi admin logs events owi admin logs location """ cli_ctx: CLIContext = ctx.obj log_dir = Path(cli_ctx.target) / ".logs" if not log_dir.exists(): cli_ctx.console.print(f"[yellow]No logs directory found at {log_dir}[/yellow]") return
known_logs = { "events": log_dir / "events.json", "sql": log_dir / "sql.log", "lexis": log_dir / "lexis.log", "errors": log_dir / "errors.log", }
if log_type == "location": cli_ctx.console.print(f"[bold]Log directory:[/bold] {log_dir}") for name, path in known_logs.items(): status = "exists" if path.exists() else "missing" cli_ctx.console.print(f" • {name}: {path} [{status}]") return
available_logs = sorted({p for p in log_dir.iterdir() if p.is_file() and p.suffix in {".log", ".json"}}) if not available_logs: cli_ctx.console.print("[yellow]No log files found[/yellow]") return
cli_ctx.console.print(f"[bold]Available logs in {log_dir}:[/bold]") for lf in available_logs: cli_ctx.console.print(f" • {lf.name}")
if log_type == "latest": selected = max(available_logs, key=lambda p: p.stat().st_mtime) else: selected = known_logs.get(log_type) if selected is None: selected = log_dir / log_type if not selected.exists(): cli_ctx.console.print(f"[red]Log file not found:[/red] {selected}") return
cli_ctx.console.print(f"\n[bold]Showing last {lines} lines from {selected.name}:[/bold]\n")
with open(selected, 'r', encoding='utf-8') as f: all_lines = f.readlines() for line in all_lines[-lines:]: # Color based on level if "ERROR" in line or "CRITICAL" in line: cli_ctx.console.print(f"[red]{line.rstrip()}[/red]") elif "WARNING" in line: cli_ctx.console.print(f"[yellow]{line.rstrip()}[/yellow]") elif "INFO" in line: cli_ctx.console.print(f"[blue]{line.rstrip()}[/blue]") else: cli_ctx.console.print(line.rstrip())
@app.command()def stats(ctx: typer.Context): """ Display CLI usage statistics. Shows command invocation counts, durations, and repository statistics. Examples: owi admin stats """ cli_ctx: CLIContext = ctx.obj from rich.table import Table import json from datetime import datetime # Stats file location stats_file = Path(cli_ctx.target) / ".stats" / "usage.json" if not stats_file.exists(): cli_ctx.console.print("[yellow]No usage statistics found.[/yellow]") cli_ctx.console.print(f"Stats will be recorded at: {stats_file}") return with open(stats_file, 'r') as f: stats_data = json.load(f) # Command usage table if "commands" in stats_data: cmd_table = Table(title="Command Usage Statistics") cmd_table.add_column("Command", style="cyan") cmd_table.add_column("Count", style="green") cmd_table.add_column("Avg Duration", style="yellow") cmd_table.add_column("Last Used", style="blue") for cmd, data in sorted(stats_data["commands"].items(), key=lambda x: x[1].get("count", 0), reverse=True): avg_duration = data.get("total_duration", 0) / max(data.get("count", 1), 1) last_used = data.get("last_used", "N/A") cmd_table.add_row( cmd, str(data.get("count", 0)), f"{avg_duration:.2f}s", last_used ) cli_ctx.console.print(cmd_table) # Repository stats if "repositories" in stats_data: repo_table = Table(title="Repository Statistics") repo_table.add_column("Repository", style="cyan") repo_table.add_column("Datasets", style="green") repo_table.add_column("Total Size", style="yellow") for repo, data in stats_data["repositories"].items(): repo_table.add_row( repo, str(data.get("dataset_count", 0)), f"{data.get('total_size', 0) / 1e9:.2f} GB" ) cli_ctx.console.print(repo_table) # Overall stats cli_ctx.console.print(f"\n[bold]Total commands executed:[/bold] {stats_data.get('total_commands', 0)}") cli_ctx.console.print(f"[bold]Stats since:[/bold] {stats_data.get('started', 'Unknown')}")
@app.command()def check( ctx: typer.Context, specifier: str = typer.Argument(..., help="Dataset specifier (e.g., 'lrz:latest')"), repair_mode: str = typer.Option( "syntax", "--repair", "-r", help="Repair mode: None, 'syntax' (metadata only), 'infer' (from filesystem)" ), update_mode: str = typer.Option( "interactive", "--update", "-u", help="Update mode: None, 'interactive', 'update'" ),): """ Check and repair dataset metadata. Validates metadata for datasets matching the specifier. Can optionally repair syntax issues or infer metadata from files. Examples: owi admin check lrz:latest owi admin check main:* --repair=infer --update=update """ cli_ctx: CLIContext = ctx.obj from rich.prompt import Prompt from owilix.core.models.dataset import fill_metadata, format_metadata from ._common.ui import input_dict
def correct_ui(metadata): """ corrects the metadata for the UI """ while True: choice = Prompt.ask("Enter action to correct ('delete','edit', 'fill', 'format', 'update', 'done') ", default="done") if choice == 'done' or choice == 'update': break elif choice == 'delete': return choice, None elif choice == 'fill': metadata = fill_metadata(metadata) elif choice == 'format': metadata = format_metadata(metadata) elif choice == 'edit': metadata = input_dict(metadata, title="Modify Dataset Metadata", console=cli_ctx.console) return choice, metadata
def _do_special_checks(d, _checks, repair_mode=None):
def _list_of_dict_of_single_char(v): if v is None: return False if isinstance(v,list) and len(v)>0 and isinstance(v[0],dict) and all([len(d.get("name", ""))==1 for d in v]): return True return False
for k in ["creators", "contributors"]: v = d.metadata.get_container(k) if v is None: continue if _list_of_dict_of_single_char(v.data): if repair_mode is not None: v.fill_defaults() _checks[k] = ("repaired", f"{k} was a dict of single char. now filled with defaults") else: _checks[k] = ("warning", f"{k} is dict of single char")
def _print_checks(_checks, indent="\t"): for k, v in _checks.items(): cli_ctx.console.print(f"{indent} {k}: {v[0]} - {v[1]}")
repair_mode = repair_mode if repair_mode != "None" else None update_mode = update_mode if update_mode != "None" else None
cli_ctx.console.print(f"Fetching datasets for specifier {specifier}") # Use owi.remote_data for listing as checking is usually against remote/local abstraction # But wait, 'check' logic can be local too. cmd/admin.py used list_remote_dataset_by_specifier # We should probably support both, but for now let's stick to what cmd/admin.py did: # It accessed self.list_remote_dataset_by_specifier which calls self.owi.remote_data.list # However, cli/admin/commands/check documentation says "remote data centers". # And logic suggests it patches metadata. spec = cli_ctx.owi.parse_specifier(specifier) access, query = split_query_access(spec.get("query")) datasets = cli_ctx.owi.remote_data.list(spec.get("data_center"), access, day=spec.get("day"), duration=spec.get("duration", 0), query=query) cli_ctx.console.print(f"Found {len(datasets)} datasets to check.") count_ok = 0 count_error = 0
for d in datasets: try: _checks = d.check_metadata() _check = lambda x: any([v[0]!="ok" for v in x.values()]) _do_special_checks(d, _checks) if any([v[0]!="ok" for v in _checks.values()]): # metadata seems to be wrong. Let's try to fix it. cli_ctx.console.print(f"[violet]WARNING[/violet]:{d.metadata.id}-{d.metadata.title} Metadata errors occured (repair mode = {repair_mode}).") if repair_mode is not None: # now patch from the metadata at the files system. _wrong_fields= [k for k,v in _checks.items() if v[0]!="ok"] d._patch_metadata(fields=_wrong_fields) # load the metadata again from the path. It could be, that server side some updates have not been performed _checks = d.check_metadata() _do_special_checks(d, _checks) if _check(_checks): # we now need to do the inferencing of the types and correct it. cli_ctx.console.print(f"\t [violet]patching did not resolve everything. Trying to infer from {'all sources (may take long):' if repair_mode=='infer' else 'metadata only'}[/violet] ") _print_checks(_checks, "\t\t") d.check_metadata(repair_mode=repair_mode) # no time to make that nicer. IdeaL: here we repair _do_special_checks(d, _checks, repair_mode=repair_mode) _checks = d.check_metadata() # check again _do_special_checks(d, _checks) if _check(_checks): cli_ctx.console.print(f"\t [violet]inference (mode ={repair_mode} could did not resolve everything. The following issues remain[/violet]") _print_checks(_checks) else: cli_ctx.console.print(f"\t [violet]successfully patched {_wrong_fields} from storage metadata [/violet] ")
if update_mode=="interactive": choice, metadata = correct_ui(d.metadata) if choice == 'update': d.metadata = metadata d.save_metadata() elif update_mode=="update": for f in _wrong_fields: cli_ctx.console.print(f"\t changed {f}: {d.metadata[f]}") d.save_metadata(force=True) else: cli_ctx.console.print(f"\t Not updating metadata on server. Use --update_mode=interactive or --update_mode=update or to update metadata on server") else: _print_checks(_checks) count_error += 1 else: cli_ctx.console.print(f"[green]OK[/green]: {d.metadata.id}-{d.metadata.title}") count_ok += 1
except Exception as e: #import traceback cli_ctx.console.print(f"[red]ERROR[/red]: {d.metadata.id}-{d.metadata.title} raised exception {e}") count_error += 1 cli_ctx.console.print(f"Check complete. {count_ok} OK, {count_error} issues found.")