from __future__ import annotations from typing import Any, Dict, Sequence import sys from . import register import pipeline as ctx from ._shared import Cmdlet, CmdletArg, SharedArgs, parse_cmdlet_args, get_field, normalize_hash from helper.logger import log from helper.store import FileStorage class Get_Url(Cmdlet): """Get url associated with files via hash+store.""" NAME = "get-url" SUMMARY = "List url associated with a file" USAGE = "@1 | get-url" ARGS = [ SharedArgs.HASH, SharedArgs.STORE, ] DETAIL = [ "- Lists all url associated with file identified by hash+store", ] def run(self, result: Any, args: Sequence[str], config: Dict[str, Any]) -> int: """Get url for file via hash+store backend.""" parsed = parse_cmdlet_args(args, self) # Extract hash and store from result or args file_hash = parsed.get("hash") or get_field(result, "hash") store_name = parsed.get("store") or get_field(result, "store") if not file_hash: log("Error: No file hash provided") return 1 if not store_name: log("Error: No store name provided") return 1 # Normalize hash file_hash = normalize_hash(file_hash) if not file_hash: log("Error: Invalid hash format") return 1 # Get backend and retrieve url try: storage = FileStorage(config) backend = storage[store_name] url = backend.get_url(file_hash) if url: for url in url: # Emit rich object for pipeline compatibility ctx.emit({ "url": url, "hash": file_hash, "store": store_name, }) return 0 else: ctx.emit("No url found") return 0 except KeyError: log(f"Error: Storage backend '{store_name}' not configured") return 1 except Exception as exc: log(f"Error retrieving url: {exc}", file=sys.stderr) return 1 # Register cmdlet register(["get-url", "get_url"])(Get_Url)