"""Delete-file cmdlet: Delete files from local storage and/or Hydrus.""" from __future__ import annotations from typing import Any, Dict, List, Sequence import sys from pathlib import Path from SYS.logger import debug, log from SYS.utils import format_bytes from Store.Folder import Folder from Store import Store from . import _shared as sh from API import HydrusNetwork as hydrus_wrapper from SYS import pipeline as ctx from SYS.result_table import ResultTable, _format_size from SYS.rich_display import stdout_console class Delete_File(sh.Cmdlet): """Class-based delete-file cmdlet with self-registration.""" def __init__(self) -> None: super().__init__( name="delete-file", summary= "Delete a file locally and/or from Hydrus, including database entries.", usage= 'delete-file [-query "hash:"] [-conserve ] [-lib-root ] [reason]', alias=["del-file"], arg=[ sh.SharedArgs.QUERY, sh.CmdletArg( "conserve", description="Choose which copy to keep: 'local' or 'hydrus'." ), sh.CmdletArg( "lib-root", description="Path to local library root for database cleanup." ), sh.CmdletArg( "reason", description="Optional reason for deletion (free text)." ), ], detail=[ "Default removes both the local file and Hydrus file.", "Use -conserve local to keep the local file, or -conserve hydrus to keep it in Hydrus.", "Database entries are automatically cleaned up for local files.", "Any remaining arguments are treated as the Hydrus reason text.", ], exec=self.run, ) self.register() def _process_single_item( self, item: Any, override_hash: str | None, conserve: str | None, lib_root: str | None, reason: str, config: Dict[str, Any], ) -> List[Dict[str, Any]]: """Process deletion for a single item. Returns display rows (for the final Rich table). Returning an empty list indicates no delete occurred. """ # Handle item as either dict or object if isinstance(item, dict): hash_hex_raw = item.get("hash_hex") or item.get("hash") target = item.get("target") or item.get("file_path") or item.get("path") title_val = item.get("title") or item.get("name") else: hash_hex_raw = sh.get_field(item, "hash_hex") or sh.get_field(item, "hash") target = ( sh.get_field(item, "target") or sh.get_field(item, "file_path") or sh.get_field(item, "path") ) title_val = sh.get_field(item, "title") or sh.get_field(item, "name") def _get_ext_from_item() -> str: try: if isinstance(item, dict): ext_val = item.get("ext") if ext_val: return str(ext_val) extra = item.get("extra") if isinstance(extra, dict) and extra.get("ext"): return str(extra.get("ext")) else: ext_val = sh.get_field(item, "ext") if ext_val: return str(ext_val) extra = sh.get_field(item, "extra") if isinstance(extra, dict) and extra.get("ext"): return str(extra.get("ext")) except Exception: pass # Fallback: infer from target path or title if it looks like a filename try: if isinstance(target, str) and target: suffix = Path(target).suffix if suffix: return suffix.lstrip(".") except Exception: pass try: if title_val: suffix = Path(str(title_val)).suffix if suffix: return suffix.lstrip(".") except Exception: pass return "" store = None if isinstance(item, dict): store = item.get("store") else: store = sh.get_field(item, "store") store_lower = str(store).lower() if store else "" backend = None try: if store: registry = Store(config) if registry.is_available(str(store)): backend = registry[str(store)] except Exception: backend = None # Determine whether the store backend is HydrusNetwork. # IMPORTANT: Hydrus instances are named by the user (e.g. 'home', 'rpi'), # so checking only the store name is unreliable. is_hydrus_store = False try: if backend is not None: from Store.HydrusNetwork import HydrusNetwork as HydrusStore is_hydrus_store = isinstance(backend, HydrusStore) except Exception: is_hydrus_store = False # Backwards-compatible fallback heuristic (older items might only carry a name). if ((not is_hydrus_store) and bool(store_lower) and ("hydrus" in store_lower or store_lower in {"home", "work"})): is_hydrus_store = True store_label = str(store) if store else "default" hydrus_prefix = f"[hydrusnetwork:{store_label}]" # For Hydrus files, the target IS the hash if is_hydrus_store and not hash_hex_raw: hash_hex_raw = target hash_hex = ( sh.normalize_hash(override_hash) if override_hash else sh.normalize_hash(hash_hex_raw) ) local_deleted = False local_target = ( isinstance(target, str) and target.strip() and not str(target).lower().startswith(("http://", "https://")) ) deleted_rows: List[Dict[str, Any]] = [] # If this item references a configured non-Hydrus store backend, prefer deleting # via the backend API. This supports store items where `path`/`target` is the hash. if conserve != "local" and store and (not is_hydrus_store): try: # Re-use an already resolved backend when available. if backend is None: registry = Store(config) if registry.is_available(str(store)): backend = registry[str(store)] if backend is not None: # Prefer hash when available. hash_candidate = sh.normalize_hash( hash_hex_raw ) if hash_hex_raw else None if not hash_candidate and isinstance(target, str): hash_candidate = sh.normalize_hash(target) resolved_path = None try: if hash_candidate and hasattr(backend, "get_file"): candidate_path = backend.get_file(hash_candidate) resolved_path = ( candidate_path if isinstance(candidate_path, Path) else None ) except Exception: resolved_path = None identifier = hash_candidate or ( str(target).strip() if isinstance(target, str) else "" ) if identifier: deleter = getattr(backend, "delete_file", None) if callable(deleter) and bool(deleter(identifier)): local_deleted = True size_bytes: int | None = None try: if (resolved_path is not None and isinstance(resolved_path, Path) and resolved_path.exists()): size_bytes = int(resolved_path.stat().st_size) except Exception: size_bytes = None deleted_rows.append( { "title": ( str(title_val).strip() if title_val else ( resolved_path.name if resolved_path else identifier ) ), "store": store_label, "hash": hash_candidate or (hash_hex or ""), "size_bytes": size_bytes, "ext": _get_ext_from_item() or ( resolved_path.suffix.lstrip(".") if resolved_path else "" ), } ) # Best-effort remove sidecars if we know the resolved path. try: if resolved_path is not None and isinstance( resolved_path, Path): for sidecar in ( resolved_path.with_suffix(".tag"), resolved_path.with_suffix(".metadata"), resolved_path.with_suffix(".notes"), ): try: if sidecar.exists() and sidecar.is_file(): sidecar.unlink() except Exception: pass except Exception: pass # Skip legacy local-path deletion below. local_target = False except Exception: pass if conserve != "local" and local_target: path = Path(str(target)) size_bytes: int | None = None try: if path.exists() and path.is_file(): size_bytes = int(path.stat().st_size) except Exception: size_bytes = None # If lib_root is provided and this is from a folder store, use the Folder class if lib_root: try: folder = Folder(Path(lib_root), name=store or "local") if folder.delete_file(str(path)): local_deleted = True deleted_rows.append( { "title": str(title_val).strip() if title_val else path.name, "store": store_label, "hash": hash_hex or sh.normalize_hash(path.stem) or "", "size_bytes": size_bytes, "ext": _get_ext_from_item() or path.suffix.lstrip("."), } ) except Exception as exc: debug(f"Folder.delete_file failed: {exc}", file=sys.stderr) # Fallback to manual deletion try: if path.exists() and path.is_file(): path.unlink() local_deleted = True deleted_rows.append( { "title": str(title_val).strip() if title_val else path.name, "store": store_label, "hash": hash_hex or sh.normalize_hash(path.stem) or "", "size_bytes": size_bytes, "ext": _get_ext_from_item() or path.suffix.lstrip("."), } ) except Exception as exc: log(f"Local delete failed: {exc}", file=sys.stderr) else: # No lib_root, just delete the file try: if path.exists() and path.is_file(): path.unlink() local_deleted = True deleted_rows.append( { "title": str(title_val).strip() if title_val else path.name, "store": store_label, "hash": hash_hex or sh.normalize_hash(path.stem) or "", "size_bytes": size_bytes, "ext": _get_ext_from_item() or path.suffix.lstrip("."), } ) except Exception as exc: log(f"Local delete failed: {exc}", file=sys.stderr) # Remove common sidecars regardless of file removal success for sidecar in ( path.with_suffix(".tag"), path.with_suffix(".metadata"), path.with_suffix(".notes"), ): try: if sidecar.exists() and sidecar.is_file(): sidecar.unlink() except Exception: pass hydrus_deleted = False should_try_hydrus = is_hydrus_store # If conserve is set to hydrus, definitely don't delete if conserve == "hydrus": should_try_hydrus = False if should_try_hydrus and hash_hex: # Prefer deleting via the resolved store backend when it is a HydrusNetwork store. # This ensures store-specific post-delete hooks run (e.g., clearing Hydrus deletion records). did_backend_delete = False try: if backend is not None: deleter = getattr(backend, "delete_file", None) if callable(deleter): did_backend_delete = bool(deleter(hash_hex, reason=reason)) except Exception: did_backend_delete = False if did_backend_delete: hydrus_deleted = True title_str = str(title_val).strip() if title_val else "" if title_str: debug( f"{hydrus_prefix} Deleted title:{title_str} hash:{hash_hex}", file=sys.stderr, ) else: debug(f"{hydrus_prefix} Deleted hash:{hash_hex}", file=sys.stderr) else: # Fallback to direct client calls. client = None if store: # Store specified: do not fall back to a global/default Hydrus client. try: registry = Store(config) backend = registry[str(store)] candidate = getattr(backend, "_client", None) if candidate is not None and hasattr(candidate, "_post"): client = candidate except Exception as exc: if not local_deleted: log( f"Hydrus client unavailable for store '{store}': {exc}", file=sys.stderr, ) return False if client is None: if not local_deleted: log( f"Hydrus client unavailable for store '{store}'", file=sys.stderr ) return False else: # No store context; use default Hydrus client. try: client = hydrus_wrapper.get_client(config) except Exception as exc: if not local_deleted: log(f"Hydrus client unavailable: {exc}", file=sys.stderr) return False if client is None: if not local_deleted: log("Hydrus client unavailable", file=sys.stderr) return False payload: Dict[str, Any] = { "hashes": [hash_hex] } if reason: payload["reason"] = reason try: client._post( "/add_files/delete_files", data=payload ) # type: ignore[attr-defined] # Best-effort clear deletion record if supported by this client. try: clearer = getattr(client, "clear_file_deletion_record", None) if callable(clearer): clearer([hash_hex]) else: client._post( "/add_files/clear_file_deletion_record", data={ "hashes": [hash_hex] } ) # type: ignore[attr-defined] except Exception: pass hydrus_deleted = True title_str = str(title_val).strip() if title_val else "" if title_str: debug( f"{hydrus_prefix} Deleted title:{title_str} hash:{hash_hex}", file=sys.stderr, ) else: debug( f"{hydrus_prefix} Deleted hash:{hash_hex}", file=sys.stderr ) except Exception: # If it's not in Hydrus (e.g. 404 or similar), that's fine if not local_deleted: return [] if hydrus_deleted and hash_hex: size_hint = None try: if isinstance(item, dict): size_hint = item.get("size_bytes") or item.get("size") else: size_hint = sh.get_field(item, "size_bytes" ) or sh.get_field(item, "size") except Exception: size_hint = None deleted_rows.append( { "title": str(title_val).strip() if title_val else "", "store": store_label, "hash": hash_hex, "size_bytes": size_hint, "ext": _get_ext_from_item(), } ) if hydrus_deleted or local_deleted: return deleted_rows log("Selected result has neither Hydrus hash nor local file target") return [] def run(self, result: Any, args: Sequence[str], config: Dict[str, Any]) -> int: """Execute delete-file command.""" if sh.should_show_help(args): log(f"Cmdlet: {self.name}\nSummary: {self.summary}\nUsage: {self.usage}") return 0 # Parse arguments override_query: str | None = None override_hash: str | None = None conserve: str | None = None lib_root: str | None = None reason_tokens: list[str] = [] i = 0 while i < len(args): token = args[i] low = str(token).lower() if low in {"-query", "--query", "query"} and i + 1 < len(args): override_query = str(args[i + 1]).strip() i += 2 continue if low in {"-conserve", "--conserve"} and i + 1 < len(args): value = str(args[i + 1]).strip().lower() if value in {"local", "hydrus"}: conserve = value i += 2 continue if low in {"-lib-root", "--lib-root", "lib-root"} and i + 1 < len(args): lib_root = str(args[i + 1]).strip() i += 2 continue reason_tokens.append(token) i += 1 override_hash = sh.parse_single_hash_query( override_query ) if override_query else None if override_query and not override_hash: log("Invalid -query value (expected hash:)", file=sys.stderr) return 1 # If no lib_root provided, try to get the first folder store from config if not lib_root: try: storage_config = config.get("storage", {}) folder_config = storage_config.get("folder", {}) if folder_config: # Get first folder store path for store_name, store_config in folder_config.items(): if isinstance(store_config, dict): path = store_config.get("path") if path: lib_root = path break except Exception: pass reason = " ".join(token for token in reason_tokens if str(token).strip()).strip() items = [] if isinstance(result, list): items = result elif result: items = [result] if not items: log("No items to delete", file=sys.stderr) return 1 success_count = 0 deleted_rows: List[Dict[str, Any]] = [] for item in items: rows = self._process_single_item( item, override_hash, conserve, lib_root, reason, config ) if rows: success_count += 1 deleted_rows.extend(rows) if deleted_rows: table = ResultTable("Deleted") table.set_no_choice(True).set_preserve_order(True) for row in deleted_rows: result_row = table.add_row() result_row.add_column("Title", row.get("title", "")) result_row.add_column("Store", row.get("store", "")) result_row.add_column("Hash", row.get("hash", "")) result_row.add_column( "Size", _format_size(row.get("size_bytes"), integer_only=False) ) result_row.add_column("Ext", row.get("ext", "")) # Display-only: print directly and do not affect selection/history. try: stdout_console().print() stdout_console().print(table) setattr(table, "_rendered_by_cmdlet", True) except Exception: pass # Ensure no stale overlay/selection carries forward. try: ctx.set_last_result_items_only([]) ctx.set_current_stage_table(None) except Exception: pass return 0 if success_count > 0 else 1 # Instantiate and register the cmdlet Delete_File()