scraper_engine.py

import asyncio

import aiohttp

import re

import urllib.parse

import random

from bs4 import BeautifulSoup

import pandas as pd

from datetime import datetime

class EmailValidator:

    """
    Handles validation of scraped emails to ensure high accuracy and zero false positives.
    Filters out common web assets, package version identifiers, placeholder addresses, etc.
    """

    def __init__(self, custom_blacklist_domains=None, custom_blacklist_emails=None):

        self.blacklist_domains = {

            'example.com', 'example.org', 'example.net', 'test.com',

            'yourdomain.com', 'domain.com', 'placeholder.com', 'tempmail.com',

            'sentry.io', 'fontawesome.com', 'bootstrap.com', 'jquery.com',

            'w3.org', 'github.com', 'google.com', 'microsoft.com',

            'contoh.com', 'contoh.id'

        }

        if custom_blacklist_domains:

            self.blacklist_domains.update(custom_blacklist_domains)

        self.blacklist_emails = {

            'email@example.com', 'user@example.com', 'your@email.com',

            'username@domain.com', 'test@test.com', 'info@example.com',

            'git@github.com', 'noreply@github.com', 'support@github.com',

            'webmaster@domain.com', 'admin@domain.com', 'postmaster@domain.com',

            'test@gmail.com', 'yourname@gmail.com', 'john.doe@gmail.com',

            'email@contoh.com', 'admin@contoh.com', 'contoh@contoh.com',

            'email@contoh.id', 'admin@contoh.id', 'contoh@contoh.id'

        }

        if custom_blacklist_emails:

            self.blacklist_emails.update(custom_blacklist_emails)

        self.email_regex = re.compile(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b')

        self.invalid_tld_extensions = {

            'png', 'jpg', 'jpeg', 'gif', 'svg', 'webp', 'ico', 'pdf', 'mp4', 'zip', 'gz',

            'tar', 'rar', 'js', 'css', 'scss', 'woff', 'woff2', 'ttf', 'eot', 'html',

            'htm', 'php', 'json', 'xml', 'map', 'md', 'txt', 'mp3', 'wav', 'avi', 'mov',

            'exe', 'dll', 'dmg', 'iso', 'bin', 'less', 'ts', 'jsx', 'tsx', 'py'

        }

    def validate(self, email: str) -> bool:

        """
        Runs comprehensive checks on the email address.
        Returns True if valid, False otherwise.
        """

        email = email.strip().lower()

        if not email:

            return False

        if not self.email_regex.match(email):

            return False

        try:

            local_part, domain_part = email.split('@', 1)

        except ValueError:

            return False

        if email in self.blacklist_emails or domain_part in self.blacklist_domains:

            return False

        if '..' in domain_part or '..' in local_part:

            return False

        if domain_part.startswith('.') or domain_part.endswith('.'):

            return False

        if local_part.startswith('.') or local_part.endswith('.'):

            return False

        domain_segments = domain_part.split('.')

        if len(domain_segments) < 2:

            return False

        tld = domain_segments[-1]

        if tld in self.invalid_tld_extensions:

            return False

        if not tld.isalpha() or len(tld) < 2:

            return False

        numeric_segments = sum(1 for s in domain_segments[:-1] if s.isdigit())

        if numeric_segments >= 2:

            return False

        if len(domain_segments) >= 3:

            if domain_segments[0].isdigit() and domain_segments[1].isdigit():

                return False

        invalid_local_parts = {'npm', 'yarn', 'bower', 'v', 'version', 'dist', 'node_modules', 'jquery', 'bootstrap'}

        if local_part in invalid_local_parts:

            return False

        return True

class HttpRequestManager:

    """
    Manages custom HTTP headers and rotates user agents to avoid bot detection and IP bans.
    """

    def __init__(self):

        self.user_agents = [

            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",

            "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",

            "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0",

            "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.2 Safari/605.1.15",

            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36 Edg/119.0.0.0",

            "Mozilla/5.0 (Linux; Android 10; K) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Mobile Safari/537.36"

        ]

    def get_headers(self) -> dict:

        return {

            "User-Agent": random.choice(self.user_agents),

            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8",

            "Accept-Language": "en-US,en;q=0.9,id;q=0.8",

            "Referer": "https://www.google.com/",

            "Connection": "keep-alive"

        }

class SearchEngineScraper:

    """
    Scrapes search engines (Bing, DuckDuckGo, Yahoo) to discover target URLs for keyword-based email scraping.
    Supports paginated results to return a massive number of target URLs.
    """

    def __init__(self):

        self.http_manager = HttpRequestManager()

    async def search(self, query: str, num_results: int = 200) -> list:

        pages_to_request = max(1, (num_results // 10) + 1)

        tasks = []

        for p in range(min(5, pages_to_request)):

            first_idx = (p * 10) + 1

            tasks.append(self._search_bing(query, first_idx))

        for p in range(min(5, pages_to_request)):

            first_idx = (p * 10) + 1

            tasks.append(self._search_yahoo(query, first_idx))

        tasks.append(self._search_duckduckgo(query))

        results = await asyncio.gather(*tasks, return_exceptions=True)

        urls = set()

        for res in results:

            if isinstance(res, list):

                for url in res:

                    try:

                        parsed = urllib.parse.urlparse(url)

                        clean_url = f"{parsed.scheme}://{parsed.netloc}{parsed.path}"

                        if parsed.netloc:

                            urls.add(clean_url.rstrip("/"))

                    except Exception:

                        urls.add(url)

        filtered_urls = []

        blacklisted_search_domains = {

            'bing.com', 'google.com', 'yahoo.com', 'duckduckgo.com', 'yandex.com',

            'wikipedia.org', 'w3.org', 'microsoft.com', 'youtube.com', 'facebook.com',

            'twitter.com', 'instagram.com', 'linkedin.com', 'pinterest.com', 'github.com',

            'npm.js', 'cloudflare.com', 'sentry.io', 'wordpress.org', 'medium.com'

        }

        for url in urls:

            try:

                domain = urllib.parse.urlparse(url).netloc.lower()

                if not any(b in domain for b in blacklisted_search_domains):

                    filtered_urls.append(url)

            except Exception:

                pass

        return list(set(filtered_urls))[:num_results]

    async def _search_bing(self, query: str, first: int) -> list:

        found_urls = []

        url = f"https://www.bing.com/search?q={urllib.parse.quote_plus(query)}&first={first}"

        try:

            headers = self.http_manager.get_headers()

            try:

                resolver = aiohttp.AsyncResolver(nameservers=["8.8.8.8", "8.8.4.4", "1.1.1.1"])

            except Exception:

                resolver = aiohttp.ThreadedResolver()

            connector = aiohttp.TCPConnector(ssl=False, resolver=resolver)

            async with aiohttp.ClientSession(headers=headers, connector=connector) as session:

                async with session.get(url, timeout=10) as response:

                    if response.status == 200:

                        html = await response.text()

                        soup = BeautifulSoup(html, 'html.parser')

                        for a in soup.find_all('a', href=True):

                            href = a['href']

                            if href.startswith(('http://', 'https://')):

                                domain = urllib.parse.urlparse(href).netloc.lower()

                                if not any(k in domain for k in ['bing.com', 'microsoft', 'live.com', 'msn.com', 'microsofttranslator.com', 'w3.org']):

                                    found_urls.append(href)

        except Exception:

            pass

        return found_urls

    async def _search_yahoo(self, query: str, first: int) -> list:

        found_urls = []

        url = f"https://search.yahoo.com/search?q={urllib.parse.quote_plus(query)}&b={first}"

        try:

            headers = self.http_manager.get_headers()

            try:

                resolver = aiohttp.AsyncResolver(nameservers=["8.8.8.8", "8.8.4.4", "1.1.1.1"])

            except Exception:

                resolver = aiohttp.ThreadedResolver()

            connector = aiohttp.TCPConnector(ssl=False, resolver=resolver)

            async with aiohttp.ClientSession(headers=headers, connector=connector) as session:

                async with session.get(url, timeout=10) as response:

                    if response.status == 200:

                        html = await response.text()

                        soup = BeautifulSoup(html, 'html.parser')

                        for a in soup.find_all('a', href=True):

                            href = a['href']

                            if href.startswith(('http://', 'https://')):

                                domain = urllib.parse.urlparse(href).netloc.lower()

                                if not any(k in domain for k in ['yahoo.com', 'yimg.com', 'flickr.com', 'w3.org']):

                                    found_urls.append(href)

        except Exception:

            pass

        return found_urls

    async def _search_duckduckgo(self, query: str) -> list:

        found_urls = []

        url = f"https://html.duckduckgo.com/html/?q={urllib.parse.quote_plus(query)}"

        try:

            headers = self.http_manager.get_headers()

            try:

                resolver = aiohttp.AsyncResolver(nameservers=["8.8.8.8", "8.8.4.4", "1.1.1.1"])

            except Exception:

                resolver = aiohttp.ThreadedResolver()

            connector = aiohttp.TCPConnector(ssl=False, resolver=resolver)

            async with aiohttp.ClientSession(headers=headers, connector=connector) as session:

                async with session.get(url, timeout=10) as response:

                    if response.status == 200:

                        html = await response.text()

                        soup = BeautifulSoup(html, 'html.parser')

                        for a in soup.find_all('a', href=True):

                            href = a['href']

                            if '/l/?' in href or 'uddg=' in href:

                                parsed = urllib.parse.urlparse(href)

                                qs = urllib.parse.parse_qs(parsed.query)

                                if 'uddg' in qs:

                                    ext_url = qs['uddg'][0]

                                    found_urls.append(ext_url)

                            elif href.startswith(('http://', 'https://')):

                                domain = urllib.parse.urlparse(href).netloc.lower()

                                if 'duckduckgo.com' not in domain and 'w3.org' not in domain:

                                    found_urls.append(href)

        except Exception:

            pass

        return found_urls

class AsyncEmailScraper:

    """
    Main asynchronous crawler that parses pages, collects emails, validates them,
    and handles links recursively up to a certain depth.
    """

    def __init__(self, start_urls, max_depth=2, max_pages=100, concurrent_connections=15,

                 internal_only=True, deobfuscate=True, validator=None, update_callback=None):

        self.start_urls = [self._normalize_url(url) for url in start_urls if url.strip()]

        self.max_depth = max_depth

        self.max_pages = max_pages

        self.concurrent_connections = concurrent_connections

        self.internal_only = internal_only

        self.deobfuscate = deobfuscate

        self.validator = validator or EmailValidator()

        self.update_callback = update_callback

        self.scraped_emails = {}

        self.visited_urls = set()

        self.crawled_count = 0

        self.error_count = 0

        self.success_count = 0

        self.status_log = []

        self.is_running = False

        self.allowed_domains = {self._get_domain(url) for url in self.start_urls if url}

        self.http_manager = HttpRequestManager()

    def _normalize_url(self, url: str) -> str:

        url = url.strip()

        if not url.startswith(('http://', 'https://')):

            url = 'https://' + url

        return url

    def _get_domain(self, url: str) -> str:

        try:

            parsed = urllib.parse.urlparse(url)

            return parsed.netloc.lower()

        except Exception:

            return ""

    def log(self, message: str):

        timestamp = datetime.now().strftime("%H:%M:%S")

        formatted = f"[{timestamp}] {message}"

        self.status_log.append(formatted)

        if self.update_callback:

            self.update_callback(formatted, self)

    async def start(self):

        self.is_running = True

        self.log(f"[INFO] Scraper initialized with {len(self.start_urls)} target URLs.")

        self.log(f"[INFO] Settings: Max Depth={self.max_depth}, Max Pages={self.max_pages}, Max Concurrency={self.concurrent_connections}")

        timeout = aiohttp.ClientTimeout(total=15, connect=5)

        try:

            resolver = aiohttp.AsyncResolver(nameservers=["8.8.8.8", "8.8.4.4", "1.1.1.1"])

        except Exception:

            resolver = aiohttp.ThreadedResolver()

        connector = aiohttp.TCPConnector(

            limit=self.concurrent_connections,

            ssl=False,

            resolver=resolver,

            family=0

        )

        async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:

            self.session = session

            self.semaphore = asyncio.Semaphore(self.concurrent_connections)

            self.queue = asyncio.Queue()

            for url in self.start_urls:

                await self.queue.put((url, 0))

            workers = []

            for _ in range(self.concurrent_connections):

                task = asyncio.create_task(self._worker())

                workers.append(task)

            while self.is_running and self.crawled_count < self.max_pages:

                if self.queue.empty() and self.semaphore._value == self.concurrent_connections:

                    break

                await asyncio.sleep(0.2)

            self.is_running = False

            for w in workers:

                w.cancel()

            await asyncio.gather(*workers, return_exceptions=True)

            self.log(f"[INFO] Crawl completed. Visited {self.crawled_count} pages. Emails scraped: {len(self.scraped_emails)}")

    async def _worker(self):

        while self.is_running:

            try:

                if self.crawled_count >= self.max_pages:

                    break

                try:

                    url, depth = await asyncio.wait_for(self.queue.get(), timeout=0.5)

                except asyncio.TimeoutError:

                    continue

                if url in self.visited_urls:

                    self.queue.task_done()

                    continue

                self.visited_urls.add(url)

                async with self.semaphore:

                    await self._crawl_page(url, depth)

                self.queue.task_done()

            except asyncio.CancelledError:

                break

            except Exception as e:

                self.log(f"[ERROR] Worker error: {str(e)}")

    async def _crawl_page(self, url: str, depth: int):

        self.crawled_count += 1

        self.log(f"[CRAWL] [{self.crawled_count}/{self.max_pages}] Fetching: {url} (Depth: {depth})")

        headers = self.http_manager.get_headers()

        try:

            async with self.session.get(url, headers=headers, timeout=10, allow_redirects=True) as response:

                if response.status != 200:

                    self.error_count += 1

                    self.log(f"[WARN] Failed: {url} (Status: {response.status})")

                    return

                self.success_count += 1

                try:

                    html = await response.text(errors='ignore')

                except Exception:

                    html = str(await response.read())

                emails = self._extract_emails(html, url)

                if emails:

                    self.log(f"[SUCCESS] Extracted {len(emails)} emails from {url}")

                if depth < self.max_depth:

                    links = self._extract_links(html, url)

                    for link in links:

                        if link not in self.visited_urls:

                            link_domain = self._get_domain(link)

                            if not link_domain:

                                continue

                            if self.internal_only:

                                if any(link_domain.endswith(d) or d.endswith(link_domain) for d in self.allowed_domains):

                                    await self.queue.put((link, depth + 1))

                            else:

                                await self.queue.put((link, depth + 1))

        except asyncio.CancelledError:

            raise

        except Exception as e:

            self.error_count += 1

            self.log(f"[ERROR] Fetch failed: {url} | Reason: {str(e)}")

    def _extract_emails(self, html: str, source_url: str) -> list:

        emails = []

        try:

            soup = BeautifulSoup(html, 'html.parser')

            for a in soup.find_all('a', href=True):

                href = a['href'].strip()

                if href.lower().startswith('mailto:'):

                    email = href[7:].split('?')[0].strip()

                    if self.validator.validate(email):

                        emails.append(email)

        except Exception:

            pass

        raw_matches = self.validator.email_regex.findall(html)

        for email in raw_matches:

            if self.validator.validate(email):

                emails.append(email)

        if self.deobfuscate:

            obfuscated_patterns = [

                r'\b[A-Za-z0-9._%+-]+\s*(?:\[at\]|\(at\)|\{at\}|@|_at_)\s*[A-Za-z0-9.-]+\s*(?:\[dot\]|\(dot\)|\{dot\}|\.|_dot_)\s*[A-Za-z]{2,}\b'

            ]

            for pattern in obfuscated_patterns:

                obf_matches = re.findall(pattern, html, re.IGNORECASE)

                for obf in obf_matches:

                    cleaned = re.sub(r'\s*(?:\[at\]|\(at\)|\{at\}|_at_)\s*', '@', obf, flags=re.IGNORECASE)

                    cleaned = re.sub(r'\s*(?:\[dot\]|\(dot\)|\{dot\}|_dot_)\s*', '.', cleaned, flags=re.IGNORECASE)

                    cleaned = cleaned.replace(' ', '')

                    if self.validator.validate(cleaned):

                        emails.append(cleaned)

        unique_emails = list(set(emails))

        for email in unique_emails:

            if email not in self.scraped_emails:

                self.scraped_emails[email] = set()

            self.scraped_emails[email].add(source_url)

        return unique_emails

    def _extract_links(self, html: str, base_url: str) -> list:

        links = []

        try:

            soup = BeautifulSoup(html, 'html.parser')

            for a in soup.find_all('a', href=True):

                href = a['href'].strip()

                if not href or href.startswith(('#', 'javascript:', 'mailto:', 'tel:')):

                    continue

                full_url = urllib.parse.urljoin(base_url, href)

                links.append(full_url)

            url_regex = re.compile(r'https?://[A-Za-z0-9.-]+\.[A-Za-z]{2,}(?:/[^\s\'"<>\(\)]*)?')

            for script in soup.find_all('script'):

                if script.string:

                    raw_urls = url_regex.findall(script.string)

                    for raw_url in raw_urls:

                        links.append(raw_url.rstrip('.,;)"\''))

        except Exception:

            pass

        valid_links = []

        for link in set(links):

            try:

                parsed = urllib.parse.urlparse(link)

                if parsed.scheme in ('http', 'https'):

                    path = parsed.path.lower()

                    if not any(path.endswith(ext) for ext in [

                        '.png', '.jpg', '.jpeg', '.gif', '.svg', '.webp', '.pdf',

                        '.css', '.js', '.woff', '.woff2', '.ttf', '.mp4', '.zip'

                    ]):

                        valid_links.append(link)

            except Exception:

                pass

        return valid_links