feat(downloads): add patreon URL-rewrite and error-pattern helpers

backend/app/tasks/downloads.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import logging
+import re
 from datetime import datetime, timedelta
 from typing import Optional
 
@@ -70,6 +71,48 @@ def _get_db_setting_sync(key: str, default=None):
     return asyncio.run(_fetch())
 
 
+# --- Patreon campaign-ID auto-resolution helpers -------------------------
+# These support auto-healing gallery-dl's "Failed to extract campaign ID"
+# failures on Patreon sources. See docs/superpowers/specs/2026-04-18-patreon-campaign-resolver-design.md.
+
+_PATREON_VANITY_RE = re.compile(
+    r"^https?://(?:www\.)?patreon\.com/(?:c/)?(?!id:)([^/?#]+)",
+    re.IGNORECASE,
+)
+_CAMPAIGN_ID_FAILURE_PATTERN = "failed to extract campaign id"
+
+
+def _effective_patreon_url(platform: str, source_url: str, metadata: Optional[dict]) -> str:
+    """Return the URL to pass to gallery-dl for this source.
+
+    For Patreon sources with a cached campaign_id in metadata, returns
+    patreon.com/id:<id> to bypass the broken vanity-URL lookup path.
+    All other cases return source_url unchanged.
+    """
+    if platform == "patreon" and metadata:
+        campaign_id = metadata.get("patreon_campaign_id")
+        if campaign_id:
+            return f"https://www.patreon.com/id:{campaign_id}"
+    return source_url
+
+
+def _looks_like_campaign_id_failure(dl_result) -> bool:
+    """True if the gallery-dl output contains the campaign-ID extraction failure marker."""
+    stdout = getattr(dl_result, "stdout", None) or ""
+    stderr = getattr(dl_result, "stderr", None) or ""
+    return _CAMPAIGN_ID_FAILURE_PATTERN in f"{stdout}\n{stderr}".lower()
+
+
+def _extract_vanity_from_url(url: str) -> Optional[str]:
+    """Return the vanity (creator slug) from a Patreon URL, or None if not extractable.
+
+    Handles patreon.com/<vanity>, patreon.com/c/<vanity>[/...], with or without www.
+    Returns None for id:<N> URLs, bare domain, and non-Patreon URLs.
+    """
+    m = _PATREON_VANITY_RE.match(url)
+    return m.group(1) if m else None
+
+
 async def _download_source_async(source_id: int, download_id: int = None) -> dict:
     """Async implementation of source download.