"""Gallery-dl wrapper service. This service provides a clean interface for executing gallery-dl downloads with support for per-source configuration overrides. """ import json import logging import subprocess import sys import tempfile import time from dataclasses import dataclass, field from enum import Enum from pathlib import Path from typing import Optional from app.config import get_settings logger = logging.getLogger(__name__) class ErrorType(str, Enum): """Error types for categorizing download failures.""" AUTH_ERROR = "auth_error" RATE_LIMITED = "rate_limited" NOT_FOUND = "not_found" ACCESS_DENIED = "access_denied" NETWORK_ERROR = "network_error" NO_NEW_CONTENT = "no_new_content" TIMEOUT = "timeout" HTTP_ERROR = "http_error" UNSUPPORTED_URL = "unsupported_url" UNKNOWN_ERROR = "unknown_error" class ContentType(str, Enum): """Content types that can be downloaded from platforms.""" # Patreon content types IMAGES = "images" IMAGE_LARGE = "image_large" ATTACHMENTS = "attachments" POSTFILE = "postfile" CONTENT = "content" # Embedded content in post body # Generic types ALL = "all" @dataclass class SourceConfig: """Per-source configuration that can override global settings. These settings are stored in the source's metadata field and merged with global defaults when executing downloads. """ # Content types to download (platform-specific) # For Patreon: ["images", "image_large", "attachments", "postfile", "content"] # For others: typically just ["all"] or specific types content_types: list[str] = field(default_factory=lambda: ["all"]) # Rate limiting overrides sleep: Optional[float] = None # Delay between downloads sleep_request: Optional[float] = None # Delay between requests # Directory/filename pattern overrides directory_pattern: Optional[str] = None filename_pattern: Optional[str] = None # Other options skip_existing: bool = True # Skip files already in archive save_metadata: bool = True # Save JSON metadata alongside files timeout: int = 3600 # Download timeout in seconds @classmethod def from_dict(cls, data: dict) -> "SourceConfig": """Create SourceConfig from a dictionary (e.g., from source.metadata).""" return cls( content_types=data.get("content_types", ["all"]), sleep=data.get("sleep"), sleep_request=data.get("sleep_request"), directory_pattern=data.get("directory_pattern"), filename_pattern=data.get("filename_pattern"), skip_existing=data.get("skip_existing", True), save_metadata=data.get("save_metadata", True), timeout=data.get("timeout", 3600), ) def to_dict(self) -> dict: """Convert to dictionary for storage.""" return { "content_types": self.content_types, "sleep": self.sleep, "sleep_request": self.sleep_request, "directory_pattern": self.directory_pattern, "filename_pattern": self.filename_pattern, "skip_existing": self.skip_existing, "save_metadata": self.save_metadata, "timeout": self.timeout, } @dataclass class DownloadResult: """Result of a gallery-dl download execution.""" success: bool url: str subscription_name: str platform: str # Output details files_downloaded: int = 0 stdout: str = "" stderr: str = "" return_code: int = 0 # Error details (if failed) error_type: Optional[ErrorType] = None error_message: Optional[str] = None # Timing duration_seconds: float = 0.0 started_at: Optional[str] = None completed_at: Optional[str] = None class GalleryDLService: """Service for executing gallery-dl downloads. Handles: - Building gallery-dl configuration with per-source overrides - Executing the subprocess with proper error handling - Parsing output and categorizing errors - Managing temporary config files """ # Platform-specific default content types and options # Directory patterns are relative - subscription_name/platform is prepended automatically PLATFORM_DEFAULTS = { "patreon": { "content_types": ["images", "image_large", "attachments", "postfile", "content"], "directory": ["{date:%Y-%m-%d}_{id}_{title[:40]}"], "filename": "{num:>02}_{filename}.{extension}", # Additional options "videos": True, # Download video content "embeds": True, # Download embedded URLs (YouTube, etc.) # Deep crawling options "cursor": True, # Enable cursor-based pagination for complete history }, "subscribestar": { "content_types": ["all"], "directory": ["{date:%Y-%m-%d}_{id}_{title[:40]}"], "filename": "{num:>02}_{filename}.{extension}", # Automatic infinite scroll pagination is built-in }, "hentaifoundry": { "content_types": ["all"], # Download all content types "directory": [], # Flat structure within platform folder "filename": "{category}_{index:>03}_{title[:50]}.{extension}", "include": "all", # Include pictures, scraps, stories - complete content # Automatic pagination through all pages (25 items/page) }, "discord": { "content_types": ["all"], "directory": ["{channel[name]}"], "filename": "{date:%Y%m%d}_{id}_{filename}.{extension}", # Additional options "embeds": True, # Download embedded URLs "stickers": True, # Download stickers "reactions": False, # Skip reaction images # Deep crawling options "threads": True, # Include thread content for complete history }, "pixiv": { "content_types": ["all"], "directory": ["{category}"], "filename": "{id}_{title[:50]}_{num:>02}.{extension}", # Content options "ugoira": True, # Download ugoira (animated) as WebM }, "deviantart": { "content_types": ["all"], "directory": [], # Flat structure within platform folder "filename": "{index:>03}_{title[:50]}.{extension}", # Content options "flat": True, # Flatten folder structure "original": True, # Download original quality "mature": True, # Include mature content "metadata": True, # Include metadata }, } def __init__(self, rate_limit: Optional[float] = None): """Initialize the GalleryDL service. Args: rate_limit: Override for download rate limit (seconds between requests). If None, uses environment default. """ self.settings = get_settings() self._rate_limit = rate_limit if rate_limit is not None else self.settings.download_rate_limit self._base_config = self._load_base_config() def _load_base_config(self) -> dict: """Load the base gallery-dl configuration file.""" config_path = Path(self.settings.config_path) / "gallery-dl.conf" if config_path.exists(): try: with open(config_path) as f: return json.load(f) except json.JSONDecodeError as e: logger.warning(f"Invalid gallery-dl.conf, using defaults: {e}") # Return sensible defaults if no config exists return self._get_default_config() def _get_default_config(self) -> dict: """Get default gallery-dl configuration.""" return { "extractor": { "base-directory": self.settings.download_path, "archive": str(Path(self.settings.config_path) / "archive.sqlite3"), "skip": True, "sleep": self._rate_limit, # Faster request rate for pagination/checking (1/4 of rate_limit, min 0.5s) "sleep-request": max(0.5, self._rate_limit / 4), "retries": 3, "timeout": 30.0, "verify": True, "postprocessors": [ { "name": "metadata", "mode": "json", "directory": ".", "filename": "{filename}.json", } ], }, "downloader": { "part": True, "part-directory": str(Path(self.settings.config_path) / "temp"), "retries": 3, "timeout": 120.0, }, "output": { "progress": True, }, } def _build_config_for_source( self, platform: str, source_config: SourceConfig, destination: str, ) -> dict: """Build a merged configuration for a specific source. Merges: base config → platform defaults → source overrides """ config = json.loads(json.dumps(self._base_config)) # Deep copy # Get platform defaults platform_defaults = self.PLATFORM_DEFAULTS.get(platform, {}) # Ensure extractor section exists if "extractor" not in config: config["extractor"] = {} # Apply base directory config["extractor"]["base-directory"] = destination # Apply rate limiting overrides if source_config.sleep is not None: config["extractor"]["sleep"] = source_config.sleep if source_config.sleep_request is not None: config["extractor"]["sleep-request"] = source_config.sleep_request # Apply skip setting config["extractor"]["skip"] = source_config.skip_existing # Configure metadata postprocessor if source_config.save_metadata: config["extractor"]["postprocessors"] = [ { "name": "metadata", "mode": "json", "directory": ".", "filename": "{filename}.json", } ] else: config["extractor"].pop("postprocessors", None) # Build platform-specific section platform_config = {} # Apply all platform defaults first (videos, embeds, stickers, etc.) for key, value in platform_defaults.items(): if key not in ["content_types", "directory", "filename"]: platform_config[key] = value # Content types (platform-specific handling) if platform == "patreon": # Patreon uses "files" array for content types if "all" in source_config.content_types: platform_config["files"] = ["images", "image_large", "attachments", "postfile", "content"] else: platform_config["files"] = source_config.content_types elif platform == "hentaifoundry": # HentaiFoundry uses "include" for content filtering if "pictures" in source_config.content_types or "all" in source_config.content_types: platform_config["include"] = "all" # Directory pattern if source_config.directory_pattern: platform_config["directory"] = source_config.directory_pattern elif "directory" in platform_defaults: platform_config["directory"] = platform_defaults["directory"] # Filename pattern if source_config.filename_pattern: platform_config["filename"] = source_config.filename_pattern elif "filename" in platform_defaults: platform_config["filename"] = platform_defaults["filename"] # Always save metadata at platform level too platform_config["metadata"] = source_config.save_metadata # Add platform config to extractor config["extractor"][platform] = platform_config return config def _categorize_error(self, return_code: int, stdout: str, stderr: str) -> tuple[ErrorType, str]: """Categorize the error based on output and return code. Priority order: 1. Check for skip messages (# prefixed lines or text patterns) → NO_NEW_CONTENT 2. Check for actual errors (auth, rate limit, 404, network, etc.) 3. Fall back to UNKNOWN_ERROR This ordering prevents false positives where debug output matches error patterns. """ combined = f"{stdout} {stderr}".lower() # === PHASE 1: Detect skip/no-new-content scenarios === # Gallery-dl indicates skipped files two ways: # 1. Lines starting with '#' in stdout (e.g., "# /data/downloads/Artist/file.png") # 2. Text patterns like "skipping", "already exists" # Count lines starting with '#' (skip messages) skip_line_count = len([ line for line in stdout.split('\n') if line.strip().startswith('#') ]) # Text-based skip indicators skip_text_indicators = ["skipping", "already exists", "archive"] has_skip_text = any(ind in combined for ind in skip_text_indicators) # Actual error indicators that override skip detection actual_error_indicators = [ "][error]", # gallery-dl error log level: [patreon][error] "exception", # Python exceptions "traceback", # Python tracebacks "download failed", "extraction failed", ] has_actual_error = any(err in combined for err in actual_error_indicators) # If we have skip evidence and no real errors, it's no_new_content if (skip_line_count > 0 or has_skip_text) and not has_actual_error: return ErrorType.NO_NEW_CONTENT, "No new content to download" # Empty output with low return codes often means nothing to download if return_code in (0, 1) and not stdout.strip(): return ErrorType.NO_NEW_CONTENT, "No new content to download" # === PHASE 2: Check for specific error types === # Auth errors - look for error context, not just HTTP codes that appear in debug logs # HTTP debug lines look like: '"GET /path HTTP/1.1" 401 None' - we need to detect actual errors auth_error_patterns = [ "unauthorized", "login required", "please login", "must be logged in", "authentication required", "session expired", "invalid cookie", "cookies are expired", "cookies have expired", "not logged in", ] # Check for 401 specifically as an HTTP error response (not in URLs) # Look for patterns like "401" followed by error context or at end of HTTP log line has_401_error = ( '" 401 ' in combined or # HTTP response: "GET /path HTTP/1.1" 401 None "401 unauthorized" in combined or "error 401" in combined or "status: 401" in combined or "status code: 401" in combined ) if has_401_error or any(pattern in combined for pattern in auth_error_patterns): return ErrorType.AUTH_ERROR, "Authentication failed - cookies may be expired or invalid" # Rate limiting - look for actual rate limit errors rate_limit_patterns = [ '" 429 ', # HTTP response "rate limit", "too many requests", "ratelimit", "throttl", ] if any(pattern in combined for pattern in rate_limit_patterns): return ErrorType.RATE_LIMITED, "Rate limited by server - try increasing sleep time" # 404 Not Found - check for HTTP 404 responses with error context # Be specific to avoid matching "not found" in debug messages like "archive entry not found" not_found_patterns = [ '" 404 ', # HTTP response log line "error 404", # Explicit error 404 "status: 404", # Status code indicator "status code: 404", # Status code indicator "404 not found", # Standard 404 message "page not found", # Common 404 page text "user not found", # User doesn't exist "creator not found", "artist not found", "content no longer available", "has been deleted", "account.*deleted", "profile.*not.*exist", ] # Also require error-level context for generic patterns to avoid false positives generic_not_found = ["does not exist", "no longer available"] has_specific_404 = any(pattern in combined for pattern in not_found_patterns) has_generic_with_error = any( pattern in combined and "][error]" in combined for pattern in generic_not_found ) if has_specific_404 or has_generic_with_error: return ErrorType.NOT_FOUND, "URL not found - artist may have changed username or deleted content" # Network errors - use specific patterns to avoid matching config values # (e.g., "timeout": 30.0 in debug output should not trigger network_error) network_error_patterns = [ "timed out", # Actual timeout occurred "read timed out", # urllib3/requests timeout "connect timed out", # Connection timeout "connection timed out", "connection refused", "connection reset", "network unreachable", "name resolution failed", "name or service not known", "nodename nor servname provided", "ssl: certificate_verify_failed", "ssl: wrong_version_number", "certificate verify failed", "[errno", # Socket errors like [Errno 111] ] if any(pattern in combined for pattern in network_error_patterns): return ErrorType.NETWORK_ERROR, "Network error - check internet connection" # 403 Forbidden / Access denied access_denied_patterns = [ '" 403 ', # HTTP response "error 403", "forbidden", "access denied", "permission denied", "tier required", "pledge required", ] if any(pattern in combined for pattern in access_denied_patterns): return ErrorType.ACCESS_DENIED, "Access denied - may need higher subscription tier" # Generic HTTP errors if "http error" in combined or "httperror" in combined: return ErrorType.HTTP_ERROR, "HTTP error occurred" # Unsupported URL if "no suitable" in combined or "unsupported" in combined or "no extractor" in combined: return ErrorType.UNSUPPORTED_URL, "URL format not supported by gallery-dl" # If we got here with return code 1 and saw some activity but no error patterns matched, # it's likely just no new content (gallery-dl returns 1 when nothing to download) if return_code == 1 and stdout.strip() and not has_actual_error: return ErrorType.NO_NEW_CONTENT, "No new content to download" # Final fallback - genuine unknown error return ErrorType.UNKNOWN_ERROR, f"Unknown error (return code: {return_code})" def _count_downloaded_files(self, stdout: str) -> int: """Count the number of files downloaded from stdout. Gallery-dl outputs one line per downloaded file. Lines starting with '#' or containing 'skip' are not downloads. """ if not stdout: return 0 count = 0 for line in stdout.strip().split("\n"): line = line.strip() if line and not line.startswith("#") and "skip" not in line.lower(): count += 1 return count async def download( self, url: str, subscription_name: str, platform: str, source_config: Optional[SourceConfig] = None, cookies_path: Optional[str] = None, auth_token: Optional[str] = None, ) -> DownloadResult: """Execute a gallery-dl download. Args: url: The URL to download from subscription_name: Name of the subscription/artist (used for destination folder) platform: Platform name (patreon, subscribestar, etc.) source_config: Optional per-source configuration overrides cookies_path: Optional path to cookies file auth_token: Optional auth token (for Discord and other token-based platforms) Returns: DownloadResult with success status and details Downloads are saved to: {download_path}/{subscription_name}/{platform}/{post_folders} """ import asyncio from datetime import datetime start_time = time.time() started_at = datetime.utcnow().isoformat() # Use default config if none provided if source_config is None: source_config = SourceConfig() # Build destination path: subscription_name/platform destination = str(Path(self.settings.download_path) / subscription_name / platform) # Build merged configuration config = self._build_config_for_source(platform, source_config, destination) # Add cookies if provided if cookies_path: config["extractor"]["cookies"] = cookies_path # Add auth token for Discord if auth_token and platform == "discord": if "discord" not in config["extractor"]: config["extractor"]["discord"] = {} config["extractor"]["discord"]["token"] = auth_token # Write temporary config file with tempfile.NamedTemporaryFile( mode="w", suffix=".json", delete=False, dir=self.settings.config_path, ) as f: json.dump(config, f, indent=2) temp_config_path = f.name try: # Build command cmd = [ sys.executable, "-m", "gallery_dl", "--config", temp_config_path, "--verbose", url, ] logger.info(f"Executing gallery-dl for {subscription_name}/{platform}: {url}") logger.debug(f"Command: {' '.join(cmd)}") # Execute subprocess # Run in thread pool to not block async event loop loop = asyncio.get_event_loop() proc = await loop.run_in_executor( None, lambda: subprocess.run( cmd, capture_output=True, text=True, timeout=source_config.timeout, ), ) duration = time.time() - start_time completed_at = datetime.utcnow().isoformat() # Log output if proc.stdout: logger.debug(f"STDOUT for {subscription_name}/{platform}:\n{proc.stdout}") if proc.stderr: logger.debug(f"STDERR for {subscription_name}/{platform}:\n{proc.stderr}") # Determine success if proc.returncode == 0: return DownloadResult( success=True, url=url, subscription_name=subscription_name, platform=platform, files_downloaded=self._count_downloaded_files(proc.stdout), stdout=proc.stdout, stderr=proc.stderr, return_code=proc.returncode, duration_seconds=duration, started_at=started_at, completed_at=completed_at, ) # Handle errors error_type, error_message = self._categorize_error( proc.returncode, proc.stdout, proc.stderr ) # NO_NEW_CONTENT is considered successful success = error_type == ErrorType.NO_NEW_CONTENT return DownloadResult( success=success, url=url, subscription_name=subscription_name, platform=platform, files_downloaded=self._count_downloaded_files(proc.stdout) if success else 0, stdout=proc.stdout, stderr=proc.stderr, return_code=proc.returncode, error_type=error_type, error_message=error_message, duration_seconds=duration, started_at=started_at, completed_at=completed_at, ) except subprocess.TimeoutExpired: duration = time.time() - start_time logger.error(f"Download timeout for {subscription_name}/{platform} after {duration:.1f}s") return DownloadResult( success=False, url=url, subscription_name=subscription_name, platform=platform, error_type=ErrorType.TIMEOUT, error_message=f"Download timed out after {source_config.timeout} seconds", duration_seconds=duration, started_at=started_at, completed_at=datetime.utcnow().isoformat(), ) except Exception as e: duration = time.time() - start_time logger.exception(f"Unexpected error downloading {subscription_name}/{platform}: {e}") return DownloadResult( success=False, url=url, subscription_name=subscription_name, platform=platform, error_type=ErrorType.UNKNOWN_ERROR, error_message=str(e), duration_seconds=duration, started_at=started_at, completed_at=datetime.utcnow().isoformat(), ) finally: # Clean up temporary config file try: Path(temp_config_path).unlink() except Exception: pass def get_platform_content_types(self, platform: str) -> list[str]: """Get available content types for a platform. Useful for UI to show what options are available. """ if platform == "patreon": return ["images", "image_large", "attachments", "postfile", "content"] elif platform == "hentaifoundry": return ["all", "pictures", "scraps", "stories"] elif platform == "subscribestar": return ["all"] # No granular control elif platform == "discord": return ["all"] # No granular control elif platform == "pixiv": return ["all", "illusts", "manga", "bookmarks"] elif platform == "deviantart": return ["all", "gallery", "scraps", "favorites"] else: return ["all"] def get_default_config_for_platform(self, platform: str) -> dict: """Get the default configuration for a platform. Useful for showing defaults in UI. """ defaults = self.PLATFORM_DEFAULTS.get(platform, {}) return { "content_types": defaults.get("content_types", ["all"]), "directory_pattern": defaults.get("directory"), "filename_pattern": defaults.get("filename"), "sleep": self._rate_limit, "sleep_request": max(0.5, self._rate_limit / 4), }