""" 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": "*/*", } # Caso 1: blob -> raw 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 # Caso 2: raw en github.com -> raw.githubusercontent.com 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 # Caso 3: releases/tag/{tag} -> archive refs/tags/{tag}.zip 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 # Caso 4: tree/{branch} -> archive/{branch}.zip (soporta cualquier rama o tag) 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 # Caso 5: Repositorio raíz -> código fuente del branch por defecto vía HEAD.zip (funciona en main, master, trunk...) 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}")