from __future__ import annotations from typing import Any, Dict, Sequence, Optional import sys from SYS.logger import log from SYS import pipeline as ctx from API import HydrusNetwork as hydrus_wrapper from . import _shared as sh Cmdlet = sh.Cmdlet CmdletArg = sh.CmdletArg SharedArgs = sh.SharedArgs normalize_hash = sh.normalize_hash fmt_bytes = sh.fmt_bytes get_hash_for_operation = sh.get_hash_for_operation fetch_hydrus_metadata = sh.fetch_hydrus_metadata should_show_help = sh.should_show_help get_field = sh.get_field from Store import Store CMDLET = Cmdlet( name="get-relationship", summary="Print relationships for the selected file (Hydrus).", usage='get-relationship [-query "hash:"]', alias=[], arg=[ SharedArgs.QUERY, SharedArgs.STORE, ], detail=[ "- Lists relationship data as returned by Hydrus.", ], ) def _run(result: Any, _args: Sequence[str], config: Dict[str, Any]) -> int: # Help if should_show_help(_args): log(f"Cmdlet: {CMDLET.name}\nSummary: {CMDLET.summary}\nUsage: {CMDLET.usage}") return 0 # Parse -query and -store override override_query: str | None = None override_store: str | None = None args_list = list(_args) i = 0 while i < len(args_list): a = args_list[i] low = str(a).lower() if low in {"-query", "--query", "query"} and i + 1 < len(args_list): override_query = str(args_list[i + 1]).strip() i += 2 continue if low in {"-store", "--store", "store"} and i + 1 < len(args_list): override_store = str(args_list[i + 1]).strip() i += 2 continue i += 1 override_hash: str | None = ( sh.parse_single_hash_query(override_query) if override_query else None ) if override_query and not override_hash: log('get-relationship requires -query "hash:"', file=sys.stderr) return 1 # Handle @N selection which creates a list if isinstance(result, list): if len(result) == 0: result = None elif len(result) > 1 and not override_hash: log( 'get-relationship expects a single item; select one row (e.g. @1) or pass -query "hash:"', file=sys.stderr, ) return 1 else: result = result[0] # Initialize results collection found_relationships = [] # List of dicts: {hash, type, title, path, store} source_title = "Unknown" # Store/hash-first subject resolution store_name: Optional[str] = override_store if not store_name: store_name = get_field(result, "store") hash_hex = ( normalize_hash(override_hash) if override_hash else normalize_hash(get_hash_for_operation(None, result)) ) if not source_title or source_title == "Unknown": source_title = ( get_field(result, "title") or get_field(result, "name") or (hash_hex[:16] + "..." if hash_hex else "Unknown") ) if not hash_hex: log('get-relationship requires -query "hash:"', file=sys.stderr) return 1 # Fetch Hydrus relationships if we have a hash. hash_hex = ( normalize_hash(override_hash) if override_hash else normalize_hash(get_hash_for_operation(None, result)) ) if hash_hex: try: client = None store_label = "hydrus" backend_obj = None if store_name: # Store specified: do not fall back to a global/default Hydrus client. store_label = str(store_name) try: store = Store(config) backend_obj = store[str(store_name)] candidate = getattr(backend_obj, "_client", None) if candidate is not None and hasattr(candidate, "get_file_relationships"): client = candidate except Exception: client = None if client is None: log( f"Hydrus client unavailable for store '{store_name}'", file=sys.stderr ) return 1 else: client = hydrus_wrapper.get_client(config) def _resolve_related_title(rel_hash: str) -> str: """Best-effort resolve a Hydrus hash to a human title. Preference order: - title: tag from the backend (fast path) - Hydrus metadata tags via fetch_hydrus_metadata - fallback to short hash """ h = normalize_hash(rel_hash) if not h: return str(rel_hash) # Prefer backend tag extraction when available. if backend_obj is not None and hasattr(backend_obj, "get_tag"): try: tag_result = backend_obj.get_tag(h) tags = ( tag_result[0] if isinstance(tag_result, tuple) and tag_result else tag_result ) if isinstance(tags, list): for t in tags: if isinstance(t, str) and t.lower().startswith("title:"): val = t.split(":", 1)[1].strip() if val: return val except Exception: pass # Fallback: fetch minimal metadata and scan for a title tag. try: meta, _ = fetch_hydrus_metadata( config, h, store_name=store_label if store_name else None, hydrus_client=client, include_service_keys_to_tags=True, include_file_url=False, include_duration=False, include_size=False, include_mime=False, ) if isinstance(meta, dict): tags_payload = meta.get("tags") tag_candidates: list[str] = [] if isinstance(tags_payload, dict): for svc_data in tags_payload.values(): if not isinstance(svc_data, dict): continue storage = svc_data.get("storage_tags") if isinstance(storage, dict): for group in storage.values(): if isinstance(group, list): tag_candidates.extend( [ str(x) for x in group if isinstance(x, str) ] ) display = svc_data.get("display_tags") if isinstance(display, list): tag_candidates.extend( [str(x) for x in display if isinstance(x, str)] ) flat = meta.get("tags_flat") if isinstance(flat, list): tag_candidates.extend( [str(x) for x in flat if isinstance(x, str)] ) for t in tag_candidates: if isinstance(t, str) and t.lower().startswith("title:"): val = t.split(":", 1)[1].strip() if val: return val except Exception: pass return h[:16] + "..." if client: rel = client.get_file_relationships(hash_hex) if rel: file_rels = rel.get("file_relationships", {}) this_file_rels = file_rels.get(hash_hex) if this_file_rels: # Map Hydrus relationship IDs to names. # For /manage_file_relationships/get_file_relationships, the Hydrus docs define: # 0=potential duplicates, 1=false positives, 3=alternates, 8=duplicates # Additionally, this endpoint includes metadata keys like 'king'/'is_king'. rel_map = { "0": "potential", "1": "false positive", "3": "alternate", "8": "duplicate", } for rel_type_id, rel_value in this_file_rels.items(): key = str(rel_type_id) # Handle metadata keys explicitly. if key in {"is_king", "king_is_on_file_domain", "king_is_local"}: continue # Some Hydrus responses provide a direct king hash under the 'king' key. if key == "king": king_hash = ( normalize_hash(rel_value) if isinstance(rel_value, str) else None ) if king_hash and king_hash != hash_hex: if not any(str(r.get("hash", "")).lower() == king_hash for r in found_relationships): found_relationships.append( { "hash": king_hash, "type": "king", "title": _resolve_related_title(king_hash), "path": None, "store": store_label, } ) continue rel_name = rel_map.get(key, f"type-{key}") # The relationship value is typically a list of hashes. if isinstance(rel_value, list): for rel_hash in rel_value: rel_hash_norm = ( normalize_hash(rel_hash) if isinstance(rel_hash, str) else None ) if not rel_hash_norm or rel_hash_norm == hash_hex: continue if not any(str(r.get("hash", "")).lower() == rel_hash_norm for r in found_relationships): found_relationships.append( { "hash": rel_hash_norm, "type": rel_name, "title": _resolve_related_title(rel_hash_norm), "path": None, "store": store_label, } ) # Defensive: sometimes the API may return a single hash string. elif isinstance(rel_value, str): rel_hash_norm = normalize_hash(rel_value) if rel_hash_norm and rel_hash_norm != hash_hex: if not any(str(r.get("hash", "")).lower() == rel_hash_norm for r in found_relationships): found_relationships.append( { "hash": rel_hash_norm, "type": rel_name, "title": _resolve_related_title(rel_hash_norm), "path": None, "store": store_label, } ) except Exception as exc: # Only log error if we didn't find local relationships either if not found_relationships: log(f"Hydrus relationships fetch failed: {exc}", file=sys.stderr) # Display results from SYS.result_table import ItemDetailView, extract_item_metadata # Prepare metadata for the detail view metadata = extract_item_metadata(result) if hash_hex: metadata["Hash"] = hash_hex # Overlays if source_title and source_title != "Unknown": metadata["Title"] = source_title table = ItemDetailView("Relationships", item_metadata=metadata ).init_command("get-relationship", []) # Sort by type then title # Custom sort order: King first, then Derivative, then others def type_sort_key(item): t = item["type"].lower() if t == "king": return 0 elif t == "derivative": return 1 elif t in {"alternative", "alternate", "alt"}: return 2 elif t == "duplicate": return 3 else: return 4 found_relationships.sort(key=lambda x: (type_sort_key(x), x["title"])) pipeline_results = [] for i, item in enumerate(found_relationships): row = table.add_row() row.add_column("Type", item["type"].title()) row.add_column("Title", item["title"]) # row.add_column("Hash", item['hash'][:16] + "...") # User requested removal row.add_column("Store", item["store"]) # Create result object for pipeline res_obj = { "title": item["title"], "hash": item["hash"], "file_hash": item["hash"], "relationship_type": item["type"], "store": item["store"], } # Target is always hash in store/hash-first mode res_obj["target"] = item["hash"] pipeline_results.append(res_obj) # Set selection args table.set_row_selection_args( i, ["-store", str(item["store"]), "-query", f"hash:{item['hash']}"] ) # Ensure empty state is still navigable/visible ctx.set_last_result_table_overlay(table, pipeline_results) from SYS.rich_display import stdout_console stdout_console().print(table) if not found_relationships: log("No relationships found.") return 0 CMDLET.exec = _run CMDLET.register()