#!/usr/bin/env python3 """ Test de téléchargement d'une vidéo Instagram en utilisant l'API de cobalt """ import subprocess import json import logging import os import time import re from urllib.parse import urlencode # Configuration du logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) 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] # Enlever les paramètres d'URL return None def download_instagram_video(url, output_dir="downloads"): """Télécharge une vidéo Instagram via l'API cobalt""" logger.info(f"🔄 Téléchargement depuis Instagram: {url}") # Créer le dossier de téléchargement s'il n'existe pas if not os.path.exists(output_dir): os.makedirs(output_dir) # Extraire l'ID du post post_id = extract_post_id(url) if not post_id: logger.error("❌ URL Instagram invalide") return None try: # Headers pour les requêtes headers = { 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7', 'Accept-Language': 'en-GB,en;q=0.9', 'Cache-Control': 'max-age=0', 'Dnt': '1', 'Priority': 'u=0, i', 'Sec-Ch-Ua': '"Chromium";v="124", "Google Chrome";v="124", "Not-A.Brand";v="99"', 'Sec-Ch-Ua-Mobile': '?0', 'Sec-Ch-Ua-Platform': 'macOS', 'Sec-Fetch-Dest': 'document', 'Sec-Fetch-Mode': 'navigate', 'Sec-Fetch-Site': 'none', 'Sec-Fetch-User': '?1', 'Upgrade-Insecure-Requests': '1', 'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36', 'Content-Type': 'application/x-www-form-urlencoded', 'X-FB-Friendly-Name': 'PolarisPostActionLoadPostQueryQuery' } # D'abord, récupérer la page HTML pour les tokens logger.info("Récupération des informations de la page...") html_cmd = [ 'curl', f'https://www.instagram.com/p/{post_id}/', '-H', 'Accept: text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8', '-H', 'Accept-Language: en-GB,en;q=0.9', '-H', f'User-Agent: {headers["User-Agent"]}', '--compressed', '-s' # Mode silencieux ] process = subprocess.Popen(html_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) stdout, stderr = process.communicate() if process.returncode != 0: logger.error(f"❌ Erreur lors de la récupération de la page: {stderr.decode()}") return None html_content = stdout.decode() # Extraire les tokens nécessaires csrf_match = re.search(r'"csrf_token":"([^"]+)"', html_content) if not csrf_match: logger.error("❌ Impossible de trouver le csrf_token") return None csrf_token = csrf_match.group(1) # Préparer la requête GraphQL variables = { "shortcode": post_id, "fetch_comment_count": "null", "fetch_related_profile_media_count": "null", "parent_comment_count": "null", "child_comment_count": "null", "fetch_like_count": "null", "fetch_tagged_user_count": "null", "fetch_preview_comment_count": "null", "has_threaded_comments": "false", "hoisted_comment_id": "null", "hoisted_reply_id": "null" } data = { "av": "0", "__d": "www", "__user": "0", "__a": "1", "__req": "3", "__hs": "19624.HYP:instagram_web_pkg.2.1..0.0", "variables": json.dumps(variables), "server_timestamps": "true", "doc_id": "10015901848480474" } # Faire la requête GraphQL logger.info("Récupération des données via GraphQL...") graphql_cmd = [ 'curl', 'https://www.instagram.com/graphql/query', '-X', 'POST', '-H', f'Accept: {headers["Accept"]}', '-H', f'Accept-Language: {headers["Accept-Language"]}', '-H', f'Content-Type: {headers["Content-Type"]}', '-H', f'User-Agent: {headers["User-Agent"]}', '-H', f'X-CSRFToken: {csrf_token}', '-H', 'X-IG-App-ID: 936619743392459', '--compressed', '--data', urlencode(data) ] process = subprocess.Popen(graphql_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) stdout, stderr = process.communicate() if process.returncode != 0: logger.error(f"❌ Erreur lors de la requête GraphQL: {stderr.decode()}") return None try: response_data = json.loads(stdout) if 'data' not in response_data: logger.error("❌ Pas de données dans la réponse GraphQL") return None media = response_data['data']['shortcode_media'] if not media['is_video']: logger.error("❌ Ce post n'est pas une vidéo") return None video_url = media['video_url'] # Télécharger la vidéo output_file = os.path.join(output_dir, f"video_{int(time.time())}.mp4") logger.info("🔄 Téléchargement de la vidéo...") download_cmd = [ 'curl', video_url, '-L', # Suivre les redirections '-o', output_file, '-H', f'User-Agent: {headers["User-Agent"]}', '--compressed', '-s' # Mode silencieux ] download_process = subprocess.Popen(download_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) stdout, stderr = download_process.communicate() if download_process.returncode == 0 and os.path.exists(output_file): logger.info(f"✅ Vidéo téléchargée: {output_file}") return output_file else: logger.error(f"❌ Erreur lors du téléchargement: {stderr.decode()}") return None except json.JSONDecodeError: logger.error("❌ Erreur de décodage JSON de la réponse GraphQL") logger.error(f"Réponse reçue: {stdout.decode()}") return None except Exception as e: logger.error(f"❌ Erreur: {str(e)}") return None 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")