| import logging |
| import os |
| import re |
| from urllib.parse import quote_plus |
| from typing import List, Union |
| import requests |
|
|
| from plugins.common import DownloadCancelled |
| from plugins.stream import download_stream_url |
|
|
| logger = logging.getLogger("worker.plugins.twitter") |
|
|
|
|
| def download_twitter( |
| url: str, |
| output_dir: str, |
| progress_fn=None, |
| cancel_event=None |
| ) -> Union[str, List[str]]: |
| """ |
| Descarga fotos, videos o gifs de posts de X / Twitter con fallback automático. |
| """ |
| if cancel_event and cancel_event.is_set(): |
| raise DownloadCancelled() |
|
|
| os.makedirs(output_dir, exist_ok=True) |
| delirius_key = os.getenv("DELIRIUS_API_KEY", "") |
|
|
| try: |
| enc_url = quote_plus(url.strip()) |
| api_url = f"https://api.delirius.online/download/twitter?url={enc_url}" |
| if delirius_key: |
| api_url += f"&apikey={delirius_key}" |
|
|
| resp = requests.get(api_url, timeout=25) |
| if resp.status_code == 200: |
| json_data = resp.json() |
| if json_data.get("status") and "data" in json_data: |
| data = json_data["data"] |
| media_list = [] |
|
|
| raw_media = [] |
| if isinstance(data, dict): |
| raw_media = data.get("media") or data.get("result") or [] |
| if not raw_media and (data.get("url") or data.get("download_url")): |
| raw_media = [data] |
| elif isinstance(data, list): |
| raw_media = data |
|
|
| tweet_id = data.get("id") if isinstance(data, dict) else "" |
| author_data = data.get("author") if isinstance(data, dict) else {} |
| username = "" |
| if isinstance(author_data, dict): |
| username = author_data.get("username", "") |
|
|
| base_prefix = f"twitter_{username}_{tweet_id}".strip("_") if (username or tweet_id) else "twitter" |
| clean_base = re.sub(r'[\/\\:\*\?"<>\|\x00-\x1f]', '_', base_prefix)[:50].strip() or "twitter" |
|
|
| for idx, item in enumerate(raw_media, 1): |
| if isinstance(item, dict): |
| m_url = item.get("url") or item.get("download_url") or item.get("dl_url") or item.get("download") |
| m_type = str(item.get("type", "")).lower() |
| else: |
| m_url = str(item) |
| m_type = "" |
|
|
| if not m_url: |
| continue |
|
|
| if "video" in m_type or ".mp4" in m_url.lower(): |
| ext = ".mp4" |
| elif "gif" in m_type or ".gif" in m_url.lower(): |
| ext = ".mp4" if ".mp4" in m_url.lower() else ".gif" |
| elif ".png" in m_url.lower(): |
| ext = ".png" |
| elif ".webp" in m_url.lower(): |
| ext = ".webp" |
| else: |
| ext = ".jpg" |
|
|
| if len(raw_media) > 1: |
| filename = f"{clean_base}_{idx}{ext}" |
| else: |
| filename = f"{clean_base}{ext}" |
|
|
| media_list.append((m_url, filename)) |
|
|
| if media_list: |
| downloaded = [] |
| for idx, (m_url, m_name) in enumerate(media_list): |
| if cancel_event and cancel_event.is_set(): |
| raise DownloadCancelled() |
| target = os.path.join(output_dir, m_name) |
| out_path, _ = download_stream_url( |
| m_url, |
| target, |
| progress_fn=progress_fn, |
| cancel_event=cancel_event |
| ) |
| if os.path.exists(out_path): |
| downloaded.append(out_path) |
| if downloaded: |
| return downloaded[0] if len(downloaded) == 1 else downloaded |
| except Exception as e: |
| logger.warning(f"Descarga directa falló para Twitter/X ({e}), intentando alternativa...") |
|
|
| raise RuntimeError("No se pudo obtener el archivo multimedia de Twitter/X para este enlace.") |
|
|
