# app/utils/metadata_enrichment.py """ Metadata enrichment from Gallery-DL sidecar JSON files. This module handles: - Detection and parsing of Gallery-DL sidecar .json files - Extraction of metadata (dates, tags, post info) from various platforms - Tag generation for source tracking (source:platform, post:platform:artist:id) - Date conflict resolution (earliest date wins) """ import json import logging import os import re from datetime import datetime from pathlib import Path from typing import Optional log = logging.getLogger("metadata-enrichment") # ============================================================================= # Sidecar File Detection # ============================================================================= def find_sidecar_json(image_path: str) -> Optional[str]: """ Find the Gallery-DL sidecar JSON file for an image. Gallery-DL creates JSON files with various naming patterns: - Patreon: {filename}.json in same directory - HentaiFoundry: {artist}-{index}-{title}.json (may differ from image filename) - UUID-based: {uuid}.json where UUID is extracted from filename Returns the path to the sidecar JSON if found, None otherwise. """ image_path = Path(image_path) parent_dir = image_path.parent stem = image_path.stem # filename without extension # Pattern 1: Same stem with .json extension (most common) # e.g., "Artist-12345-title.jpg" -> "Artist-12345-title.json" stem_json = parent_dir / f"{stem}.json" if stem_json.exists(): return str(stem_json) # Pattern 2: Direct {filename}.json (e.g., image.jpg -> image.jpg.json) direct_json = image_path.with_suffix(image_path.suffix + ".json") if direct_json.exists(): return str(direct_json) # Pattern 3: Extract UUID from filename and look for {uuid}.json # Common patterns: "01_UUID", "UUID", "{num}_{UUID}" uuid_pattern = re.compile(r'([A-F0-9]{8}-[A-F0-9]{4}-[A-F0-9]{4}-[A-F0-9]{4}-[A-F0-9]{12})', re.IGNORECASE) match = uuid_pattern.search(stem) if match: uuid = match.group(1) uuid_json = parent_dir / f"{uuid}.json" if uuid_json.exists(): return str(uuid_json) # Pattern 4: Just the stem without prefix numbers # e.g., "01_filename" -> look for "filename.json" stem_no_prefix = re.sub(r'^\d+_', '', stem) if stem_no_prefix != stem: alt_json = parent_dir / f"{stem_no_prefix}.json" if alt_json.exists(): return str(alt_json) # Pattern 5: Extract numeric ID from filename and find JSON with same ID # Handles HentaiFoundry where image is "hentaifoundry_1000220_title.jpg" # but JSON is "Artist-1000220-title.json" # Look for a 5-7 digit number that could be a post ID id_match = re.search(r'[_-](\d{5,8})[_-]', stem) if id_match: post_id = id_match.group(1) # Search for any JSON file in the directory containing this ID for json_file in parent_dir.glob("*.json"): if f"-{post_id}-" in json_file.name or f"_{post_id}_" in json_file.name: return str(json_file) return None def load_sidecar_metadata(json_path: str) -> Optional[dict]: """Load and parse a Gallery-DL sidecar JSON file.""" try: with open(json_path, "r", encoding="utf-8") as f: return json.load(f) except (json.JSONDecodeError, IOError) as e: log.warning(f"Failed to load sidecar JSON {json_path}: {e}") return None # ============================================================================= # Metadata Extraction (Platform-specific) # ============================================================================= def extract_platform(metadata: dict) -> Optional[str]: """Extract the platform name from metadata.""" # Check common fields if "category" in metadata: return metadata["category"].lower() # Infer from URL patterns url = metadata.get("url", "") or metadata.get("post_url", "") if "patreon.com" in url: return "patreon" if "hentai-foundry" in url or "hentaifoundry" in url.lower(): return "hentaifoundry" if "subscribestar" in url: return "subscribestar" if "discord.com" in url: return "discord" return None def extract_artist(metadata: dict) -> Optional[str]: """Extract the artist/creator name from metadata.""" # Try various common fields if "creator" in metadata and isinstance(metadata["creator"], dict): return metadata["creator"].get("vanity") or metadata["creator"].get("full_name") if "author_name" in metadata: return metadata["author_name"] if "artist" in metadata: return metadata["artist"] # For hentaifoundry if "user" in metadata: return metadata["user"] return None def extract_post_id(metadata: dict) -> Optional[str]: """Extract the post ID from metadata.""" # Patreon, general if "id" in metadata: return str(metadata["id"]) # HentaiFoundry uses "index" as the post ID if "index" in metadata: return str(metadata["index"]) if "post_id" in metadata: return str(metadata["post_id"]) if "message_id" in metadata: # Discord return str(metadata["message_id"]) return None def extract_post_title(metadata: dict) -> Optional[str]: """Extract the post title from metadata.""" return metadata.get("title") def extract_date(metadata: dict) -> Optional[datetime]: """ Extract the publication/post date from metadata. Tries multiple common date fields and formats. """ date_fields = ["published_at", "date", "created_at", "timestamp"] for field in date_fields: if field not in metadata: continue value = metadata[field] if not value: continue # Handle various formats parsed = _parse_date(value) if parsed: return parsed return None def _parse_date(value) -> Optional[datetime]: """Parse a date value that could be string, int, or already datetime.""" if isinstance(value, datetime): return value if isinstance(value, (int, float)): # Unix timestamp try: return datetime.fromtimestamp(value) except (ValueError, OSError): pass if isinstance(value, str): # Try common formats formats = [ "%Y-%m-%dT%H:%M:%S.%f%z", # ISO with microseconds and timezone "%Y-%m-%dT%H:%M:%S%z", # ISO with timezone "%Y-%m-%dT%H:%M:%S.%f", # ISO with microseconds "%Y-%m-%dT%H:%M:%S", # ISO basic "%Y-%m-%d %H:%M:%S", # Space-separated "%Y-%m-%d", # Date only ] # Handle timezone offset with colon (Python < 3.11 compat) # e.g., "2023-08-01T04:20:02.000+00:00" -> "2023-08-01T04:20:02.000+0000" normalized = re.sub(r'([+-]\d{2}):(\d{2})$', r'\1\2', value) for fmt in formats: try: return datetime.strptime(normalized, fmt) except ValueError: continue # Try without timezone info # Strip timezone suffix for naive parsing stripped = re.sub(r'[+-]\d{2}:?\d{2}$', '', value) stripped = re.sub(r'Z$', '', stripped) for fmt in formats: try: return datetime.strptime(stripped, fmt) except ValueError: continue return None def extract_content_tags(metadata: dict) -> list[str]: """ Extract content tags from metadata (not to be confused with DB tags). These are tags the creator applied to the post. """ tags = [] # Direct tags field if "tags" in metadata and isinstance(metadata["tags"], list): tags.extend(metadata["tags"]) return [str(t) for t in tags if t] # ============================================================================= # Tag Generation # ============================================================================= def generate_source_tags(metadata: dict, artist_override: Optional[str] = None) -> list[dict]: """ Generate tags from Gallery-DL metadata. Returns a list of tag dictionaries with 'name' and 'kind' keys. Tag format ensures uniqueness: - source:{platform} (kind='source') - post:{platform}:{artist}:{post_id} (kind='post') """ tags = [] platform = extract_platform(metadata) artist = artist_override or extract_artist(metadata) post_id = extract_post_id(metadata) # Source platform tag if platform: tags.append({ "name": f"source:{platform}", "kind": "source" }) # Post tag (unique across platform + artist + post_id) if platform and artist and post_id: # Sanitize artist name for tag (lowercase, replace spaces) safe_artist = artist.lower().replace(" ", "_") tags.append({ "name": f"post:{platform}:{safe_artist}:{post_id}", "kind": "post" }) return tags # ============================================================================= # Date Conflict Resolution # ============================================================================= def resolve_date_conflict(existing_date: Optional[datetime], new_date: Optional[datetime]) -> Optional[datetime]: """ Resolve date conflicts by keeping the earliest date. Args: existing_date: The currently stored date (may be None) new_date: The new date from metadata (may be None) Returns: The earliest non-None date, or None if both are None """ if existing_date is None: return new_date if new_date is None: return existing_date # Return the earlier date return min(existing_date, new_date) # ============================================================================= # Main Enrichment Function # ============================================================================= def enrich_from_sidecar(image_path: str, existing_metadata: Optional[dict] = None) -> dict: """ Enrich image metadata from Gallery-DL sidecar JSON. Args: image_path: Path to the image file existing_metadata: Optional dict with existing metadata (e.g., from EXIF) Returns: Dictionary with enriched metadata: { "platform": str or None, "artist": str or None, "post_id": str or None, "post_title": str or None, "taken_at": datetime or None, "tags": [{"name": str, "kind": str}, ...], "raw_metadata": dict or None (the full sidecar content) } """ result = { "platform": None, "artist": None, "post_id": None, "post_title": None, "taken_at": None, "tags": [], "raw_metadata": None, } # Find and load sidecar sidecar_path = find_sidecar_json(image_path) if not sidecar_path: return result metadata = load_sidecar_metadata(sidecar_path) if not metadata: return result result["raw_metadata"] = metadata # Extract fields result["platform"] = extract_platform(metadata) result["artist"] = extract_artist(metadata) result["post_id"] = extract_post_id(metadata) result["post_title"] = extract_post_title(metadata) # Extract date and resolve conflicts with existing sidecar_date = extract_date(metadata) existing_date = existing_metadata.get("taken_at") if existing_metadata else None result["taken_at"] = resolve_date_conflict(existing_date, sidecar_date) # Generate tags result["tags"] = generate_source_tags(metadata, result["artist"]) return result