#!/usr/bin/env python3 """ Test de téléchargement d'une vidéo Instagram en utilisant la même technique que cobalt.tools """ import subprocess import json import logging import os import time import re import random import string import uuid from urllib.parse import urlencode # Configuration du logging logging.basicConfig(level=logging.DEBUG) logger = logging.getLogger(__name__) def generate_device_id(): """Génère un identifiant de device Android comme le fait cobalt.tools""" # Format : android-XXXXXXXXXXXXXXXX (16 caractères hex) device_id = 'android-' + ''.join(random.choices(string.hexdigits.lower(), k=16)) return device_id def generate_uuid(): """Génère un UUID v4 pour les requêtes""" return str(uuid.uuid4()) def extract_post_id(url): """Extrait l'ID du post Instagram depuis l'URL""" patterns = [ r'instagram\.com/p/([^/]+)', r'instagram\.com/reel/([^/]+)', ] for pattern in patterns: match = re.search(pattern, url) if match: return match.group(1).split('/')[0] return None def get_video_cdn_url(post_url): """Récupère l'URL CDN de la vidéo comme le fait cobalt.tools""" post_id = extract_post_id(post_url) if not post_id: logger.error("❌ URL Instagram invalide") return None try: # Headers de base utilisés par cobalt.tools user_agent = "Mozilla/5.0 (Linux; Android 11; SM-G998B Build/RP1A.200720.012; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/121.0.6167.143 Mobile Safari/537.36 Instagram 314.0.0.22.106 Android (30/11; 560dpi; 1440x2932; samsung; SM-G998B; p3s; exynos2100; en_US; 556277901)" device_id = generate_device_id() mid = generate_uuid() ig_did = generate_uuid() ds_user_id = str(random.randint(1000000, 9999999)) session_id = generate_uuid() # Headers utilisés par l'app Instagram headers = { 'User-Agent': user_agent, 'Accept-Language': 'en-US', 'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8', 'Accept-Encoding': 'gzip, deflate', 'X-IG-Capabilities': '3brTvx0=', 'X-IG-Connection-Type': 'WIFI', 'X-IG-App-ID': '567067343352427', 'X-IG-Device-ID': device_id, 'X-IG-Android-ID': device_id, # Headers d'authentification 'Cookie': f'mid={mid}; ig_did={ig_did}; ds_user_id={ds_user_id}; sessionid={session_id}', 'Authorization': 'Bearer IGT:2:eyJkc191c2VyX2lkIjoiNDU0NzU0MjM4OSIsInNlc3Npb25pZCI6IjQ1NDc1NDIzODk6MmJCbjJHY3Z4UVlYWkk6MTc6MTU6MTY5MDY4MDAwMDowLjc1NjQ4OTQ4In0=', } # Première requête pour obtenir l'App Token init_url = f"https://i.instagram.com/api/v1/media/{post_id}/info/" init_cmd = [ 'curl', init_url, '-H', f'User-Agent: {headers["User-Agent"]}', '-H', f'Accept-Language: {headers["Accept-Language"]}', '-H', f'Content-Type: {headers["Content-Type"]}', '-H', f'X-IG-Capabilities: {headers["X-IG-Capabilities"]}', '-H', f'X-IG-Connection-Type: {headers["X-IG-Connection-Type"]}', '-H', f'X-IG-App-ID: {headers["X-IG-App-ID"]}', '-H', f'X-IG-Device-ID: {headers["X-IG-Device-ID"]}', '-H', f'X-IG-Android-ID: {headers["X-IG-Android-ID"]}', '-c', 'instagram_cookies.txt', '-s' ] logger.debug("Récupération des informations de la vidéo...") process = subprocess.Popen(init_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) stdout, stderr = process.communicate() if process.returncode != 0: logger.error(f"❌ Erreur lors de la requête info: {stderr.decode()}") return None response_data = json.loads(stdout) logger.debug(f"Réponse API: {json.dumps(response_data, indent=2)}") # Extraire l'URL CDN de la réponse try: items = response_data.get('items', []) if not items: logger.error("❌ Pas de média trouvé") return None video_versions = items[0].get('video_versions', []) if not video_versions: logger.error("❌ Pas de version vidéo trouvée") return None # Prendre la meilleure qualité best_video = max(video_versions, key=lambda x: x.get('width', 0) * x.get('height', 0)) video_url = best_video.get('url') if not video_url: logger.error("❌ URL vidéo non trouvée") return None logger.info("✅ URL CDN trouvée !") return video_url except (KeyError, IndexError, TypeError) as e: logger.error(f"❌ Erreur lors de l'extraction de l'URL: {str(e)}") return None except Exception as e: logger.error(f"❌ Erreur: {str(e)}") return None finally: # Nettoyer les fichiers temporaires if os.path.exists('instagram_cookies.txt'): os.remove('instagram_cookies.txt') def download_video(video_url, output_dir="downloads"): """Télécharge la vidéo depuis l'URL CDN""" if not os.path.exists(output_dir): os.makedirs(output_dir) output_file = os.path.join(output_dir, f"video_{int(time.time())}.mp4") try: # Headers pour la requête de téléchargement headers = { 'User-Agent': 'Instagram 219.0.0.12.117 Android', 'Accept': '*/*', 'Accept-Language': 'en-US', 'Accept-Encoding': 'gzip, deflate', 'Range': 'bytes=0-', 'Connection': 'keep-alive' } # Construire la commande curl cmd = [ 'curl', video_url, '-L', # Suivre les redirections '-o', output_file, '--retry', '3', # Réessayer 3 fois '--retry-delay', '2', # Attendre 2 secondes entre les essais ] # Ajouter les headers for header, value in headers.items(): cmd.extend(['-H', f'{header}: {value}']) logger.info("🔄 Téléchargement de la vidéo...") process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) stdout, stderr = process.communicate() if process.returncode == 0 and os.path.exists(output_file): if os.path.getsize(output_file) > 0: logger.info(f"✅ Vidéo téléchargée: {output_file}") return output_file else: logger.error("❌ Fichier téléchargé vide") os.remove(output_file) return None else: logger.error(f"❌ Erreur lors du téléchargement: {stderr.decode()}") return None except Exception as e: logger.error(f"❌ Erreur: {str(e)}") if os.path.exists(output_file): os.remove(output_file) return None def download_instagram_video(url): """Télécharge une vidéo Instagram en utilisant la technique de cobalt.tools""" logger.info(f"🔄 Téléchargement depuis Instagram: {url}") # 1. Récupérer l'URL CDN cdn_url = get_video_cdn_url(url) if not cdn_url: logger.error("❌ Impossible de récupérer l'URL CDN") return None # 2. Télécharger la vidéo return download_video(cdn_url) if __name__ == "__main__": url = "https://www.instagram.com/reel/C2_Kh6mPmGl/" result = download_instagram_video(url) if result: print(f"\n✅ Vidéo téléchargée avec succès: {result}") else: print("\n❌ Échec du téléchargement")