This commit is contained in:
2026-02-02 19:49:07 -08:00
parent 8d22ec5a81
commit 1e0000ae19
13 changed files with 297 additions and 988 deletions

View File

@@ -13,6 +13,7 @@ from typing import Any, Dict, List, Sequence, Set
from urllib.parse import parse_qs, urlparse
from SYS.logger import log
from SYS.utils import extract_hydrus_hash_from_url
from SYS import pipeline as ctx
from SYS.config import resolve_output_dir
@@ -64,17 +65,8 @@ def _extract_url(item: Any) -> str:
def _extract_hash_from_hydrus_file_url(url: str) -> str:
try:
parsed = urlparse(str(url))
if not (parsed.path or "").endswith("/get_files/file"):
return ""
qs = parse_qs(parsed.query or "")
h = (qs.get("hash") or [""])[0]
if isinstance(h, str) and _SHA256_RE.fullmatch(h.strip()):
return h.strip().lower()
except Exception:
pass
return ""
"""Extract hash from Hydrus URL using centralized utility."""
return extract_hydrus_hash_from_url(url) or ""
def _hydrus_instance_names(config: Dict[str, Any]) -> Set[str]:

View File

@@ -43,7 +43,18 @@ class Get_Metadata(Cmdlet):
@staticmethod
def _extract_imported_ts(meta: Dict[str, Any]) -> Optional[int]:
"""Extract an imported timestamp from metadata if available."""
"""Extract an imported timestamp from metadata if available.
Attempts to parse imported timestamp from metadata dict in multiple formats:
- Numeric Unix timestamp (int/float)
- ISO format string (e.g., "2024-01-15T10:30:00")
Args:
meta: Metadata dictionary from backend (e.g., from get_metadata())
Returns:
Unix timestamp as integer if found, None otherwise
"""
if not isinstance(meta, dict):
return None
@@ -65,7 +76,17 @@ class Get_Metadata(Cmdlet):
@staticmethod
def _format_imported(ts: Optional[int]) -> str:
"""Format timestamp as readable string."""
"""Format Unix timestamp as human-readable date string (UTC).
Converts Unix timestamp to YYYY-MM-DD HH:MM:SS format.
Used for displaying file import dates to users.
Args:
ts: Unix timestamp (integer) or None
Returns:
Formatted date string (e.g., "2024-01-15 10:30:00") or empty string if invalid
"""
if not ts:
return ""
try:
@@ -91,7 +112,30 @@ class Get_Metadata(Cmdlet):
ext: Optional[str] = None,
) -> Dict[str,
Any]:
"""Build a table row dict with metadata fields."""
"""Build a normalized metadata row dict for display and piping.
Converts raw metadata fields into a standardized row format suitable for:
- Display in result tables
- Piping to downstream cmdlets
- JSON serialization
Args:
title: File or resource title
store: Backend store name (e.g., "hydrus", "local")
path: File path or resource identifier
mime: MIME type (e.g., "image/jpeg", "video/mp4")
size_bytes: File size in bytes
dur_seconds: Duration in seconds (for video/audio)
imported_ts: Unix timestamp when item was imported
url: List of known URLs associated with file
hash_value: File hash (SHA256 or other)
pages: Number of pages (for PDFs)
tag: List of tags applied to file
ext: File extension (e.g., "jpg", "mp4")
Returns:
Dictionary with normalized metadata fields and display columns
"""
size_mb = None
size_int: Optional[int] = None
if size_bytes is not None:
@@ -151,7 +195,15 @@ class Get_Metadata(Cmdlet):
@staticmethod
def _add_table_body_row(table: Table, row: Dict[str, Any]) -> None:
"""Add a single row to the ResultTable using the prepared columns."""
"""Add a single metadata row to the result table.
Extracts column values from row dict and adds to result table using
standard column ordering (Hash, MIME, Size, Duration/Pages).
Args:
table: Result table to add row to
row: Metadata row dict (from _build_table_row)
"""
columns = row.get("columns") if isinstance(row, dict) else None
lookup: Dict[str,
Any] = {}
@@ -173,7 +225,25 @@ class Get_Metadata(Cmdlet):
row_obj.add_column("Duration(s)", "")
def run(self, result: Any, args: Sequence[str], config: Dict[str, Any]) -> int:
"""Main execution entry point."""
"""Execute get-metadata cmdlet - retrieve and display file metadata.
Queries a storage backend (Hydrus, local, etc.) for file metadata using hash.
Extracts tags embedded in metadata response (avoiding duplicate API calls).
Displays metadata in rich detail panel and result table.
Allows piping (@N) to other cmdlets for chaining operations.
Optimizations:
- Extracts tags from metadata response (no separate get_tag() call)
- Single HTTP request to backends per file
Args:
result: Piped input (dict with optional hash/store/title/tag fields)
args: Command line arguments ([-query "hash:..."] [-store backend])
config: Application configuration dict
Returns:
0 on success, 1 on error (no metadata found, backend unavailable, etc.)
"""
# Parse arguments
parsed = parse_cmdlet_args(args, self)

View File

@@ -318,8 +318,24 @@ def _emit_tags_as_table(
) -> None:
"""Emit tags as TagItem objects and display via ResultTable.
This replaces _print_tag_list to make tags pipe-able.
Stores the table via ctx.set_last_result_table_overlay (or ctx.set_last_result_table) for downstream @ selection.
Displays tags in a rich detail panel with file context (hash, title, URL, etc).
Creates a table of individual tag items to allow selection and downstream piping.
Preserves all metadata from subject (URLs, extensions, etc.) through to display.
Makes tags @-selectable via ctx.set_last_result_table() for chaining:
- get-tag @1 | delete-tag (remove a specific tag)
- get-tag @2 | add-url (add URL to tagged file)
Args:
tags_list: List of tag strings to display
file_hash: SHA256 hash of file
store: Backend name (e.g., "hydrus", "local", "url")
service_name: Tag service name (if from Hydrus)
config: Application configuration
item_title: Optional file title to display
path: Optional file path
subject: Full context object (should preserve original metadata)
quiet: If True, don't display (emit-only mode)
"""
from SYS.result_table import ItemDetailView, extract_item_metadata

View File

@@ -389,7 +389,27 @@ class search_file(Cmdlet):
# --- Execution ------------------------------------------------------
def run(self, result: Any, args: Sequence[str], config: Dict[str, Any]) -> int:
"""Search storage backends for files."""
"""Search storage backends for files by various criteria.
Supports searching by:
- Hash (-query "hash:...")
- Title (-query "title:...")
- Tag (-query "tag:...")
- URL (-query "url:...")
- Other backend-specific fields
Optimizations:
- Extracts tags from metadata response (avoids duplicate API calls)
- Only calls get_tag() separately for backends that don't include tags
Args:
result: Piped input (typically empty for new search)
args: Search criteria and options
config: Application configuration
Returns:
0 on success, 1 on error
"""
if should_show_help(args):
log(f"Cmdlet: {self.name}\nSummary: {self.summary}\nUsage: {self.usage}")
return 0
@@ -698,20 +718,43 @@ class search_file(Cmdlet):
except Exception:
meta_obj = {}
# Extract tags from metadata response instead of separate get_tag() call
# Metadata already includes tags if fetched with include_service_keys_to_tags=True
tags_list: List[str] = []
try:
tag_result = resolved_backend.get_tag(h)
if isinstance(tag_result, tuple) and tag_result:
maybe_tags = tag_result[0]
else:
maybe_tags = tag_result
if isinstance(maybe_tags, list):
tags_list = [
str(t).strip() for t in maybe_tags
if isinstance(t, str) and str(t).strip()
]
except Exception:
tags_list = []
# First try to extract from metadata tags dict
metadata_tags = meta_obj.get("tags")
if isinstance(metadata_tags, dict):
for service_data in metadata_tags.values():
if isinstance(service_data, dict):
display_tags = service_data.get("display_tags", {})
if isinstance(display_tags, dict):
for tag_list in display_tags.values():
if isinstance(tag_list, list):
tags_list = [
str(t).strip() for t in tag_list
if isinstance(t, str) and str(t).strip()
]
break
if tags_list:
break
# Fallback: if metadata didn't include tags, call get_tag() separately
# (This maintains compatibility with backends that don't include tags in metadata)
if not tags_list:
try:
tag_result = resolved_backend.get_tag(h)
if isinstance(tag_result, tuple) and tag_result:
maybe_tags = tag_result[0]
else:
maybe_tags = tag_result
if isinstance(maybe_tags, list):
tags_list = [
str(t).strip() for t in maybe_tags
if isinstance(t, str) and str(t).strip()
]
except Exception:
tags_list = []
title_from_tag: Optional[str] = None
try: