""" TikTok-Automatisierung - Hauptklasse für TikTok-Automatisierungsfunktionalität """ import time import random from typing import Dict, List, Any, Optional, Tuple from browser.playwright_manager import PlaywrightManager from browser.playwright_extensions import PlaywrightExtensions from social_networks.base_automation import BaseAutomation from utils.password_generator import PasswordGenerator from utils.username_generator import UsernameGenerator from utils.birthday_generator import BirthdayGenerator from utils.human_behavior import HumanBehavior from utils.logger import setup_logger # Importiere Helferklassen from .tiktok_registration import TikTokRegistration from .tiktok_login import TikTokLogin from .tiktok_verification import TikTokVerification from .tiktok_ui_helper import TikTokUIHelper from .tiktok_utils import TikTokUtils # Konfiguriere Logger logger = setup_logger("tiktok_automation") class TikTokAutomation(BaseAutomation): """ Hauptklasse für die TikTok-Automatisierung. Implementiert die Registrierung und Anmeldung bei TikTok. """ 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", enhanced_stealth: bool = True, fingerprint_noise: float = 0.5, window_position = None, fingerprint = None, auto_close_browser: bool = False): """ Initialisiert die TikTok-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 enhanced_stealth: Ob erweiterter Stealth-Modus aktiviert werden soll fingerprint_noise: Menge an Rauschen für Fingerprint-Verschleierung (0.0-1.0) window_position: Optional - Fensterposition als Tuple (x, y) fingerprint: Optional - Vordefinierter Browser-Fingerprint auto_close_browser: Ob Browser automatisch geschlossen werden soll (Standard: False) """ # Initialisiere die Basisklasse super().__init__( headless=headless, use_proxy=use_proxy, proxy_type=proxy_type, save_screenshots=save_screenshots, screenshots_dir=screenshots_dir, slowmo=slowmo, debug=debug, email_domain=email_domain, window_position=window_position, auto_close_browser=auto_close_browser ) # Stealth-Modus-Einstellungen self.enhanced_stealth = enhanced_stealth self.fingerprint_noise = max(0.0, min(1.0, fingerprint_noise)) # Initialisiere Helferklassen self.registration = TikTokRegistration(self) self.login = TikTokLogin(self) self.verification = TikTokVerification(self) self.ui_helper = TikTokUIHelper(self) self.utils = TikTokUtils(self) # Zusätzliche Hilfsklassen self.password_generator = PasswordGenerator() self.username_generator = UsernameGenerator() self.birthday_generator = BirthdayGenerator() self.human_behavior = HumanBehavior(speed_factor=0.8, randomness=0.6) # Nutze übergebenen Fingerprint wenn vorhanden self.provided_fingerprint = fingerprint logger.info("TikTok-Automatisierung initialisiert") def _initialize_browser(self) -> bool: """ Initialisiert den Browser mit den entsprechenden Einstellungen. Diese Methode überschreibt die Methode der Basisklasse, um den erweiterten Fingerprint-Schutz zu aktivieren. 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() # Erweiterten Fingerprint-Schutz aktivieren, wenn gewünscht if self.enhanced_stealth: # Erstelle Extensions-Objekt extensions = PlaywrightExtensions(self.browser) # Methoden anhängen extensions.hook_into_playwright_manager() # Fingerprint-Schutz aktivieren mit angepasster Konfiguration if self.provided_fingerprint: # Nutze den bereitgestellten Fingerprint logger.info("Verwende bereitgestellten Fingerprint für Account-Erstellung") # Konvertiere Dict zu BrowserFingerprint wenn nötig if isinstance(self.provided_fingerprint, dict): from domain.entities.browser_fingerprint import BrowserFingerprint fingerprint_obj = BrowserFingerprint.from_dict(self.provided_fingerprint) else: fingerprint_obj = self.provided_fingerprint # Wende Fingerprint über FingerprintProtection an from browser.fingerprint_protection import FingerprintProtection protection = FingerprintProtection( context=self.browser.context, fingerprint_config=fingerprint_obj ) protection.apply_to_context(self.browser.context) logger.info(f"Fingerprint {fingerprint_obj.fingerprint_id} angewendet") else: # Fallback: Zufällige Fingerprint-Konfiguration fingerprint_config = { "noise_level": self.fingerprint_noise, "canvas_noise": True, "audio_noise": True, "webgl_noise": True, "hardware_concurrency": random.choice([4, 6, 8]), "device_memory": random.choice([4, 8]), "timezone_id": "Europe/Berlin" } success = self.browser.enable_enhanced_fingerprint_protection(fingerprint_config) if success: logger.info("Erweiterter Fingerprint-Schutz erfolgreich aktiviert") else: logger.warning("Erweiterter Fingerprint-Schutz konnte nicht aktiviert werden") 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 register_account(self, full_name: str, age: int, registration_method: str = "email", phone_number: str = None, **kwargs) -> Dict[str, Any]: """ Registriert einen neuen TikTok-Account. 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 """ logger.info(f"Starte TikTok-Account-Registrierung für '{full_name}' via {registration_method}") self._emit_customer_log(f"🎵 TikTok-Account wird erstellt für: {full_name}") try: # Initialisiere Browser, falls noch nicht geschehen if not self.browser or not hasattr(self.browser, 'page'): if not self._initialize_browser(): return {"success": False, "error": "Browser konnte nicht initialisiert werden"} # Rotiere Fingerprint vor der Hauptaktivität, um Erkennung weiter zu erschweren if self.enhanced_stealth and hasattr(self.browser, 'rotate_fingerprint'): self.browser.rotate_fingerprint() logger.info("Browser-Fingerprint vor der Registrierung rotiert") # Delegiere die Hauptregistrierungslogik an die Registration-Klasse result = self.registration.register_account(full_name, age, registration_method, phone_number, **kwargs) # Nehme Abschlussfoto auf self._take_screenshot(f"registration_finished_{int(time.time())}") # Aktualisiere Status self.status.update(result) return result except Exception as e: error_msg = f"Unerwarteter Fehler bei der Account-Registrierung: {str(e)}" logger.error(error_msg, exc_info=True) # Fehler-Screenshot self._take_screenshot(f"registration_error_{int(time.time())}") # Aktualisiere Status self.status.update({ "success": False, "error": error_msg, "stage": "error" }) return self.status finally: # Browser schließen self._close_browser() def login_account(self, username_or_email: str, password: str, **kwargs) -> Dict[str, Any]: """ Meldet sich bei einem bestehenden TikTok-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 """ logger.info(f"Starte TikTok-Login für '{username_or_email}'") try: # Initialisiere Browser, falls noch nicht geschehen if not self.browser or not hasattr(self.browser, 'page'): if not self._initialize_browser(): return {"success": False, "error": "Browser konnte nicht initialisiert werden"} # Rotiere Fingerprint vor dem Login if self.enhanced_stealth and hasattr(self.browser, 'rotate_fingerprint'): self.browser.rotate_fingerprint() logger.info("Browser-Fingerprint vor dem Login rotiert") # Delegiere die Hauptlogin-Logik an die Login-Klasse result = self.login.login_account(username_or_email, password, **kwargs) # Nehme Abschlussfoto auf self._take_screenshot(f"login_finished_{int(time.time())}") # Aktualisiere Status self.status.update(result) return result except Exception as e: error_msg = f"Unerwarteter Fehler beim Login: {str(e)}" logger.error(error_msg, exc_info=True) # Fehler-Screenshot self._take_screenshot(f"login_error_{int(time.time())}") # Aktualisiere Status self.status.update({ "success": False, "error": error_msg, "stage": "error" }) return self.status finally: # Browser schließen self._close_browser() def verify_account(self, verification_code: str, **kwargs) -> Dict[str, Any]: """ Verifiziert einen TikTok-Account mit einem Bestätigungscode. Args: verification_code: Der Bestätigungscode **kwargs: Weitere optionale Parameter Returns: Dict[str, Any]: Ergebnis der Verifizierung mit Status """ logger.info(f"Starte TikTok-Account-Verifizierung mit Code: {verification_code}") try: # Initialisiere Browser, falls noch nicht geschehen if not self.browser or not hasattr(self.browser, 'page'): if not self._initialize_browser(): return {"success": False, "error": "Browser konnte nicht initialisiert werden"} # Delegiere die Hauptverifizierungslogik an die Verification-Klasse result = self.verification.verify_account(verification_code, **kwargs) # Nehme Abschlussfoto auf self._take_screenshot(f"verification_finished_{int(time.time())}") # Aktualisiere Status self.status.update(result) return result except Exception as e: error_msg = f"Unerwarteter Fehler bei der Account-Verifizierung: {str(e)}" logger.error(error_msg, exc_info=True) # Fehler-Screenshot self._take_screenshot(f"verification_error_{int(time.time())}") # Aktualisiere Status self.status.update({ "success": False, "error": error_msg, "stage": "error" }) return self.status finally: # Browser schließen self._close_browser() def get_fingerprint_status(self) -> Dict[str, Any]: """ Gibt den aktuellen Status des Fingerprint-Schutzes zurück. Returns: Dict[str, Any]: Status des Fingerprint-Schutzes """ if not self.enhanced_stealth or not hasattr(self.browser, 'get_fingerprint_status'): return { "active": False, "message": "Erweiterter Fingerprint-Schutz ist nicht aktiviert" } return self.browser.get_fingerprint_status() def get_current_page(self): """ Gibt die aktuelle Playwright Page-Instanz zurück. Returns: Page: Die aktuelle Seite oder None """ if self.browser and hasattr(self.browser, 'page'): return self.browser.page return None def get_session_data(self) -> Dict[str, Any]: """ Extrahiert Session-Daten (Cookies, LocalStorage, etc.) aus dem aktuellen Browser. Returns: Dict[str, Any]: Session-Daten """ if not self.is_browser_open(): return {} try: return { "cookies": self.browser.page.context.cookies(), "local_storage": self.browser.page.evaluate("() => Object.assign({}, window.localStorage)"), "session_storage": self.browser.page.evaluate("() => Object.assign({}, window.sessionStorage)"), "url": self.browser.page.url } except Exception as e: logger.error(f"Fehler beim Extrahieren der Session-Daten: {e}") return {}