captcha_solver.py

"""
Advanced CAPTCHA Solving Integration
Supports multiple third-party CAPTCHA solving services and OCR fallbacks
"""

import os
import time
import base64
import requests
from typing import Dict, Optional, Union
from dotenv import load_dotenv

load_dotenv()

class CaptchaSolver:
    """
    Advanced CAPTCHA solver with multiple service integrations
    """
    
    def __init__(self):
        # API keys for different services
        self.twocaptcha_key = os.environ.get("TWOCAPTCHA_API_KEY")
        self.anticaptcha_key = os.environ.get("ANTICAPTCHA_API_KEY")
        self.capmonster_key = os.environ.get("CAPMONSTER_API_KEY")
        
        # Service endpoints
        self.services = {
            "2captcha": {
                "submit": "http://2captcha.com/in.php",
                "result": "http://2captcha.com/res.php"
            },
            "anti-captcha": {
                "submit": "https://api.anti-captcha.com/createTask",
                "result": "https://api.anti-captcha.com/getTaskResult"
            },
            "capmonster": {
                "submit": "https://api.capmonster.cloud/createTask",
                "result": "https://api.capmonster.cloud/getTaskResult"
            }
        }
    
    def solve_recaptcha_v2(self, site_key: str, page_url: str, service: str = "2captcha") -> Optional[str]:
        """
        Solve reCAPTCHA v2 using third-party service
        
        Args:
            site_key: reCAPTCHA site key
            page_url: URL where CAPTCHA is located
            service: Service to use (2captcha, anti-captcha, capmonster)
            
        Returns:
            CAPTCHA solution token or None if failed
        """
        
        if service == "2captcha" and self.twocaptcha_key:
            return self._solve_2captcha_recaptcha(site_key, page_url)
        elif service == "anti-captcha" and self.anticaptcha_key:
            return self._solve_anticaptcha_recaptcha(site_key, page_url)
        elif service == "capmonster" and self.capmonster_key:
            return self._solve_capmonster_recaptcha(site_key, page_url)
        else:
            print(f"❌ Service {service} not available or API key missing")
            return None
    
    def solve_image_captcha(self, image_data: Union[str, bytes], service: str = "2captcha") -> Optional[str]:
        """
        Solve image-based CAPTCHA
        
        Args:
            image_data: Base64 encoded image or raw bytes
            service: Service to use
            
        Returns:
            CAPTCHA solution text or None if failed
        """
        
        if isinstance(image_data, bytes):
            image_data = base64.b64encode(image_data).decode('utf-8')
        
        if service == "2captcha" and self.twocaptcha_key:
            return self._solve_2captcha_image(image_data)
        elif service == "anti-captcha" and self.anticaptcha_key:
            return self._solve_anticaptcha_image(image_data)
        else:
            # Fallback to OCR
            return self._solve_with_ocr(image_data)
    
    def solve_hcaptcha(self, site_key: str, page_url: str, service: str = "2captcha") -> Optional[str]:
        """
        Solve hCaptcha using third-party service
        """
        
        if service == "2captcha" and self.twocaptcha_key:
            return self._solve_2captcha_hcaptcha(site_key, page_url)
        elif service == "anti-captcha" and self.anticaptcha_key:
            return self._solve_anticaptcha_hcaptcha(site_key, page_url)
        else:
            print(f"❌ hCaptcha service {service} not available")
            return None
    
    def _solve_2captcha_recaptcha(self, site_key: str, page_url: str) -> Optional[str]:
        """Solve reCAPTCHA using 2captcha service"""
        
        try:
            # Submit CAPTCHA
            submit_data = {
                'key': self.twocaptcha_key,
                'method': 'userrecaptcha',
                'googlekey': site_key,
                'pageurl': page_url,
                'json': 1
            }
            
            response = requests.post(self.services["2captcha"]["submit"], data=submit_data)
            result = response.json()
            
            if result['status'] != 1:
                print(f"❌ 2captcha submit failed: {result.get('error_text', 'Unknown error')}")
                return None
            
            captcha_id = result['request']
            print(f"🔄 2captcha CAPTCHA submitted, ID: {captcha_id}")
            
            # Wait for solution
            for attempt in range(30):  # Wait up to 5 minutes
                time.sleep(10)
                
                result_data = {
                    'key': self.twocaptcha_key,
                    'action': 'get',
                    'id': captcha_id,
                    'json': 1
                }
                
                response = requests.get(self.services["2captcha"]["result"], params=result_data)
                result = response.json()
                
                if result['status'] == 1:
                    print("✅ 2captcha reCAPTCHA solved successfully!")
                    return result['request']
                elif result['request'] == 'CAPCHA_NOT_READY':
                    continue
                else:
                    print(f"❌ 2captcha error: {result.get('error_text', 'Unknown error')}")
                    return None
            
            print("❌ 2captcha timeout waiting for solution")
            return None
            
        except Exception as e:
            print(f"❌ 2captcha exception: {e}")
            return None
    
    def _solve_2captcha_image(self, image_data: str) -> Optional[str]:
        """Solve image CAPTCHA using 2captcha"""
        
        try:
            submit_data = {
                'key': self.twocaptcha_key,
                'method': 'base64',
                'body': image_data,
                'json': 1
            }
            
            response = requests.post(self.services["2captcha"]["submit"], data=submit_data)
            result = response.json()
            
            if result['status'] != 1:
                print(f"❌ 2captcha image submit failed: {result.get('error_text', 'Unknown error')}")
                return None
            
            captcha_id = result['request']
            print(f"🔄 2captcha image CAPTCHA submitted, ID: {captcha_id}")
            
            # Wait for solution
            for attempt in range(20):  # Wait up to 3+ minutes
                time.sleep(10)
                
                result_data = {
                    'key': self.twocaptcha_key,
                    'action': 'get',
                    'id': captcha_id,
                    'json': 1
                }
                
                response = requests.get(self.services["2captcha"]["result"], params=result_data)
                result = response.json()
                
                if result['status'] == 1:
                    print("✅ 2captcha image CAPTCHA solved successfully!")
                    return result['request']
                elif result['request'] == 'CAPCHA_NOT_READY':
                    continue
                else:
                    print(f"❌ 2captcha error: {result.get('error_text', 'Unknown error')}")
                    return None
            
            print("❌ 2captcha timeout waiting for image solution")
            return None
            
        except Exception as e:
            print(f"❌ 2captcha image exception: {e}")
            return None
    
    def _solve_2captcha_hcaptcha(self, site_key: str, page_url: str) -> Optional[str]:
        """Solve hCaptcha using 2captcha service"""
        
        try:
            submit_data = {
                'key': self.twocaptcha_key,
                'method': 'hcaptcha',
                'sitekey': site_key,
                'pageurl': page_url,
                'json': 1
            }
            
            response = requests.post(self.services["2captcha"]["submit"], data=submit_data)
            result = response.json()
            
            if result['status'] != 1:
                print(f"❌ 2captcha hCaptcha submit failed: {result.get('error_text', 'Unknown error')}")
                return None
            
            captcha_id = result['request']
            print(f"🔄 2captcha hCaptcha submitted, ID: {captcha_id}")
            
            # Wait for solution
            for attempt in range(30):
                time.sleep(10)
                
                result_data = {
                    'key': self.twocaptcha_key,
                    'action': 'get',
                    'id': captcha_id,
                    'json': 1
                }
                
                response = requests.get(self.services["2captcha"]["result"], params=result_data)
                result = response.json()
                
                if result['status'] == 1:
                    print("✅ 2captcha hCaptcha solved successfully!")
                    return result['request']
                elif result['request'] == 'CAPCHA_NOT_READY':
                    continue
                else:
                    print(f"❌ 2captcha error: {result.get('error_text', 'Unknown error')}")
                    return None
            
            return None
            
        except Exception as e:
            print(f"❌ 2captcha hCaptcha exception: {e}")
            return None
    
    def _solve_with_ocr(self, image_data: str) -> Optional[str]:
        """
        Fallback OCR solution using Pytesseract
        Works for simple text CAPTCHAs but not reCAPTCHA
        """
        
        try:
            import pytesseract
            from PIL import Image
            import io
            
            # Decode base64 image
            image_bytes = base64.b64decode(image_data)
            image = Image.open(io.BytesIO(image_bytes))
            
            # Preprocess image for better OCR
            image = image.convert('L')  # Convert to grayscale
            
            # Extract text using OCR
            text = pytesseract.image_to_string(image, config='--psm 8 -c tessedit_char_whitelist=0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz')
            text = text.strip()
            
            if text:
                print(f"✅ OCR solved CAPTCHA: {text}")
                return text
            else:
                print("❌ OCR could not extract text from CAPTCHA")
                return None
                
        except ImportError:
            print("❌ OCR dependencies not installed (pytesseract, Pillow)")
            return None
        except Exception as e:
            print(f"❌ OCR exception: {e}")
            return None
    
    def get_balance(self, service: str = "2captcha") -> Optional[float]:
        """Get account balance for CAPTCHA service"""
        
        try:
            if service == "2captcha" and self.twocaptcha_key:
                response = requests.get(
                    "http://2captcha.com/res.php",
                    params={'key': self.twocaptcha_key, 'action': 'getbalance'}
                )
                
                if response.text.startswith('ERROR'):
                    print(f"❌ 2captcha balance error: {response.text}")
                    return None
                else:
                    balance = float(response.text)
                    print(f"💰 2captcha balance: ${balance}")
                    return balance
            
        except Exception as e:
            print(f"❌ Balance check error: {e}")
            return None

# Global solver instance
captcha_solver = CaptchaSolver()

def solve_captcha_for_page(page, captcha_type: str = "recaptcha") -> Optional[str]:
    """
    Detect and solve CAPTCHA on a Playwright page
    
    Args:
        page: Playwright page object
        captcha_type: Type of CAPTCHA (recaptcha, hcaptcha, image)
        
    Returns:
        CAPTCHA solution token or None
    """
    
    try:
        if captcha_type == "recaptcha":
            # Look for reCAPTCHA
            site_key_element = page.locator('[data-sitekey]').first
            if site_key_element.is_visible():
                site_key = site_key_element.get_attribute('data-sitekey')
                page_url = page.url
                
                print(f"🤖 Detected reCAPTCHA with site key: {site_key}")
                solution = captcha_solver.solve_recaptcha_v2(site_key, page_url)
                
                if solution:
                    # Inject solution into page
                    page.evaluate(f'document.getElementById("g-recaptcha-response").innerHTML = "{solution}";')
                    return solution
        
        elif captcha_type == "hcaptcha":
            # Look for hCaptcha
            site_key_element = page.locator('[data-sitekey]').first
            if site_key_element.is_visible():
                site_key = site_key_element.get_attribute('data-sitekey')
                page_url = page.url
                
                print(f"🤖 Detected hCaptcha with site key: {site_key}")
                solution = captcha_solver.solve_hcaptcha(site_key, page_url)
                
                if solution:
                    # Inject solution into page
                    page.evaluate(f'document.querySelector("[name=h-captcha-response]").value = "{solution}";')
                    return solution
        
        elif captcha_type == "image":
            # Look for image CAPTCHA
            captcha_img = page.locator('img[src*="captcha"], img[alt*="captcha"], .captcha img').first
            if captcha_img.is_visible():
                # Take screenshot of CAPTCHA image
                image_data = captcha_img.screenshot()
                image_b64 = base64.b64encode(image_data).decode('utf-8')
                
                print("🖼️ Detected image CAPTCHA")
                solution = captcha_solver.solve_image_captcha(image_b64)
                
                if solution:
                    # Find input field and enter solution
                    captcha_input = page.locator('input[name*="captcha"], input[id*="captcha"], .captcha input').first
                    if captcha_input.is_visible():
                        captcha_input.fill(solution)
                        return solution
        
        return None
        
    except Exception as e:
        print(f"❌ CAPTCHA solving error: {e}")
        return None

# Test function
def test_captcha_solver():
    """Test CAPTCHA solver functionality"""
    
    print("🧪 Testing CAPTCHA Solver")
    print("=" * 40)
    
    # Check balances
    captcha_solver.get_balance("2captcha")
    
    print("\n💡 CAPTCHA Solver Ready!")
    print("✅ Supports: reCAPTCHA v2, hCaptcha, Image CAPTCHAs")
    print("✅ Services: 2captcha, Anti-Captcha, CapMonster")
    print("✅ OCR Fallback: Pytesseract for simple text CAPTCHAs")

if __name__ == "__main__":
    test_captcha_solver()