Fix #7: Auto-solve hCaptcha with integrated solver & retry logic

udio_wrapper/__init__.py

@@ -9,25 +9,80 @@
 import requests
 import os
 import time
+from .hcaptcha_solver import HCaptchaSolver
 
 class UdioWrapper:
     API_BASE_URL = "https://www.udio.com/api"
 
-    def __init__(self, auth_token):
+    def __init__(self, auth_token, captcha_api_key=None, captcha_service="capsolver"):
         self.auth_token = auth_token
         self.all_track_ids = []
-
-    def make_request(self, url, method, data=None, headers=None):
-        try:
-            if method == 'POST':
-                response = requests.post(url, headers=headers, json=data)
+        # Initialize hCaptcha solver
+        self._captcha_solver = HCaptchaSolver()
+        # Set API key in env if provided via constructor
+        if captcha_api_key:
+            if captcha_service == "2captcha":
+                os.environ["CAPTCHA_API_KEY"] = captcha_api_key
             else:
-                response = requests.get(url, headers=headers)
-            response.raise_for_status()
-            return response
-        except requests.exceptions.RequestException as e:
-            print(f"Error making {method} request to {url}: {e}")
-            return None
+                os.environ["CAPSOLVER_API_KEY"] = captcha_api_key
+
+    def _inject_captcha_header(self, headers):
+        """Inject h-captcha-response header if a captcha token is available."""
+        if headers is None:
+            headers = {}
+        # Only add captcha header if not already present
+        if "h-captcha-response" not in headers:
+            token = self._captcha_solver.get_token()
+            if token:
+                headers = dict(headers)  # Don't mutate caller's dict
+                headers["h-captcha-response"] = token
+        return headers
+
+    def make_request(self, url, method, data=None, headers=None, max_retries=2):
+        for attempt in range(max_retries + 1):
+            try:
+                # Add captcha token for POST requests
+                if method == "POST":
+                    headers = self._inject_captcha_header(headers)
+
+                if method == "POST":
+                    response = requests.post(url, headers=headers, json=data)
+                else:
+                    response = requests.get(url, headers=headers)
+
+                # Check for captcha-related failures
+                if response.status_code in (403, 500, 503) and attempt < max_retries:
+                    print(f"[hCaptcha] {response.status_code} error detected "
+                          f"(attempt {attempt + 1}/{max_retries + 1}), "
+                          f"refreshing captcha token...")
+                    self._captcha_solver.clear_cache()
+                    time.sleep(2)
+                    continue
+
+                response.raise_for_status()
+                return response
+
+            except requests.exceptions.HTTPError as e:
+                if attempt < max_retries:
+                    print(f"[hCaptcha] HTTP error, retrying "
+                          f"(attempt {attempt + 1}/{max_retries + 1}): {e}")
+                    self._captcha_solver.clear_cache()
+                    time.sleep((attempt + 1) * 2)
+                    continue
+                print(f"Error making {method} request to {url}: {e}")
+                return None
+
+            except requests.exceptions.RequestException as e:
+                if attempt < max_retries:
+                    print(f"[hCaptcha] Request error, retrying "
+                          f"(attempt {attempt + 1}/{max_retries + 1}): {e}")
+                    time.sleep((attempt + 1) * 2)
+                    continue
+                print(f"Error making {method} request to {url}: {e}")
+                return None
+
+        print(f"Error: All {max_retries + 1} attempts failed for {url}")
+        return None
 
     def get_headers(self, get_request=False):
         headers = {

udio_wrapper/hcaptcha_solver.py

@@ -0,0 +1,273 @@
+"""
+hCaptcha auto-solver for UdioWrapper.
+Solves issue #7 - 500 Server Error caused by hCaptcha challenges.
+
+Automatically detects hCaptcha failures from Udio's API,
+extracts the site key dynamically, and solves via:
+  1. Capsolver  (CAPSOLVER_API_KEY env var)
+  2. 2captcha    (CAPTCHA_API_KEY env var)
+  3. Manual      (paste token from browser)
+
+Usage:
+    from .hcaptcha_solver import HCaptchaSolver
+    solver = HCaptchaSolver()
+    token = solver.get_token()
+"""
+
+import os
+import re
+import time
+import requests
+
+
+class HCaptchaSolver:
+    """
+    Handles hCaptcha solving for Udio API requests.
+    Caches tokens and site keys to minimize API calls.
+    """
+
+    UDIO_URL = "https://www.udio.com"
+    # Known Udio hCaptcha sitekey (fallback)
+    FALLBACK_SITEKEY = "4c672d35-0701-42b2-88c3-78380b0db560"
+
+    def __init__(self):
+        self._token_cache = None
+        self._token_ttl = 0
+        self._sitekey_cache = None
+        self._sitekey_ttl = 0
+        self._cache_ttl = 110  # hCaptcha tokens usually last ~120s
+
+    def _is_cached(self, cache_val, ttl):
+        """Check if a cached value is still valid."""
+        return cache_val is not None and time.time() < ttl
+
+    def get_sitekey(self):
+        """Get the hCaptcha sitekey from Udio's homepage or cache."""
+        if self._is_cached(self._sitekey_cache, self._sitekey_ttl):
+            return self._sitekey_cache
+
+        try:
+            resp = requests.get(
+                self.UDIO_URL,
+                timeout=15,
+                headers={
+                    "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
+                    "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36"
+                },
+            )
+            key = self._extract_sitekey(resp.text)
+            if key:
+                self._sitekey_cache = key
+                self._sitekey_ttl = time.time() + 3600  # Cache for 1 hour
+                print(f"[hCaptcha] Site key found: {key}")
+                return key
+        except Exception as e:
+            print(f"[hCaptcha] Failed to fetch site key: {e}")
+
+        # Use fallback
+        print(f"[hCaptcha] Using fallback site key: {self.FALLBACK_SITEKEY}")
+        self._sitekey_cache = self.FALLBACK_SITEKEY
+        self._sitekey_ttl = time.time() + 3600
+        return self.FALLBACK_SITEKEY
+
+    def _extract_sitekey(self, html):
+        """Extract hCaptcha sitekey from HTML using regex patterns."""
+        patterns = [
+            r'data-sitekey["\']?\s*[:=]\s*["\']([a-f0-9-]{20,})["\']',
+            r'websiteKey["\']?\s*[:=]\s*["\']([a-f0-9-]{20,})["\']',
+            r'sitekey["\']?\s*[:=]\s*["\']([a-f0-9-]{20,})["\']',
+            r'hcaptcha.*?["\']([a-f0-9-]{36})["\']',
+            r'["\']([a-f0-9-]{36})["\'].*?hcaptcha',
+        ]
+        for pattern in patterns:
+            match = re.search(pattern, html, re.IGNORECASE)
+            if match:
+                return match.group(1)
+        return None
+
+    def get_token(self, force_refresh=False):
+        """
+        Get a valid hCaptcha token. Returns None if solving fails.
+        Uses cached token when available unless force_refresh=True.
+        """
+        if not force_refresh and self._is_cached(self._token_cache, self._token_ttl):
+            return self._token_cache
+
+        sitekey = self.get_sitekey()
+        token = None
+
+        # Try Capsolver first (more reliable for hCaptcha)
+        api_key = os.environ.get("CAPSOLVER_API_KEY") or os.environ.get("CAPTCHA_API_KEY")
+        if api_key:
+            token = self._solve_via_service(sitekey, api_key)
+
+        # Try manual input as fallback
+        if not token:
+            token = self._manual_input()
+
+        if token:
+            self._token_cache = token
+            self._token_ttl = time.time() + self._cache_ttl
+
+        return token
+
+    def _solve_via_service(self, sitekey, api_key):
+        """Solve captcha via Capsolver or 2captcha."""
+        # Detect service type
+        capsolver_key = os.environ.get("CAPSOLVER_API_KEY")
+        captcha_key = os.environ.get("CAPTCHA_API_KEY")
+
+        if capsolver_key:
+            token = self._solve_capsolver(sitekey, capsolver_key)
+            if token:
+                return token
+            # Fall through to 2captcha if capsolver failed and key available
+
+        if captcha_key:
+            token = self._solve_2captcha(sitekey, captcha_key)
+            if token:
+                return token
+
+        return None
+
+    def _solve_capsolver(self, sitekey, api_key):
+        """Solve hCaptcha using Capsolver API."""
+        create_url = "https://api.capsolver.com/createTask"
+        result_url = "https://api.capsolver.com/getTaskResult"
+
+        try:
+            # Create task
+            resp = requests.post(
+                create_url,
+                json={
+                    "clientKey": api_key,
+                    "task": {
+                        "type": "HCaptchaTaskProxyLess",
+                        "websiteURL": self.UDIO_URL,
+                        "websiteKey": sitekey,
+                    },
+                },
+                timeout=15,
+            )
+            data = resp.json()
+            if data.get("errorId", 0) != 0:
+                print(f"[hCaptcha] Capsolver create error: {data.get('errorDescription', data)}")
+                return None
+
+            task_id = data.get("taskId")
+            if not task_id:
+                return None
+
+            # Poll for result
+            for _ in range(60):
+                time.sleep(2)
+                resp = requests.post(
+                    result_url,
+                    json={"clientKey": api_key, "taskId": task_id},
+                    timeout=10,
+                )
+                data = resp.json()
+
+                if data.get("status") == "ready":
+                    solution = data.get("solution", {})
+                    token = solution.get("gRecaptchaResponse") or solution.get("token")
+                    if token:
+                        print("[hCaptcha] Solved via Capsolver")
+                        return token
+
+                if data.get("errorId", 0) != 0:
+                    print(f"[hCaptcha] Capsolver poll error: {data.get('errorDescription', data)}")
+                    return None
+
+            print("[hCaptcha] Capsolver timed out")
+        except Exception as e:
+            print(f"[hCaptcha] Capsolver exception: {e}")
+        return None
+
+    def _solve_2captcha(self, sitekey, api_key):
+        """Solve hCaptcha using 2captcha API."""
+        submit_url = "https://2captcha.com/in.php"
+        result_url = "https://2captcha.com/res.php"
+
+        try:
+            # Submit task
+            resp = requests.post(
+                submit_url,
+                data={
+                    "key": api_key,
+                    "method": "hcaptcha",
+                    "sitekey": sitekey,
+                    "pageurl": self.UDIO_URL,
+                    "json": 1,
+                },
+                timeout=15,
+            )
+            data = resp.json()
+            if data.get("status") != 1:
+                print(f"[hCaptcha] 2captcha submit error: {data}")
+                return None
+
+            captcha_id = data.get("request")
+            if not captcha_id or captcha_id.startswith("ERROR"):
+                print(f"[hCaptcha] 2captcha error: {data.get('request')}")
+                return None
+
+            # Poll for result
+            for _ in range(60):
+                time.sleep(5)
+                resp = requests.get(
+                    result_url,
+                    params={
+                        "key": api_key,
+                        "action": "get",
+                        "id": captcha_id,
+                        "json": 1,
+                    },
+                    timeout=10,
+                )
+                data = resp.json()
+
+                if data.get("status") == 1:
+                    token = data.get("request")
+                    if token:
+                        print("[hCaptcha] Solved via 2captcha")
+                        return token
+
+                if data.get("request") == "ERROR_CAPTCHA_UNSOLVABLE":
+                    print("[hCaptcha] 2captcha: captcha unsolvable")
+                    return None
+
+            print("[hCaptcha] 2captcha timed out")
+        except Exception as e:
+            print(f"[hCaptcha] 2captcha exception: {e}")
+        return None
+
+    def _manual_input(self):
+        """Prompt user to manually solve captcha and paste token."""
+        print()
+        print("=" * 60)
+        print("[hCaptcha] No CAPSOLVER_API_KEY or CAPTCHA_API_KEY set.")
+        print("[hCaptcha] To auto-solve, set one of these environment variables:")
+        print("  export CAPSOLVER_API_KEY=your_key_here")
+        print("  export CAPTCHA_API_KEY=your_key_here")
+        print()
+        print("[hCaptcha] Manual: Visit https://www.udio.com in your browser,")
+        print("  solve the captcha, then:")
+        print("  1. Open DevTools (F12) -> Network tab")
+        print("  2. Find a request to udio.com/api/")
+        print("  3. Look for 'h-captcha-response' in request headers")
+        print("  4. Copy the token value")
+        print("=" * 60)
+        try:
+            token = input("Paste hCaptcha token (or Enter to skip): ").strip()
+            if token:
+                print("[hCaptcha] Manual token accepted")
+                return token
+        except (EOFError, KeyboardInterrupt):
+            pass
+        return None
+
+    def clear_cache(self):
+        """Clear cached tokens to force a fresh solve."""
+        self._token_cache = None
+        self._token_ttl = 0