# 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) - Pixiv: {id}_p{num}.json for image named {id}_{title}_{num}.png - 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: Pixiv naming pattern # Image: {id}_{title}_{num}.png (e.g., "115358744_Koseki Bijou (Bunny)_00.png") # JSON: {id}_p{num}.json (e.g., "115358744_p0.json") # Extract the leading numeric ID and trailing index pixiv_match = re.match(r'^(\d{6,12})_.*_(\d{2})$', stem) if pixiv_match: pixiv_id = pixiv_match.group(1) pixiv_num = int(pixiv_match.group(2)) # Convert "00" to 0 pixiv_json = parent_dir / f"{pixiv_id}_p{pixiv_num}.json" if pixiv_json.exists(): return str(pixiv_json) # Pattern 4: 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 5: 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 6: 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 find_archive_sidecar(archive_path: str) -> Optional[str]: """ Find the Gallery-DL sidecar JSON file for an archive. Gallery-DL names sidecars based on the archive's content filename, which may differ from the archive filename due to prefixes added during download (e.g., "02_HolyMeh July 2023.rar" -> "HolyMeh July 2023.json"). Args: archive_path: Path to the archive file (e.g., "02_Filename.rar") Returns: Path to the sidecar JSON if found, None otherwise. """ archive_path = Path(archive_path) parent_dir = archive_path.parent stem = archive_path.stem # filename without extension # Pattern 1: Exact stem match (most common case) # e.g., "HolyMeh July 2023.rar" -> "HolyMeh July 2023.json" stem_json = parent_dir / f"{stem}.json" if stem_json.exists(): return str(stem_json) # Pattern 2: Strip numeric prefix from archive name # e.g., "02_HolyMeh July 2023.rar" -> "HolyMeh July 2023.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 3: Look for any JSON in the directory that references this archive # by checking the 'file_name' or 'filename' field in the JSON for json_file in parent_dir.glob("*.json"): try: with open(json_file, "r", encoding="utf-8") as f: data = json.load(f) # Check if this JSON references our archive # Gallery-DL uses 'file.file_name' or 'filename' + 'extension' file_info = data.get("file", {}) json_filename = file_info.get("file_name", "") if not json_filename: # Try filename + extension pattern fn = data.get("filename", "") ext = data.get("extension", "") if fn and ext: json_filename = f"{fn}.{ext}" # Compare with archive filename (with or without prefix) archive_filename = archive_path.name archive_filename_no_prefix = re.sub(r'^\d+_', '', archive_filename) if json_filename and ( json_filename == archive_filename or json_filename == archive_filename_no_prefix ): return str(json_file) except (json.JSONDecodeError, IOError): continue 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 (covers Pixiv which uses "category": "pixiv") 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" if "pixiv.net" in url or "pximg.net" in url: return "pixiv" 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 Pixiv - user is an object with name and account fields if "user" in metadata and isinstance(metadata["user"], dict): # Prefer account (username) over display name for consistency return metadata["user"].get("account") or metadata["user"].get("name") # For hentaifoundry - user is a string if "user" in metadata and isinstance(metadata["user"], str): 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.""" title = metadata.get("title") # Return None for empty strings if title and isinstance(title, str) and title.strip(): return title.strip() return None def extract_description(metadata: dict) -> Optional[str]: """ Extract the post description/content from metadata. Platform-specific fields: - Patreon: 'content' (may be HTML) - SubscribeStar: 'content' (may be HTML) - HentaiFoundry: 'description' (plain text) - Pixiv: 'caption' (plain text or HTML) """ # Try description first (HentaiFoundry uses this) if "description" in metadata: desc = metadata["description"] if desc and isinstance(desc, str) and desc.strip(): return desc.strip() # Try caption (Pixiv uses this) if "caption" in metadata: caption = metadata["caption"] if caption and isinstance(caption, str) and caption.strip(): return caption.strip() # Try content (Patreon/SubscribeStar use this) if "content" in metadata: content = metadata["content"] if content and isinstance(content, str) and content.strip(): return content.strip() # Try teaser_text as fallback (Patreon) if "teaser_text" in metadata: teaser = metadata["teaser_text"] if teaser and isinstance(teaser, str) and teaser.strip(): return teaser.strip() return None def extract_source_url(metadata: dict) -> Optional[str]: """ Extract the URL to the original post. Platform-specific fields: - Patreon: 'url' (direct post link) - SubscribeStar: construct from author_name and post_id - HentaiFoundry: construct from user and index - Pixiv: construct from id """ platform = extract_platform(metadata) # Construct URL for Pixiv (do this first to avoid using the pximg.net image URL) if platform == "pixiv": post_id = metadata.get("id") if post_id: return f"https://www.pixiv.net/artworks/{post_id}" # Direct URL field (Patreon) - but not for Pixiv where 'url' is the image URL if "url" in metadata: url = metadata["url"] if url and isinstance(url, str) and url.startswith("http"): # Skip pximg.net URLs (these are image URLs, not post URLs) if "pximg.net" not in url: return url # Construct URL for SubscribeStar if platform == "subscribestar": author = metadata.get("author_name") post_id = metadata.get("post_id") if author and post_id: return f"https://subscribestar.adult/{author}/posts/{post_id}" # Construct URL for HentaiFoundry if platform == "hentaifoundry": user = metadata.get("user") or metadata.get("artist") index = metadata.get("index") if user and index: return f"https://www.hentai-foundry.com/pictures/user/{user}/{index}" return None def extract_attachment_count(metadata: dict) -> int: """ Extract the number of attachments/images in the post. """ # Pixiv: use page_count field if "page_count" in metadata: return int(metadata["page_count"]) # Patreon: check images array or post_metadata.image_order if "images" in metadata and isinstance(metadata["images"], list): return len(metadata["images"]) if "post_metadata" in metadata: pm = metadata["post_metadata"] if isinstance(pm, dict) and "image_order" in pm: return len(pm["image_order"]) # Default to 1 if we have a file if "filename" in metadata or "file" in metadata: return 1 return 0 def extract_date(metadata: dict) -> Optional[datetime]: """ Extract the publication/post date from metadata. Tries multiple common date fields and formats. """ # Pixiv uses create_date, others use published_at, date, etc. date_fields = ["published_at", "create_date", "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 # Normalize timezone awareness before comparison # If one is aware and one is naive, strip timezone from the aware one existing_aware = existing_date.tzinfo is not None new_aware = new_date.tzinfo is not None if existing_aware and not new_aware: # Strip timezone from existing to compare existing_date = existing_date.replace(tzinfo=None) elif new_aware and not existing_aware: # Strip timezone from new to compare new_date = new_date.replace(tzinfo=None) # 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, "description": str or None, "source_url": str or None, "attachment_count": int, "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, "description": None, "source_url": None, "attachment_count": 0, "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) result["description"] = extract_description(metadata) result["source_url"] = extract_source_url(metadata) result["attachment_count"] = extract_attachment_count(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