AI-Powered Web Scraping Without API Costs

import asyncio
import aiohttp
from transformers import AutoTokenizer, AutoModelForCausalLM
import torch
import json

class AIWebScraper:
    def __init__(self, model_name="meta-llama/Llama-3.2-11B-Instruct"):
        """Initialize with a real open-source model"""
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.model = AutoModelForCausalLM.from_pretrained(
            model_name,
            torch_dtype=torch.float16,  # Use FP16 to save memory
            device_map="auto"
        )
        
    async def fetch_with_proxy(self, url, proxy=None):
        """Fetch page content through proxy"""
        async with aiohttp.ClientSession() as session:
            proxy_url = f"http://{proxy}" if proxy else None
            async with session.get(url, proxy=proxy_url) as response:
                return await response.text()
    
    def extract_data(self, html_content, extraction_prompt):
        """Use AI to extract structured data from HTML"""
        # Truncate HTML to fit context window
        max_html_length = 8000  # Conservative limit
        if len(html_content) > max_html_length:
            html_content = html_content[:max_html_length]
        
        prompt = f"""Extract the following information from this HTML:
{extraction_prompt}

HTML Content:
{html_content}

Return valid JSON only:"""
        
        inputs = self.tokenizer(prompt, return_tensors="pt", truncation=True)
        
        with torch.no_grad():
            outputs = self.model.generate(
                **inputs,
                max_new_tokens=500,
                temperature=0.1,  # Low temperature for consistency
                do_sample=True
            )
        
        response = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
        
        # Extract JSON from response
        try:
            json_start = response.find('{')
            json_end = response.rfind('}') + 1
            json_str = response[json_start:json_end]
            return json.loads(json_str)
        except:
            return {"error": "Failed to parse AI response", "raw": response}

# Usage Example
async def main():
    scraper = AIWebScraper()
    
    # Configure mobile proxy (example)
    mobile_proxy = "your-mobile-proxy.com:8080"
    
    # Fetch page
    html = await scraper.fetch_with_proxy(
        "https://example.com/products",
        proxy=mobile_proxy
    )
    
    # Define what to extract
    extraction_prompt = """
    - Product names
    - Prices (number only)
    - Availability status
    """
    
    # Extract with AI
    data = scraper.extract_data(html, extraction_prompt)
    print(json.dumps(data, indent=2))

if __name__ == "__main__":
    asyncio.run(main())