# social_networks/x/x_verification.py """ X (Twitter) Verification - Klasse für Account-Verifizierungsprozesse bei X """ import time import re from typing import Dict, Any, Optional, List from .x_selectors import XSelectors from .x_workflow import XWorkflow from utils.logger import setup_logger # Konfiguriere Logger logger = setup_logger("x_verification") class XVerification: """ Klasse für die Verifizierung von X-Konten. Behandelt verschiedene Verifizierungsmethoden und Sicherheitsabfragen. """ def __init__(self, automation): """ Initialisiert die X-Verifizierung. Args: automation: Referenz auf die Hauptautomatisierungsklasse """ self.automation = automation self.selectors = XSelectors() self.workflow = XWorkflow.get_verification_workflow() logger.debug("X-Verifizierung initialisiert") def verify_account(self, verification_code: str = None, **kwargs) -> Dict[str, Any]: """ Führt den Verifizierungsprozess für einen X-Account durch. Args: verification_code: Optionaler Verifizierungscode **kwargs: Weitere Parameter (method, phone_number, email) Returns: Dict[str, Any]: Ergebnis der Verifizierung """ logger.info("Starte X-Account-Verifizierung") try: # Prüfe welche Art von Verifizierung benötigt wird verification_type = self._detect_verification_type() if not verification_type: logger.info("Keine Verifizierung erforderlich") return { "success": True, "message": "Keine Verifizierung erforderlich" } logger.info(f"Verifizierungstyp erkannt: {verification_type}") self.automation._emit_customer_log(f"🔐 Verifizierung erforderlich: {verification_type}") # Führe entsprechende Verifizierung durch if verification_type == "email": return self._handle_email_verification(verification_code, **kwargs) elif verification_type == "phone": return self._handle_phone_verification(verification_code, **kwargs) elif verification_type == "captcha": return self._handle_captcha_verification() elif verification_type == "arkose": return self._handle_arkose_challenge() else: return { "success": False, "error": f"Unbekannter Verifizierungstyp: {verification_type}" } except Exception as e: error_msg = f"Unerwarteter Fehler bei der Verifizierung: {str(e)}" logger.error(error_msg, exc_info=True) return { "success": False, "error": error_msg, "stage": "exception" } def _detect_verification_type(self) -> Optional[str]: """ Erkennt welche Art von Verifizierung benötigt wird. Returns: Optional[str]: Verifizierungstyp oder None """ try: page = self.automation.browser.page # E-Mail-Verifizierung if page.is_visible(self.selectors.REGISTRATION["verification_code_input"], timeout=2000): return "email" # Telefon-Verifizierung if page.is_visible(self.selectors.VERIFICATION["phone_verification_input"], timeout=2000): return "phone" # Captcha if page.is_visible(self.selectors.VERIFICATION["captcha_frame"], timeout=2000): return "captcha" # Arkose Challenge if page.is_visible(self.selectors.VERIFICATION["challenge_frame"], timeout=2000): return "arkose" # Prüfe auf Text-Hinweise if page.is_visible('text="Verifiziere deine E-Mail"', timeout=1000): return "email" if page.is_visible('text="Verifiziere deine Telefonnummer"', timeout=1000): return "phone" return None except Exception as e: logger.error(f"Fehler bei Verifizierungstyp-Erkennung: {e}") return None def _handle_email_verification(self, verification_code: str = None, **kwargs) -> Dict[str, Any]: """ Behandelt E-Mail-Verifizierung. Args: verification_code: Verifizierungscode **kwargs: Weitere Parameter (email) Returns: Dict[str, Any]: Ergebnis der Verifizierung """ try: page = self.automation.browser.page # Wenn kein Code übergeben wurde, versuche ihn abzurufen if not verification_code: email = kwargs.get("email") if not email: return { "success": False, "error": "E-Mail-Adresse für Verifizierung fehlt" } self.automation._emit_customer_log("📧 Warte auf Verifizierungs-E-Mail...") verification_code = self._retrieve_email_code(email) if not verification_code: return { "success": False, "error": "Konnte keinen Verifizierungscode aus E-Mail abrufen" } # Code eingeben self.automation._emit_customer_log(f"✍️ Gebe Verifizierungscode ein: {verification_code}") if not self._enter_verification_code(verification_code): return { "success": False, "error": "Fehler beim Eingeben des Verifizierungscodes" } # Bestätigen if not self._submit_verification(): return { "success": False, "error": "Fehler beim Bestätigen der Verifizierung" } # Warte auf Erfolg if self._check_verification_success(): logger.info("E-Mail-Verifizierung erfolgreich") self.automation._emit_customer_log("✅ E-Mail erfolgreich verifiziert!") return { "success": True, "method": "email", "code": verification_code } else: return { "success": False, "error": "Verifizierung fehlgeschlagen" } except Exception as e: logger.error(f"Fehler bei E-Mail-Verifizierung: {e}") return { "success": False, "error": f"E-Mail-Verifizierung fehlgeschlagen: {str(e)}" } def _handle_phone_verification(self, verification_code: str = None, **kwargs) -> Dict[str, Any]: """ Behandelt Telefon-Verifizierung. Args: verification_code: Verifizierungscode **kwargs: Weitere Parameter (phone_number) Returns: Dict[str, Any]: Ergebnis der Verifizierung """ try: page = self.automation.browser.page # Telefonnummer eingeben wenn erforderlich phone_number = kwargs.get("phone_number") if phone_number and page.is_visible(self.selectors.VERIFICATION["phone_verification_input"]): logger.info(f"Gebe Telefonnummer ein: {phone_number}") phone_input = page.wait_for_selector(self.selectors.VERIFICATION["phone_verification_input"]) self.automation.human_behavior.type_text(phone_input, phone_number) # Code senden if page.is_visible(self.selectors.VERIFICATION["send_code_button"]): page.click(self.selectors.VERIFICATION["send_code_button"]) self.automation.human_behavior.random_delay(2, 3) # Wenn kein Code übergeben wurde, warte auf manuelle Eingabe if not verification_code: self.automation._emit_customer_log("📱 Bitte gib den SMS-Code manuell ein...") # Warte bis Code eingegeben wurde (max 5 Minuten) start_time = time.time() while time.time() - start_time < 300: # 5 Minuten code_input = page.query_selector(self.selectors.VERIFICATION["verification_code_input"]) if code_input: current_value = code_input.get_attribute("value") if current_value and len(current_value) >= 6: verification_code = current_value break time.sleep(2) if not verification_code: return { "success": False, "error": "Kein SMS-Code eingegeben (Timeout)" } else: # Code eingeben self.automation._emit_customer_log(f"✍️ Gebe SMS-Code ein: {verification_code}") if not self._enter_verification_code(verification_code): return { "success": False, "error": "Fehler beim Eingeben des SMS-Codes" } # Bestätigen if not self._submit_verification(): return { "success": False, "error": "Fehler beim Bestätigen der Telefon-Verifizierung" } # Warte auf Erfolg if self._check_verification_success(): logger.info("Telefon-Verifizierung erfolgreich") self.automation._emit_customer_log("✅ Telefonnummer erfolgreich verifiziert!") return { "success": True, "method": "phone", "phone_number": phone_number } else: return { "success": False, "error": "Telefon-Verifizierung fehlgeschlagen" } except Exception as e: logger.error(f"Fehler bei Telefon-Verifizierung: {e}") return { "success": False, "error": f"Telefon-Verifizierung fehlgeschlagen: {str(e)}" } def _handle_captcha_verification(self) -> Dict[str, Any]: """ Behandelt Captcha-Verifizierung. Returns: Dict[str, Any]: Ergebnis der Verifizierung """ try: logger.warning("Captcha-Verifizierung erkannt") self.automation._emit_customer_log("🤖 Captcha erkannt - manuelle Lösung erforderlich") # Screenshot für Debugging self.automation._take_screenshot("captcha_challenge") # Hier könnte Integration mit Captcha-Solving-Service erfolgen # Für jetzt: Warte auf manuelle Lösung return { "success": False, "error": "Captcha-Lösung erforderlich - bitte manuell lösen", "type": "captcha", "manual_intervention_required": True } except Exception as e: logger.error(f"Fehler bei Captcha-Behandlung: {e}") return { "success": False, "error": f"Captcha-Behandlung fehlgeschlagen: {str(e)}" } def _handle_arkose_challenge(self) -> Dict[str, Any]: """ Behandelt Arkose Labs Challenge. Returns: Dict[str, Any]: Ergebnis der Verifizierung """ try: logger.warning("Arkose Challenge erkannt") self.automation._emit_customer_log("🛡️ Arkose Challenge erkannt - erweiterte Verifizierung erforderlich") # Screenshot für Debugging self.automation._take_screenshot("arkose_challenge") return { "success": False, "error": "Arkose Challenge erkannt - manuelle Intervention erforderlich", "type": "arkose", "manual_intervention_required": True } except Exception as e: logger.error(f"Fehler bei Arkose Challenge: {e}") return { "success": False, "error": f"Arkose Challenge fehlgeschlagen: {str(e)}" } def _retrieve_email_code(self, email: str) -> Optional[str]: """ Ruft Verifizierungscode aus E-Mail ab. Args: email: E-Mail-Adresse Returns: Optional[str]: Verifizierungscode oder None """ try: logger.info(f"Rufe Verifizierungscode für {email} ab") # Warte auf E-Mail self.automation.human_behavior.random_delay(5, 10) # Hole E-Mails emails = self.automation.email_handler.get_emails( email, subject_filter="X Verifizierungscode" ) if not emails: # Versuche alternative Betreff-Filter emails = self.automation.email_handler.get_emails( email, subject_filter="Twitter" ) if not emails: logger.error("Keine Verifizierungs-E-Mail gefunden") return None # Extrahiere Code aus der neuesten E-Mail latest_email = emails[0] subject = latest_email.get('subject', '') body = latest_email.get('body', '') # Suche nach 6-stelligem Code # Erst im Betreff code_match = re.search(r'(\d{6})', subject) if code_match: code = code_match.group(1) logger.info(f"Code aus Betreff extrahiert: {code}") return code # Dann im Body code_match = re.search(r'(\d{6})', body) if code_match: code = code_match.group(1) logger.info(f"Code aus E-Mail-Body extrahiert: {code}") return code logger.error("Konnte keinen Code aus E-Mail extrahieren") return None except Exception as e: logger.error(f"Fehler beim Abrufen des E-Mail-Codes: {e}") return None def _enter_verification_code(self, code: str) -> bool: """ Gibt einen Verifizierungscode ein. Args: code: Verifizierungscode Returns: bool: True bei Erfolg """ try: page = self.automation.browser.page # Finde Code-Eingabefeld code_selectors = [ self.selectors.REGISTRATION["verification_code_input"], self.selectors.VERIFICATION["verification_code_input"], 'input[autocomplete="one-time-code"]', 'input[name="code"]' ] for selector in code_selectors: try: code_input = page.wait_for_selector(selector, timeout=3000) if code_input: logger.info(f"Code-Eingabefeld gefunden: {selector}") self.automation.human_behavior.type_text(code_input, code) self.automation.human_behavior.random_delay(0.5, 1) return True except: continue logger.error("Kein Code-Eingabefeld gefunden") return False except Exception as e: logger.error(f"Fehler beim Eingeben des Codes: {e}") return False def _submit_verification(self) -> bool: """ Bestätigt die Verifizierung. Returns: bool: True bei Erfolg """ try: page = self.automation.browser.page # Submit-Button Selektoren submit_selectors = [ 'button:has-text("Weiter")', 'button:has-text("Bestätigen")', 'button:has-text("Verifizieren")', 'button:has-text("Next")', 'button:has-text("Verify")', 'button:has-text("Submit")' ] for selector in submit_selectors: if page.is_visible(selector, timeout=2000): logger.info(f"Submit-Button gefunden: {selector}") page.click(selector) self.automation.human_behavior.random_delay(2, 3) return True # Alternativ: Enter drücken page.keyboard.press("Enter") logger.info("Enter gedrückt zur Bestätigung") return True except Exception as e: logger.error(f"Fehler beim Bestätigen: {e}") return False def _check_verification_success(self) -> bool: """ Prüft ob Verifizierung erfolgreich war. Returns: bool: True bei Erfolg """ try: page = self.automation.browser.page # Warte auf Weiterleitung oder Erfolgsmeldung self.automation.human_behavior.random_delay(2, 3) # Erfolgsindikatoren success_indicators = [ # Fehlen von Verifizierungsfeldern lambda: not page.is_visible(self.selectors.REGISTRATION["verification_code_input"], timeout=2000), # Vorhandensein von Account-Elementen lambda: page.is_visible(self.selectors.NAVIGATION["home_link"], timeout=2000), # URL-Änderung lambda: "/home" in page.url or "/welcome" in page.url ] for indicator in success_indicators: if indicator(): logger.info("Verifizierung erfolgreich") return True # Prüfe auf Fehlermeldungen error_msg = self.automation.ui_helper.check_for_errors() if error_msg: logger.error(f"Verifizierungsfehler: {error_msg}") return False return False except Exception as e: logger.error(f"Fehler bei Erfolgsprüfung: {e}") return False