src/core/web_loader.py

#
# SPDX-FileCopyrightText: Hadad <hadad@linuxmail.org>
# SPDX-License-Identifier: Apache-2.0
#

import random
import threading
from collections import deque
from config import (
    OS,
    OCTETS,
    BROWSERS,
    CHROME_VERSIONS,
    FIREFOX_VERSIONS,
    SAFARI_VERSIONS,
    EDGE_VERSIONS,
    DOMAINS,
    PROTOCOLS,
    SEARCH_ENGINES,
    KEYWORDS,
    COUNTRIES,
    LANGUAGES,
    TIMEZONES
)

class WebLoader:
    def __init__(self):
        self.ipv4_pool = deque(maxlen=1000)
        self.ipv6_pool = deque(maxlen=1000)
        self.user_agent_pool = deque(maxlen=500)
        self.origin_pool = deque(maxlen=500)
        self.referrer_pool = deque(maxlen=500)
        self.location_pool = deque(maxlen=500)
        self.lock = threading.Lock()
        self.running = True
        
    def generate_ipv4(self):
        while len(self.ipv4_pool) < 1000 and self.running:
            octet = random.choice(OCTETS)
            ip = f"{octet}.{random.randint(0, 255)}.{random.randint(0, 255)}.{random.randint(1, 254)}"
            with self.lock:
                self.ipv4_pool.append(ip)
            
    def generate_ipv6(self):
        while len(self.ipv6_pool) < 1000 and self.running:
            segments = []
            for _ in range(8):
                segments.append(f"{random.randint(0, 65535):04x}")
            ip = ":".join(segments)
            with self.lock:
                self.ipv6_pool.append(ip)
            
    def generate_user_agents(self):
        os_list = OS
        browsers = BROWSERS
        chrome_versions = CHROME_VERSIONS
        firefox_versions = FIREFOX_VERSIONS
        safari_versions = SAFARI_VERSIONS
        edge_versions = EDGE_VERSIONS
        
        while len(self.user_agent_pool) < 500 and self.running:
            browser = random.choice(browsers)
            os_string = random.choice(os_list)
            
            if browser == "Chrome":
                version = random.choice(chrome_versions)
                ua = f"Mozilla/5.0 ({os_string}) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{version} Safari/537.36"
            elif browser == "Firefox":
                version = random.choice(firefox_versions)
                ua = f"Mozilla/5.0 ({os_string}) Gecko/20100101 Firefox/{version}"
            elif browser == "Safari":
                version = random.choice(safari_versions)
                webkit_version = f"{600 + random.randint(0, 15)}.{random.randint(1, 9)}.{random.randint(1, 20)}"
                ua = f"Mozilla/5.0 ({os_string}) AppleWebKit/{webkit_version} (KHTML, like Gecko) Version/{version} Safari/{webkit_version}"
            elif browser == "Edge":
                version = random.choice(edge_versions)
                ua = f"Mozilla/5.0 ({os_string}) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{version.split('.')[0]}.0.0.0 Safari/537.36 Edg/{version}"
            else:
                version = f"{random.randint(70, 100)}.0.{random.randint(3000, 5000)}.{random.randint(50, 150)}"
                ua = f"Mozilla/5.0 ({os_string}) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{version} Safari/537.36 OPR/{random.randint(80, 106)}.0.0.0"
                
            with self.lock:
                self.user_agent_pool.append(ua)
            
    def generate_origins(self):
        domains = DOMAINS
        protocols = PROTOCOLS
        
        while len(self.origin_pool) < 500 and self.running:
            protocol = random.choice(protocols)
            domain = random.choice(domains)
            origin = f"{protocol}{domain}"
            with self.lock:
                self.origin_pool.append(origin)
            
    def generate_referrers(self):
        search_engines = SEARCH_ENGINES
        keywords = KEYWORDS
        
        while len(self.referrer_pool) < 500 and self.running:
            engine = random.choice(search_engines)
            keyword = random.choice(keywords)
            referrer = f"{engine}{keyword}"
            with self.lock:
                self.referrer_pool.append(referrer)
            
    def generate_locations(self):
        countries = COUNTRIES
        languages = LANGUAGES
        timezones = TIMEZONES
        
        while len(self.location_pool) < 500 and self.running:
            country = random.choice(countries)
            language = random.choice(languages)
            timezone = random.choice(timezones)
            location = {
                "country": country,
                "language": language,
                "timezone": timezone
            }
            with self.lock:
                self.location_pool.append(location)
            
    def get_ipv4(self):
        with self.lock:
            if self.ipv4_pool:
                return self.ipv4_pool[random.randint(0, len(self.ipv4_pool) - 1)]
        return f"{random.randint(1, 223)}.{random.randint(0, 255)}.{random.randint(0, 255)}.{random.randint(1, 254)}"
        
    def get_ipv6(self):
        with self.lock:
            if self.ipv6_pool:
                return self.ipv6_pool[random.randint(0, len(self.ipv6_pool) - 1)]
        segments = [f"{random.randint(0, 65535):04x}" for _ in range(8)]
        return ":".join(segments)
        
    def get_user_agent(self):
        with self.lock:
            if self.user_agent_pool:
                return self.user_agent_pool[random.randint(0, len(self.user_agent_pool) - 1)]
        return "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
        
    def get_origin(self):
        with self.lock:
            if self.origin_pool:
                return self.origin_pool[random.randint(0, len(self.origin_pool) - 1)]
        return "https://www.google.com"
        
    def get_referrer(self):
        with self.lock:
            if self.referrer_pool:
                return self.referrer_pool[random.randint(0, len(self.referrer_pool) - 1)]
        return "https://www.google.com/search?q=search"
        
    def get_location(self):
        with self.lock:
            if self.location_pool:
                return self.location_pool[random.randint(0, len(self.location_pool) - 1)]
        return {
            "country": "US",
            "language": "en-US",
            "timezone": "America/New_York"
        }
        
    def start_engine(self):
        threads = [
            threading.Thread(target=self.generate_ipv4, daemon=True),
            threading.Thread(target=self.generate_ipv6, daemon=True),
            threading.Thread(target=self.generate_user_agents, daemon=True),
            threading.Thread(target=self.generate_origins, daemon=True),
            threading.Thread(target=self.generate_referrers, daemon=True),
            threading.Thread(target=self.generate_locations, daemon=True)
        ]
        for thread in threads:
            thread.start()
            
    def stop(self):
        self.running = False

web_loader = WebLoader()
web_loader.start_engine()