File size: 2,568 Bytes
b90715b 7678719 ccb5cde | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | class DownloadCancelled(Exception):
"""Excepción para cancelar descargas en vuelo."""
def __init__(self, message: str = "Descarga cancelada por solicitud del usuario"):
super().__init__(message)
def unwrap_redirect_url(url: str, timeout: int = 6) -> str:
"""
Desempaqueta URLs envueltas o enlaces cortos de redirección (ej. getsnap.link, thedylink.com)
utilizando únicamente cabeceras HTTP nativas y análisis de parámetros estándar.
NO utiliza ninguna API externa ni servicios de terceros.
"""
if not url or not isinstance(url, str):
return url
curr = url.strip()
if not curr.startswith(("http://", "https://")):
return curr
from urllib.parse import parse_qs, urljoin, urlparse
try:
parsed = urlparse(curr)
host = (parsed.hostname or "").lower()
except Exception:
return curr
wrapper_domains = ("getsnap.link", "thedylink.com", "snaptube", "bit.ly", "tinyurl.com", "t.co")
if not any(d in host for d in wrapper_domains):
return curr
import requests
session = requests.Session()
session.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
})
try:
for _ in range(5):
parsed = urlparse(curr)
host = (parsed.hostname or "").lower()
qs = parse_qs(parsed.query)
for key in ("url", "link", "target", "dest"):
val = qs.get(key)
if val and val[0].startswith(("http://", "https://")):
curr = val[0]
break
if not any(d in host for d in wrapper_domains):
break
try:
r = session.get(curr, allow_redirects=False, timeout=timeout)
loc = r.headers.get("Location")
if not loc:
break
if loc.startswith("intent://"):
intent_p = urlparse(loc.replace("intent://", "http://", 1))
intent_qs = parse_qs(intent_p.query)
if intent_qs.get("url"):
curr = intent_qs["url"][0]
break
elif loc.startswith(("http://", "https://")):
curr = loc
else:
curr = urljoin(curr, loc)
except Exception:
break
except Exception:
pass
finally:
session.close()
return curr
|