import logging import os import re from urllib.parse import quote_plus from typing import Optional import requests from plugins.common import DownloadCancelled from plugins.stream import download_stream_url logger = logging.getLogger("worker.plugins.facebook") def _clean_fb_url(url: str) -> str: """Elimina parámetros de rastreo superfluos de URLs de Facebook.""" u = (url or "").strip() if "?" in u: base, query = u.split("?", 1) if "watch" in base.lower(): from urllib.parse import parse_qs, urlencode qs = parse_qs(query) keep = {k: v for k, v in qs.items() if k in ("v", "id")} if keep: return f"{base}?{urlencode(keep, doseq=True)}" return base return u def _resolve_canonical_fb_url(url: str) -> str: """Resuelve redirecciones cortas de Facebook (fb.watch, share/v, share/r) a la URL canónica.""" u = _clean_fb_url(url) if any(m in u.lower() for m in ["fb.watch", "fb.me", "/share/v/", "/share/r/"]): try: sess = requests.Session() sess.headers.update({ "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", "Accept-Language": "en-US,en;q=0.9", }) r = sess.head(u, allow_redirects=True, timeout=10) if r.status_code < 400 and r.url and r.url != u: return _clean_fb_url(r.url) except Exception as e: logger.debug(f"[Facebook] No se pudo resolver URL canónica mediante HEAD: {e}") try: r = requests.get(u, stream=True, allow_redirects=True, timeout=10) if r.status_code < 400 and r.url and r.url != u: return _clean_fb_url(r.url) except Exception: pass return u def download_facebook( url: str, output_dir: str, progress_fn=None, cancel_event=None, api_key: Optional[str] = None, ) -> str: """ Descarga videos y reels de Facebook con fallback automático (Delirius API + yt-dlp). """ if cancel_event and cancel_event.is_set(): raise DownloadCancelled() os.makedirs(output_dir, exist_ok=True) delirius_key = api_key or os.getenv("DELIRIUS_API_KEY", "") errors = [] canonical_url = _resolve_canonical_fb_url(url) urls_to_try = [canonical_url] if canonical_url != url: urls_to_try.append(url) # 1. Intentar con Delirius API for target_url in urls_to_try: if cancel_event and cancel_event.is_set(): raise DownloadCancelled() try: enc_url = quote_plus(target_url.strip()) api_url = f"https://api.delirius.online/download/facebook?url={enc_url}" if delirius_key: api_url += f"&apikey={delirius_key}" logger.info(f"[Facebook] Consultando Delirius API para: {target_url}") resp = requests.get(api_url, timeout=(10, 45)) if resp.status_code == 200: json_data = resp.json() if json_data.get("status"): video_url = None title = "facebook_video" if isinstance(json_data.get("list"), list) and json_data["list"]: hd_items = [ x for x in json_data["list"] if isinstance(x, dict) and ("hd" in str(x.get("quality", "")).lower() or "720" in str(x.get("quality", "")) or "1080" in str(x.get("quality", ""))) ] if hd_items: video_url = hd_items[0].get("url") if not video_url: video_url = json_data["list"][0].get("url") elif isinstance(json_data.get("data"), dict): data = json_data["data"] title = data.get("title") or title video_url = data.get("hd") or data.get("sd") or data.get("url") or data.get("download") elif isinstance(json_data.get("data"), list) and json_data["data"]: video_url = json_data["data"][0].get("url") if video_url: clean_name = re.sub(r'[\/\\:\*\?"<>\|\x00-\x1f]', '_', str(title))[:60].strip() or "facebook_video" target = os.path.join(output_dir, f"{clean_name}.mp4") try: out_path, _ = download_stream_url( video_url, target, progress_fn=progress_fn, cancel_event=cancel_event ) if os.path.exists(out_path) and os.path.getsize(out_path) > 0: return out_path except DownloadCancelled: raise except Exception as dl_err: logger.warning(f"[Facebook] Stream normal falló ({dl_err}), intentando con User-Agent de bot...") try: out_path, _ = download_stream_url( video_url, target, progress_fn=progress_fn, cancel_event=cancel_event, headers={"User-Agent": "TelegramBot (like TwitterBot)"} ) if os.path.exists(out_path) and os.path.getsize(out_path) > 0: return out_path except DownloadCancelled: raise except Exception as dl_err2: errors.append(f"Stream error: {dl_err2}") else: msg = json_data.get("msg") or json_data.get("message") or "status false" errors.append(f"Delirius ({resp.status_code}): {msg}") else: errors.append(f"Delirius HTTP {resp.status_code}") except DownloadCancelled: raise except Exception as e: errors.append(f"Delirius: {e}") logger.warning(f"[Facebook] Error con Delirius API: {e}") # 2. Fallback con yt-dlp if cancel_event and cancel_event.is_set(): raise DownloadCancelled() try: import yt_dlp logger.info(f"[Facebook] Intentando fallback con yt-dlp para: {url}") out_template = os.path.join(output_dir, "%(title).60s_%(id)s.%(ext)s") ydl_opts = { "outtmpl": out_template, "format": "best[ext=mp4]/best/b", "quiet": True, "no_warnings": True, "noplaylist": True, "socket_timeout": 30, "retries": 3, } def _ytdlp_hook(d): if cancel_event and cancel_event.is_set(): raise DownloadCancelled() if progress_fn and d.get("status") == "downloading": downloaded_b = d.get("downloaded_bytes", 0) total_b = d.get("total_bytes") or d.get("total_bytes_estimate", 0) if total_b: progress_fn(downloaded_b, total_b) ydl_opts["progress_hooks"] = [_ytdlp_hook] with yt_dlp.YoutubeDL(ydl_opts) as ydl: info = ydl.extract_info(url, download=True) if info: cand = ydl.prepare_filename(info) mp4_cand = os.path.splitext(cand)[0] + ".mp4" for c in [mp4_cand, cand]: if os.path.exists(c) and os.path.getsize(c) > 0: return c for root, _, files in os.walk(output_dir): for fn in files: fp = os.path.join(root, fn) if os.path.exists(fp) and os.path.getsize(fp) > 0 and not fp.endswith(".part"): return fp except DownloadCancelled: raise except Exception as ytdlp_err: errors.append(f"yt-dlp: {ytdlp_err}") logger.warning(f"[Facebook] Falló fallback con yt-dlp: {ytdlp_err}") detail = " | ".join(errors) if errors else "Servidores no disponibles" raise RuntimeError(f"No se pudo obtener el archivo multimedia de Facebook para este enlace ({detail}).")