"""Utilities that drive the modern Textual UI menus and presets.""" from __future__ import annotations import sys from dataclasses import dataclass from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Sequence BASE_DIR = Path(__file__).resolve().parent ROOT_DIR = BASE_DIR.parent for path in (ROOT_DIR, BASE_DIR): str_path = str(path) if str_path not in sys.path: sys.path.insert(0, str_path) import metadata @dataclass(slots=True) class PipelinePreset: """Simple descriptor for a reusable pipeline.""" label: str description: str pipeline: str PIPELINE_PRESETS: List[PipelinePreset] = [ PipelinePreset( label="Download → Merge → Local", description="Use download-data with playlist auto-selection, merge the pieces, tag, then import into local storage.", pipeline='download-data "" | merge-file | add-tag | add-file -storage local', ), PipelinePreset( label="Download → Hydrus", description="Fetch media, auto-tag, and push directly into Hydrus.", pipeline='download-data "" | merge-file | add-tag | add-file -storage hydrus', ), PipelinePreset( label="Search Local Library", description="Run search-file against the local library and emit a result table for further piping.", pipeline='search-file -library local -query ""', ), ] def load_tags(file_path: Path) -> List[str]: """Read tags for a file using metadata.py as the single source of truth.""" try: return metadata.read_tags_from_file(file_path) except Exception: return [] def group_tags_by_namespace(tags: Sequence[str]) -> Dict[str, List[str]]: """Return tags grouped by namespace for quick UI summaries.""" grouped: Dict[str, List[str]] = {} for tag in metadata.normalize_tags(list(tags)): namespace, value = metadata.split_tag(tag) key = namespace or "_untagged" grouped.setdefault(key, []).append(value) for items in grouped.values(): items.sort() return grouped def build_metadata_snapshot(file_path: Path) -> Dict[str, Any]: """Load any available sidecar metadata for the selected file.""" snapshot: Dict[str, Any] = { "file": str(file_path), "tags": group_tags_by_namespace(load_tags(file_path)), } try: sidecar = metadata._derive_sidecar_path(file_path) if sidecar.is_file(): title, tags, notes = metadata._read_sidecar_metadata(sidecar) snapshot["sidecar"] = { "title": title, "tags": group_tags_by_namespace(tags), "notes": notes, } except Exception: snapshot["sidecar"] = None return snapshot def summarize_result(result: Dict[str, Any]) -> str: """Build a one-line summary for a pipeline result row.""" title = result.get("title") or result.get("identifier") or result.get("file_path") source = result.get("source") or result.get("cmdlet") or "result" return f"{source}: {title}" if title else source def normalize_tags(tags: Iterable[str]) -> List[str]: """Expose metadata.normalize_tags for callers that imported the old helper.""" return metadata.normalize_tags(list(tags))