""" 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 from domain.value_objects.browser_protection_style import BrowserProtectionStyle, ProtectionLevel # 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", session_manager=None, window_position: Optional[Tuple[int, int]] = None, auto_close_browser: bool = False): """ 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 session_manager: Optional - Session Manager für Ein-Klick-Login window_position: Optional - Fensterposition als Tuple (x, y) auto_close_browser: Ob Browser automatisch geschlossen werden soll (Standard: False) """ self.headless = headless self.use_proxy = use_proxy self.proxy_type = proxy_type self.session_manager = session_manager self.save_screenshots = save_screenshots self.slowmo = slowmo self.debug = debug self.email_domain = email_domain self.auto_close_browser = auto_close_browser # 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 # Session-bezogene Attribute self.current_session = None self.current_fingerprint = None self.session_restored = False # Fensterposition self.window_position = window_position # Status und Ergebnis der Automatisierung self.status = { "success": False, "stage": "initialized", "error": None, "account_data": {} } # Customer log callback (wird vom Worker Thread gesetzt) self.customer_log_callback = None # Status update callback für Login-Progress self.status_update_callback = None self.log_update_callback = None # Debug-Logging if self.debug: logging.getLogger().setLevel(logging.DEBUG) logger.info(f"Basis-Automatisierung initialisiert (Proxy: {use_proxy}, Typ: {proxy_type})") def set_customer_log_callback(self, callback): """Setzt den Callback für kundenfreundliche Log-Nachrichten.""" self.customer_log_callback = callback def _emit_customer_log(self, message: str): """Sendet eine kundenfreundliche Log-Nachricht.""" if self.customer_log_callback: self.customer_log_callback(message) def _initialize_browser(self) -> bool: """ Initialisiert den Browser mit den entsprechenden Einstellungen. Returns: bool: True bei Erfolg, False bei Fehler """ try: self._emit_customer_log("🔄 Sichere Verbindung wird aufgebaut...") # Proxy-Konfiguration, falls aktiviert proxy_config = None if self.use_proxy: self._emit_customer_log("🌐 Optimale Verbindung wird ausgewählt...") 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") # Prüfe ob Session wiederhergestellt werden soll if self.current_session and self.current_fingerprint: # Verwende Session-Aware Playwright Manager from browser.session_aware_playwright_manager import SessionAwarePlaywrightManager self.browser = SessionAwarePlaywrightManager( session=self.current_session, fingerprint=self.current_fingerprint, headless=self.headless, proxy=proxy_config, browser_type="chromium", screenshots_dir=self.screenshots_dir, slowmo=self.slowmo, window_position=self.window_position ) # Browser mit Session starten self.browser.start_with_session() self.session_restored = True logger.info("Browser mit wiederhergestellter Session initialisiert") else: # Normaler Browser ohne Session self.browser = PlaywrightManager( headless=self.headless, proxy=proxy_config, browser_type="chromium", screenshots_dir=self.screenshots_dir, slowmo=self.slowmo, window_position=self.window_position ) # Browser starten self.browser.start() logger.info("Browser erfolgreich initialisiert") # Browser-Schutz anwenden wenn nicht headless if not self.headless: self._emit_customer_log("🛡️ Sicherheitseinstellungen werden konfiguriert...") # TEMPORÄR DEAKTIVIERT zum Testen # self._apply_browser_protection() logger.info("Browser-Schutz wurde temporär deaktiviert") self._emit_customer_log("✅ Verbindung erfolgreich hergestellt") 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. Berücksichtigt die auto_close_browser Einstellung. """ if self.auto_close_browser and self.browser: self.browser.close() self.browser = None logger.info("Browser automatisch geschlossen") elif self.browser and not self.auto_close_browser: logger.info("Browser bleibt geöffnet (auto_close_browser=False)") def close_browser(self) -> None: """ Explizite Methode zum manuellen Schließen des Browsers. Ignoriert die auto_close_browser Einstellung. """ if self.browser: self.browser.close() self.browser = None logger.info("Browser manuell geschlossen") def is_browser_open(self) -> bool: """ Prüft, ob der Browser noch geöffnet ist. Returns: bool: True wenn Browser geöffnet ist, False sonst """ return self.browser is not None and hasattr(self.browser, 'page') def get_browser(self) -> Optional[PlaywrightManager]: """ Gibt die Browser-Instanz zurück für weitere Operationen. Returns: Optional[PlaywrightManager]: Browser-Instanz oder None """ return self.browser 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 _send_status_update(self, status: str) -> None: """ Sendet ein Status-Update über den Callback. Args: status: Status-Nachricht """ if self.status_update_callback: try: self.status_update_callback(status) except Exception as e: logger.error(f"Fehler beim Senden des Status-Updates: {e}") def _send_log_update(self, message: str) -> None: """ Sendet ein Log-Update über den Callback. Args: message: Log-Nachricht """ if self.log_update_callback: try: self.log_update_callback(message) except Exception as e: logger.error(f"Fehler beim Senden des Log-Updates: {e}") # Auch über customer_log_callback senden für Kompatibilität if self.customer_log_callback: try: self.customer_log_callback(message) except Exception as e: logger.error(f"Fehler beim Senden des Customer-Log-Updates: {e}") 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 _apply_browser_protection(self): """Wendet Browser-Schutz an, um versehentliche Interaktionen zu verhindern.""" try: # Lade Schutz-Einstellungen aus stealth_config.json import json from pathlib import Path protection_config = None try: config_file = Path(__file__).parent.parent / "config" / "stealth_config.json" if config_file.exists(): with open(config_file, 'r', encoding='utf-8') as f: stealth_config = json.load(f) protection_config = stealth_config.get("browser_protection", {}) except Exception as e: logger.warning(f"Konnte Browser-Schutz-Konfiguration nicht laden: {e}") # Nutze Konfiguration oder Standardwerte if protection_config and protection_config.get("enabled", True): level_mapping = { "none": ProtectionLevel.NONE, "light": ProtectionLevel.LIGHT, "medium": ProtectionLevel.MEDIUM, "strong": ProtectionLevel.STRONG } protection_style = BrowserProtectionStyle( level=level_mapping.get(protection_config.get("level", "medium"), ProtectionLevel.MEDIUM), show_border=protection_config.get("show_border", True), show_badge=protection_config.get("show_badge", True), blur_effect=protection_config.get("blur_effect", False), opacity=protection_config.get("opacity", 0.1), badge_text=protection_config.get("badge_text", "🔒 Account wird erstellt - Bitte nicht eingreifen"), badge_position=protection_config.get("badge_position", "top-right"), border_color=protection_config.get("border_color", "rgba(255, 0, 0, 0.5)") ) # Wende Schutz an if hasattr(self.browser, 'apply_protection'): self.browser.apply_protection(protection_style) logger.info("Browser-Schutz aktiviert") except Exception as e: # Browser-Schutz ist optional, Fehler nicht kritisch logger.warning(f"Browser-Schutz konnte nicht aktiviert werden: {str(e)}") 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 login_with_session(self, session_data: Dict[str, Any]) -> Dict[str, Any]: """ Template Method für Ein-Klick-Login mit gespeicherter Session. Args: session_data: Session-Daten vom OneClickLoginUseCase Returns: Dict mit Login-Ergebnis """ try: # Session und Fingerprint setzen self.current_session = session_data.get('session') self.current_fingerprint = session_data.get('fingerprint') # Browser mit Session initialisieren if not self._initialize_browser(): return { 'success': False, 'error': 'Browser-Initialisierung fehlgeschlagen' } # Plattform-spezifische Login-Seite aufrufen login_url = self._get_login_url() self.browser.page.goto(login_url) # Warte kurz für Session-Wiederherstellung time.sleep(2) # Prüfe ob Login erfolgreich if self._check_logged_in_state(): logger.info("Ein-Klick-Login erfolgreich") return { 'success': True, 'message': 'Erfolgreich mit Session eingeloggt', 'session_restored': True } else: logger.warning("Session abgelaufen oder ungültig") return { 'success': False, 'error': 'Session abgelaufen', 'requires_manual_login': True } except Exception as e: logger.error(f"Fehler beim Session-Login: {e}") return { 'success': False, 'error': str(e) } finally: self._close_browser() def create_with_session_persistence(self, **kwargs) -> Dict[str, Any]: """ Template Method für Account-Erstellung mit Session-Speicherung. Args: **kwargs: Plattformspezifische Parameter Returns: Dict mit Ergebnis und Session-Daten """ # Normale Account-Erstellung result = self.register_account(**kwargs) if result.get('success') and self.browser: try: # Session-Daten extrahieren from browser.session_aware_playwright_manager import SessionAwarePlaywrightManager if isinstance(self.browser, SessionAwarePlaywrightManager): session = self.browser.save_current_session() platform_data = self.browser.extract_platform_session_data( self._get_platform_name() ) result['session_data'] = { 'session': session, 'platform_data': platform_data, 'fingerprint': self.current_fingerprint } logger.info("Session-Daten für späteren Login gespeichert") except Exception as e: logger.error(f"Fehler beim Speichern der Session: {e}") return result def _get_login_url(self) -> str: """ Gibt die Login-URL der Plattform zurück. Kann von Unterklassen überschrieben werden. """ # Standard-URLs für bekannte Plattformen urls = { 'instagram': 'https://www.instagram.com/accounts/login/', 'facebook': 'https://www.facebook.com/', 'twitter': 'https://twitter.com/login', 'tiktok': 'https://www.tiktok.com/login' } platform = self._get_platform_name().lower() return urls.get(platform, '') def _check_logged_in_state(self) -> bool: """ Prüft ob der Benutzer eingeloggt ist. Kann von Unterklassen überschrieben werden. """ # Basis-Implementation: Prüfe auf typische Login-Indikatoren logged_in_indicators = [ 'logout', 'log out', 'sign out', 'abmelden', 'profile', 'profil', 'dashboard', 'home' ] # Prüfe URL current_url = self.browser.page.url.lower() if any(indicator in current_url for indicator in ['home', 'feed', 'dashboard']): return True # Prüfe Seiteninhalte return self._check_for_text_on_page(logged_in_indicators, threshold=0.7) def _get_platform_name(self) -> str: """ Gibt den Namen der Plattform zurück. Sollte von Unterklassen überschrieben werden. """ # Versuche aus Klassennamen zu extrahieren class_name = self.__class__.__name__.lower() if 'instagram' in class_name: return 'instagram' elif 'facebook' in class_name: return 'facebook' elif 'twitter' in class_name: return 'twitter' elif 'tiktok' in class_name: return 'tiktok' return 'unknown' 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 get_browser_context_data(self) -> Dict[str, Any]: """ Extrahiert Browser-Context-Daten für Session-Speicherung. Returns: Dict[str, Any]: Browser-Context-Daten """ try: if not self.browser or not hasattr(self.browser, 'page'): return {} # Cookies extrahieren cookies = self.browser.page.context.cookies() # User Agent und Viewport user_agent = self.browser.page.evaluate("() => navigator.userAgent") viewport = self.browser.page.viewport_size # URL current_url = self.browser.page.url return { 'cookies': cookies, 'user_agent': user_agent, 'viewport_size': viewport, 'current_url': current_url, 'context_id': getattr(self.browser.page.context, '_guid', None) } except Exception as e: logger.error(f"Fehler beim Extrahieren der Browser-Context-Daten: {e}") return {} def extract_platform_session_data(self, platform: str) -> Dict[str, Any]: """ Extrahiert plattform-spezifische Session-Daten. Args: platform: Zielplattform Returns: Dict[str, Any]: Plattform-Session-Daten """ try: if not self.browser or not hasattr(self.browser, 'page'): return {} # Local Storage extrahieren local_storage = {} try: local_storage_script = """ () => { const storage = {}; for (let i = 0; i < localStorage.length; i++) { const key = localStorage.key(i); storage[key] = localStorage.getItem(key); } return storage; } """ local_storage = self.browser.page.evaluate(local_storage_script) except Exception as e: logger.warning(f"Konnte Local Storage nicht extrahieren: {e}") # Session Storage extrahieren session_storage = {} try: session_storage_script = """ () => { const storage = {}; for (let i = 0; i < sessionStorage.length; i++) { const key = sessionStorage.key(i); storage[key] = sessionStorage.getItem(key); } return storage; } """ session_storage = self.browser.page.evaluate(session_storage_script) except Exception as e: logger.warning(f"Konnte Session Storage nicht extrahieren: {e}") return { 'platform': platform, 'local_storage': local_storage, 'session_storage': session_storage, 'url': self.browser.page.url, 'created_at': time.time() } except Exception as e: logger.error(f"Fehler beim Extrahieren der Plattform-Session-Daten: {e}") return {} 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.")