from __future__ import annotations from typing import Any, Dict, Sequence import json import sys from helper.logger import debug, log import sqlite3 from pathlib import Path import models import pipeline as ctx from helper import hydrus as hydrus_wrapper from ._shared import Cmdlet, CmdletArg, normalize_hash def _delete_database_entry(db_path: Path, file_path: str) -> bool: """Delete file and related entries from local library database. Args: db_path: Path to the library.db file file_path: Exact file path string as stored in database Returns: True if successful, False otherwise """ try: if not db_path.exists(): log(f"Database not found at {db_path}", file=sys.stderr) return False conn = sqlite3.connect(db_path) cursor = conn.cursor() log(f"Searching database for file_path: {file_path}", file=sys.stderr) # Find the file_id using the exact file_path cursor.execute('SELECT id FROM files WHERE file_path = ?', (file_path,)) result = cursor.fetchone() if not result: log(f"ERROR: File path not found in database", file=sys.stderr) log(f"Expected: {file_path}", file=sys.stderr) # Debug: show sample entries cursor.execute('SELECT id, file_path FROM files LIMIT 3') samples = cursor.fetchall() if samples: log(f"Sample DB entries:", file=sys.stderr) for fid, fpath in samples: log(f"{fid}: {fpath}", file=sys.stderr) conn.close() return False file_id = result[0] log(f"Found file_id={file_id}, deleting all related records", file=sys.stderr) # Delete related records cursor.execute('DELETE FROM metadata WHERE file_id = ?', (file_id,)) meta_count = cursor.rowcount cursor.execute('DELETE FROM tags WHERE file_id = ?', (file_id,)) tags_count = cursor.rowcount cursor.execute('DELETE FROM notes WHERE file_id = ?', (file_id,)) notes_count = cursor.rowcount cursor.execute('DELETE FROM files WHERE id = ?', (file_id,)) files_count = cursor.rowcount conn.commit() conn.close() log(f"Deleted: metadata={meta_count}, tags={tags_count}, notes={notes_count}, files={files_count}", file=sys.stderr) return True except Exception as exc: log(f"Database cleanup failed: {exc}", file=sys.stderr) import traceback traceback.print_exc(file=sys.stderr) return False def _process_single_item(item: Any, override_hash: str | None, conserve: str | None, lib_root: str | None, reason: str, config: Dict[str, Any]) -> bool: """Process deletion for a single item.""" # 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") origin = item.get("origin") else: hash_hex_raw = getattr(item, "hash_hex", None) or getattr(item, "hash", None) target = getattr(item, "target", None) origin = getattr(item, "origin", None) # For Hydrus files, the target IS the hash if origin and origin.lower() == "hydrus" and not hash_hex_raw: hash_hex_raw = target hash_hex = normalize_hash(override_hash) if override_hash else normalize_hash(hash_hex_raw) local_deleted = False local_target = isinstance(target, str) and target.strip() and not str(target).lower().startswith(("http://", "https://")) if conserve != "local" and local_target: path = Path(str(target)) file_path_str = str(target) # Keep the original string for DB matching try: if path.exists() and path.is_file(): path.unlink() local_deleted = True if ctx._PIPE_ACTIVE: ctx.emit(f"Removed local file: {path}") log(f"Deleted: {path.name}", file=sys.stderr) 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(".tags"), path.with_suffix(".tags.txt"), path.with_suffix(".metadata"), path.with_suffix(".notes")): try: if sidecar.exists() and sidecar.is_file(): sidecar.unlink() except Exception: pass # Clean up database entry if library root provided - do this regardless of file deletion success if lib_root: lib_root_path = Path(lib_root) db_path = lib_root_path / ".downlow_library.db" if _delete_database_entry(db_path, file_path_str): if ctx._PIPE_ACTIVE: ctx.emit(f"Removed database entry: {path.name}") log(f"Database entry cleaned up", file=sys.stderr) local_deleted = True # Mark as deleted if DB cleanup succeeded else: log(f"Database entry not found or cleanup failed for {file_path_str}", file=sys.stderr) else: debug(f"No lib_root provided, skipping database cleanup", file=sys.stderr) hydrus_deleted = False if conserve != "hydrus" and hash_hex: 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 else: if client is None: if not local_deleted: log("Hydrus client unavailable", file=sys.stderr) return False else: 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] hydrus_deleted = True preview = hash_hex[:12] + ('…' if len(hash_hex) > 12 else '') debug(f"Deleted from Hydrus: {preview}…", file=sys.stderr) except Exception as exc: log(f"Hydrus delete failed: {exc}", file=sys.stderr) if not local_deleted: return False if hydrus_deleted and hash_hex: preview = hash_hex[:12] + ('…' if len(hash_hex) > 12 else '') if ctx._PIPE_ACTIVE: if reason: ctx.emit(f"Deleted {preview} (reason: {reason}).") else: ctx.emit(f"Deleted {preview}.") if hydrus_deleted or local_deleted: return True log("Selected result has neither Hydrus hash nor local file target") return False def _run(result: Any, args: Sequence[str], config: Dict[str, Any]) -> int: # Help try: if any(str(a).lower() in {"-?", "/?", "--help", "-h", "help", "--cmdlet"} for a in args): log(json.dumps(CMDLET, ensure_ascii=False, indent=2)) return 0 except Exception: pass 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 {"-hash", "--hash", "hash"} and i + 1 < len(args): override_hash = 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 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 for item in items: if _process_single_item(item, override_hash, conserve, lib_root, reason, config): success_count += 1 return 0 if success_count > 0 else 1 CMDLET = Cmdlet( name="delete-file", summary="Delete a file locally and/or from Hydrus, including database entries.", usage="delete-file [-hash ] [-conserve ] [-lib-root ] [reason]", aliases=["del-file"], args=[ CmdletArg("hash", description="Override the Hydrus file hash (SHA256) to target instead of the selected result."), CmdletArg("conserve", description="Choose which copy to keep: 'local' or 'hydrus'."), CmdletArg("lib-root", description="Path to local library root for database cleanup."), CmdletArg("reason", description="Optional reason for deletion (free text)."), ], details=[ "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.", ], )