| """ |
| Direct Link Generator / Multi-Host Resolver para Hugging Face Workers. |
| Extrae enlaces de descarga directa para KrakenFiles, 1Fichier, SolidFiles, |
| WeTransfer, Linkbox, DoodStream, Streamtape, Upload.ee, EasyUpload, Streamvid, OneDrive, etc. |
| """ |
|
|
| import logging |
| import os |
| import re |
| from typing import Dict, Optional, Tuple |
| from urllib.parse import parse_qs, quote, unquote, urlparse |
|
|
| import cloudscraper |
| from lxml.etree import HTML |
| import requests |
|
|
| logger = logging.getLogger("hf_worker.plugins.direct_link") |
|
|
| REAL_DEBRID_API = os.getenv("REAL_DEBRID_API", "").strip() |
| DEBRID_LINK_API = os.getenv("DEBRID_LINK_API", "").strip() |
|
|
| DEFAULT_TIMEOUT = (15, 30) |
|
|
|
|
| class DirectLinkError(Exception): |
| """Excepción al resolver enlaces directos de servidores multi-host.""" |
| pass |
|
|
|
|
| def _get_scraper() -> cloudscraper.CloudScraper: |
| return cloudscraper.create_scraper( |
| browser={"browser": "chrome", "platform": "linux", "mobile": False} |
| ) |
|
|
|
|
| HOST_MAPPINGS = { |
| "krakenfiles": ["krakenfiles.com"], |
| "1fichier": ["1fichier.com", "alterupload.com", "cjoint.net", "desfichiers.com", "dfichiers.com"], |
| "solidfiles": ["solidfiles.com"], |
| "uploadee": ["upload.ee"], |
| "wetransfer": ["wetransfer.com", "we.tl"], |
| "linkbox": ["linkbox.to", "linkbox.co"], |
| "streamtape": ["streamtape.com", "streamtape.co", "streamtape.cc", "streamtape.to", "streamtape.net", "streamta.pe", "streamtape.xyz"], |
| "doodstream": ["dood.watch", "doodstream.com", "dood.to", "dood.so", "dood.cx", "dood.la", "dood.ws", "dood.sh", "doodstream.co", "dood.pm", "dood.wf", "dood.re", "dood.video", "dooood.com", "dood.yt", "doods.yt", "dood.stream", "doods.pro"], |
| "easyupload": ["easyupload.io"], |
| "streamvid": ["streamvid.net"], |
| "letsupload": ["letsupload.io", "letsupload.cc"], |
| "onedrive": ["1drv.ms"], |
| "filepress": ["filepress.click", "filepress.store"], |
| "github": ["github.com", "raw.githubusercontent.com", "gist.github.com", "gist.githubusercontent.com"], |
| } |
|
|
|
|
| def get_supported_host_name(url: str) -> Optional[str]: |
| if not url or not isinstance(url, str): |
| return None |
| try: |
| domain = (urlparse(url.strip()).hostname or "").lower() |
| except Exception: |
| return None |
|
|
| if not domain: |
| return None |
|
|
| for host_name, domains in HOST_MAPPINGS.items(): |
| if any(domain == d or domain.endswith("." + d) for d in domains): |
| return host_name |
|
|
| if REAL_DEBRID_API or DEBRID_LINK_API: |
| return "debrid" |
|
|
| return None |
|
|
|
|
| def is_supported_direct_host(url: str) -> bool: |
| return get_supported_host_name(url) is not None |
|
|
|
|
| def _resolve_krakenfiles(url: str) -> Tuple[str, Dict[str, str]]: |
| with requests.Session() as session: |
| res = session.get(url, timeout=DEFAULT_TIMEOUT) |
| res.raise_for_status() |
| html = HTML(res.text) |
| post_url = html.xpath('//form[@id="dl-form"]/@action') |
| token = html.xpath('//input[@id="dl-token"]/@value') |
| if not post_url or not token: |
| raise DirectLinkError("KrakenFiles: No se encontró el formulario o token de descarga.") |
|
|
| target = f"https:{post_url[0]}" if post_url[0].startswith("//") else post_url[0] |
| json_resp = session.post(target, data={"token": token[0]}, timeout=DEFAULT_TIMEOUT).json() |
| if json_resp.get("status") != "ok" or not json_resp.get("url"): |
| raise DirectLinkError(f"KrakenFiles: Enlace no disponible ({json_resp.get('status')}).") |
| return json_resp["url"], {} |
|
|
|
|
| def _resolve_1fichier(url: str) -> Tuple[str, Dict[str, str]]: |
| pswd = None |
| clean_url = url |
| if "::" in url: |
| parts = url.split("::") |
| clean_url = parts[0] |
| pswd = parts[1] |
|
|
| scraper = _get_scraper() |
| data = {"pass": pswd} if pswd else {} |
| req = scraper.post(clean_url, data=data, timeout=DEFAULT_TIMEOUT) |
| if req.status_code == 404: |
| raise DirectLinkError("1Fichier: El archivo no existe o fue eliminado.") |
|
|
| html = HTML(req.text) |
| dl_url = html.xpath('//a[@class="ok btn-general btn-orange"]/@href') |
| if dl_url: |
| return dl_url[0], {} |
|
|
| warns = html.xpath('//div[@class="ct_warn"]') |
| for w in warns: |
| txt = (w.text or "").lower() |
| if "you must wait" in txt or "devez attendre" in txt: |
| raise DirectLinkError("1Fichier: Límite temporal alcanzado. Espera unos minutos.") |
| if "protect access" in txt or "password" in txt: |
| raise DirectLinkError("1Fichier: El archivo requiere contraseña (agrega ::password al enlace).") |
|
|
| raise DirectLinkError("1Fichier: No se pudo obtener el enlace de descarga.") |
|
|
|
|
| def _resolve_solidfiles(url: str) -> Tuple[str, Dict[str, str]]: |
| scraper = _get_scraper() |
| res = scraper.get(url, timeout=DEFAULT_TIMEOUT) |
| res.raise_for_status() |
| m = re.search(r'viewerOptions\'\,\s*(.*?)\)\;', res.text) |
| if not m: |
| raise DirectLinkError("SolidFiles: No se pudo localizar la configuración del visor.") |
| import json |
| opts = json.loads(m.group(1)) |
| dl_url = opts.get("downloadUrl") |
| if dl_url: |
| return dl_url, {} |
| raise DirectLinkError("SolidFiles: Enlace de descarga no encontrado.") |
|
|
|
|
| def _resolve_uploadee(url: str) -> Tuple[str, Dict[str, str]]: |
| scraper = _get_scraper() |
| res = scraper.get(url, timeout=DEFAULT_TIMEOUT) |
| res.raise_for_status() |
| html = HTML(res.text) |
| link = html.xpath("//a[@id='d_l']/@href") |
| if link: |
| return link[0], {} |
| raise DirectLinkError("Upload.ee: Enlace de descarga no encontrado.") |
|
|
|
|
| def _resolve_wetransfer(url: str) -> Tuple[str, Dict[str, str]]: |
| scraper = _get_scraper() |
| resp = scraper.get(url, timeout=DEFAULT_TIMEOUT) |
| final_url = resp.url |
| parts = final_url.rstrip("/").split("/") |
| sec_hash = parts[-1] |
| transfer_id = parts[-2] |
| api_url = f"https://wetransfer.com/api/v4/transfers/{transfer_id}/download" |
| res = scraper.post( |
| api_url, |
| json={"security_hash": sec_hash, "intent": "entire_transfer"}, |
| timeout=DEFAULT_TIMEOUT, |
| ).json() |
| if "direct_link" in res: |
| return res["direct_link"], {} |
| msg = res.get("message") or res.get("error") or "enlace no disponible" |
| raise DirectLinkError(f"WeTransfer: {msg}") |
|
|
|
|
| def _resolve_linkbox(url: str) -> Tuple[str, Dict[str, str]]: |
| scraper = _get_scraper() |
| item_id = url.rstrip("/").split("/")[-1] |
| api = f"https://www.linkbox.to/api/file/detail?itemId={item_id}" |
| res = scraper.get(api, timeout=DEFAULT_TIMEOUT).json() |
| data = res.get("data") or {} |
| item_info = data.get("itemInfo") or {} |
| raw_url = item_info.get("url") |
| if not raw_url: |
| raise DirectLinkError("Linkbox: Archivo no disponible o URL no encontrada.") |
| name = quote(item_info.get("name", "archivo")) |
| endpoint_path = raw_url.split("/", 3)[-1] |
| return f"https://wdl.nuplink.net/{endpoint_path}&filename={name}", {} |
|
|
|
|
| def _resolve_streamtape(url: str) -> Tuple[str, Dict[str, str]]: |
| parts = url.rstrip("/").split("/") |
| video_id = parts[4] if len(parts) >= 6 else parts[-1] |
| with requests.Session() as session: |
| res = session.get(url, timeout=DEFAULT_TIMEOUT) |
| res.raise_for_status() |
| html = HTML(res.text) |
| script = html.xpath("//script[contains(text(),'ideoooolink')]/text()") |
| if not script: |
| raise DirectLinkError("Streamtape: No se encontró script de video.") |
| link = re.findall(r"(&expires\S+)'", script[0]) |
| if not link: |
| raise DirectLinkError("Streamtape: Parámetro de expiración no encontrado.") |
| direct_url = f"https://streamtape.com/get_video?id={video_id}{link[-1]}" |
| return direct_url, {"Referer": "https://streamtape.com/"} |
|
|
|
|
| def _resolve_doodstream(url: str) -> Tuple[str, Dict[str, str]]: |
| if "/e/" in url: |
| url = url.replace("/e/", "/d/") |
| parsed = urlparse(url) |
| scraper = _get_scraper() |
| res = scraper.get(url, timeout=DEFAULT_TIMEOUT) |
| html = HTML(res.text) |
| btn = html.xpath("//div[@class='download-content']//a/@href") |
| if not btn: |
| raise DirectLinkError("DoodStream: Botón de descarga no disponible en la página.") |
| token_url = f"{parsed.scheme}://{parsed.hostname}{btn[0]}" |
| res_token = scraper.get(token_url, timeout=DEFAULT_TIMEOUT) |
| m = re.search(r"window\.open\('(\S+)'", res_token.text) |
| if not m: |
| raise DirectLinkError("DoodStream: Enlace final no encontrado.") |
| direct_url = m.group(1) |
| return direct_url, {"Referer": f"{parsed.scheme}://{parsed.hostname}/"} |
|
|
|
|
| def _resolve_onedrive(url: str) -> Tuple[str, Dict[str, str]]: |
| scraper = _get_scraper() |
| resp = scraper.get(url, timeout=DEFAULT_TIMEOUT) |
| final_url = resp.url |
| q = parse_qs(urlparse(final_url).query) |
| resid = q.get("resid") |
| authkey = q.get("authkey") |
| if not resid or not authkey: |
| raise DirectLinkError("OneDrive: Enlace 1drv.ms no contiene resid o authkey válidos.") |
| drive_id = resid[0].split("!", 1)[0] |
| api_url = f"https://api.onedrive.com/v1.0/drives/{drive_id}/items/{resid[0]}?$select=id,@content.downloadUrl&authKey={authkey[0]}" |
| meta = scraper.get(api_url, timeout=DEFAULT_TIMEOUT).json() |
| dl = meta.get("@content.downloadUrl") |
| if dl: |
| return dl, {} |
| raise DirectLinkError("OneDrive: No se obtuvo @content.downloadUrl de la API.") |
|
|
|
|
| def _resolve_github(url: str) -> Tuple[str, Dict[str, str]]: |
| """ |
| Normaliza y resuelve URLs de GitHub a enlaces directos de descarga: |
| - blob/ -> raw.githubusercontent.com (evita descargar la página HTML) |
| - raw/ -> raw.githubusercontent.com |
| - tree/ -> archive .zip de la rama |
| - repo root -> archive .zip de la rama main |
| - releases/download/... -> enlace directo asegurando User-Agent de navegador |
| """ |
| clean = str(url or "").strip() |
| parsed = urlparse(clean) |
| path = parsed.path.strip("/") |
| parts = path.split("/") |
|
|
| headers = { |
| "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": "*/*", |
| } |
|
|
| |
| if len(parts) >= 4 and parts[2] == "blob": |
| owner, repo = parts[0], parts[1] |
| branch = parts[3] |
| file_path = "/".join(parts[4:]) |
| return f"https://raw.githubusercontent.com/{owner}/{repo}/{branch}/{file_path}", headers |
|
|
| |
| if len(parts) >= 4 and parts[2] == "raw": |
| owner, repo = parts[0], parts[1] |
| branch = parts[3] |
| file_path = "/".join(parts[4:]) |
| return f"https://raw.githubusercontent.com/{owner}/{repo}/{branch}/{file_path}", headers |
|
|
| |
| if len(parts) >= 5 and parts[2] == "releases" and parts[3] == "tag": |
| owner, repo = parts[0], parts[1] |
| tag = parts[4] |
| return f"https://github.com/{owner}/{repo}/archive/refs/tags/{tag}.zip", headers |
|
|
| |
| if len(parts) >= 4 and parts[2] == "tree": |
| owner, repo = parts[0], parts[1] |
| branch = "/".join(parts[3:]) |
| return f"https://github.com/{owner}/{repo}/archive/{branch}.zip", headers |
|
|
| |
| if len(parts) == 2 and parts[0] and parts[1]: |
| owner, repo = parts[0], parts[1] |
| return f"https://github.com/{owner}/{repo}/archive/HEAD.zip", headers |
|
|
| return clean, headers |
|
|
|
|
| def resolve_direct_link(url: str) -> Tuple[str, Dict[str, str]]: |
| clean_url = str(url or "").strip() |
| host_name = get_supported_host_name(clean_url) |
|
|
| if host_name == "github": |
| return _resolve_github(clean_url) |
| elif host_name == "krakenfiles": |
| return _resolve_krakenfiles(clean_url) |
| elif host_name == "1fichier": |
| return _resolve_1fichier(clean_url) |
| elif host_name == "solidfiles": |
| return _resolve_solidfiles(clean_url) |
| elif host_name == "uploadee": |
| return _resolve_uploadee(clean_url) |
| elif host_name == "wetransfer": |
| return _resolve_wetransfer(clean_url) |
| elif host_name == "linkbox": |
| return _resolve_linkbox(clean_url) |
| elif host_name == "streamtape": |
| return _resolve_streamtape(clean_url) |
| elif host_name == "doodstream": |
| return _resolve_doodstream(clean_url) |
| elif host_name == "onedrive": |
| return _resolve_onedrive(clean_url) |
|
|
| if REAL_DEBRID_API: |
| try: |
| scraper = _get_scraper() |
| resp = scraper.post( |
| f"https://api.real-debrid.com/rest/1.0/unrestrict/link?auth_token={REAL_DEBRID_API}", |
| data={"link": clean_url}, |
| timeout=DEFAULT_TIMEOUT, |
| ).json() |
| if "download" in resp: |
| return resp["download"], {} |
| except Exception as e: |
| logger.warning(f"Error en Real-Debrid para {clean_url}: {e}") |
|
|
| if DEBRID_LINK_API: |
| try: |
| scraper = _get_scraper() |
| resp = scraper.post( |
| f"https://debrid-link.com/api/v2/downloader/add?access_token={DEBRID_LINK_API}", |
| data={"url": clean_url}, |
| timeout=DEFAULT_TIMEOUT, |
| ).json() |
| if resp.get("success") and isinstance(resp.get("value"), dict): |
| return resp["value"]["downloadUrl"], {} |
| except Exception as e: |
| logger.warning(f"Error en Debrid-Link para {clean_url}: {e}") |
|
|
| raise DirectLinkError(f"No hay extractor disponible para {clean_url}") |
|
|
