| 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) |
|
|
| |
| 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}") |
|
|
| |
| 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}).") |
|
|
