#!/usr/bin/env python3 """ Script de test pour le scraping Instagram combinant Selenium et cobalt.tools """ from selenium import webdriver from selenium.webdriver.chrome.options import Options from selenium.webdriver.common.by import By from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC import json import logging import time import os import subprocess import sqlite3 from datetime import datetime # Configuration du logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class InstagramScraper: def __init__(self): """Initialise la base de données et le scraper""" self.setup_database() def setup_database(self): """Crée la base de données si elle n'existe pas""" self.conn = sqlite3.connect('instagram_cache.db') cursor = self.conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS video_cache ( url TEXT PRIMARY KEY, duration INTEGER, description TEXT, keywords TEXT, last_updated TIMESTAMP ) ''') self.conn.commit() def get_cached_data(self, url): """Récupère les données en cache si elles existent""" cursor = self.conn.cursor() cursor.execute('SELECT * FROM video_cache WHERE url = ?', (url,)) result = cursor.fetchone() if result: logger.info("✅ Données trouvées en cache") return { 'url': result[0], 'duration': result[1], 'description': result[2], 'keywords': result[3].split(',') if result[3] else [], 'last_updated': result[4] } return None def save_to_cache(self, data): """Sauvegarde les données dans le cache""" cursor = self.conn.cursor() cursor.execute(''' INSERT OR REPLACE INTO video_cache (url, duration, description, keywords, last_updated) VALUES (?, ?, ?, ?, ?) ''', ( data['url'], data['duration'], data['description'], ','.join(data['keywords']) if data.get('keywords') else None, datetime.now().isoformat() )) self.conn.commit() logger.info("✅ Données sauvegardées en cache") def setup_driver(self): """Configure le driver Selenium""" chrome_options = Options() chrome_options.add_argument('--headless') chrome_options.add_argument('--no-sandbox') chrome_options.add_argument('--disable-dev-shm-usage') chrome_options.add_argument('--disable-blink-features=AutomationControlled') chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"]) chrome_options.add_experimental_option('useAutomationExtension', False) driver = webdriver.Chrome(options=chrome_options) driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})") return driver def get_description_with_selenium(self, url): """Récupère la description avec Selenium""" driver = self.setup_driver() try: logger.info(f"🔍 Récupération de la description pour: {url}") driver.get(url) time.sleep(3) meta_tags = driver.find_elements(By.TAG_NAME, 'meta') description = None for meta in meta_tags: property_value = meta.get_attribute('property') if property_value and ('description' in property_value.lower()): description = meta.get_attribute('content') if description: logger.info(f"✅ Description trouvée: {description[:100]}...") # Recherche des mots-clés keywords = ["Palia", "CROSSPLAY"] found_keywords = [kw for kw in keywords if kw in description] if found_keywords: logger.info(f"✅ Mots-clés trouvés: {', '.join(found_keywords)}") return description, found_keywords return None, [] except Exception as e: logger.error(f"❌ Erreur Selenium: {str(e)}") return None, [] finally: driver.quit() def get_duration_with_cobalt(self, url): """Récupère la durée via l'API de cobalt.tools""" try: logger.info("🔄 Récupération via cobalt.tools...") # Simuler une requête comme le site web headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36', 'Accept': 'application/json', 'Origin': 'https://cobalt.tools', 'Referer': 'https://cobalt.tools/', } # Construire la commande curl avec les bons headers curl_cmd = ['curl', '-X', 'POST'] for k, v in headers.items(): curl_cmd.extend(['-H', f'{k}: {v}']) # Ajouter les données POST data = { "url": url, "aFormat": "mp3", "filenamePattern": "classic", "isAudioOnly": False, "isAudioMuted": False, "dubLang": False, "vQuality": "720p", # Qualité minimale pour la durée "isTrimEnabled": False, "isUrlBypass": False, "timeFrom": "00:00", "timeTo": "00:00" } curl_cmd.extend([ '-H', 'Content-Type: application/json', '-d', json.dumps(data), 'https://cobalt.tools/api/json' ]) # Exécuter curl et récupérer la réponse process = subprocess.Popen(curl_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) stdout, stderr = process.communicate() if process.returncode == 0: response = json.loads(stdout) if 'metadata' in response and 'duration' in response['metadata']: duration = int(float(response['metadata']['duration'])) logger.info(f"✅ Durée trouvée: {duration} secondes") return duration elif 'error' in response: logger.error(f"❌ Erreur cobalt.tools: {response['error']}") else: logger.error("❌ Pas de durée dans la réponse") else: logger.error(f"❌ Erreur curl: {stderr.decode()}") return None except Exception as e: logger.error(f"❌ Erreur cobalt.tools: {str(e)}") return None def get_video_data(self, url): """Récupère toutes les données d'une vidéo Instagram""" # Vérifier d'abord le cache cached_data = self.get_cached_data(url) if cached_data: return cached_data # Si pas en cache, récupérer les données description, keywords = self.get_description_with_selenium(url) duration = self.get_duration_with_cobalt(url) # Préparer les données data = { 'url': url, 'duration': duration, 'description': description, 'keywords': keywords } # Sauvegarder dans le cache self.save_to_cache(data) return data def main(): """Fonction principale""" url = "https://www.instagram.com/reel/DMpc37-tJdF/" scraper = InstagramScraper() result = scraper.get_video_data(url) if result: print("\n📊 Résultats:") print(json.dumps(result, indent=2, ensure_ascii=False)) if __name__ == "__main__": main()