Auto CAPTCHA VLM: Local Vision Models for Automated CAPTCHA Solving

import torch
from transformers import TrOCRProcessor, VisionEncoderDecoderModel
from PIL import Image
import numpy as np
import cv2

class CaptchaSolver:
    """Vision model-based CAPTCHA solver"""

    def __init__(self, model_name='microsoft/trocr-base-printed'):
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
        print(f"Using device: {self.device}")

        # Load pre-trained TrOCR model
        self.processor = TrOCRProcessor.from_pretrained(model_name)
        self.model = VisionEncoderDecoderModel.from_pretrained(model_name).to(self.device)
        self.model.eval()

        # Fine-tuned character set for CAPTCHAs
        self.charset = '0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz'

    def preprocess_image(self, image_path):
        """Enhanced preprocessing for CAPTCHA images"""
        # Load image
        img = cv2.imread(image_path)
        if img is None:
            raise ValueError(f"Failed to load image: {image_path}")

        # Convert to grayscale
        gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)

        # Apply adaptive thresholding
        thresh = cv2.adaptiveThreshold(
            gray, 255,
            cv2.ADAPTIVE_THRESH_GAUSSIAN_C,
            cv2.THRESH_BINARY_INV, 11, 2
        )

        # Remove noise
        kernel = np.ones((2,2), np.uint8)
        opening = cv2.morphologyEx(thresh, cv2.MORPH_OPEN, kernel)

        # Dilate to connect character parts
        kernel = np.ones((3,3), np.uint8)
        dilated = cv2.dilate(opening, kernel, iterations=1)

        # Convert back to PIL Image
        pil_image = Image.fromarray(dilated)

        return pil_image

    def denoise_image(self, pil_image):
        """Advanced denoising using ML-based approach"""
        img_array = np.array(pil_image)

        # Apply bilateral filter for edge-preserving denoising
        denoised = cv2.bilateralFilter(img_array, 9, 75, 75)

        # Remove small noise with median filter
        denoised = cv2.medianBlur(denoised, 3)

        return Image.fromarray(denoised)

    def segment_characters(self, pil_image):
        """Character segmentation for complex CAPTCHAs"""
        img_array = np.array(pil_image)

        # Find contours
        contours, _ = cv2.findContours(
            img_array,
            cv2.RETR_EXTERNAL,
            cv2.CHAIN_APPROX_SIMPLE
        )

        # Filter and sort contours left to right
        char_contours = []
        for contour in contours:
            x, y, w, h = cv2.boundingRect(contour)
            if w > 5 and h > 10 and w < 50 and h < 50:  # Filter noise
                char_contours.append((x, y, w, h))

        # Sort by x position
        char_contours = sorted(char_contours, key=lambda c: c[0])

        # Extract character images
        char_images = []
        for x, y, w, h in char_contours:
            char_img = img_array[y:y+h, x:x+w]
            char_images.append(Image.fromarray(char_img))

        return char_images

    def solve_captcha(self, image_path):
        """Main CAPTCHA solving pipeline"""
        try:
            # Preprocess image
            processed = self.preprocess_image(image_path)

            # Denoise
            denoised = self.denoise_image(processed)

            # Prepare for model
            pixel_values = self.processor(images=denoised, return_tensors="pt").pixel_values.to(self.device)

            # Generate text prediction
            with torch.no_grad():
                generated_ids = self.model.generate(pixel_values, max_length=10)
                generated_text = self.processor.batch_decode(generated_ids, skip_special_tokens=True)[0]

            # Post-process: filter to charset
            result = ''.join(c for c in generated_text if c in self.charset)

            confidence = self._calculate_confidence(pixel_values, generated_ids)

            return {
                'text': result,
                'confidence': confidence,
                'method': 'trocr_vision_model'
            }

        except Exception as e:
            print(f"Error solving CAPTCHA: {e}")
            return {
                'text': None,
                'confidence': 0.0,
                'error': str(e)
            }

    def _calculate_confidence(self, pixel_values, generated_ids):
        """Calculate confidence score for the prediction"""
        with torch.no_grad():
            outputs = self.model(
                pixel_values=pixel_values,
                labels=generated_ids
            )
            # Convert loss to confidence
            loss = outputs.loss.item()
            confidence = max(0.0, min(1.0, 1.0 - (loss / 5.0)))
        return confidence

# Batch processing for performance
class BatchCaptchaSolver:
    """Solve multiple CAPTCHAs in parallel on GPU"""

    def __init__(self, batch_size=8):
        self.solver = CaptchaSolver()
        self.batch_size = batch_size

    async def solve_batch(self, image_paths):
        """Solve multiple CAPTCHAs in parallel"""
        results = []
        for image_path in image_paths:
            result = self.solver.solve_captcha(image_path)
            results.append(result)
        return results

import torch
import gc
from contextlib import contextmanager

class MemoryEfficientCaptchaSolver:
    """Memory-optimized CAPTCHA solver with explicit cleanup"""

    def __init__(self, model_name='microsoft/trocr-base-printed'):
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
        self.model_name = model_name
        self.load_model()

        # Performance tracking
        self.solve_count = 0
        self.reload_interval = 500  # Reload model every 500 solves

    def load_model(self):
        """Load model with memory-efficient settings"""
        self.processor = TrOCRProcessor.from_pretrained(self.model_name)
        self.model = VisionEncoderDecoderModel.from_pretrained(
            self.model_name,
            torch_dtype=torch.float16 if self.device.type == 'cuda' else torch.float32
        ).to(self.device)

        # Disable gradients for inference
        for param in self.model.parameters():
            param.requires_grad = False

        self.model.eval()

    @contextmanager
    def torch_context(self):
        """Context manager for tensor cleanup"""
        yield
        # Explicit cleanup
        if torch.cuda.is_available():
            torch.cuda.empty_cache()
        gc.collect()

    def solve_with_cleanup(self, image_path):
        """Solve CAPTCHA with explicit memory cleanup"""
        try:
            with self.torch_context():
                # Load and preprocess
                image = Image.open(image_path).convert('RGB')
                pixel_values = self.processor(
                    images=image,
                    return_tensors="pt"
                ).pixel_values.to(self.device)

                # Inference with no_grad
                with torch.no_grad():
                    generated_ids = self.model.generate(
                        pixel_values,
                        max_length=10,
                        num_beams=4,
                        early_stopping=True
                    )

                    # Decode
                    generated_text = self.processor.batch_decode(
                        generated_ids,
                        skip_special_tokens=True
                    )[0]

                # Explicitly delete tensors
                del pixel_values
                del generated_ids

                self.solve_count += 1

                # Periodic model reload to clear accumulated memory
                if self.solve_count % self.reload_interval == 0:
                    self.reload_model()

                return {
                    'text': generated_text,
                    'success': True
                }

        except Exception as e:
            print(f"Solve error: {e}")
            # Force cleanup on error
            gc.collect()
            if torch.cuda.is_available():
                torch.cuda.empty_cache()
            return {
                'text': None,
                'success': False,
                'error': str(e)
            }

    def reload_model(self):
        """Reload model to clear accumulated memory"""
        print(f"Reloading model after {self.solve_count} solves...")

        # Delete old model
        del self.model
        del self.processor

        # Force cleanup
        gc.collect()
        if torch.cuda.is_available():
            torch.cuda.empty_cache()

        # Load fresh model
        self.load_model()

    def get_memory_stats(self):
        """Get current GPU memory usage"""
        if torch.cuda.is_available():
            allocated = torch.cuda.memory_allocated() / 1024**3  # GB
            reserved = torch.cuda.memory_reserved() / 1024**3
            return {
                'allocated_gb': allocated,
                'reserved_gb': reserved,
                'solve_count': self.solve_count
            }
        return {'error': 'CUDA not available'}

import torch
from transformers import CLIPProcessor, CLIPModel
from torchvision import transforms
from PIL import Image
import numpy as np

class RecaptchaImageSolver:
    """Solve reCAPTCHA v2 image challenges using multimodal models"""

    def __init__(self):
        self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

        # Load CLIP for vision-language understanding
        self.clip_processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")
        self.clip_model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32").to(self.device)
        self.clip_model.eval()

        # Common reCAPTCHA challenge categories
        self.categories = [
            "bicycle", "bus", "car", "motorcycle", "airplane",
            "boat", "train", "truck", "traffic light",
            "fire hydrant", "stop sign", "parking meter",
            "bench", "chair", "couch", "potted plant",
            "bed", "dining table", "toilet", "tv",
            "laptop", "mouse", "remote", "keyboard",
            "cell phone", "microwave", "oven", "toaster",
            "sink", "refrigerator", "book", "clock",
            "vase", "scissors", "teddy bear", "hair drier"
        ]

    async def solve_image_challenge(self, challenge_text, image_paths):
        """Solve reCAPTCHA v2 image challenge"""
        # Parse challenge text to identify target
        target_category = self._parse_challenge_text(challenge_text)

        if target_category not in self.categories:
            print(f"Unknown category: {target_category}")
            return None

        # Score each image
        image_scores = []
        for img_path in image_paths:
            score = await self._score_image(img_path, target_category)
            image_scores.append({
                'path': img_path,
                'score': score,
                'selected': score > 0.7  # Threshold
            })

        return image_scores

    def _parse_challenge_text(self, challenge_text):
        """Extract target category from challenge text"""
        # Examples: "Select all bicycles", "Click on traffic lights"
        text_lower = challenge_text.lower()

        for category in self.categories:
            if category.replace('_', ' ') in text_lower:
                return category

        # Fallback: use CLIP to find closest category
        return self._find_closest_category(challenge_text)

    async def _score_image(self, image_path, target_category):
        """Score image against target category using CLIP"""
        try:
            image = Image.open(image_path).convert('RGB')

            # Prepare inputs
            text_inputs = [f"a photo of a {target_category}",
                          f"a {target_category}",
                          target_category]

            inputs = self.clip_processor(
                text=text_inputs,
                images=image,
                return_tensors="pt",
                padding=True
            ).to(self.device)

            # Get similarity scores
            with torch.no_grad():
                outputs = self.clip_model(**inputs)
                logits_per_image = outputs.logits_per_image
                probs = logits_per_image.softmax(dim=-1)

            # Maximum probability across text prompts
            max_prob = probs.cpu().numpy().max()

            return float(max_prob)

        except Exception as e:
            print(f"Error scoring {image_path}: {e}")
            return 0.0

    def _find_closest_category(self, challenge_text):
        """Find closest matching category using CLIP"""
        # Create text embeddings for challenge and categories
        text_inputs = [challenge_text] + [f"a {cat}" for cat in self.categories]

        inputs = self.clip_processor(
            text=text_inputs,
            return_tensors="pt",
            padding=True
        ).to(self.device)

        with torch.no_grad():
            text_features = self.clip_model.get_text_features(**inputs)

        # Find closest category
        challenge_features = text_features[0:1]
        category_features = text_features[1:]

        similarities = torch.cosine_similarity(
            challenge_features,
            category_features
        )

        closest_idx = similarities.argmax().item()
        return self.categories[closest_idx]

    async def select_images(self, challenge_text, image_grid):
        """Select images from a reCAPTCHA grid"""
        image_paths = self._extract_grid_images(image_grid)
        results = await self.solve_image_challenge(challenge_text, image_paths)

        if results:
            selected_indices = [
                i for i, r in enumerate(results)
                if r['selected']
            ]
            return selected_indices

        return None