""" 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.")