253 lines
9.3 KiB
Python
253 lines
9.3 KiB
Python
"""
|
||
videoclub.py — Listado diario de videos disponibles en el catalogo RealDebrid
|
||
de Torrentio (addon de Stremio).
|
||
|
||
Consultas:
|
||
https://torrentio.strem.fun/<config>/catalog/other/torrentio-realdebrid.json
|
||
|
||
Hace snapshot diario, detecta novedades respecto al dia anterior y genera
|
||
un listado markdown. Con --notify envia resumen por Telegram.
|
||
"""
|
||
import argparse
|
||
import json
|
||
import re
|
||
import sys
|
||
import urllib.request
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
BASE = Path(__file__).resolve().parent
|
||
sys.path.insert(0, str(BASE))
|
||
|
||
import config
|
||
|
||
CATALOGO = "https://torrentio.strem.fun/{config}/catalog/other/torrentio-realdebrid.json"
|
||
|
||
# Patrones para clasificar titulos
|
||
YEAR_RE = re.compile(r"\((19\d\d|20\d\d)\)")
|
||
YEAR_BARE_RE = re.compile(r"\.(19\d\d|20\d\d)\.")
|
||
SERIE_RE = re.compile(r"(S\d{1,2}E\d{1,3}|Cap[\.\s]|cap[\.\s]|Temporada|temp\.|\bS0?1E\d|\dx\d\d)", re.IGNORECASE)
|
||
EPISODE_RE = re.compile(r"\.S\d{1,2}E\d{1,3}\.", re.IGNORECASE)
|
||
ADULTO_RE = re.compile(r"(\.XXX\.|XXX\.|private\.|brazzers|onlytarts|spankbang|xhamster|porntrex|missav|sexandsubmission|evilangel|monstersofcock|itsanal|thelifeerotic)", re.IGNORECASE)
|
||
SERIE_PALABRAS = {"star trek", "stargate", "heidi", "euphoria", "from", "the pitt",
|
||
"supergirl", "monarch", "paradise", "desaparecida", "vanished",
|
||
"strange new worlds", "star trek strange", "the studio",
|
||
"one punch man", "conan", "detective conan"}
|
||
EXTRA_SUCIO = ["wolfmax4k.com", "pctfenix", "pctfenix.com", "yts", "yts.mx",
|
||
"rarbg", "verpeliculasonline", "viruse", "proyecto", "kikorip",
|
||
"camrip", "telesync", "hdtv", "bluray", "remux", "webrip", "bdrip",
|
||
"repack", "proper", "xtrem"]
|
||
|
||
|
||
def clean_filename(name):
|
||
"""Limpia un nombre de fichero y devuelve (titulo_limpio, year, tipo)."""
|
||
name = re.sub(r"\.(mkv|mp4|avi|mov|webm|m4v)$", "", name.strip(), flags=re.IGNORECASE)
|
||
year = None
|
||
m = YEAR_RE.search(name)
|
||
if m:
|
||
year = m.group(1)
|
||
name = name.replace(m.group(0), " ")
|
||
else:
|
||
m = YEAR_BARE_RE.search(name)
|
||
if m:
|
||
year = m.group(1)
|
||
name = name.replace(m.group(0), " ")
|
||
|
||
es_serie = bool(SERIE_RE.search(name)) or any(p in name.lower() for p in SERIE_PALABRAS)
|
||
es_adulto = bool(ADULTO_RE.search(name))
|
||
|
||
# Quitar etiquetas entre corchetes y otros ruidos
|
||
name = re.sub(r"\[[^\]]*\]", " ", name)
|
||
name = re.sub(r"[\(\)\[\]]", " ", name)
|
||
name = name.replace("_", " ").replace(".", " ")
|
||
# Colapsar espacios
|
||
name = re.sub(r"\s+", " ", name).strip()
|
||
# Quitar palabras sueltas de calidad/formato residuales
|
||
tokens = [t for t in name.split(" ") if t.lower() not in EXTRA_SUCIO]
|
||
name = " ".join(tokens).strip(" -–")
|
||
name = re.sub(r"\s+", " ", name).strip()
|
||
|
||
tipo = "adulto" if es_adulto else ("serie" if es_serie else "pelicula")
|
||
return name, year, tipo
|
||
|
||
|
||
def fetch_catalogo():
|
||
"""Descarga el catalogo y devuelve lista de items [{id,name,tipo,clean,year,enlace}]."""
|
||
url = CATALOGO.format(config=config.TORRENTIO_CONFIG_WITH_TOKEN)
|
||
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
|
||
with urllib.request.urlopen(req, timeout=40) as r:
|
||
data = json.loads(r.read().decode("utf-8"))
|
||
|
||
# Enlaces directos de descarga de RealDebrid (1 sola llamada al listado)
|
||
enlaces = {}
|
||
try:
|
||
req2 = urllib.request.Request(
|
||
"https://api.real-debrid.com/rest/1.0/torrents",
|
||
headers={"Authorization": f"Bearer {config.TORRENTIO_RD_TOKEN}"})
|
||
with urllib.request.urlopen(req2, timeout=40) as r2:
|
||
for t in json.loads(r2.read().decode("utf-8")):
|
||
links = t.get("links") or []
|
||
if links:
|
||
enlaces[t["id"]] = links[0]
|
||
except Exception as e:
|
||
print(f"[RD] No se pudieron obtener enlaces: {e}")
|
||
|
||
items = []
|
||
for m in data.get("metas", []):
|
||
clean, year, tipo = clean_filename(m.get("name", ""))
|
||
rd_id = m.get("id", "").split(":", 1)[-1]
|
||
items.append({
|
||
"id": m.get("id"),
|
||
"name": m.get("name"),
|
||
"clean": clean,
|
||
"year": year,
|
||
"tipo": tipo,
|
||
"enlace": enlaces.get(rd_id),
|
||
})
|
||
return items
|
||
|
||
|
||
def cargar_snapshot(fecha):
|
||
p = BASE / "snapshots" / f"{fecha}.json"
|
||
if not p.exists():
|
||
return None
|
||
with open(p, encoding="utf-8") as f:
|
||
return json.load(f)
|
||
|
||
|
||
def guardar_snapshot(fecha, items):
|
||
p = BASE / "snapshots" / f"{fecha}.json"
|
||
with open(p, "w", encoding="utf-8") as f:
|
||
json.dump({"fecha": fecha, "items": items}, f, ensure_ascii=False, indent=2)
|
||
|
||
|
||
def diff_novedades(hoy, ayer):
|
||
"""Devuelve items de hoy que no estaban en el snapshot de ayer."""
|
||
if not ayer:
|
||
return hoy
|
||
ids_ayer = {i["id"] for i in ayer["items"]}
|
||
return [i for i in hoy if i["id"] not in ids_ayer]
|
||
|
||
|
||
def agrupar(items):
|
||
g = {"pelicula": [], "serie": [], "adulto": [], "otro": []}
|
||
for it in items:
|
||
tipo = it["tipo"]
|
||
if tipo not in g:
|
||
tipo = "otro"
|
||
g[tipo].append(it)
|
||
for k in g:
|
||
g[k].sort(key=lambda x: (x["clean"] or "").lower())
|
||
return g
|
||
|
||
|
||
def generar_markdown(fecha, items, novedades):
|
||
g = agrupar(items)
|
||
gn = agrupar(novedades)
|
||
lineas = [f"# Videoclub — {fecha}", ""]
|
||
lineas.append(f"**Total disponibles: {len(items)}** | "
|
||
f"Novedades: {len(novedades)}")
|
||
lineas.append("")
|
||
for seccion, clave in [("Novedades", None)]:
|
||
if novedades:
|
||
lineas.append(f"## Novedades ({len(novedades)})")
|
||
lineas.append("")
|
||
for it in novedades:
|
||
marca = {"pelicula": "🎬", "serie": "📺", "adulto": "🔞", "otro": "📁"}.get(it["tipo"], "📁")
|
||
y = f" ({it['year']})" if it["year"] else ""
|
||
link = f" `{it['enlace']}`" if it.get("enlace") else ""
|
||
lineas.append(f"- {marca} {it['clean']}{y}{link}")
|
||
lineas.append("")
|
||
for nombre, clave in [("Películas", "pelicula"), ("Series", "serie"),
|
||
("Adulto", "adulto"), ("Otros", "otro")]:
|
||
if g[clave]:
|
||
lineas.append(f"## {nombre} ({len(g[clave])})")
|
||
lineas.append("")
|
||
for it in g[clave]:
|
||
y = f" ({it['year']})" if it["year"] else ""
|
||
link = f" `{it['enlace']}`" if it.get("enlace") else ""
|
||
lineas.append(f"- {it['clean']}{y}{link}")
|
||
lineas.append("")
|
||
return "\n".join(lineas)
|
||
|
||
|
||
def enlace_realdebrid(rd_id):
|
||
"""Devuelve el enlace directo de descarga para un id de torrent RD."""
|
||
import urllib.parse
|
||
rd_id = rd_id.split(":", 1)[-1]
|
||
tok = config.TORRENTIO_RD_TOKEN
|
||
url = f"https://api.real-debrid.com/rest/1.0/torrents/info/{urllib.parse.quote(rd_id)}"
|
||
req = urllib.request.Request(url, headers={"Authorization": f"Bearer {tok}"})
|
||
with urllib.request.urlopen(req, timeout=30) as r:
|
||
info = json.loads(r.read().decode("utf-8"))
|
||
links = info.get("links") or []
|
||
if not links:
|
||
print(f"[RD] Sin enlace descargable para {rd_id}")
|
||
return None
|
||
return links[0]
|
||
|
||
|
||
def buscar_item(items, termino):
|
||
"""Busca un item del catalogo por titulo (case-insensitive) o por id."""
|
||
t = termino.strip().lower()
|
||
for it in items:
|
||
if it["id"].lower() == t or it["id"].lower().replace("realdebrid:", "") == t:
|
||
return it
|
||
if t in (it.get("clean") or "").lower() or t in (it.get("name") or "").lower():
|
||
return it
|
||
return None
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description="Videoclub: listado diario RealDebrid")
|
||
ap.add_argument("--notify", action="store_true",
|
||
help="Enviar resumen de novedades por Telegram")
|
||
ap.add_argument("--fecha", help="Fecha del snapshot (YYYY-MM-DD), por defecto hoy")
|
||
ap.add_argument("--descargar", metavar="ID|TITULO",
|
||
help="Anadir un titulo del catalogo a Download Station del NAS")
|
||
args = ap.parse_args()
|
||
|
||
if args.descargar:
|
||
items = fetch_catalogo()
|
||
it = buscar_item(items, args.descargar)
|
||
if not it:
|
||
print(f"[NAS] No se encontro '{args.descargar}' en el catalogo")
|
||
return 1
|
||
enlace = enlace_realdebrid(it["id"])
|
||
if not enlace:
|
||
return 1
|
||
print(f"[NAS] {it['clean']} -> {enlace[:80]}")
|
||
import nas_dl
|
||
ok = nas_dl.add_torrent(enlace)
|
||
return 0 if ok else 1
|
||
|
||
fecha = args.fecha or datetime.now().strftime("%Y-%m-%d")
|
||
items = fetch_catalogo()
|
||
# Snapshot de referencia: el mas reciente con fecha distinta de hoy
|
||
ref = None
|
||
snaps = sorted((BASE / "snapshots").glob("*.json"))
|
||
for s in reversed(snaps):
|
||
if s.stem != fecha:
|
||
with open(s, encoding="utf-8") as f:
|
||
ref = json.load(f)
|
||
break
|
||
guardar_snapshot(fecha, items)
|
||
novedades = diff_novedades(items, ref)
|
||
|
||
md = generar_markdown(fecha, items, novedades)
|
||
listado = BASE / "listados" / f"{fecha}.md"
|
||
listado.write_text(md, encoding="utf-8")
|
||
|
||
print(f"Videoclub {fecha}: {len(items)} disponibles, {len(novedades)} novedades")
|
||
print(f"Listado: {listado}")
|
||
|
||
if args.notify:
|
||
from tg_notify import enviar_novedades
|
||
enviar_novedades(fecha, items, novedades)
|
||
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|