Email Scraper Python

import scrapy
import re
import os
import datetime
import csv
from scrapy_playwright.page import PageMethod


class EmailSpider(scrapy.Spider):
    name = "email_spider"

    custom_settings = {
        "ROBOTSTXT_OBEY": False,
        "DOWNLOAD_DELAY": 1,
        "USER_AGENT": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/115.0.0.0 Safari/537.36",
        "DEFAULT_REQUEST_HEADERS": {
            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
            "Accept-Language": "en-US,en;q=0.9",
            "Accept-Encoding": "gzip, deflate, br",
        },
        "PLAYWRIGHT_BROWSER_TYPE": "chromium",
        "PLAYWRIGHT_LAUNCH_OPTIONS": {
            "headless": False,
            "args": [
                "--no-sandbox",
                "--disable-blink-features=AutomationControlled",
                "--disable-infobars",
                "--disable-extensions",
                "--disable-dev-shm-usage",
                "--disable-gpu",
                "--window-size=1920,1080",
            ],
        },
        "DOWNLOAD_HANDLERS": {
            "http": "scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler",
            "https": "scrapy_playwright.handler.ScrapyPlaywrightDownloadHandler",
        },
        "TWISTED_REACTOR": "twisted.internet.asyncioreactor.AsyncioSelectorReactor",
        "PLAYWRIGHT_DEFAULT_NAVIGATION_TIMEOUT": 60000,
        "DOWNLOAD_TIMEOUT": 90,
    }

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.found_emails = set()
        self.rows = []

    def start_requests(self):
        try:
            with open("urls.txt", "r") as f:
                urls = [line.strip() for line in f if line.strip()]
        except FileNotFoundError:
            raise ValueError("urls.txt nicht gefunden. Lege eine Datei mit URLs im Projektordner an.")

        for url in urls:
            self.logger.info(f"Starte URL: {url}")
            yield scrapy.Request(
                url=url,
                callback=self.parse,
                errback=self.handle_error,
                meta={
                    "playwright": True,
                    "playwright_include_page": True,
                    "playwright_page_methods": [
                        PageMethod("add_init_script", self._stealth_script()),
                        PageMethod("eval_on_selector_all", "[id*='cookie'], [id*='consent'], [id*='overlay']", "els => els.forEach(el => el.remove())"),
                        PageMethod("wait_for_selector", "body"),
                        PageMethod("evaluate", """() => {
                            const link = Array.from(document.querySelectorAll("a")).find(a =>
                                a.href.toLowerCase().includes("kontakt") ||
                                a.href.toLowerCase().includes("impressum")
                            );
                            if (link) link.click();
                        }"""),
                        PageMethod("wait_for_timeout", 2000),
                    ],
                }
            )

    def parse(self, response):
        self.logger.info(f"Analysiere Seite: {response.url}")
        text_content = " ".join(response.xpath('//body//text()').getall())

        raw_emails = re.findall(
            r"[a-zA-Z0-9._%+-]+\s?(?:@|[\[\(]{1}at[\]\)]{1}|\s+at\s+)\s?[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}",
            text_content
        )
        cleaned_emails = set(
            re.sub(r"\s?(?:\[\s*at\s*\]|\(at\)|\sat\s)", "@", e.replace(" ", ""))
            for e in raw_emails
        )

        new_emails = cleaned_emails - self.found_emails
        if new_emails:
            self.logger.info(f"✅ {len(new_emails)} neue E-Mails auf {response.url} gefunden")
        else:
            self.logger.info(f"❌ Keine E-Mails auf {response.url} gefunden")

        for email in new_emails:
            self.found_emails.add(email)
            self.rows.append({"source": response.url, "email": email})
            yield {"source": response.url, "email": email}

    def handle_error(self, failure):
        self.logger.warning(f"❌ Fehler bei {failure.request.url}: {failure.value}")

    def _stealth_script(self):
        return """
        () => {
            Object.defineProperty(navigator, 'webdriver', {get: () => false});
            Object.defineProperty(navigator, 'plugins', {get: () => [1, 2, 3, 4, 5]});
            Object.defineProperty(navigator, 'languages', {get: () => ['en-US', 'en']});
            Object.defineProperty(navigator, 'platform', {get: () => 'Win32'});
            Object.defineProperty(navigator, 'vendor', {get: () => 'Google Inc.'});
            const originalQuery = window.navigator.permissions.query;
            window.navigator.permissions.query = (parameters) => (
                parameters.name === 'notifications'
                    ? Promise.resolve({ state: Notification.permission })
                    : originalQuery(parameters)
            );
            return true;
        }
        """

    def closed(self, reason):
        self.logger.info("🚀 Scraper abgeschlossen, lade direkt als Google Sheet hoch ...")
        try:
            from drive_upload import upload_to_drive
            timestamp = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M")
            filename = f"emails_{timestamp}.csv"

            with open(filename, "w", newline="", encoding="utf-8") as f:
                writer = csv.DictWriter(f, fieldnames=["source", "email"])
                writer.writeheader()
                for row in self.rows:
                    writer.writerow(row)

            upload_to_drive(filename)
            os.remove(filename)

        except Exception as e:
            self.logger.error(f"❌ Fehler beim Google Drive Upload: {e}")