import redis
import asyncio
import json
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
import os
import time
from typing import Dict, Any, Optional

# Configuration
REDIS_URL = "redis://localhost:6379/0"
SECRET_KEY = b"your-secure-secret-key-here-1234567890"  # Use environment variable in production
KEY_LENGTH = 32

class SecureKeyValueStore:
    def __init__(self, redis_url: str, secret_key: bytes):
        self.redis_client = redis.from_url(redis_url)
        self.fernet = Fernet(secret_key)
        self._lock = asyncio.Lock()
        self.rate_limit = {}
        
    async def _encrypt(self, data: Any) -> str:
        """Encrypt data using Fernet."""
        json_data = json.dumps(data).encode('utf-8')
        return self.fernet.encrypt(json_data).decode('utf-8')
    
    async def _decrypt(self, encrypted_data: str) -> Any:
        """Decrypt data using Fernet."""
        try:
            decrypted = self.fernet.decrypt(encrypted_data.encode('utf-8'))
            return json.loads(decrypted.decode('utf-8'))
        except Exception as e:
            logger.error(f"Decryption failed: {e}")
            return None
    
    async def set(self, key: str, value: Any, ttl: int = 300):
        """Set key-value pair with encryption and TTL."""
        encrypted_value = await self._encrypt(value)
        async with self._lock:
            await self.redis_client.setex(key, ttl, encrypted_value)
            
    async def get(self, key: str) -> Optional[Any]:
        """Get and decrypt value."""
        raw_value = await self.redis_client.get(key)
        if raw_value:
            return await self._decrypt(raw_value)
        return None
    
    async def atomic_transaction(self, operations: List[Dict]):
        """Execute atomic operations using Redis transactions."""
        pipe = self.redis_client.pipeline()
        for op in operations:
            if op['type'] == 'set':
                encrypted = await self._encrypt(op['value'])
                pipe.setex(op['key'], op.get('ttl', 300), encrypted)
            elif op['type'] == 'delete':
                pipe.delete(op['key'])
        await pipe.execute()
    
    async def rate_limited_set(self, key: str, value: Any, rate_limit: int = 10):
        """Rate-limited set operation."""
        now = time.time()
        if key not in self.rate_limit:
            self.rate_limit[key] = []
        self.rate_limit[key] = [t for t in self.rate_limit[key] if t > now - 60]
        
        if len(self.rate_limit[key]) >= rate_limit:
            raise Exception("Rate limit exceeded")
        
        self.rate_limit[key].append(now)
        await self.set(key, value)

# Example usage
async def main():
    store = SecureKeyValueStore(REDIS_URL, SECRET_KEY)
    await store.set("user:1", {"name": "John", "age": 30})
    data = await store.get("user:1")
    print(data)
    await store.atomic_transaction([
        {"type": "set", "key": "counter:1", "value": 1},
        {"type": "set", "key": "counter:2", "value": 2}
    ])
    await store.rate_limited_set("api_call", {"count": 1}, rate_limit=5)

# Run the example
asyncio.run(main())

import asyncio
import aiohttp
import redis
import json
import random
from typing import Dict, Any, List
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import time

# Configuration
REDIS_URL = "redis://localhost:6379/0"
PROXIES = ["https://proxy1:8080", "https://proxy2:8080"]
HEADERS = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}

class AsyncWebScraper:
    def __init__(self, redis_url: str):
        self.redis_client = redis.from_url(redis_url)
        self.session = None
        self.proxy = None
        
    async def setup_session(self):
        """Setup aiohttp session with proxy."""
        self.session = aiohttp.ClientSession()
        
    async def get_with_proxy(self, url: str) -> str:
        """Fetch URL with random proxy."""
        self.proxy = random.choice(PROXIES)
        headers = HEADERS.copy()
        headers["Accept"] = "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"
        
        try:
            async with self.session.get(url, headers=headers, proxy=self.proxy) as response:
                return await response.text()
        except Exception as e:
            print(f"Request failed: {e}")
            return None
            
    async def scrape_with_selenium(self, url: str) -> str:
        """Scrape dynamic content using Selenium."""
        options = Options()
        options.add_argument("--headless")
        options.add_argument("--no-sandbox")
        options.add_argument("--disable-dev-shm-usage")
        
        driver = webdriver.Chrome(options=options)
        try:
            driver.get(url)
            WebDriverWait(driver, 10).until(
                EC.presence_of_element_located((By.TAG_NAME, "body"))
            )
            return driver.page_source
        finally:
            driver.quit()
            
    async def process_task(self, task_id: str, url: str):
        """Process individual scraping task."""
        # Rate limiting
        await asyncio.sleep(random.uniform(1, 3))
        
        # Try HTTP first, fallback to Selenium
        html = await self.get_with_proxy(url)
        if not html:
            html = await self.scrape_with_selenium(url)
            
        # Store result
        if html:
            await self.redis_client.set(f"result:{task_id}", html)
            
    async def worker_loop(self):
        """Worker that processes tasks from Redis queue."""
        while True:
            task = await self.redis_client.brpop("scraping_queue", timeout=5)
            if task:
                task_id, url = task[1].decode().split(":")
                await self.process_task(task_id, url)

# Example usage
async def main():
    scraper = AsyncWebScraper(REDIS_URL)
    await scraper.setup_session()
    
    # Add tasks to queue
    for i in range(5):
        await scraper.redis_client.lpush("scraping_queue", f"{i}:https://example.com")
    
    # Start worker
    await scraper.worker_loop()

asyncio.run(main())