Initial commit

social_networks/base_automation.py

@@ -0,0 +1,487 @@
+"""
+Basis-Automatisierungsklasse für soziale Netzwerke
+"""
+
+import os
+import logging
+import time
+import random
+from typing import Dict, List, Optional, Any, Tuple
+from abc import ABC, abstractmethod
+
+from browser.playwright_manager import PlaywrightManager
+from utils.proxy_rotator import ProxyRotator
+from utils.email_handler import EmailHandler
+from utils.text_similarity import TextSimilarity, fuzzy_find_element, click_fuzzy_button
+
+# Konfiguriere Logger
+logger = logging.getLogger("base_automation")
+
+class BaseAutomation(ABC):
+    """
+    Abstrakte Basisklasse für die Automatisierung von sozialen Netzwerken.
+    Definiert die gemeinsame Schnittstelle für alle Implementierungen.
+    """
+    
+    def __init__(self, 
+                 headless: bool = False, 
+                 use_proxy: bool = False,
+                 proxy_type: str = None,
+                 save_screenshots: bool = True,
+                 screenshots_dir: str = None,
+                 slowmo: int = 0,
+                 debug: bool = False,
+                 email_domain: str = "z5m7q9dk3ah2v1plx6ju.com"):
+        """
+        Initialisiert die Basis-Automatisierung.
+        
+        Args:
+            headless: Ob der Browser im Headless-Modus ausgeführt werden soll
+            use_proxy: Ob ein Proxy verwendet werden soll
+            proxy_type: Proxy-Typ ("ipv4", "ipv6", "mobile") oder None für zufälligen Typ
+            save_screenshots: Ob Screenshots gespeichert werden sollen
+            screenshots_dir: Verzeichnis für Screenshots
+            slowmo: Verzögerung zwischen Aktionen in Millisekunden (nützlich für Debugging)
+            debug: Ob Debug-Informationen angezeigt werden sollen
+            email_domain: Domain für generierte E-Mail-Adressen
+        """
+        self.headless = headless
+        self.use_proxy = use_proxy
+        self.proxy_type = proxy_type
+        self.save_screenshots = save_screenshots
+        self.slowmo = slowmo
+        self.debug = debug
+        self.email_domain = email_domain
+        
+        # Verzeichnis für Screenshots
+        base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+        self.screenshots_dir = screenshots_dir or os.path.join(base_dir, "logs", "screenshots")
+        os.makedirs(self.screenshots_dir, exist_ok=True)
+        
+        # Initialisiere Hilfsklassen
+        self.proxy_rotator = ProxyRotator()
+        self.email_handler = EmailHandler()
+        
+        # Initialisiere TextSimilarity für robustes UI-Element-Matching
+        self.text_similarity = TextSimilarity(default_threshold=0.75)
+        
+        # Playwright-Manager wird bei Bedarf initialisiert
+        self.browser = None
+        
+        # Status und Ergebnis der Automatisierung
+        self.status = {
+            "success": False,
+            "stage": "initialized",
+            "error": None,
+            "account_data": {}
+        }
+        
+        # Debug-Logging
+        if self.debug:
+            logging.getLogger().setLevel(logging.DEBUG)
+            
+        logger.info(f"Basis-Automatisierung initialisiert (Proxy: {use_proxy}, Typ: {proxy_type})")
+    
+    def _initialize_browser(self) -> bool:
+        """
+        Initialisiert den Browser mit den entsprechenden Einstellungen.
+        
+        Returns:
+            bool: True bei Erfolg, False bei Fehler
+        """
+        try:
+            # Proxy-Konfiguration, falls aktiviert
+            proxy_config = None
+            if self.use_proxy:
+                proxy_config = self.proxy_rotator.get_proxy(self.proxy_type)
+                if not proxy_config:
+                    logger.warning(f"Kein Proxy vom Typ '{self.proxy_type}' verfügbar, verwende direkten Zugriff")
+            
+            # Browser initialisieren
+            self.browser = PlaywrightManager(
+                headless=self.headless,
+                proxy=proxy_config,
+                browser_type="chromium",
+                screenshots_dir=self.screenshots_dir,
+                slowmo=self.slowmo
+            )
+            
+            # Browser starten
+            self.browser.start()
+            
+            logger.info("Browser erfolgreich initialisiert")
+            return True
+            
+        except Exception as e:
+            logger.error(f"Fehler bei der Browser-Initialisierung: {e}")
+            self.status["error"] = f"Browser-Initialisierungsfehler: {str(e)}"
+            return False
+    
+    def _close_browser(self) -> None:
+        """Schließt den Browser und gibt Ressourcen frei."""
+        if self.browser:
+            self.browser.close()
+            self.browser = None
+            logger.info("Browser geschlossen")
+    
+    def _take_screenshot(self, name: str) -> Optional[str]:
+        """
+        Erstellt einen Screenshot der aktuellen Seite.
+        
+        Args:
+            name: Name für den Screenshot (ohne Dateierweiterung)
+            
+        Returns:
+            Optional[str]: Pfad zum erstellten Screenshot oder None bei Fehler
+        """
+        if not self.save_screenshots:
+            return None
+        
+        try:
+            if self.browser and hasattr(self.browser, 'take_screenshot'):
+                return self.browser.take_screenshot(name)
+            
+        except Exception as e:
+            logger.warning(f"Fehler beim Erstellen eines Screenshots: {e}")
+        
+        return None
+    
+    def _random_delay(self, min_seconds: float = 1.0, max_seconds: float = 3.0) -> None:
+        """
+        Führt eine zufällige Wartezeit aus, um menschliches Verhalten zu simulieren.
+        
+        Args:
+            min_seconds: Minimale Wartezeit in Sekunden
+            max_seconds: Maximale Wartezeit in Sekunden
+        """
+        delay = random.uniform(min_seconds, max_seconds)
+        logger.debug(f"Zufällige Wartezeit: {delay:.2f} Sekunden")
+        time.sleep(delay)
+    
+    def _fill_field_fuzzy(self, field_labels: List[str], value: str, fallback_selector: str = None) -> bool:
+        """
+        Füllt ein Formularfeld mit Fuzzy-Text-Matching aus.
+        
+        Args:
+            field_labels: Liste mit möglichen Bezeichnungen des Feldes
+            value: Einzugebender Wert
+            fallback_selector: CSS-Selektor für Fallback
+            
+        Returns:
+            bool: True bei Erfolg, False bei Fehler
+        """
+        # Versuche, das Feld mit Fuzzy-Matching zu finden
+        field = fuzzy_find_element(self.browser.page, field_labels, selector_type="input", threshold=0.6, wait_time=3000)
+        
+        if field:
+            try:
+                field.fill(value)
+                return True
+            except Exception as e:
+                logger.warning(f"Fehler beim Ausfüllen des Feldes mit Fuzzy-Match: {e}")
+                # Fallback auf normales Ausfüllen
+        
+        # Fallback: Versuche mit dem angegebenen Selektor
+        if fallback_selector:
+            field_success = self.browser.fill_form_field(fallback_selector, value)
+            if field_success:
+                return True
+        
+        return False
+    
+    def _click_button_fuzzy(self, button_texts: List[str], fallback_selector: str = None) -> bool:
+        """
+        Klickt einen Button mit Fuzzy-Text-Matching.
+        
+        Args:
+            button_texts: Liste mit möglichen Button-Texten
+            fallback_selector: CSS-Selektor für Fallback
+            
+        Returns:
+            bool: True bei Erfolg, False bei Fehler
+        """
+        # Versuche, den Button mit Fuzzy-Matching zu finden
+        success = click_fuzzy_button(self.browser.page, button_texts, threshold=0.6, timeout=3000)
+        
+        if success:
+            return True
+        
+        # Fallback: Versuche mit dem angegebenen Selektor
+        if fallback_selector:
+            return self.browser.click_element(fallback_selector)
+        
+        return False
+    
+    def _find_element_by_text(self, texts: List[str], selector_type: str = "any", threshold: float = 0.7) -> Optional[Any]:
+        """
+        Findet ein Element basierend auf Textähnlichkeit.
+        
+        Args:
+            texts: Liste mit möglichen Texten
+            selector_type: Art des Elements ("button", "link", "input", "any")
+            threshold: Ähnlichkeitsschwellenwert
+            
+        Returns:
+            Das gefundene Element oder None
+        """
+        return fuzzy_find_element(self.browser.page, texts, selector_type, threshold, wait_time=3000)
+    
+    def _check_for_text_on_page(self, texts: List[str], threshold: float = 0.7) -> bool:
+        """
+        Prüft, ob ein Text auf der Seite vorhanden ist.
+        
+        Args:
+            texts: Liste mit zu suchenden Texten
+            threshold: Ähnlichkeitsschwellenwert
+            
+        Returns:
+            True wenn einer der Texte gefunden wurde, False sonst
+        """
+        # Hole den gesamten Seiteninhalt
+        try:
+            page_content = self.browser.page.content()
+            if not page_content:
+                return False
+            
+            # Versuche, Text im HTML zu finden (einfache Suche)
+            for text in texts:
+                if text.lower() in page_content.lower():
+                    return True
+            
+            # Wenn nicht gefunden, versuche über alle sichtbaren Textelemente
+            elements = self.browser.page.query_selector_all("p, h1, h2, h3, h4, h5, h6, span, div, button, a, label")
+            
+            for element in elements:
+                element_text = element.inner_text()
+                if not element_text:
+                    continue
+                
+                element_text = element_text.strip()
+                
+                # Prüfe die Textähnlichkeit mit jedem der gesuchten Texte
+                for text in texts:
+                    if self.text_similarity.is_similar(text, element_text, threshold=threshold):
+                        return True
+            
+            return False
+            
+        except Exception as e:
+            logger.error(f"Fehler beim Prüfen auf Text auf der Seite: {e}")
+            return False
+    
+    def _check_for_error(self, error_selectors: List[str], error_texts: List[str]) -> Optional[str]:
+        """
+        Prüft, ob Fehlermeldungen angezeigt werden.
+        
+        Args:
+            error_selectors: Liste mit CSS-Selektoren für Fehlermeldungen
+            error_texts: Liste mit typischen Fehlertexten
+            
+        Returns:
+            Die Fehlermeldung oder None, wenn keine Fehler gefunden wurden
+        """
+        try:
+            # Prüfe selektoren
+            for selector in error_selectors:
+                element = self.browser.wait_for_selector(selector, timeout=2000)
+                if element:
+                    error_text = element.text_content()
+                    if error_text:
+                        return error_text.strip()
+            
+            # Fuzzy-Suche nach Fehlermeldungen
+            elements = self.browser.page.query_selector_all("p, div[role='alert'], span.error, .error-message")
+            
+            for element in elements:
+                element_text = element.inner_text()
+                if not element_text:
+                    continue
+                
+                element_text = element_text.strip()
+                
+                # Prüfe, ob der Text einem Fehlermuster ähnelt
+                for error_text in error_texts:
+                    if self.text_similarity.is_similar(error_text, element_text, threshold=0.6):
+                        return element_text
+            
+            return None
+            
+        except Exception as e:
+            logger.error(f"Fehler beim Prüfen auf Fehlermeldungen: {e}")
+            return None
+    
+    def _attempt_ocr_fallback(self, action_name: str, target_text: str = None, value: str = None) -> bool:
+        """
+        Versucht, eine Aktion mit OCR-Fallback durchzuführen, wenn Playwright fehlschlägt.
+        
+        Args:
+            action_name: Name der Aktion ("click", "type", "select")
+            target_text: Text, nach dem gesucht werden soll
+            value: Wert, der eingegeben werden soll (bei "type" oder "select")
+            
+        Returns:
+            bool: True bei Erfolg, False bei Fehler
+        """
+        # Diese Methode wird in abgeleiteten Klassen implementiert
+        logger.warning(f"OCR-Fallback für '{action_name}' wurde aufgerufen, aber nicht implementiert")
+        return False
+    
+    def _rotate_proxy(self) -> bool:
+        """
+        Rotiert den Proxy und aktualisiert die Browser-Sitzung.
+        
+        Returns:
+            bool: True bei Erfolg, False bei Fehler
+        """
+        if not self.use_proxy:
+            return False
+            
+        try:
+            # Browser schließen
+            self._close_browser()
+            
+            # Proxy rotieren
+            new_proxy = self.proxy_rotator.rotate_proxy(self.proxy_type)
+            if not new_proxy:
+                logger.warning("Konnte Proxy nicht rotieren")
+                return False
+            
+            # Browser neu initialisieren
+            success = self._initialize_browser()
+            
+            logger.info(f"Proxy rotiert zu: {new_proxy['server']}")
+            return success
+            
+        except Exception as e:
+            logger.error(f"Fehler bei der Proxy-Rotation: {e}")
+            return False
+    
+    def _generate_random_email(self, length: int = 10) -> str:
+        """
+        Generiert eine zufällige E-Mail-Adresse.
+        
+        Args:
+            length: Länge des lokalen Teils der E-Mail
+            
+        Returns:
+            str: Die generierte E-Mail-Adresse
+        """
+        import string
+        local_chars = string.ascii_lowercase + string.digits
+        local_part = ''.join(random.choice(local_chars) for _ in range(length))
+        return f"{local_part}@{self.email_domain}"
+    
+    def _get_confirmation_code(self, email_address: str, search_criteria: str, 
+                              max_attempts: int = 30, delay_seconds: int = 2) -> Optional[str]:
+        """
+        Ruft einen Bestätigungscode aus einer E-Mail ab.
+        
+        Args:
+            email_address: E-Mail-Adresse, an die der Code gesendet wurde
+            search_criteria: Suchkriterium für die E-Mail
+            max_attempts: Maximale Anzahl an Versuchen
+            delay_seconds: Verzögerung zwischen Versuchen in Sekunden
+            
+        Returns:
+            Optional[str]: Der Bestätigungscode oder None, wenn nicht gefunden
+        """
+        logger.info(f"Suche nach Bestätigungscode für {email_address}")
+        
+        code = self.email_handler.get_confirmation_code(
+            expected_email=email_address,
+            search_criteria=search_criteria,
+            max_attempts=max_attempts,
+            delay_seconds=delay_seconds
+        )
+        
+        if code:
+            logger.info(f"Bestätigungscode gefunden: {code}")
+        else:
+            logger.warning(f"Kein Bestätigungscode für {email_address} gefunden")
+            
+        return code
+    
+    def _is_text_similar(self, text1: str, text2: str, threshold: float = None) -> bool:
+        """
+        Prüft, ob zwei Texte ähnlich sind.
+        
+        Args:
+            text1: Erster Text
+            text2: Zweiter Text
+            threshold: Ähnlichkeitsschwellenwert (None für Standardwert)
+            
+        Returns:
+            True wenn die Texte ähnlich sind, False sonst
+        """
+        return self.text_similarity.is_similar(text1, text2, threshold)
+    
+    @abstractmethod
+    def register_account(self, full_name: str, age: int, registration_method: str = "email", 
+                        phone_number: str = None, **kwargs) -> Dict[str, Any]:
+        """
+        Registriert einen neuen Account im sozialen Netzwerk.
+        
+        Args:
+            full_name: Vollständiger Name für den Account
+            age: Alter des Benutzers
+            registration_method: "email" oder "phone"
+            phone_number: Telefonnummer (nur bei registration_method="phone")
+            **kwargs: Weitere optionale Parameter
+            
+        Returns:
+            Dict[str, Any]: Ergebnis der Registrierung mit Status und Account-Daten
+        """
+        pass
+    
+    @abstractmethod
+    def login_account(self, username_or_email: str, password: str, **kwargs) -> Dict[str, Any]:
+        """
+        Meldet sich bei einem bestehenden Account an.
+        
+        Args:
+            username_or_email: Benutzername oder E-Mail-Adresse
+            password: Passwort
+            **kwargs: Weitere optionale Parameter
+            
+        Returns:
+            Dict[str, Any]: Ergebnis der Anmeldung mit Status
+        """
+        pass
+    
+    @abstractmethod
+    def verify_account(self, verification_code: str, **kwargs) -> Dict[str, Any]:
+        """
+        Verifiziert einen Account mit einem Bestätigungscode.
+        
+        Args:
+            verification_code: Der Bestätigungscode
+            **kwargs: Weitere optionale Parameter
+            
+        Returns:
+            Dict[str, Any]: Ergebnis der Verifizierung mit Status
+        """
+        pass
+    
+    def get_status(self) -> Dict[str, Any]:
+        """
+        Gibt den aktuellen Status der Automatisierung zurück.
+        
+        Returns:
+            Dict[str, Any]: Aktueller Status
+        """
+        return self.status
+    
+    def __enter__(self):
+        """Kontext-Manager-Eintritt."""
+        self._initialize_browser()
+        return self
+    
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        """Kontext-Manager-Austritt."""
+        self._close_browser()
+
+
+# Wenn direkt ausgeführt, zeige Informationen zur Klasse
+if __name__ == "__main__":
+    print("Dies ist eine abstrakte Basisklasse und kann nicht direkt instanziiert werden.")
+    print("Bitte verwende eine konkrete Implementierung wie InstagramAutomation.")