import os import threading import http.server import socketserver import json import time import sys from playwright.sync_api import sync_playwright PORT = 3002 DIRECTORY = "." SESSION_DIR = "./snapchat_session" MESSAGES_FILE = "decrypted_messages.json" IS_HEADLESS = sys.platform.startswith("linux") # Headless on VPS, windowed on local Mac class ScraperService: def __init__(self): self.latest_data = {"chat_list": [], "messages": []} self.running = False self.target_chat = None # the name of the chat to open self.current_open_chat = None def get_latest_data(self): return self.latest_data def set_target_chat(self, name): self.target_chat = name # Also clear current messages so we don't show old ones while loading self.latest_data["messages"] = [] with open(MESSAGES_FILE, "w", encoding="utf-8") as f: json.dump(self.latest_data, f, indent=2, ensure_ascii=False) def start(self): if not self.running: self.running = True self.thread = threading.Thread(target=self._run_scraper_loop, daemon=True) self.thread.start() def _run_scraper_loop(self): with sync_playwright() as p: print(f"[*] Démarrage du scraper en arrière-plan (Headless: {IS_HEADLESS})...") context = p.chromium.launch_persistent_context( user_data_dir=SESSION_DIR, channel="chrome", headless=IS_HEADLESS, viewport={"width": 1280, "height": 720}, args=["--disable-blink-features=AutomationControlled"], ignore_default_args=["--enable-automation"], user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36" ) if len(context.pages) > 0: page = context.pages[0] else: page = context.new_page() print("[*] Navigation vers Snapchat Web...") try: page.goto("https://web.snapchat.com/", timeout=60000, wait_until="domcontentloaded") except Exception as e: print("[!] Erreur lors de la navigation, mais on continue :", e) print("[*] Scraper actif. En attente des messages...") while self.running: # Handle navigation request if self.target_chat and self.target_chat != self.current_open_chat: try: print(f"[*] Tentative d'ouverture de la conversation avec {self.target_chat}...") # Look for a div that contains the target_chat text and click it # Since React virtualization reuses divs, we must search text js_click = f""" () => {{ let container = document.querySelector('.ReactVirtualized__Grid__innerScrollContainer'); if (container) {{ for (let item of container.children) {{ if (item.innerText.includes("{self.target_chat}")) {{ item.click(); return true; }} }} }} return false; }} """ success = page.evaluate(js_click) if success: print(f"[*] Clic effectué sur la conversation de {self.target_chat}.") self.current_open_chat = self.target_chat time.sleep(2) # wait for chat to load else: print(f"[!] Impossible de trouver {self.target_chat} dans la liste.") except Exception as e: print(f"[!] Erreur lors du clic: {e}") js_code = """ () => { let chat_list = []; let messages = []; // 1. Extraire la Sidebar (liste des conversations) // Snapchat Web utilise 'ReactVirtualized' pour la liste de gauche. let container = document.querySelector('.ReactVirtualized__Grid__innerScrollContainer'); if (container) { for (let item of container.children) { let text = item.innerText.trim(); let img = item.querySelector('img'); let avatar = img ? img.src : ''; if (text.length > 0) { // Nettoyer le texte car les emojis sont souvent sur une ligne séparée let lines = text.split('\\n').filter(l => l.trim() !== '' && l !== '😊' && l !== 'Répondre'); chat_list.push({ text: lines.join('\\n'), avatar: avatar }); } } } // 2. Extraire les Messages de la conversation ouverte // Les messages sont toujours dans des balises
  • let allLis = document.querySelectorAll('li'); allLis.forEach(li => { let text = li.innerText.trim(); let media = []; li.querySelectorAll('img, video, audio').forEach(el => { if (el.src && !el.src.includes('avatar') && !el.src.includes('bitmoji') && !el.src.includes('pravatar')) { media.push(el.src); } }); if (text.length > 0 || media.length > 0) { messages.push({ text: text, media: media }); } }); // Deduplicate return { chat_list: Array.from(new Set(chat_list.map(JSON.stringify))).map(JSON.parse), messages: Array.from(new Set(messages.map(JSON.stringify))).map(JSON.parse) }; } """ try: extracted = page.evaluate(js_code) if extracted is not None: # Convert blob: URLs to data URIs extracted = self._resolve_blob_urls(page, extracted) if extracted != self.latest_data: self.latest_data = extracted print(f"[*] Mise à jour: {len(extracted.get('chat_list',[]))} chats, {len(extracted.get('messages',[]))} messages") with open(MESSAGES_FILE, "w", encoding="utf-8") as f: json.dump(extracted, f, indent=2, ensure_ascii=False) except Exception as e: print(f"[!] Erreur scraping: {e}") time.sleep(3) def _resolve_blob_urls(self, page, data): """Convert blob: URLs in messages to base64 data URIs so they work outside the browser.""" messages = data.get("messages", []) for msg in messages: new_media = [] for url in msg.get("media", []): if url.startswith("blob:"): try: data_uri = page.evaluate(""" async (blobUrl) => { try { const response = await fetch(blobUrl); const blob = await response.blob(); return new Promise((resolve, reject) => { const reader = new FileReader(); reader.onloadend = () => resolve(reader.result); reader.onerror = reject; reader.readAsDataURL(blob); }); } catch(e) { return null; } } """, url) if data_uri: new_media.append(data_uri) except Exception: pass else: new_media.append(url) msg["media"] = new_media return data scraper_service = ScraperService() class Handler(http.server.SimpleHTTPRequestHandler): def __init__(self, *args, **kwargs): super().__init__(*args, directory=DIRECTORY, **kwargs) def end_headers(self): if self.path.endswith('.json'): self.send_header('Cache-Control', 'no-cache, no-store, must-revalidate') self.send_header('Pragma', 'no-cache') self.send_header('Expires', '0') self.send_header('Access-Control-Allow-Origin', '*') super().end_headers() def do_POST(self): if self.path == '/open_chat': content_length = int(self.headers.get('Content-Length', 0)) if content_length > 0: body = self.rfile.read(content_length).decode('utf-8') try: data = json.loads(body) name = data.get('name') if name: scraper_service.set_target_chat(name) self.send_response(200) self.send_header('Content-Type', 'application/json') self.send_header('Access-Control-Allow-Origin', '*') self.end_headers() self.wfile.write(b'{"status":"ok"}') return except json.JSONDecodeError: pass self.send_response(400) self.end_headers() else: self.send_response(404) self.end_headers() def log_message(self, format, *args): try: msg = format % args if 'decrypted_messages.json' in msg: return except Exception: pass super().log_message(format, *args) if __name__ == "__main__": socketserver.TCPServer.allow_reuse_address = True scraper_service.start() with socketserver.TCPServer(("", PORT), Handler) as httpd: print("=======================================================") print(f"[*] Serveur UI démarré : http://localhost:{PORT}/snapchat_ui.html") print("[*] Le scraper s'exécute automatiquement en arrière-plan.") print("=======================================================\n") try: httpd.serve_forever() except KeyboardInterrupt: print("[*] Arrêt du serveur...")