2297 lines
95 KiB
Python
2297 lines
95 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
AETN Video Downloader - Web App
|
||
Supports: Lifetime Movie Club, History Vault, A&E Crime Central, Lifetime
|
||
Run: python3 app.py
|
||
Open: http://localhost:5000
|
||
"""
|
||
|
||
import os, re, time, json, uuid, threading, subprocess, glob, sys, struct, importlib.util, csv
|
||
from datetime import datetime, timezone
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
from html import unescape
|
||
from urllib.parse import urljoin, urlparse
|
||
from flask import Flask, render_template, request, jsonify
|
||
from selenium import webdriver
|
||
from selenium.webdriver import ActionChains
|
||
from selenium.webdriver.firefox.options import Options
|
||
from selenium.webdriver.common.keys import Keys
|
||
|
||
app = Flask(__name__)
|
||
|
||
# ── HBO Max (Max) constants ────────────────────────────────────────────────────
|
||
_SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
HBO_MAX_PROFILE_DIR = os.path.join(_SCRIPT_DIR, "hbo-max-downloader", "chrome_profile_hbomax")
|
||
HBO_MAX_FIRST_RUN = os.path.join(HBO_MAX_PROFILE_DIR, ".first_run_done")
|
||
HBO_MAX_BIN_DIR = os.path.join(_SCRIPT_DIR, "hbo-max-downloader", "bin")
|
||
HBO_MAX_AUTH_URL = "https://auth.max.com/"
|
||
HBOMAX_CDP_BLOCKED = ["*.png","*.jpg","*.jpeg","*.gif","*.webp","*.svg","*.woff","*.woff2","*.ttf"]
|
||
# Shared DRM decrypt tools (Max, Discovery+ both use Widevine/CENC). Real Linux binaries — the
|
||
# original hbo-max-downloader script called Windows .exe files that don't exist in this container.
|
||
MP4DECRYPT_BIN = "/usr/local/bin/mp4decrypt"
|
||
DRM_TOOLS_OK = os.path.exists(MP4DECRYPT_BIN)
|
||
|
||
# ── Storage ────────────────────────────────────────────────────────────────────
|
||
HISTORY_FILE = os.path.expanduser("~/.aetn_downloader_history.json")
|
||
QUEUE_FILE = os.path.expanduser("~/.aetn_downloader_queue.json")
|
||
downloads = {}
|
||
history = []
|
||
download_semaphore = threading.Semaphore(3) # max 3 concurrent downloads total
|
||
browser_semaphore = threading.Semaphore(1) # max 1 concurrent Firefox session
|
||
|
||
def load_history():
|
||
global history
|
||
if os.path.exists(HISTORY_FILE):
|
||
try:
|
||
with open(HISTORY_FILE) as f:
|
||
history = json.load(f)
|
||
except:
|
||
history = []
|
||
|
||
state_lock = threading.Lock() # every download thread saves state; unlocked writes corrupted the file
|
||
|
||
def write_json_atomic(path, data, **dump_args):
|
||
"""Write to a temp file, then swap it in, so a crash or restart mid-write can't leave a broken file."""
|
||
with state_lock:
|
||
tmp = f"{path}.{os.getpid()}.{threading.get_ident()}.tmp" # unique per process, so two can't clash
|
||
with open(tmp, "w") as f:
|
||
json.dump(data, f, **dump_args)
|
||
os.replace(tmp, path)
|
||
|
||
def save_history():
|
||
write_json_atomic(HISTORY_FILE, history[-200:], indent=2)
|
||
|
||
def save_queue():
|
||
queue = []
|
||
# copies: download threads add and change entries while this runs
|
||
for d in list(downloads.values()):
|
||
entry = {k: v for k, v in list(d.items())
|
||
if k not in ('process', 'subtitle_process', 'video_drm_key', 'audio_drm_key')}
|
||
queue.append(entry)
|
||
write_json_atomic(QUEUE_FILE, queue)
|
||
|
||
def load_queue():
|
||
if not os.path.exists(QUEUE_FILE):
|
||
return
|
||
try:
|
||
with open(QUEUE_FILE) as f:
|
||
queue = json.load(f)
|
||
threads = []
|
||
for entry in queue:
|
||
if entry.get('status') in ('complete', 'cancelled'):
|
||
continue
|
||
entry['status'] = 'queued'
|
||
entry['progress'] = 0
|
||
entry['message'] = 'Requeued after restart'
|
||
entry['speed'] = '--'
|
||
entry['eta'] = '--'
|
||
downloads[entry['id']] = entry
|
||
threads.append(threading.Thread(
|
||
target=run_download,
|
||
args=(entry['id'], entry['url'], entry['output_path'], entry['site_type']),
|
||
daemon=True
|
||
))
|
||
for thread in threads: # started only once every entry is in place
|
||
thread.start()
|
||
restored = len(threads)
|
||
if restored:
|
||
print(f" Restored {restored} download(s) from previous session")
|
||
except Exception as e:
|
||
print(f" Could not restore queue: {e}")
|
||
|
||
load_history()
|
||
|
||
# ── Site detection ─────────────────────────────────────────────────────────────
|
||
def detect_site(url):
|
||
if "lifetimemovieclub.com" in url:
|
||
return "svod", "Lifetime Movie Club"
|
||
elif "historyvault.com" in url:
|
||
return "svod", "History Vault"
|
||
elif "aecrimecentral.com" in url:
|
||
return "svod", "A&E Crime Central"
|
||
elif "mylifetime.com" in url:
|
||
return "lifetime", "Lifetime"
|
||
elif "play.history.com" in url:
|
||
return "lifetime", "History"
|
||
elif "primevideo.com" in url or ("amazon.com" in url and "/gp/video" in url):
|
||
return "ytdlp", "Prime Video"
|
||
elif "max.com" in url or "hbomax.com" in url:
|
||
return "hbomax", "Max"
|
||
elif "discoveryplus.com" in url:
|
||
return "discoveryplus", "Discovery+"
|
||
return None, None
|
||
|
||
def parse_title(url):
|
||
# HBO Max URLs contain UUIDs — use a placeholder; title updated after page load
|
||
if "max.com" in url or "hbomax.com" in url:
|
||
uid = url.rstrip('/').split('/')[-1][:8]
|
||
title = f"Max Video {uid}"
|
||
return title, f"Max_Video_{uid}"
|
||
# Episode URLs (/shows/<show>/season-1/episode-2) end in "episode-2", so name by the show
|
||
show = re.search(r'/shows/([^/]+)/season-\d+/episode-\d+', url)
|
||
if show:
|
||
title = ' '.join(w.capitalize() for w in show.group(1).split('-'))
|
||
return title, title.replace(' ', '_')
|
||
# play.discoveryplus.com/video/watch/<uuid> — no title in the URL; updated after page load
|
||
watch = re.search(r'discoveryplus\.com/video/watch/([a-f0-9-]+)', url)
|
||
if watch:
|
||
uid = watch.group(1)[:8]
|
||
title = f"Discovery Video {uid}"
|
||
return title, f"Discovery_Video_{uid}"
|
||
# Older-style Discovery+ URLs are /video/<show-slug>/<episode-slug> — combine for a real title
|
||
dplus = re.match(r'.*discoveryplus\.com/(?:[a-z]{2}/)?video/([^/]+)/([^/?#]+)', url)
|
||
if dplus:
|
||
show, episode = (' '.join(w.capitalize() for w in part.split('-')) for part in dplus.groups())
|
||
title = f"{show} - {episode}"
|
||
return title, title.replace(' ', '_')
|
||
slug = re.sub(r'/$', '', url)
|
||
slug = re.sub(r'/full-movie|/full-episode', '', slug)
|
||
slug = slug.split('/')[-1].replace('-', ' ')
|
||
title = ' '.join(w.capitalize() for w in slug.split())
|
||
safe = title.replace(' ', '_')
|
||
return title, safe
|
||
|
||
def get_episode_tag(url):
|
||
season = re.search(r'season-(\d+)', url)
|
||
episode = re.search(r'episode-(\d+)', url)
|
||
if season and episode:
|
||
return f"S{int(season.group(1)):02d}E{int(episode.group(1)):02d}"
|
||
return None
|
||
|
||
# ── Library naming + comparison (Plex style: "Title (Year).mp4") ───────────────
|
||
MEDIA_ROOT = os.environ.get("AETV_MEDIA_ROOT", "/mnt/storage")
|
||
LIST_DIR = os.environ.get("AETV_LIST_DIR", os.path.join(_SCRIPT_DIR, "lmc-scraper"))
|
||
# Where each source's movies and TV episodes are saved (folders under MEDIA_ROOT)
|
||
SOURCES = {
|
||
"lifetime": {"label": "Lifetime", "movies": "Lifetime", "tv": "LIFETIME_TV"},
|
||
"history": {"label": "History", "movies": "History", "tv": "History_TV"},
|
||
"crime": {"label": "Crime Central", "movies": "Crime_Central", "tv": "CRIME_CENTRAL_TV"},
|
||
"discovery": {"label": "Discovery+", "movies": "Discovery", "tv": "DISCOVERY_TV"},
|
||
}
|
||
# Episode lists written by lmc-scraper/extract-svod-urls.py
|
||
EPISODE_FILES = {
|
||
"lifetime": ["lifetimemovieclub_episodes.csv"],
|
||
"history": ["historyvault_episodes.csv", "playhistory_episodes.csv"],
|
||
"crime": ["aecrimecentral_episodes.csv"],
|
||
"discovery": [], # no episode scraper yet — Show Check reports 0 shows instead of crashing
|
||
}
|
||
META_CACHE_FILE = os.path.expanduser("~/.aetn_movie_meta.json")
|
||
VIDEO_EXTS = (".mp4", ".mkv", ".m4v", ".avi", ".mov")
|
||
meta_cache = {}
|
||
meta_lock = threading.Lock() # the library check reads pages from several threads at once
|
||
|
||
def load_meta_cache():
|
||
global meta_cache
|
||
try:
|
||
with open(META_CACHE_FILE) as f:
|
||
meta_cache = json.load(f)
|
||
except (OSError, ValueError):
|
||
meta_cache = {}
|
||
|
||
def save_meta_cache():
|
||
with open(META_CACHE_FILE, "w") as f:
|
||
json.dump(meta_cache, f, indent=2)
|
||
|
||
load_meta_cache()
|
||
|
||
def norm_title(text):
|
||
text = re.sub(r"['’]", "", text.lower().replace("&", " and "))
|
||
text = re.sub(r"[^a-z0-9]+", " ", text).strip()
|
||
return re.sub(r"^the ", "", text)
|
||
|
||
def safe_filename(text):
|
||
text = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "", text)
|
||
return re.sub(r"\s+", " ", text).strip(" .")
|
||
|
||
def movie_filename(title, year=None):
|
||
name = safe_filename(title)
|
||
return f"{name} ({year})" if year else name
|
||
|
||
def parse_library_name(filename):
|
||
"""'A Nanny's Revenge (2024).mp4' -> ('a nannys revenge', 2024). Year is None if absent."""
|
||
stem = os.path.splitext(filename)[0].replace("_", " ")
|
||
m = re.search(r"\s*\((\d{4})\)\s*$", stem)
|
||
if not m:
|
||
return norm_title(stem), None
|
||
# tolerate names like 'Dying To Belong 2021 (2021)' where the year is repeated
|
||
return norm_title(re.sub(rf"\s+{m.group(1)}$", "", stem[:m.start()])), int(m.group(1))
|
||
|
||
def scan_library(folder):
|
||
"""Returns {normalized_title: {years}} for every video file in folder."""
|
||
found = {}
|
||
try:
|
||
names = os.listdir(folder)
|
||
except OSError:
|
||
return found
|
||
for name in names:
|
||
if name.lower().endswith(VIDEO_EXTS):
|
||
key, year = parse_library_name(name)
|
||
found.setdefault(key, set()).add(year)
|
||
return found
|
||
|
||
def source_for(url):
|
||
"""Which SOURCES entry a URL belongs to, or None for sites that have no folder."""
|
||
if "play.history.com" in url or "historyvault.com" in url:
|
||
return "history"
|
||
if "aecrimecentral.com" in url:
|
||
return "crime"
|
||
if "mylifetime.com" in url or "lifetimemovieclub.com" in url:
|
||
return "lifetime"
|
||
if "discoveryplus.com" in url:
|
||
return "discovery"
|
||
return None
|
||
|
||
def source_dir(source, kind):
|
||
return os.path.join(MEDIA_ROOT, SOURCES[source][kind])
|
||
|
||
_episode_cache = {}
|
||
|
||
def load_episodes(source):
|
||
"""Returns ({(show, season, episode): row}, {url: row}) from the scraped episode lists.
|
||
Where two sites list the same episode, the one that needs no login is kept."""
|
||
files = [os.path.join(LIST_DIR, name) for name in EPISODE_FILES[source]]
|
||
stamp = tuple(os.path.getmtime(f) if os.path.exists(f) else 0 for f in files)
|
||
cached = _episode_cache.get(source)
|
||
if cached and cached[0] == stamp:
|
||
return cached[1]
|
||
by_key, by_url = {}, {}
|
||
for path in files:
|
||
try:
|
||
with open(path, newline="") as f:
|
||
reader = list(csv.DictReader(f))
|
||
except OSError:
|
||
continue
|
||
for r in reader:
|
||
try:
|
||
row = {"series": r["series"], "season": int(r["season"]), "episode": int(r["episode"]),
|
||
"title": r["title"], "url": r["url"], "needs_login": r["needs_login"] == "True"}
|
||
except (ValueError, KeyError):
|
||
continue
|
||
by_url[row["url"]] = row
|
||
key = (norm_title(row["series"]), row["season"], row["episode"])
|
||
old = by_key.get(key)
|
||
if old is None or (old["needs_login"] and not row["needs_login"]):
|
||
by_key[key] = row
|
||
_episode_cache[source] = (stamp, (by_key, by_url))
|
||
return by_key, by_url
|
||
|
||
def find_episode(url):
|
||
"""(source, row) for an episode URL found in the scraped lists, else (None, None)."""
|
||
for source in EPISODE_FILES:
|
||
row = load_episodes(source)[1].get(url)
|
||
if row:
|
||
return source, row
|
||
return None, None
|
||
|
||
def episode_relpath(row):
|
||
"""'Show/Season 01/Show - S01E02 - Title.mp4'"""
|
||
show = safe_filename(row["series"])
|
||
name = f"{show} - S{row['season']:02d}E{row['episode']:02d}"
|
||
if row["title"]:
|
||
name += f" - {safe_filename(row['title'])}"
|
||
return os.path.join(show, f"Season {row['season']:02d}", name + ".mp4")
|
||
|
||
def scan_tv_library(folder):
|
||
"""{(show, season, episode)} for every video under folder/<show>/ named like '... S01E02 ...'."""
|
||
have = set()
|
||
try:
|
||
shows = os.listdir(folder)
|
||
except OSError:
|
||
return have
|
||
for show in shows:
|
||
for root, _dirs, files in os.walk(os.path.join(folder, show)):
|
||
for name in files:
|
||
m = re.search(r"[Ss](\d{1,2})[Ee](\d{1,3})", name)
|
||
if m and name.lower().endswith(VIDEO_EXTS):
|
||
have.add((norm_title(show), int(m.group(1)), int(m.group(2))))
|
||
return have
|
||
|
||
def library_status(library, title, year):
|
||
"""'have', 'year_differs' (same title, other year), or 'missing'.
|
||
A year within one of a library file counts as the same movie: sites often use the
|
||
air date, which lands a year off the release year."""
|
||
years = library.get(norm_title(title))
|
||
if not years:
|
||
return "missing"
|
||
if year is None or any(y is None or abs(y - year) <= 1 for y in years):
|
||
return "have"
|
||
return "year_differs"
|
||
|
||
def extract_page_meta(html, url=""):
|
||
"""Reads {'title', 'year'} for the movie out of a page's Next.js Apollo state."""
|
||
m = re.search(r'<script id="__NEXT_DATA__"[^>]*>(.*?)</script>', html, re.S)
|
||
if not m:
|
||
return None
|
||
try:
|
||
state = json.loads(m.group(1)).get("props", {}).get("apolloState", {})
|
||
except ValueError:
|
||
return None
|
||
slug = re.sub(r"/full-[a-z]+$", "", url.rstrip("/")).split("/")[-1] # '/x/full-movie' is the same title as '/x'
|
||
|
||
def pick(prefix):
|
||
found = [v for k, v in state.items() if k.startswith(prefix) and v.get("title")]
|
||
for entity in found:
|
||
if entity.get("slug") == slug:
|
||
return entity
|
||
return found[0] if len(found) == 1 else None
|
||
|
||
movie, special, video = pick("Movie:"), pick("Special:"), pick("Video:")
|
||
entity = movie or special or video
|
||
if not entity:
|
||
return None
|
||
year = None
|
||
pub = (movie or special or {}).get("pubDate") or ""
|
||
if re.match(r"\d{4}", pub):
|
||
year = int(pub[:4])
|
||
elif video and not special and video.get("originalAirDate"):
|
||
# a special's only date is often just when it aired, not when it was made, so it's left out
|
||
year = datetime.fromtimestamp(video["originalAirDate"] / 1000, timezone.utc).year
|
||
return {"title": unescape(entity["title"]), "year": year}
|
||
|
||
COOKIE_FILE = os.path.expanduser("~/.aetn_cookies.json")
|
||
|
||
def saved_cookie_header(url):
|
||
"""Cookie header for url's site from the file save_cookies.sh writes ('' if none)."""
|
||
host = url.split("/")[2]
|
||
try:
|
||
with open(COOKIE_FILE) as f:
|
||
cookies = json.load(f)
|
||
except (OSError, ValueError):
|
||
return ""
|
||
return "; ".join(f"{c['name']}={c['value']}" for c in cookies
|
||
if c.get("domain") and host.endswith(c["domain"].lstrip(".")))
|
||
|
||
def get_movie_meta(url, html=None):
|
||
"""Title + year for a movie URL. Lifetime, History Vault and Crime Central pages are
|
||
public (Crime Central has no year); other sites need a logged-in page, so pass html
|
||
captured from the browser."""
|
||
if url in meta_cache:
|
||
return meta_cache[url]
|
||
if html is None:
|
||
headers = {"User-Agent": "Mozilla/5.0"}
|
||
cookie = saved_cookie_header(url)
|
||
if cookie:
|
||
headers["Cookie"] = cookie
|
||
public = any(site in url for site in ("mylifetime.com", "play.history.com", "historyvault.com",
|
||
"aecrimecentral.com", "lifetimemovieclub.com"))
|
||
if not public and not cookie:
|
||
return None
|
||
try:
|
||
import requests as _req
|
||
html = _req.get(url, headers=headers, timeout=20).text
|
||
except Exception:
|
||
return None
|
||
meta = extract_page_meta(html, url)
|
||
if meta:
|
||
with meta_lock:
|
||
meta_cache[url] = meta
|
||
save_meta_cache()
|
||
return meta
|
||
|
||
def get_firefox_profile():
|
||
appdata = os.environ.get("APPDATA")
|
||
profile_roots = [
|
||
os.path.expanduser("~/Library/Application Support/Firefox/Profiles"),
|
||
os.path.expanduser("~/.mozilla/firefox"),
|
||
]
|
||
if appdata:
|
||
profile_roots.insert(0, os.path.join(appdata, "Mozilla", "Firefox", "Profiles"))
|
||
|
||
profiles = []
|
||
for profile_dir in profile_roots:
|
||
profiles.extend(glob.glob(os.path.join(profile_dir, "*.default-release")))
|
||
profiles.extend(glob.glob(os.path.join(profile_dir, "*.default")))
|
||
return profiles[0] if profiles else None
|
||
|
||
def build_request_headers(referer_url):
|
||
return f"User-Agent: Mozilla/5.0\r\nReferer: {referer_url}\r\n"
|
||
|
||
def probe_stream(m3u8_url, referer_url):
|
||
"""Returns (is_playable, has_audio, has_video)."""
|
||
probe = subprocess.run([
|
||
"ffprobe", "-v", "error",
|
||
"-extension_picky", "0", # ad-stitched playlists have segments without a video file extension
|
||
"-headers", build_request_headers(referer_url),
|
||
"-show_entries", "stream=codec_type",
|
||
"-show_entries", "format=format_name,duration",
|
||
"-of", "json",
|
||
m3u8_url
|
||
], capture_output=True, text=True)
|
||
if probe.returncode != 0:
|
||
return False, False, False
|
||
try:
|
||
data = json.loads(probe.stdout or "{}")
|
||
except Exception:
|
||
return False, False, False
|
||
format_info = data.get("format", {})
|
||
format_name = (format_info.get("format_name") or "").lower()
|
||
is_playable = "hls" in format_name or bool(format_info.get("duration"))
|
||
streams = data.get("streams", [])
|
||
codec_types = {s.get("codec_type") for s in streams}
|
||
return is_playable, "audio" in codec_types, "video" in codec_types
|
||
|
||
def choose_av_streams(candidates, referer_url):
|
||
"""Returns (video_url, audio_url). audio_url may be None if muxed."""
|
||
unique = list(dict.fromkeys(candidates))
|
||
ranked = sorted(unique, key=lambda u: (
|
||
"drm" in u.lower() or "fairplay" in u.lower() or "widevine" in u.lower(),
|
||
"ad" in u.lower() or "ads" in u.lower() or "stitched" in u.lower(),
|
||
))
|
||
|
||
muxed, video_only, audio_only = [], [], []
|
||
for url in ranked:
|
||
ok, has_audio, has_video = probe_stream(url, referer_url)
|
||
if not ok:
|
||
continue
|
||
if has_video and has_audio:
|
||
muxed.append(url)
|
||
elif has_video:
|
||
video_only.append(url)
|
||
elif has_audio:
|
||
audio_only.append(url)
|
||
|
||
if muxed:
|
||
return muxed[0], None
|
||
if video_only:
|
||
return video_only[0], (audio_only[0] if audio_only else None)
|
||
return None, None
|
||
|
||
def nudge_video_player(driver):
|
||
script = """
|
||
const selectors = [
|
||
'button[aria-label*="Play" i]',
|
||
'button[title*="Play" i]',
|
||
'[data-testid*="play" i]',
|
||
'[class*="play" i]',
|
||
'.vjs-big-play-button',
|
||
'video'
|
||
];
|
||
|
||
const clickIfVisible = (el) => {
|
||
if (!el) return false;
|
||
const rect = el.getBoundingClientRect();
|
||
if (rect.width === 0 || rect.height === 0) return false;
|
||
el.scrollIntoView({ block: 'center', inline: 'center' });
|
||
try { el.click(); return true; } catch (e) {}
|
||
try {
|
||
el.dispatchEvent(new MouseEvent('click', { bubbles: true, cancelable: true, view: window }));
|
||
return true;
|
||
} catch (e) {}
|
||
return false;
|
||
};
|
||
|
||
const overlaySelectors = [
|
||
'button[aria-label*="Accept" i]',
|
||
'button[aria-label*="Agree" i]',
|
||
'button[title*="Accept" i]',
|
||
'#onetrust-accept-btn-handler',
|
||
'.onetrust-close-btn-handler',
|
||
'[data-testid*="close" i]',
|
||
'[class*="close" i]'
|
||
];
|
||
|
||
for (const selector of overlaySelectors) {
|
||
const overlay = document.querySelector(selector);
|
||
if (clickIfVisible(overlay)) break;
|
||
}
|
||
|
||
for (const selector of selectors) {
|
||
const el = document.querySelector(selector);
|
||
if (!el) continue;
|
||
if (el.tagName && el.tagName.toLowerCase() === 'video') {
|
||
try {
|
||
el.muted = true;
|
||
const playResult = el.play && el.play();
|
||
if (playResult && typeof playResult.catch === 'function') {
|
||
playResult.catch(() => {});
|
||
}
|
||
} catch (e) {}
|
||
}
|
||
if (clickIfVisible(el)) return true;
|
||
}
|
||
|
||
const media = document.querySelector('video');
|
||
if (media) {
|
||
media.scrollIntoView({ block: 'center', inline: 'center' });
|
||
try {
|
||
media.muted = true;
|
||
const playResult = media.play && media.play();
|
||
if (playResult && typeof playResult.catch === 'function') {
|
||
playResult.catch(() => {});
|
||
}
|
||
return true;
|
||
} catch (e) {}
|
||
}
|
||
return false;
|
||
"""
|
||
try:
|
||
return bool(driver.execute_script(script))
|
||
except Exception:
|
||
return False
|
||
|
||
def collect_m3u8_candidates(driver):
|
||
script = """
|
||
return window.performance.getEntriesByType('resource')
|
||
.map(e => e.name)
|
||
.filter(name => name && name.includes('.m3u8'));
|
||
"""
|
||
try:
|
||
return driver.execute_script(script) or []
|
||
except Exception:
|
||
return []
|
||
|
||
def collect_candidates_from_frames(driver):
|
||
candidates = []
|
||
try:
|
||
candidates.extend(collect_m3u8_candidates(driver))
|
||
frame_count = len(driver.find_elements("tag name", "iframe"))
|
||
for index in range(frame_count):
|
||
try:
|
||
driver.switch_to.default_content()
|
||
frames = driver.find_elements("tag name", "iframe")
|
||
if index >= len(frames):
|
||
continue
|
||
driver.switch_to.frame(frames[index])
|
||
nudge_video_player(driver)
|
||
candidates.extend(collect_m3u8_candidates(driver))
|
||
except Exception:
|
||
continue
|
||
finally:
|
||
try:
|
||
driver.switch_to.default_content()
|
||
except Exception:
|
||
pass
|
||
return list(dict.fromkeys(candidates))
|
||
|
||
def send_playback_keys(driver):
|
||
try:
|
||
body = driver.find_element("tag name", "body")
|
||
body.click()
|
||
body.send_keys(Keys.SPACE)
|
||
body.send_keys("k")
|
||
ActionChains(driver).send_keys(Keys.SPACE).perform()
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
def choose_path(dialog_type):
|
||
try:
|
||
import tkinter as tk
|
||
from tkinter import filedialog
|
||
except Exception:
|
||
return None
|
||
|
||
root = tk.Tk()
|
||
root.withdraw()
|
||
root.attributes("-topmost", True)
|
||
root.lift()
|
||
root.focus_force()
|
||
try:
|
||
if dialog_type == "folder":
|
||
return filedialog.askdirectory(title="Choose download folder", parent=root) or None
|
||
return filedialog.askopenfilename(
|
||
title="Choose a video file",
|
||
parent=root,
|
||
filetypes=[
|
||
("MP4 files", "*.mp4"),
|
||
("MKV files", "*.mkv"),
|
||
("All video files", "*.mp4 *.mkv *.avi *.mov *.m4v"),
|
||
("All files", "*.*"),
|
||
]
|
||
) or None
|
||
finally:
|
||
root.destroy()
|
||
|
||
# ── HBO Max (Chrome / CDP) ─────────────────────────────────────────────────────
|
||
def _build_chrome_driver():
|
||
from selenium.webdriver.chrome.options import Options as ChromeOptions
|
||
from selenium.webdriver.chrome.service import Service as ChromeService
|
||
opts = ChromeOptions()
|
||
opts.add_argument("--disable-gpu")
|
||
opts.add_argument("--no-sandbox")
|
||
opts.add_argument("--window-size=1920,1080")
|
||
opts.add_argument("--lang=en-US")
|
||
opts.add_argument(
|
||
"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"
|
||
)
|
||
opts.add_argument(f"--user-data-dir={HBO_MAX_PROFILE_DIR}")
|
||
opts.add_argument("--profile-directory=Default")
|
||
opts.add_argument("--log-level=3")
|
||
opts.add_argument("--disable-logging")
|
||
opts.add_argument("--no-first-run")
|
||
opts.add_argument("--no-default-browser-check")
|
||
caps = opts.to_capabilities()
|
||
caps["goog:loggingPrefs"] = {"performance": "ALL"}
|
||
try:
|
||
from webdriver_manager.chrome import ChromeDriverManager
|
||
service = ChromeService(ChromeDriverManager().install())
|
||
except ImportError:
|
||
service = ChromeService()
|
||
return webdriver.Chrome(service=service, options=opts)
|
||
|
||
_HBOMAX_UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
||
"AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36")
|
||
_HBOMAX_TARGET_HEIGHT = 1080 # preferred resolution
|
||
|
||
def _hbomax_http_size(session, url):
|
||
"""Return (reachable, size_bytes|None) via HEAD then byte-range fallback."""
|
||
try:
|
||
r = session.head(url, allow_redirects=True, timeout=15)
|
||
if r.status_code in (200, 206):
|
||
cl = r.headers.get("Content-Length")
|
||
return True, int(cl) if cl and cl.strip().isdigit() else None
|
||
if r.status_code in (403, 405, 501):
|
||
gr = session.get(url, headers={"Range": "bytes=0-0"}, timeout=15)
|
||
if gr.status_code in (200, 206):
|
||
cr = gr.headers.get("Content-Range", "")
|
||
if "/" in cr:
|
||
try:
|
||
return True, int(cr.split("/")[-1].strip())
|
||
except Exception:
|
||
pass
|
||
cl = gr.headers.get("Content-Length")
|
||
return True, int(cl) if cl and cl.strip().isdigit() else None
|
||
except Exception:
|
||
pass
|
||
return False, None
|
||
|
||
def _hbomax_parse_dims(blob):
|
||
"""Parse (width, height) from MP4 moov/trak/tkhd boxes in a bytes blob."""
|
||
be32 = lambda b: struct.unpack(">I", b)[0]
|
||
be64 = lambda b: struct.unpack(">Q", b)[0]
|
||
|
||
def iter_boxes(data, start=0, end=None):
|
||
if end is None:
|
||
end = len(data)
|
||
off = start
|
||
while off + 8 <= end:
|
||
size = be32(data[off:off+4]); typ = data[off+4:off+8]
|
||
hdr = 8
|
||
if size == 1:
|
||
if off + 16 > end: return
|
||
size = be64(data[off+8:off+16]); hdr = 16
|
||
box_end = end if size == 0 else off + size
|
||
if box_end > end: return
|
||
yield typ, off + hdr, box_end
|
||
off = box_end
|
||
|
||
for t, ps, pe in iter_boxes(blob):
|
||
if t != b"moov":
|
||
continue
|
||
moov = blob[ps:pe]
|
||
for t2, ps2, pe2 in iter_boxes(moov):
|
||
if t2 != b"trak":
|
||
continue
|
||
trak = moov[ps2:pe2]
|
||
is_video = w = h = None
|
||
for t3, ps3, pe3 in iter_boxes(trak):
|
||
if t3 == b"mdia":
|
||
mdia = trak[ps3:pe3]
|
||
for t4, ps4, pe4 in iter_boxes(mdia):
|
||
if t4 == b"hdlr" and len(mdia) >= ps4 + 8 and mdia[ps4+4:ps4+8] == b"vide":
|
||
is_video = True
|
||
elif t3 == b"tkhd":
|
||
tk = trak[ps3:pe3]
|
||
try:
|
||
ver = tk[0]; off = 4
|
||
off += (16 + 8) if ver == 1 else (8 + 4)
|
||
off += 4 + (8 if ver == 1 else 4) + 8 + 4 + 36
|
||
if off + 8 <= len(tk):
|
||
w = int(struct.unpack(">I", tk[off:off+4])[0] / 65536)
|
||
h = int(struct.unpack(">I", tk[off+4:off+8])[0] / 65536)
|
||
except Exception:
|
||
pass
|
||
if is_video and w and h:
|
||
return w, h
|
||
return None, None
|
||
|
||
def _hbomax_probe_dims(session, url):
|
||
"""Fetch start + end chunks and try to read resolution from moov box."""
|
||
for rng in (f"bytes=0-{2*1024*1024-1}", f"bytes=-{4*1024*1024}"):
|
||
try:
|
||
r = session.get(url, headers={"Range": rng}, timeout=15)
|
||
if r.status_code in (200, 206):
|
||
w, h = _hbomax_parse_dims(r.content)
|
||
if w and h:
|
||
return w, h
|
||
except Exception:
|
||
pass
|
||
return None, None
|
||
|
||
def _auto_select_hbomax(video_urls, audio_urls):
|
||
"""
|
||
Pick video at _HBOMAX_TARGET_HEIGHT (1080p). Falls back to the closest
|
||
resolution below target, then above, then largest file. Also picks the
|
||
largest audio file.
|
||
"""
|
||
import requests as _req
|
||
sess = _req.Session()
|
||
sess.headers.update({"User-Agent": _HBOMAX_UA})
|
||
|
||
# -- Video --
|
||
best_v = None
|
||
if video_urls:
|
||
variants = []
|
||
for url in video_urls:
|
||
ok, size = _hbomax_http_size(sess, url)
|
||
if not ok:
|
||
continue
|
||
w, h = _hbomax_probe_dims(sess, url)
|
||
variants.append({"url": url, "w": w, "h": h, "size": size or 0})
|
||
|
||
if variants:
|
||
target = _HBOMAX_TARGET_HEIGHT
|
||
exact = [v for v in variants if v["h"] == target]
|
||
below = [v for v in variants if v["h"] and v["h"] < target]
|
||
above = [v for v in variants if v["h"] and v["h"] > target]
|
||
unknown = [v for v in variants if not v["h"]]
|
||
|
||
if exact:
|
||
best_v = max(exact, key=lambda v: v["size"])["url"]
|
||
elif below:
|
||
# closest below target, then largest file among that height
|
||
best_h = max(v["h"] for v in below)
|
||
pool = [v for v in below if v["h"] == best_h]
|
||
best_v = max(pool, key=lambda v: v["size"])["url"]
|
||
elif above:
|
||
# lowest above target (prefer closer to 1080p over 4K)
|
||
best_h = min(v["h"] for v in above)
|
||
pool = [v for v in above if v["h"] == best_h]
|
||
best_v = max(pool, key=lambda v: v["size"])["url"]
|
||
else:
|
||
best_v = max(unknown, key=lambda v: v["size"])["url"]
|
||
|
||
# -- Audio: largest file --
|
||
best_a = None
|
||
if audio_urls:
|
||
scored = []
|
||
for url in audio_urls:
|
||
ok, size = _hbomax_http_size(sess, url)
|
||
if ok:
|
||
scored.append((size or 0, url))
|
||
if scored:
|
||
best_a = max(scored, key=lambda x: x[0])[1]
|
||
|
||
return best_v, best_a
|
||
|
||
def _extract_hbomax_title(driver):
|
||
try:
|
||
from selenium.webdriver.support.ui import WebDriverWait
|
||
from selenium.webdriver.common.by import By
|
||
from selenium.webdriver.support import expected_conditions as EC
|
||
xpath = "//*[contains(@class,'StyledVisiblyHiddenLabel')]"
|
||
el = WebDriverWait(driver, 10).until(
|
||
EC.presence_of_element_located((By.XPATH, xpath))
|
||
)
|
||
return (el.get_attribute("textContent") or "").strip()
|
||
except Exception:
|
||
return ""
|
||
|
||
def _format_hbomax_title(raw):
|
||
if not raw:
|
||
return None
|
||
safe = re.sub(r'[\\/:*?"<>|,]', '', raw).strip()
|
||
safe = re.sub(r'\s+', '_', safe).rstrip('.')
|
||
return safe or None
|
||
|
||
def capture_hbomax(url, download_id):
|
||
update_download(download_id, status="capturing", message="Opening Chrome to capture Max stream...")
|
||
|
||
if not os.path.exists(HBO_MAX_FIRST_RUN):
|
||
update_download(download_id, status="error",
|
||
message="Not logged into Max. Use the 'Login to Max' button first.")
|
||
return [], [], None
|
||
|
||
os.makedirs(HBO_MAX_PROFILE_DIR, exist_ok=True)
|
||
driver = None
|
||
try:
|
||
driver = _build_chrome_driver()
|
||
|
||
update_download(download_id, status="capturing", message="Navigating to Max page...")
|
||
driver.get(url)
|
||
|
||
# Wait for page load
|
||
try:
|
||
from selenium.webdriver.support.ui import WebDriverWait
|
||
WebDriverWait(driver, 30).until(
|
||
lambda d: d.execute_script("return document.readyState") == "complete"
|
||
)
|
||
except Exception:
|
||
time.sleep(10)
|
||
|
||
# Try to grab the episode title before starting capture
|
||
episode_title = _extract_hbomax_title(driver)
|
||
|
||
# Enable CDP network capture
|
||
driver.execute_cdp_cmd("Network.enable", {})
|
||
try:
|
||
driver.execute_cdp_cmd("Network.setBlockedURLs", {"urls": HBOMAX_CDP_BLOCKED})
|
||
except Exception:
|
||
pass
|
||
|
||
nudge_video_player(driver)
|
||
|
||
update_download(download_id, status="capturing", message="Capturing Max network traffic (15s)...")
|
||
time.sleep(15)
|
||
|
||
logs = driver.get_log("performance")
|
||
seen, video_urls, audio_urls = set(), [], []
|
||
for entry in logs:
|
||
try:
|
||
msg = json.loads(entry["message"])["message"]
|
||
if msg.get("method") != "Network.requestWillBeSent":
|
||
continue
|
||
req_url = msg.get("params", {}).get("request", {}).get("url", "")
|
||
if req_url in seen:
|
||
continue
|
||
seen.add(req_url)
|
||
if re.search(r'/v\d+\.mp4\b', req_url):
|
||
video_urls.append(req_url)
|
||
elif re.search(r'/a\d+\.mp4\b', req_url):
|
||
audio_urls.append(req_url)
|
||
except Exception:
|
||
pass
|
||
|
||
return video_urls, audio_urls, episode_title
|
||
|
||
except Exception as e:
|
||
update_download(download_id, status="error", message=f"Chrome error: {e}")
|
||
return [], [], None
|
||
finally:
|
||
if driver:
|
||
try:
|
||
driver.quit()
|
||
except Exception:
|
||
pass
|
||
|
||
def _download_mp4(url, path, download_id, label):
|
||
try:
|
||
import requests as _req
|
||
r = _req.get(url, stream=True, timeout=60,
|
||
headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"})
|
||
r.raise_for_status()
|
||
total = int(r.headers.get("content-length", 0))
|
||
done = 0
|
||
with open(path, "wb") as f:
|
||
for chunk in r.iter_content(chunk_size=1024 * 1024):
|
||
if chunk:
|
||
f.write(chunk)
|
||
done += len(chunk)
|
||
if total:
|
||
pct = min(99, done / total * 100)
|
||
update_download(download_id, progress=round(pct, 1),
|
||
size=f"{total/(1024*1024):.1f} MiB",
|
||
message=f"{label}...")
|
||
return True
|
||
except Exception as e:
|
||
update_download(download_id, status="error", message=f"Download error: {e}")
|
||
return False
|
||
|
||
def _drm_decrypt_merge(video_enc, audio_enc, video_key, audio_key, out_path, download_id, label="video"):
|
||
"""Decrypts a Widevine/CENC-encrypted video+audio pair (KID:KEY hex, from an outside tool —
|
||
this app has no way to obtain them itself) and merges them into one playable file."""
|
||
out_dir = os.path.dirname(out_path)
|
||
video_dec = os.path.join(out_dir, "._drm_video_dec.mp4")
|
||
audio_dec = os.path.join(out_dir, "._drm_audio_dec.mp4")
|
||
|
||
update_download(download_id, message=f"Decrypting {label} video...")
|
||
r = subprocess.run([MP4DECRYPT_BIN, "--key", video_key, video_enc, video_dec], capture_output=True)
|
||
if r.returncode != 0:
|
||
update_download(download_id, status="error",
|
||
message=f"Video decrypt failed — check the key: {r.stderr.decode(errors='replace')[:200]}")
|
||
return False
|
||
|
||
update_download(download_id, message=f"Decrypting {label} audio...")
|
||
r = subprocess.run([MP4DECRYPT_BIN, "--key", audio_key, audio_enc, audio_dec], capture_output=True)
|
||
if r.returncode != 0:
|
||
update_download(download_id, status="error",
|
||
message=f"Audio decrypt failed — check the key: {r.stderr.decode(errors='replace')[:200]}")
|
||
return False
|
||
|
||
update_download(download_id, message=f"Merging {label} video and audio...")
|
||
r = subprocess.run([
|
||
"ffmpeg", "-y",
|
||
"-i", video_dec, "-i", audio_dec,
|
||
"-c:v", "copy", "-c:a", "copy",
|
||
"-map", "0:v:0", "-map", "1:a:0",
|
||
out_path
|
||
], capture_output=True)
|
||
|
||
for f in (video_dec, audio_dec):
|
||
try:
|
||
os.remove(f)
|
||
except Exception:
|
||
pass
|
||
|
||
if r.returncode != 0:
|
||
update_download(download_id, status="error", message=f"ffmpeg merge failed for {label}")
|
||
return False
|
||
return True
|
||
|
||
def download_hbomax(url, output_path, download_id, video_drm_key="", audio_drm_key=""):
|
||
video_urls, audio_urls, episode_title = capture_hbomax(url, download_id)
|
||
|
||
if not video_urls:
|
||
if downloads.get(download_id, {}).get("status") != "error":
|
||
update_download(download_id, status="error",
|
||
message="No video URLs captured from Max. Try again or check login.")
|
||
return False
|
||
|
||
# Update title from page if we got one
|
||
if episode_title:
|
||
safe = _format_hbomax_title(episode_title)
|
||
if safe:
|
||
new_path = os.path.join(os.path.dirname(output_path), safe + ".mp4")
|
||
output_path = new_path
|
||
update_download(download_id, title=episode_title, output_path=output_path)
|
||
|
||
best_v, best_a = _auto_select_hbomax(video_urls, audio_urls)
|
||
out_dir = os.path.dirname(output_path)
|
||
stem = os.path.splitext(os.path.basename(output_path))[0]
|
||
has_drm = bool(video_drm_key and audio_drm_key)
|
||
|
||
if has_drm:
|
||
video_tmp = os.path.join(out_dir, f"_{stem}_venc.mp4")
|
||
audio_tmp = os.path.join(out_dir, f"_{stem}_aenc.mp4") if best_a else None
|
||
else:
|
||
video_tmp = os.path.join(out_dir, f"{stem}_video.mp4")
|
||
audio_tmp = os.path.join(out_dir, f"{stem}_audio.mp4") if best_a else None
|
||
|
||
update_download(download_id, status="downloading", message="Downloading Max video...")
|
||
if not _download_mp4(best_v, video_tmp, download_id, "Downloading Max video"):
|
||
return False
|
||
|
||
if best_a and audio_tmp:
|
||
update_download(download_id, message="Downloading Max audio...")
|
||
if not _download_mp4(best_a, audio_tmp, download_id, "Downloading Max audio"):
|
||
return False
|
||
|
||
if has_drm:
|
||
ok = _drm_decrypt_merge(video_tmp, audio_tmp or "", video_drm_key, audio_drm_key, output_path, download_id, label="Max")
|
||
for f in (video_tmp, audio_tmp):
|
||
if f:
|
||
try:
|
||
os.remove(f)
|
||
except Exception:
|
||
pass
|
||
return ok
|
||
else:
|
||
if not best_a:
|
||
try:
|
||
os.rename(video_tmp, output_path)
|
||
except Exception:
|
||
pass
|
||
else:
|
||
update_download(download_id, message=
|
||
"Saved encrypted video+audio separately (no DRM keys). Provide keys to decrypt.")
|
||
return True
|
||
|
||
# ── Discovery+ (DRM, same tier as Max) ──────────────────────────────────────────
|
||
# Discovery+ now streams through the same encrypted backend as Max (media.max.com), so it needs
|
||
# the same "download encrypted, decrypt with a key from an outside tool" flow as Max above — but
|
||
# unlike Max, it doesn't need Chrome: the login is read from the same Firefox profile and cookie
|
||
# file every other site here uses, and yt-dlp itself resolves the manifest into direct file URLs.
|
||
DISCOVERY_TARGET_HEIGHT = 1080
|
||
|
||
def _discovery_cookie_domains(cookies):
|
||
"""Cookie domain -> a real, resolvable host to visit before Selenium will accept it. Discovery+'s
|
||
session-token cookie is scoped to 'prd.api.discoveryplus.com', which itself doesn't resolve —
|
||
only its subdomains do."""
|
||
override = {"prd.api.discoveryplus.com": "default.any-any.prd.api.discoveryplus.com"}
|
||
by_domain = {}
|
||
for c in cookies:
|
||
host = c.get("domain", "").lstrip(".")
|
||
by_domain.setdefault(override.get(host, host), []).append(c)
|
||
return by_domain
|
||
|
||
def _authenticate_discoveryplus(driver):
|
||
"""Logs a fresh Selenium session into Discovery+ using the saved cookie file."""
|
||
try:
|
||
with open(os.path.expanduser("~/.aetn_cookies.json")) as f:
|
||
cookies = [c for c in json.load(f) if "discoveryplus" in c.get("domain", "")]
|
||
except (OSError, ValueError):
|
||
return False
|
||
added = 0
|
||
for host, group in _discovery_cookie_domains(cookies).items():
|
||
try:
|
||
driver.get(f"https://{host}/")
|
||
except Exception:
|
||
continue # a handful of cookie domains are unreachable directly; the rest still apply
|
||
for c in group:
|
||
c = dict(c); c.pop("sameSite", None); c.pop("expiry", None)
|
||
try:
|
||
driver.add_cookie(c); added += 1
|
||
except Exception:
|
||
pass
|
||
return added > 0
|
||
|
||
def capture_discoveryplus_manifest(url, download_id):
|
||
"""Opens a Discovery+ video page with the saved login and returns the DASH manifest URL the
|
||
player requests, or None."""
|
||
update_download(download_id, status="capturing", message="Opening Firefox to find the Discovery+ stream...")
|
||
profile_path = get_firefox_profile()
|
||
if not profile_path:
|
||
update_download(download_id, status="error", message="Firefox profile not found")
|
||
return None
|
||
|
||
tmp_profile = _copy_firefox_profile(profile_path)
|
||
options = Options()
|
||
options.add_argument("-profile"); options.add_argument(tmp_profile); options.add_argument("-no-remote")
|
||
driver = None
|
||
try:
|
||
driver = webdriver.Firefox(options=options)
|
||
if not _authenticate_discoveryplus(driver):
|
||
update_download(download_id, status="error",
|
||
message="No Discovery+ login found — refresh the saved cookies (see the Login panel)")
|
||
return None
|
||
|
||
driver.get(url)
|
||
update_download(download_id, status="capturing", message="Waiting for playback to start...")
|
||
for i in range(30):
|
||
time.sleep(1)
|
||
if i in (2, 5, 9):
|
||
nudge_video_player(driver)
|
||
send_playback_keys(driver)
|
||
manifest = driver.execute_script("""
|
||
return performance.getEntriesByType('resource').map(e => e.name)
|
||
.find(n => n.includes('.mpd')) || null;
|
||
""")
|
||
if manifest:
|
||
return manifest
|
||
update_download(download_id, status="error",
|
||
message="Timed out waiting for a stream — the account may not be logged in, or the video needs a click to start")
|
||
return None
|
||
except Exception as e:
|
||
update_download(download_id, status="error", message=f"Firefox error: {e}")
|
||
return None
|
||
finally:
|
||
if driver:
|
||
try:
|
||
driver.quit()
|
||
except Exception:
|
||
pass
|
||
try:
|
||
import shutil
|
||
shutil.rmtree(tmp_profile, ignore_errors=True)
|
||
except Exception:
|
||
pass
|
||
|
||
def _discoveryplus_formats(manifest_url, referer):
|
||
"""Asks yt-dlp to resolve a DASH manifest into its list of (still-encrypted) direct file URLs."""
|
||
r = subprocess.run([
|
||
"yt-dlp", "--allow-unplayable-formats", "-J",
|
||
"--add-header", f"Referer: {referer}",
|
||
manifest_url,
|
||
], capture_output=True, text=True, timeout=60)
|
||
if r.returncode != 0:
|
||
return []
|
||
try:
|
||
return json.loads(r.stdout).get("formats", [])
|
||
except ValueError:
|
||
return []
|
||
|
||
def _select_discoveryplus_streams(formats):
|
||
"""Picks the video closest to 1080p (preferring higher when no exact match) and the best audio."""
|
||
video = [f for f in formats if f.get("vcodec") not in (None, "none") and f.get("height")]
|
||
audio = [f for f in formats if f.get("acodec") not in (None, "none") and f.get("vcodec") in (None, "none")]
|
||
best_v = best_a = None
|
||
if video:
|
||
target = DISCOVERY_TARGET_HEIGHT
|
||
exact = [f for f in video if f["height"] == target]
|
||
below = [f for f in video if f["height"] < target]
|
||
above = [f for f in video if f["height"] > target]
|
||
if exact:
|
||
pool = exact
|
||
elif below:
|
||
best_h = max(f["height"] for f in below)
|
||
pool = [f for f in below if f["height"] == best_h]
|
||
else:
|
||
pool = above
|
||
best_v = max(pool, key=lambda f: f.get("tbr") or 0)["url"]
|
||
if audio:
|
||
best_a = max(audio, key=lambda f: f.get("tbr") or 0)["url"]
|
||
return best_v, best_a
|
||
|
||
def download_discoveryplus(url, output_path, download_id, video_drm_key="", audio_drm_key=""):
|
||
manifest = capture_discoveryplus_manifest(url, download_id)
|
||
if not manifest:
|
||
return False
|
||
|
||
update_download(download_id, message="Reading available quality levels...")
|
||
formats = _discoveryplus_formats(manifest, url)
|
||
best_v, best_a = _select_discoveryplus_streams(formats)
|
||
if not best_v:
|
||
update_download(download_id, status="error", message="No downloadable video stream found in the manifest")
|
||
return False
|
||
|
||
out_dir = os.path.dirname(output_path)
|
||
stem = os.path.splitext(os.path.basename(output_path))[0]
|
||
has_drm = bool(video_drm_key and audio_drm_key)
|
||
suffix = "_venc" if has_drm else "_video"
|
||
video_tmp = os.path.join(out_dir, f"_{stem}{suffix}.mp4")
|
||
audio_tmp = os.path.join(out_dir, f"_{stem}{'_aenc' if has_drm else '_audio'}.mp4") if best_a else None
|
||
|
||
update_download(download_id, status="downloading", message="Downloading Discovery+ video...")
|
||
if not _download_mp4(best_v, video_tmp, download_id, "Downloading Discovery+ video"):
|
||
return False
|
||
if best_a and audio_tmp:
|
||
update_download(download_id, message="Downloading Discovery+ audio...")
|
||
if not _download_mp4(best_a, audio_tmp, download_id, "Downloading Discovery+ audio"):
|
||
return False
|
||
|
||
if has_drm:
|
||
ok = _drm_decrypt_merge(video_tmp, audio_tmp or "", video_drm_key, audio_drm_key, output_path, download_id, label="Discovery+")
|
||
for f in (video_tmp, audio_tmp):
|
||
if f:
|
||
try:
|
||
os.remove(f)
|
||
except Exception:
|
||
pass
|
||
return ok
|
||
else:
|
||
if not best_a:
|
||
try:
|
||
os.rename(video_tmp, output_path)
|
||
except Exception:
|
||
pass
|
||
else:
|
||
update_download(download_id, message=
|
||
"Saved encrypted video+audio separately (no DRM keys). Provide keys to decrypt.")
|
||
return True
|
||
|
||
# ── M3U8 capture via Selenium ──────────────────────────────────────────────────
|
||
def _copy_firefox_profile(src):
|
||
"""Copy profile to a fresh temp dir, silently skipping any locked files."""
|
||
import shutil, tempfile
|
||
tmp = tempfile.mkdtemp(prefix="aetn_ff_")
|
||
for name in os.listdir(src):
|
||
s = os.path.join(src, name)
|
||
d = os.path.join(tmp, name)
|
||
try:
|
||
if os.path.isdir(s):
|
||
shutil.copytree(s, d)
|
||
else:
|
||
shutil.copy2(s, d)
|
||
except (PermissionError, OSError):
|
||
pass # skip files locked by the running Firefox
|
||
return tmp
|
||
|
||
def capture_m3u8(url, download_id):
|
||
update_download(download_id, status="capturing", message="Waiting for browser slot...")
|
||
with browser_semaphore:
|
||
return _capture_m3u8_inner(url, download_id)
|
||
|
||
def _capture_m3u8_inner(url, download_id):
|
||
update_download(download_id, status="capturing", message="Opening browser to capture stream...")
|
||
|
||
profile_path = get_firefox_profile()
|
||
if not profile_path:
|
||
update_download(download_id, status="error", message="Firefox profile not found")
|
||
return None
|
||
|
||
tmp_profile = _copy_firefox_profile(profile_path)
|
||
options = Options()
|
||
options.add_argument("-profile")
|
||
options.add_argument(tmp_profile)
|
||
options.add_argument("-no-remote") # never connect to an existing Firefox instance
|
||
|
||
driver = None
|
||
try:
|
||
driver = webdriver.Firefox(options=options)
|
||
|
||
cookie_file = os.path.expanduser("~/.aetn_cookies.json")
|
||
if os.path.exists(cookie_file):
|
||
with open(cookie_file) as f:
|
||
saved_cookies = json.load(f)
|
||
domain = url.split('/')[2]
|
||
driver.get(f"https://{domain}")
|
||
time.sleep(2)
|
||
for cookie in saved_cookies:
|
||
try:
|
||
cookie_domain = cookie.get('domain', '').lstrip('.')
|
||
if cookie_domain in domain or domain in cookie_domain:
|
||
cookie.pop('sameSite', None)
|
||
cookie.pop('expiry', None)
|
||
driver.add_cookie(cookie)
|
||
except:
|
||
pass
|
||
|
||
driver.get(url)
|
||
update_download(download_id, status="capturing", message="Waiting for video player to initialize...")
|
||
|
||
for i in range(60):
|
||
time.sleep(1)
|
||
if i in (1, 3, 6, 10):
|
||
update_download(download_id, status="capturing", message="Trying to start playback...")
|
||
nudge_video_player(driver)
|
||
send_playback_keys(driver)
|
||
elif i == 15:
|
||
update_download(download_id, status="capturing", message="If Firefox is open, click Play in the video window...")
|
||
perf = collect_candidates_from_frames(driver)
|
||
if perf:
|
||
video_url, audio_url = choose_av_streams(perf, url)
|
||
if video_url:
|
||
try:
|
||
meta = get_movie_meta(url, driver.page_source) if ('/movies/' in url or '/specials/' in url) else None
|
||
if meta:
|
||
update_download(download_id, meta_title=meta["title"], meta_year=meta["year"])
|
||
except Exception:
|
||
pass # naming falls back to the URL slug
|
||
msg = "Stream captured! Starting download..."
|
||
if audio_url:
|
||
msg = "Video + audio streams captured! Starting download..."
|
||
update_download(download_id, status="capturing", message=msg)
|
||
return video_url, audio_url
|
||
|
||
update_download(download_id, status="error", message="Timed out waiting for a playable stream URL")
|
||
return None, None
|
||
except Exception as e:
|
||
update_download(download_id, status="error", message=f"Browser error: {str(e)}")
|
||
return None, None
|
||
finally:
|
||
if driver:
|
||
try:
|
||
driver.quit()
|
||
except:
|
||
pass
|
||
try:
|
||
import shutil
|
||
shutil.rmtree(tmp_profile, ignore_errors=True)
|
||
except Exception:
|
||
pass
|
||
|
||
# ── yt-dlp download (Lifetime) ─────────────────────────────────────────────────
|
||
NETSCAPE_COOKIES = os.path.expanduser("~/.aetn_cookies.txt")
|
||
cookie_lock = threading.Lock()
|
||
|
||
def yt_dlp_cookie_args():
|
||
"""yt-dlp cookie option: the saved login file (converted to yt-dlp's format) when there is
|
||
one, else the local Firefox profile, which a container has none of."""
|
||
with cookie_lock:
|
||
try:
|
||
if not (os.path.exists(NETSCAPE_COOKIES) and
|
||
os.path.getmtime(NETSCAPE_COOKIES) >= os.path.getmtime(COOKIE_FILE)):
|
||
with open(COOKIE_FILE) as f:
|
||
cookies = json.load(f)
|
||
lines = ["# Netscape HTTP Cookie File"]
|
||
for c in cookies:
|
||
domain = c.get("domain")
|
||
if not domain:
|
||
continue
|
||
lines.append("\t".join([
|
||
("#HttpOnly_" if c.get("httpOnly") else "") + domain,
|
||
"TRUE" if domain.startswith(".") else "FALSE",
|
||
c.get("path", "/"),
|
||
"TRUE" if c.get("secure") else "FALSE",
|
||
str(int(c.get("expiry") or 0)),
|
||
c["name"], c["value"],
|
||
]))
|
||
with open(NETSCAPE_COOKIES, "w") as f:
|
||
f.write("\n".join(lines) + "\n")
|
||
os.chmod(NETSCAPE_COOKIES, 0o600)
|
||
except (OSError, ValueError, KeyError):
|
||
return ["--cookies-from-browser", "firefox"]
|
||
return ["--cookies", NETSCAPE_COOKIES]
|
||
|
||
def download_ytdlp(url, output_path, download_id):
|
||
update_download(download_id, status="downloading", message="Fetching stream info...")
|
||
|
||
probe = subprocess.run([
|
||
"yt-dlp",
|
||
*yt_dlp_cookie_args(),
|
||
"--simulate",
|
||
"--print", "%(title)s",
|
||
url
|
||
], capture_output=True, text=True)
|
||
|
||
if probe.returncode != 0:
|
||
error_text = (probe.stderr or probe.stdout or "").strip()
|
||
message = error_text.splitlines()[-1] if error_text else "yt-dlp could not read this page"
|
||
update_download(download_id, status="error", message=message[:240])
|
||
return False
|
||
|
||
update_download(download_id, status="downloading", message="Downloading best available format...")
|
||
|
||
cmd = [
|
||
"yt-dlp",
|
||
"-f", "bestvideo*+bestaudio/best",
|
||
*yt_dlp_cookie_args(),
|
||
"--merge-output-format", "mp4",
|
||
"--output", output_path,
|
||
"--newline",
|
||
url
|
||
]
|
||
|
||
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
|
||
downloads[download_id]['process'] = process
|
||
|
||
for line in process.stdout:
|
||
parse_ytdlp_progress(line.strip(), download_id)
|
||
|
||
process.wait()
|
||
if process.returncode != 0:
|
||
update_download(download_id, status="error", message="yt-dlp download failed")
|
||
return process.returncode == 0
|
||
|
||
def parse_ytdlp_progress(line, download_id):
|
||
match = re.search(r'(\d+\.?\d*)%.*?(\d+\.?\d*\s*\w+iB).*?(\d+\.?\d*\s*\w+iB/s).*?ETA\s+(\S+)', line)
|
||
if match:
|
||
update_download(download_id,
|
||
progress=float(match.group(1)),
|
||
size=match.group(2),
|
||
speed=match.group(3),
|
||
eta=match.group(4)
|
||
)
|
||
return
|
||
|
||
lowered = line.lower()
|
||
if "error:" in lowered:
|
||
update_download(download_id, status="error", message=line[:240])
|
||
|
||
# ── ffmpeg download (SVOD) ─────────────────────────────────────────────────────
|
||
# ── HLS ad removal ─────────────────────────────────────────────────────────────
|
||
# Sites like MyLifetime serve an ad-stitched playlist: the movie's own segments plus ad breaks
|
||
# served from Google's ad servers. The movie itself never comes from those hosts.
|
||
AD_HOSTS = ("googlevideo.com", "doubleclick.net")
|
||
|
||
def hls_text(url, referer_url):
|
||
import requests as _req
|
||
resp = _req.get(url, headers={"User-Agent": "Mozilla/5.0", "Referer": referer_url}, timeout=30)
|
||
resp.raise_for_status()
|
||
return resp.text, resp.url # resp.url is the address after redirects, which relative links follow
|
||
|
||
def hls_attr(line, name):
|
||
m = re.search(rf'[:,]{name}=(?:"([^"]*)"|([^,]*))', line)
|
||
return (m.group(1) if m.group(1) is not None else m.group(2)) if m else None
|
||
|
||
def hls_best_variant(master_url, text):
|
||
"""(video playlist URL, audio playlist URL or None) for the highest-bandwidth variant."""
|
||
lines = text.splitlines()
|
||
variants = []
|
||
for i, line in enumerate(lines):
|
||
if line.startswith("#EXT-X-STREAM-INF"):
|
||
uri = next((l for l in lines[i + 1:] if l and not l.startswith("#")), None)
|
||
if uri:
|
||
variants.append((int(hls_attr(line, "BANDWIDTH") or 0), hls_attr(line, "AUDIO"), urljoin(master_url, uri)))
|
||
if not variants:
|
||
return None
|
||
_bandwidth, group, video = max(variants, key=lambda v: v[0])
|
||
audio = None
|
||
for line in lines:
|
||
if (line.startswith("#EXT-X-MEDIA") and hls_attr(line, "TYPE") == "AUDIO"
|
||
and hls_attr(line, "GROUP-ID") == group and hls_attr(line, "URI")):
|
||
audio = urljoin(master_url, hls_attr(line, "URI"))
|
||
if hls_attr(line, "DEFAULT") == "YES":
|
||
break
|
||
return video, audio
|
||
|
||
def hls_without_ads(playlist_url, text, dest):
|
||
"""Writes dest: a copy of a media playlist with the ad segments removed and every address
|
||
made absolute. Returns (segments_removed, seconds_kept)."""
|
||
lines = text.splitlines()
|
||
out, removed, kept, i = [], 0, 0.0, 0
|
||
while i < len(lines):
|
||
line = lines[i]
|
||
if line.startswith("#EXTINF"):
|
||
j = i + 1
|
||
while j < len(lines) and lines[j].startswith("#"): # tags between the duration and the address
|
||
j += 1
|
||
seg = urljoin(playlist_url, lines[j]) if j < len(lines) else ""
|
||
if any(host in urlparse(seg).netloc for host in AD_HOSTS):
|
||
removed += 1
|
||
else:
|
||
out.extend(lines[i:j])
|
||
out.append(seg)
|
||
kept += float(line.split(":", 1)[1].split(",")[0])
|
||
i = j + 1
|
||
continue
|
||
if 'URI="' in line:
|
||
line = re.sub(r'URI="([^"]*)"', lambda m: f'URI="{urljoin(playlist_url, m.group(1))}"', line)
|
||
out.append(line)
|
||
i += 1
|
||
with open(dest, "w") as f:
|
||
f.write("\n".join(out) + "\n")
|
||
return removed, kept
|
||
|
||
def prepare_hls(video_url, audio_url, referer_url):
|
||
"""Returns (video input, audio input, seconds, temp dir). A stream that carries ad breaks is
|
||
rewritten as local playlists without them; anything else comes back unchanged, so it
|
||
downloads exactly as it did before."""
|
||
import tempfile, shutil
|
||
tmpdir = None
|
||
try:
|
||
text, base = hls_text(video_url, referer_url)
|
||
audio_pl = audio_url
|
||
if "#EXT-X-STREAM-INF" in text:
|
||
best = hls_best_variant(base, text)
|
||
if not best:
|
||
return video_url, audio_url, None, None
|
||
video_pl, audio_pl = best[0], best[1] or audio_url
|
||
text, base = hls_text(video_pl, referer_url)
|
||
tmpdir = tempfile.mkdtemp(prefix="aetn_hls_")
|
||
video_path = os.path.join(tmpdir, "video.m3u8")
|
||
removed, seconds = hls_without_ads(base, text, video_path)
|
||
audio_path = None
|
||
if audio_pl:
|
||
audio_path = os.path.join(tmpdir, "audio.m3u8")
|
||
audio_text, audio_base = hls_text(audio_pl, referer_url)
|
||
removed += hls_without_ads(audio_base, audio_text, audio_path)[0]
|
||
if not removed:
|
||
shutil.rmtree(tmpdir, ignore_errors=True)
|
||
return video_url, audio_url, None, None
|
||
return video_path, audio_path, seconds, tmpdir
|
||
except Exception:
|
||
if tmpdir:
|
||
shutil.rmtree(tmpdir, ignore_errors=True)
|
||
return video_url, audio_url, None, None
|
||
|
||
def download_ffmpeg(m3u8_url, output_path, download_id, referer_url="https://watch.lifetimemovieclub.com/", audio_url=None):
|
||
update_download(download_id, status="downloading", message="Preparing stream...")
|
||
|
||
headers = build_request_headers(referer_url)
|
||
m3u8_url, audio_url, kept_seconds, tmpdir = prepare_hls(m3u8_url, audio_url, referer_url)
|
||
if tmpdir:
|
||
update_download(download_id, message="Ad breaks removed — starting ffmpeg download...")
|
||
else:
|
||
update_download(download_id, message="Starting ffmpeg download...")
|
||
|
||
duration = kept_seconds
|
||
if not duration:
|
||
probe = subprocess.run([
|
||
"ffprobe", "-v", "quiet", "-print_format", "json", "-show_format",
|
||
"-extension_picky", "0",
|
||
"-headers", headers,
|
||
m3u8_url
|
||
], capture_output=True, text=True)
|
||
try:
|
||
duration = float(json.loads(probe.stdout).get('format', {}).get('duration', 0))
|
||
except:
|
||
pass
|
||
|
||
# extension_picky: ad-stitched segments have no video file extension; the whitelist lets
|
||
# ffmpeg follow a local playlist out to the http segments. ffmpeg rejects -headers for a
|
||
# local file, and the segment servers don't need them.
|
||
hls_opts = ["-extension_picky", "0", "-protocol_whitelist", "file,http,https,tcp,tls,crypto"]
|
||
if not tmpdir:
|
||
hls_opts += ["-headers", headers]
|
||
if audio_url:
|
||
cmd = [
|
||
"ffmpeg",
|
||
*hls_opts, "-i", m3u8_url,
|
||
*hls_opts, "-i", audio_url,
|
||
"-c", "copy", "-map", "0:v", "-map", "1:a", "-y",
|
||
"-progress", "pipe:1", "-nostats", output_path
|
||
]
|
||
else:
|
||
cmd = [
|
||
"ffmpeg",
|
||
*hls_opts,
|
||
"-i", m3u8_url, "-c", "copy", "-y",
|
||
"-progress", "pipe:1", "-nostats", output_path
|
||
]
|
||
|
||
process = subprocess.Popen(
|
||
cmd,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.STDOUT,
|
||
text=True,
|
||
bufsize=1
|
||
)
|
||
downloads[download_id]['process'] = process
|
||
|
||
for line in process.stdout:
|
||
key, _, val = line.strip().partition('=')
|
||
if key == 'out_time_ms' and duration:
|
||
try:
|
||
progress = min(100, (int(val) / 1_000_000 / duration) * 100)
|
||
update_download(download_id, progress=round(progress, 1))
|
||
except:
|
||
pass
|
||
elif key == 'total_size':
|
||
try:
|
||
update_download(download_id, size=f"{int(val)/(1024*1024):.1f} MiB")
|
||
except:
|
||
pass
|
||
elif key == 'speed':
|
||
update_download(download_id, speed=val.strip())
|
||
elif "error" in line.lower():
|
||
update_download(download_id, message=line.strip()[:240])
|
||
|
||
process.wait()
|
||
if tmpdir:
|
||
import shutil
|
||
shutil.rmtree(tmpdir, ignore_errors=True)
|
||
if process.returncode != 0:
|
||
current_message = downloads.get(download_id, {}).get("message", "").strip()
|
||
failure_message = current_message or "ffmpeg download failed"
|
||
update_download(download_id, status="error", message=failure_message[:240])
|
||
return process.returncode == 0
|
||
|
||
# ── Captions carried inside the video ──────────────────────────────────────────
|
||
# These sites broadcast closed captions (CEA-608) inside the video stream itself, so nothing needs
|
||
# transcribing: ffmpeg can read them out and save them as a subtitle file Plex picks up.
|
||
NO_CAPTIONS_FILE = os.path.expanduser("~/.aetn_no_captions.json") # {video path: mtime} known to have none
|
||
caption_lock = threading.Lock()
|
||
caption_job = {"running": False, "folder": "", "total": 0, "done": 0, "saved": 0, "none": 0, "errors": 0, "current": ""}
|
||
|
||
def clean_caption_srt(raw):
|
||
"""ffmpeg's caption decoder wraps each cue in <font> tags and positioning codes. Keeps the words.
|
||
Returns (srt text, number of cues)."""
|
||
cues = []
|
||
for block in re.split(r"\n\s*\n", raw.strip()):
|
||
lines = block.strip().splitlines()
|
||
if len(lines) < 3 or "-->" not in lines[1]:
|
||
continue
|
||
text = re.sub(r"<[^>]+>", "", "\n".join(lines[2:]))
|
||
text = re.sub(r"\{\\[^}]*\}", "", text).replace("\\h", " ")
|
||
text = "\n".join(line.strip() for line in text.splitlines() if line.strip())
|
||
if text:
|
||
cues.append((lines[1].strip(), text))
|
||
return "".join(f"{i}\n{timing}\n{text}\n\n" for i, (timing, text) in enumerate(cues, 1)), len(cues)
|
||
|
||
def caption_path(video_path):
|
||
return os.path.splitext(video_path)[0] + ".en.srt"
|
||
|
||
def has_subtitle_file(video_path):
|
||
stem = os.path.splitext(video_path)[0]
|
||
return any(os.path.exists(stem + ext) for ext in (".en.srt", ".eng.srt", ".srt"))
|
||
|
||
def extract_captions(video_path):
|
||
"""Saves a video's embedded captions as 'Title.en.srt' beside it. Returns how many caption
|
||
cues were written, or 0 if the video carries none."""
|
||
import tempfile, shutil
|
||
tmpdir = tempfile.mkdtemp(prefix="aetn_cc_")
|
||
try:
|
||
link = os.path.join(tmpdir, "in.mp4")
|
||
os.symlink(video_path, link) # a plain name avoids quoting trouble in ffmpeg's filter syntax
|
||
raw_path = os.path.join(tmpdir, "raw.srt")
|
||
result = subprocess.run(
|
||
["ffmpeg", "-v", "error", "-f", "lavfi", "-i", f"movie=filename={link}[out0+subcc]",
|
||
"-map", "0:s:0", "-c:s", "srt", raw_path],
|
||
capture_output=True, text=True, timeout=1800)
|
||
if not os.path.exists(raw_path):
|
||
if "matches no streams" in result.stderr:
|
||
return 0
|
||
raise RuntimeError((result.stderr.strip().splitlines() or ["ffmpeg failed"])[-1][:160])
|
||
with open(raw_path, errors="replace") as f:
|
||
text, cues = clean_caption_srt(f.read())
|
||
if not cues:
|
||
return 0
|
||
dest = caption_path(video_path)
|
||
with open(dest + ".tmp", "w", encoding="utf-8") as f:
|
||
f.write(text)
|
||
os.replace(dest + ".tmp", dest)
|
||
return cues
|
||
finally:
|
||
shutil.rmtree(tmpdir, ignore_errors=True)
|
||
|
||
def caption_folders():
|
||
"""Folders the caption run may work on: every source's movie and TV folder."""
|
||
return {source_dir(key, kind): f"{s['label']} {'movies' if kind == 'movies' else 'TV'}"
|
||
for key, s in SOURCES.items() for kind in ('movies', 'tv')}
|
||
|
||
def run_caption_job(folders):
|
||
try:
|
||
with open(NO_CAPTIONS_FILE) as f:
|
||
none_known = json.load(f)
|
||
except (OSError, ValueError):
|
||
none_known = {}
|
||
todo = []
|
||
for folder in folders:
|
||
for root, _dirs, files in os.walk(folder):
|
||
for name in sorted(files):
|
||
path = os.path.join(root, name)
|
||
if name.startswith('.') or not name.lower().endswith(VIDEO_EXTS) or has_subtitle_file(path):
|
||
continue
|
||
if none_known.get(path) == os.path.getmtime(path):
|
||
continue # already tried, and this exact file has no captions
|
||
todo.append(path)
|
||
with caption_lock:
|
||
caption_job.update(total=len(todo), done=0, saved=0, none=0, errors=0, current="")
|
||
|
||
def work(path):
|
||
with caption_lock:
|
||
caption_job['current'] = os.path.basename(path)
|
||
try:
|
||
cues = extract_captions(path)
|
||
outcome = 'saved' if cues else 'none'
|
||
if not cues:
|
||
none_known[path] = os.path.getmtime(path)
|
||
except Exception as e:
|
||
print(f" Captions failed for {path}: {e}")
|
||
outcome = 'errors'
|
||
with caption_lock:
|
||
caption_job[outcome] += 1
|
||
caption_job['done'] += 1
|
||
|
||
try:
|
||
with ThreadPoolExecutor(max_workers=2) as pool:
|
||
list(pool.map(work, todo))
|
||
write_json_atomic(NO_CAPTIONS_FILE, none_known)
|
||
finally:
|
||
with caption_lock:
|
||
caption_job.update(running=False, current="")
|
||
|
||
# ── Whisper subtitle generation ────────────────────────────────────────────────
|
||
WHISPER_MODEL = os.environ.get("AETV_WHISPER_MODEL", "medium") # "small" is roughly 3x faster, a little less accurate
|
||
whisper_slot = threading.Semaphore(1) # one transcription at a time; each already uses every CPU core
|
||
|
||
def generate_subtitles(download_id, video_path):
|
||
update_download(download_id, subtitle_status="queued",
|
||
subtitle_message="Waiting for Whisper — one video at a time...")
|
||
with whisper_slot:
|
||
if download_id not in downloads:
|
||
return # cleared while it was waiting
|
||
_generate_subtitles(download_id, video_path)
|
||
|
||
def _generate_subtitles(download_id, video_path):
|
||
import tempfile, shutil
|
||
update_download(download_id,
|
||
subtitle_status="running",
|
||
subtitle_message=f"Starting Whisper ({WHISPER_MODEL} model) — this can take an hour or more..."
|
||
)
|
||
save_queue()
|
||
|
||
video_path = os.path.normpath(video_path)
|
||
# Work in a temp folder so a half-finished run never leaves a .wav or .srt in the library
|
||
tmpdir = tempfile.mkdtemp(prefix="aetn_whisper_")
|
||
wav_path = os.path.join(tmpdir, "audio.wav")
|
||
srt_path = os.path.join(tmpdir, "audio.srt")
|
||
|
||
try:
|
||
# Extract audio to WAV first — avoids Whisper's broken Windows pipe
|
||
update_download(download_id, subtitle_message="Extracting audio...")
|
||
extract = subprocess.run([
|
||
"ffmpeg", "-y", "-i", video_path,
|
||
"-vn", "-acodec", "pcm_s16le", "-ar", "16000", "-ac", "1",
|
||
wav_path
|
||
], capture_output=True, text=True)
|
||
|
||
if extract.returncode != 0 or not os.path.exists(wav_path):
|
||
update_download(download_id, subtitle_status="error",
|
||
subtitle_message="Audio extraction failed — file may have no audio stream")
|
||
return
|
||
|
||
cmd = [
|
||
sys.executable, "-m", "whisper", wav_path,
|
||
"--model", WHISPER_MODEL,
|
||
"--language", "en",
|
||
"--output_format", "srt",
|
||
"--output_dir", tmpdir
|
||
]
|
||
|
||
update_download(download_id, subtitle_message=f"Transcribing with the {WHISPER_MODEL} model...")
|
||
process = subprocess.Popen(
|
||
cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
|
||
text=True, bufsize=1
|
||
)
|
||
downloads[download_id]['subtitle_process'] = process
|
||
|
||
for line in process.stdout:
|
||
line = line.strip()
|
||
if '-->' in line:
|
||
update_download(download_id, subtitle_message=f"Transcribing... {line[:30]}")
|
||
|
||
process.wait()
|
||
|
||
if process.returncode != 0:
|
||
update_download(download_id, subtitle_status="error", subtitle_message="Whisper failed")
|
||
return
|
||
|
||
if not os.path.exists(srt_path):
|
||
update_download(download_id, subtitle_status="error",
|
||
subtitle_message="SRT file not found after transcription")
|
||
return
|
||
|
||
final_path = caption_path(video_path) # 'Title.en.srt', the name Plex reads as English
|
||
shutil.move(srt_path, final_path + ".tmp")
|
||
os.replace(final_path + ".tmp", final_path)
|
||
|
||
update_download(download_id,
|
||
subtitle_status="complete",
|
||
subtitle_message=f"Subtitles saved: {os.path.basename(final_path)}",
|
||
srt_path=final_path
|
||
)
|
||
save_queue()
|
||
|
||
except Exception as e:
|
||
update_download(download_id, subtitle_status="error", subtitle_message=str(e))
|
||
finally:
|
||
shutil.rmtree(tmpdir, ignore_errors=True)
|
||
|
||
# ── Download state management ──────────────────────────────────────────────────
|
||
def update_download(download_id, **kwargs):
|
||
if download_id in downloads:
|
||
downloads[download_id].update(kwargs)
|
||
try:
|
||
save_queue()
|
||
except Exception as e: # a failed save must never fail the download itself
|
||
print(f" Could not save the queue: {e}")
|
||
|
||
def terminate_download_processes(entry):
|
||
for key in ('process', 'subtitle_process'):
|
||
proc = entry.get(key)
|
||
if proc:
|
||
try:
|
||
proc.terminate()
|
||
except:
|
||
pass
|
||
|
||
def apply_page_meta(download_id, output_path):
|
||
"""Renames the target to 'Title (Year).mp4' once the page's real title/year is known.
|
||
Returns the path to write to, or None if that movie is already in the library."""
|
||
entry = downloads.get(download_id, {})
|
||
title = entry.get("meta_title")
|
||
if not title:
|
||
return output_path
|
||
year = entry.get("meta_year")
|
||
folder = os.path.dirname(output_path)
|
||
new_path = os.path.join(folder, movie_filename(title, year) + ".mp4")
|
||
if os.path.exists(new_path) or library_status(scan_library(folder), title, year) == "have":
|
||
update_download(download_id, status="error", title=title, message="Already in library — skipped")
|
||
return None
|
||
update_download(download_id, output_path=new_path, title=title)
|
||
return new_path
|
||
|
||
def part_path(path):
|
||
"""Where a download is written until it finishes: a hidden file beside the final one, so Plex
|
||
and the 'already there' checks never see a half-written movie."""
|
||
return os.path.join(os.path.dirname(path), "." + os.path.basename(path)[:-4] + ".part.mp4")
|
||
|
||
def run_download(download_id, url, output_path, site_type):
|
||
update_download(download_id, status="queued", message="Waiting for a free download slot...")
|
||
with download_semaphore:
|
||
if download_id not in downloads:
|
||
return # cancelled or cleared while it was waiting for a slot
|
||
try:
|
||
success = False
|
||
if site_type == "hbomax":
|
||
video_drm_key = downloads[download_id].get('video_drm_key', '')
|
||
audio_drm_key = downloads[download_id].get('audio_drm_key', '')
|
||
success = download_hbomax(url, output_path, download_id, video_drm_key, audio_drm_key)
|
||
elif site_type == "discoveryplus":
|
||
video_drm_key = downloads[download_id].get('video_drm_key', '')
|
||
audio_drm_key = downloads[download_id].get('audio_drm_key', '')
|
||
success = download_discoveryplus(url, part_path(output_path), download_id, video_drm_key, audio_drm_key)
|
||
if success and os.path.exists(part_path(output_path)):
|
||
os.replace(part_path(output_path), output_path)
|
||
elif site_type == "svod":
|
||
video_url, audio_url = capture_m3u8(url, download_id)
|
||
if video_url:
|
||
output_path = apply_page_meta(download_id, output_path)
|
||
if output_path:
|
||
success = download_ffmpeg(video_url, part_path(output_path), download_id, referer_url=url, audio_url=audio_url)
|
||
elif site_type == "ytdlp":
|
||
success = download_ytdlp(url, part_path(output_path), download_id)
|
||
else:
|
||
success = download_ytdlp(url, part_path(output_path), download_id)
|
||
if not success and site_type == "lifetime":
|
||
current_message = downloads.get(download_id, {}).get("message", "")
|
||
# Don't bother with browser fallback for cable-auth-only content
|
||
if "participating tv provider" in current_message.lower() or "adobe pass" in current_message.lower():
|
||
update_download(download_id, status="error",
|
||
message="Requires cable TV login (Adobe Pass) — not accessible without provider credentials")
|
||
return
|
||
fallback_needed = (
|
||
"show not found in a&e feed" in current_message.lower() or
|
||
"yt-dlp could not read this page" in current_message.lower() or
|
||
"yt-dlp download failed" in current_message.lower()
|
||
)
|
||
if fallback_needed:
|
||
update_download(
|
||
download_id,
|
||
status="capturing",
|
||
message="yt-dlp failed, trying browser capture fallback..."
|
||
)
|
||
video_url, audio_url = capture_m3u8(url, download_id)
|
||
if video_url:
|
||
output_path = apply_page_meta(download_id, output_path)
|
||
if output_path:
|
||
success = download_ffmpeg(video_url, part_path(output_path), download_id, referer_url=url, audio_url=audio_url)
|
||
|
||
if success and os.path.exists(part_path(output_path)):
|
||
os.replace(part_path(output_path), output_path) # only now does it appear in the library
|
||
if success:
|
||
update_download(download_id, progress=100, message="Saving captions...")
|
||
try:
|
||
cues = extract_captions(output_path)
|
||
if cues:
|
||
update_download(download_id, subtitle_status="complete", srt_path=caption_path(output_path),
|
||
subtitle_message=f"{cues} caption lines saved")
|
||
except Exception as e: # the movie itself downloaded fine
|
||
update_download(download_id, subtitle_status="error", subtitle_message=f"Captions failed: {e}"[:200])
|
||
update_download(download_id,
|
||
status="complete", progress=100,
|
||
message="Download complete!",
|
||
completed_at=datetime.now().isoformat()
|
||
)
|
||
entry = {k: v for k, v in downloads[download_id].items() if k != 'process'}
|
||
history.append(entry)
|
||
save_history()
|
||
else:
|
||
if downloads[download_id].get('status') != 'error':
|
||
update_download(download_id, status="error", message="Download failed")
|
||
except Exception as e:
|
||
update_download(download_id, status="error", message=str(e))
|
||
finally:
|
||
# a failed or cancelled download leaves nothing behind in the library folder
|
||
if output_path and not (success and os.path.exists(output_path)):
|
||
try:
|
||
os.remove(part_path(output_path))
|
||
except OSError:
|
||
pass
|
||
|
||
# ── Routes ─────────────────────────────────────────────────────────────────────
|
||
@app.route('/')
|
||
def index():
|
||
return render_template('index.html', home=os.path.expanduser('~'))
|
||
|
||
# (name shown in the page, list file, which SOURCES folder its movies are saved to)
|
||
MOVIE_LISTS = [
|
||
('Lifetime', 'mylifetime_lifetime_urls.txt', 'lifetime'),
|
||
('LMN', 'mylifetime_lmn_urls.txt', 'lifetime'),
|
||
('Movie Club', 'lifetimemovieclub_movie_urls.txt', 'lifetime'),
|
||
('Crime Central', 'aecrimecentral_movie_urls.txt', 'crime'),
|
||
('History Vault', 'historyvault_movie_urls.txt', 'history'),
|
||
('History', 'playhistory_movie_urls.txt', 'history'),
|
||
]
|
||
|
||
def read_list(name):
|
||
try:
|
||
with open(os.path.join(LIST_DIR, name)) as f:
|
||
return [line.strip() for line in f if line.strip()]
|
||
except OSError:
|
||
return []
|
||
|
||
LOGIN_SITES = {
|
||
'Lifetime Movie Club': 'lifetimemovieclub.com',
|
||
'History Vault': 'historyvault.com',
|
||
'A&E Crime Central': 'aecrimecentral.com',
|
||
'History': 'history.com',
|
||
'Lifetime': 'mylifetime.com',
|
||
'Discovery+': 'discoveryplus.com',
|
||
}
|
||
|
||
@app.route('/api/config')
|
||
def get_config():
|
||
"""What this server can do, so the page only shows controls that work here."""
|
||
try:
|
||
with open(COOKIE_FILE) as f:
|
||
cookies = json.load(f)
|
||
saved_at = datetime.fromtimestamp(os.path.getmtime(COOKIE_FILE)).isoformat()
|
||
except (OSError, ValueError):
|
||
cookies, saved_at = [], None
|
||
sites = [{'name': label,
|
||
'cookies': sum(1 for c in cookies if c.get('domain', '').lstrip('.').endswith(domain))}
|
||
for label, domain in LOGIN_SITES.items()]
|
||
return jsonify({
|
||
'sources': [{'key': key, 'label': s['label'], 'movies': source_dir(key, 'movies'),
|
||
'tv': source_dir(key, 'tv')} for key, s in SOURCES.items()],
|
||
'movie_lists': [{'label': label, 'count': len(read_list(name))}
|
||
for label, name, _source in MOVIE_LISTS if read_list(name)],
|
||
'cookies_saved_at': saved_at,
|
||
'login_sites': sites,
|
||
'caption_folders': [{'path': p, 'label': label} for p, label in caption_folders().items()],
|
||
'whisper': importlib.util.find_spec('whisper') is not None,
|
||
'max': os.path.isdir(HBO_MAX_BIN_DIR),
|
||
'discovery_drm': DRM_TOOLS_OK,
|
||
})
|
||
|
||
@app.route('/api/captions', methods=['POST'])
|
||
def start_caption_job():
|
||
"""Saves embedded captions as .en.srt files for every video in a folder (or 'all') that has none."""
|
||
choice = ((request.json or {}).get('folder') or '').strip()
|
||
choices = caption_folders()
|
||
if choice == 'all':
|
||
folders = [p for p in choices if os.path.isdir(p)]
|
||
elif choice in choices and os.path.isdir(choice):
|
||
folders = [choice]
|
||
else:
|
||
return jsonify({'error': 'Unknown folder'}), 400
|
||
with caption_lock:
|
||
if caption_job['running']:
|
||
return jsonify({'error': 'A caption run is already in progress'}), 409
|
||
caption_job.update(running=True, folder=choice, total=0, done=0, saved=0, none=0, errors=0, current="")
|
||
threading.Thread(target=run_caption_job, args=(folders,), daemon=True).start()
|
||
return jsonify({'ok': True})
|
||
|
||
@app.route('/api/captions/status')
|
||
def caption_status():
|
||
with caption_lock:
|
||
return jsonify(dict(caption_job))
|
||
|
||
@app.route('/api/compare')
|
||
def compare_library():
|
||
"""Checks the movie lists (all, or the one named by ?list=) against the files already in
|
||
each list's movie folder."""
|
||
wanted = request.args.get('list', '').strip()
|
||
movie_lists = [m for m in MOVIE_LISTS if not wanted or m[0] == wanted]
|
||
if not movie_lists:
|
||
return jsonify({'error': f'Unknown list: {wanted}'}), 400
|
||
|
||
entries, seen = [], set() # (list name, folder source, url)
|
||
for label, name, source in movie_lists:
|
||
for url in read_list(name):
|
||
if url not in seen:
|
||
seen.add(url)
|
||
entries.append((label, source, url))
|
||
all_urls = [url for _label, _source, url in entries]
|
||
with ThreadPoolExecutor(max_workers=4) as pool:
|
||
metas = dict(zip(all_urls, pool.map(get_movie_meta, all_urls)))
|
||
|
||
libraries, folders, per_list, seen_titles = {}, [], {}, set()
|
||
missing, year_differs, unresolved = [], [], []
|
||
have = 0
|
||
for label, source, url in entries:
|
||
stats = per_list.setdefault(label, {'label': label, 'checked': 0, 'have': 0, 'missing': 0,
|
||
'year_differs': 0, 'gone': 0, 'dupes': 0})
|
||
stats['checked'] += 1
|
||
meta = metas[url]
|
||
if not meta:
|
||
unresolved.append(url)
|
||
stats['gone'] += 1
|
||
continue
|
||
title_key = (source, norm_title(meta['title']), meta['year'])
|
||
if title_key in seen_titles: # already listed from another list, e.g. a special on both History sites
|
||
stats['dupes'] += 1
|
||
continue
|
||
seen_titles.add(title_key)
|
||
if source not in libraries:
|
||
folder = source_dir(source, 'movies')
|
||
libraries[source] = scan_library(folder)
|
||
folders.append({'path': folder, 'files': sum(len(years) for years in libraries[source].values())})
|
||
library = libraries[source]
|
||
status = library_status(library, meta['title'], meta['year'])
|
||
entry = {'title': meta['title'], 'year': meta['year'], 'url': url, 'source': label}
|
||
if status == 'have':
|
||
have += 1
|
||
stats['have'] += 1
|
||
elif status == 'year_differs':
|
||
entry['library_years'] = sorted(y for y in library[norm_title(meta['title'])] if y)
|
||
year_differs.append(entry)
|
||
stats['year_differs'] += 1
|
||
else:
|
||
missing.append(entry)
|
||
stats['missing'] += 1
|
||
|
||
return jsonify({
|
||
'folders': folders,
|
||
'lists': list(per_list.values()),
|
||
'checked': len(all_urls),
|
||
'have': have,
|
||
'missing': missing,
|
||
'year_differs': year_differs,
|
||
'unresolved': unresolved,
|
||
})
|
||
|
||
@app.route('/api/shows')
|
||
def compare_shows():
|
||
"""Per-show episode counts for one source, against its TV folder."""
|
||
source = request.args.get('source', '')
|
||
if source not in SOURCES:
|
||
return jsonify({'error': 'Unknown source'}), 400
|
||
folder = source_dir(source, 'tv')
|
||
have = scan_tv_library(folder)
|
||
by_key = load_episodes(source)[0]
|
||
|
||
shows = {}
|
||
for key, row in by_key.items():
|
||
show = shows.setdefault(key[0], {'show': row['series'], 'total': 0, 'have': 0,
|
||
'locked': 0, 'missing_urls': []})
|
||
show['total'] += 1
|
||
if key in have:
|
||
show['have'] += 1
|
||
elif row['needs_login']:
|
||
show['locked'] += 1
|
||
else:
|
||
show['missing_urls'].append(row['url'])
|
||
rows = sorted(shows.values(), key=lambda s: (-len(s['missing_urls']), s['show'].lower()))
|
||
return jsonify({
|
||
'source': source, 'label': SOURCES[source]['label'], 'folder': folder,
|
||
'episodes': len(by_key), 'have': sum(s['have'] for s in rows),
|
||
'shows': rows,
|
||
# False for sources with no episode scraper (needs a real login just to see the catalog) —
|
||
# the page shouldn't show "0 shows" as if something's broken
|
||
'has_episode_list': bool(EPISODE_FILES.get(source)),
|
||
})
|
||
|
||
@app.route('/api/start', methods=['POST'])
|
||
def start_download():
|
||
data = request.json
|
||
url = data.get('url', '').strip()
|
||
output_dir = (data.get('output_dir') or '').strip() # blank = choose the folder by site
|
||
|
||
if not url:
|
||
return jsonify({'error': 'No URL provided'}), 400
|
||
|
||
site_type, site_label = detect_site(url)
|
||
if not site_type:
|
||
return jsonify({'error': 'Unsupported URL'}), 400
|
||
|
||
source = source_for(url)
|
||
if not output_dir and not source:
|
||
return jsonify({'error': f'{site_label} has no default folder — enter a Save to folder'}), 400
|
||
|
||
def root_for(kind):
|
||
"""The folder to save into: the typed one, else the site's own. Never creates the root itself,
|
||
so a folder that isn't mounted into the container is reported instead of silently made."""
|
||
root = output_dir or source_dir(source, kind)
|
||
return root if os.path.isdir(root) else None
|
||
|
||
title, safe_title = parse_title(url)
|
||
episode_tag = get_episode_tag(url)
|
||
meta = None
|
||
episode_source, episode = find_episode(url) if episode_tag else (None, None)
|
||
if episode:
|
||
root = root_for('tv')
|
||
if not root:
|
||
return jsonify({'error': f"Folder {output_dir or source_dir(source, 'tv')} doesn't exist on the server"}), 400
|
||
output_path = os.path.join(root, episode_relpath(episode))
|
||
title = f"{episode['series']} S{episode['season']:02d}E{episode['episode']:02d}"
|
||
if episode['title']:
|
||
title += f" - {episode['title']}"
|
||
key = (norm_title(episode['series']), episode['season'], episode['episode'])
|
||
if os.path.exists(output_path) or key in scan_tv_library(root):
|
||
return jsonify({'skipped': True, 'title': title})
|
||
os.makedirs(os.path.dirname(output_path), exist_ok=True)
|
||
else:
|
||
if episode_tag:
|
||
safe_title = f"{safe_title}_{episode_tag}"
|
||
elif '/movies/' in url or '/specials/' in url:
|
||
meta = get_movie_meta(url)
|
||
if meta:
|
||
title, safe_title = meta['title'], movie_filename(meta['title'], meta['year'])
|
||
root = root_for('tv' if episode_tag else 'movies')
|
||
if not root:
|
||
return jsonify({'error': f"Folder {output_dir or source_dir(source, 'movies')} doesn't exist on the server"}), 400
|
||
output_path = os.path.join(root, f"{safe_title}.mp4")
|
||
|
||
if os.path.exists(output_path):
|
||
return jsonify({'skipped': True, 'title': title})
|
||
if meta and library_status(scan_library(root), meta['title'], meta['year']) == 'have':
|
||
return jsonify({'skipped': True, 'title': title})
|
||
|
||
video_drm_key = data.get('video_drm_key', '').strip()
|
||
audio_drm_key = data.get('audio_drm_key', '').strip()
|
||
|
||
download_id = str(uuid.uuid4())[:8]
|
||
downloads[download_id] = {
|
||
'id': download_id, 'url': url,
|
||
'title': title + (f" ({episode_tag})" if episode_tag and not episode else ""),
|
||
'site': site_label, 'site_type': site_type,
|
||
'output_path': output_path, 'status': 'queued',
|
||
'progress': 0, 'speed': '--', 'size': '--', 'eta': '--',
|
||
'message': 'Queued', 'started_at': datetime.now().isoformat(),
|
||
'completed_at': None, 'subtitle_status': None,
|
||
'subtitle_message': None, 'srt_path': None, 'burned_path': None,
|
||
'video_drm_key': video_drm_key, 'audio_drm_key': audio_drm_key,
|
||
}
|
||
|
||
save_queue()
|
||
thread = threading.Thread(target=run_download, args=(download_id, url, output_path, site_type), daemon=True)
|
||
thread.start()
|
||
|
||
return jsonify({'id': download_id, 'title': downloads[download_id]['title']})
|
||
|
||
@app.route('/api/import', methods=['POST'])
|
||
def import_video():
|
||
data = request.json
|
||
video_path = data.get('path', '').strip()
|
||
|
||
if not video_path:
|
||
return jsonify({'error': 'No path provided'}), 400
|
||
|
||
if not os.path.exists(video_path):
|
||
return jsonify({'error': f'File not found: {video_path}'}), 400
|
||
|
||
ext = os.path.splitext(video_path)[1].lower()
|
||
if ext not in ('.mp4', '.mkv'):
|
||
return jsonify({'error': 'Only .mp4 and .mkv files are supported'}), 400
|
||
|
||
title = os.path.splitext(os.path.basename(video_path))[0].replace('_', ' ')
|
||
size_mb = os.path.getsize(video_path) / (1024 * 1024)
|
||
|
||
download_id = str(uuid.uuid4())[:8]
|
||
downloads[download_id] = {
|
||
'id': download_id, 'url': '',
|
||
'title': title, 'site': 'Imported',
|
||
'site_type': 'imported', 'output_path': video_path,
|
||
'status': 'complete', 'progress': 100,
|
||
'speed': '--', 'size': f"{size_mb:.1f} MiB", 'eta': '--',
|
||
'message': 'Imported from disk',
|
||
'started_at': datetime.now().isoformat(),
|
||
'completed_at': datetime.now().isoformat(),
|
||
'subtitle_status': None, 'subtitle_message': None,
|
||
'srt_path': None, 'burned_path': None,
|
||
}
|
||
|
||
# Check if srt already exists
|
||
srt_path = os.path.splitext(video_path)[0] + ".srt"
|
||
if os.path.exists(srt_path):
|
||
downloads[download_id]['srt_path'] = srt_path
|
||
downloads[download_id]['subtitle_status'] = 'complete'
|
||
downloads[download_id]['subtitle_message'] = f"Existing subtitles found: {os.path.basename(srt_path)}"
|
||
|
||
save_queue()
|
||
entry = {k: v for k, v in downloads[download_id].items() if k != 'process'}
|
||
history.append(entry)
|
||
save_history()
|
||
|
||
return jsonify({'id': download_id, 'title': title})
|
||
|
||
@app.route('/api/subtitles/<download_id>', methods=['POST'])
|
||
def start_subtitles(download_id):
|
||
if download_id not in downloads:
|
||
# Try to restore from history
|
||
match = next((h for h in history if h.get('id') == download_id), None)
|
||
if not match:
|
||
return jsonify({'error': 'Download not found'}), 404
|
||
downloads[download_id] = dict(match)
|
||
|
||
d = downloads[download_id]
|
||
if d.get('subtitle_status') == 'running':
|
||
return jsonify({'error': 'Already generating subtitles'}), 400
|
||
|
||
video_path = d.get('output_path')
|
||
if not video_path or not os.path.exists(video_path):
|
||
return jsonify({'error': 'Video file not found'}), 400
|
||
|
||
update_download(download_id, subtitle_status='queued', subtitle_message='Subtitle generation queued...')
|
||
thread = threading.Thread(target=generate_subtitles, args=(download_id, video_path), daemon=True)
|
||
thread.start()
|
||
|
||
return jsonify({'ok': True})
|
||
|
||
|
||
def videos_without_subtitles(folders):
|
||
"""Every video under the folders that has no subtitle file beside it (hidden and partial files skipped)."""
|
||
found = []
|
||
for folder in folders:
|
||
for root, _dirs, files in os.walk(folder):
|
||
for name in sorted(files):
|
||
path = os.path.join(root, name)
|
||
if not name.startswith('.') and name.lower().endswith(VIDEO_EXTS) and not has_subtitle_file(path):
|
||
found.append(path)
|
||
return found
|
||
|
||
@app.route('/api/batch_subtitles', methods=['POST'])
|
||
def batch_subtitles():
|
||
"""Whisper transcription for every video in a caption folder (or 'all') that still has no subtitle
|
||
file. Run 'Extract captions' first so only videos without built-in captions are left. With
|
||
count_only it just reports how many there are."""
|
||
data = request.json or {}
|
||
choice = (data.get('folder') or '').strip()
|
||
choices = caption_folders()
|
||
if choice == 'all':
|
||
folders = [p for p in choices if os.path.isdir(p)]
|
||
elif choice in choices and os.path.isdir(choice):
|
||
folders = [choice]
|
||
else:
|
||
return jsonify({'error': 'Unknown folder'}), 400
|
||
if importlib.util.find_spec('whisper') is None:
|
||
return jsonify({'error': 'Whisper is not installed on this server'}), 400
|
||
|
||
todo = videos_without_subtitles(folders)
|
||
if data.get('count_only'):
|
||
return jsonify({'count': len(todo)})
|
||
|
||
for video_path in todo:
|
||
title = os.path.splitext(os.path.basename(video_path))[0].replace('_', ' ')
|
||
size_mb = os.path.getsize(video_path) / (1024 * 1024)
|
||
download_id = str(uuid.uuid4())[:8]
|
||
downloads[download_id] = {
|
||
'id': download_id, 'url': '',
|
||
'title': title, 'site': 'Batch',
|
||
'site_type': 'imported', 'output_path': video_path,
|
||
'status': 'complete', 'progress': 100,
|
||
'speed': '--', 'size': f"{size_mb:.1f} MiB", 'eta': '--',
|
||
'message': 'Imported from disk',
|
||
'started_at': datetime.now().isoformat(),
|
||
'completed_at': datetime.now().isoformat(),
|
||
'subtitle_status': 'queued',
|
||
'subtitle_message': 'Queued for Whisper...',
|
||
'srt_path': None, 'burned_path': None,
|
||
}
|
||
threading.Thread(target=generate_subtitles, args=(download_id, video_path), daemon=True).start()
|
||
save_queue()
|
||
return jsonify({'queued': len(todo)})
|
||
|
||
@app.route('/api/status')
|
||
def get_status():
|
||
active = [
|
||
{k: v for k, v in d.items() if k not in ('process', 'subtitle_process')}
|
||
for d in downloads.values()
|
||
]
|
||
active.sort(key=lambda x: x.get('started_at', ''), reverse=True)
|
||
return jsonify({'downloads': active, 'history': list(reversed(history[-50:]))})
|
||
|
||
@app.route('/api/cancel/<download_id>', methods=['POST'])
|
||
def cancel_download(download_id):
|
||
if download_id in downloads:
|
||
terminate_download_processes(downloads[download_id])
|
||
downloads.pop(download_id, None)
|
||
save_queue()
|
||
return jsonify({'ok': True})
|
||
return jsonify({'error': 'Not found'}), 404
|
||
|
||
@app.route('/api/clear_downloads', methods=['POST'])
|
||
def clear_downloads_route():
|
||
for entry in downloads.values():
|
||
terminate_download_processes(entry)
|
||
downloads.clear()
|
||
save_queue()
|
||
return jsonify({'ok': True})
|
||
|
||
@app.route('/api/clear_history', methods=['POST'])
|
||
def clear_history_route():
|
||
global history
|
||
history = []
|
||
save_history()
|
||
return jsonify({'ok': True})
|
||
|
||
@app.route('/api/browse', methods=['POST'])
|
||
def browse_folder():
|
||
path = choose_path("folder") or os.path.expanduser('~/video_output')
|
||
return jsonify({'path': path})
|
||
|
||
@app.route('/api/browse_file', methods=['POST'])
|
||
def browse_file():
|
||
return jsonify({'path': choose_path("file")})
|
||
|
||
@app.route('/api/refresh_cookies', methods=['POST'])
|
||
def refresh_cookies():
|
||
helper = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'save_cookies.py')
|
||
subprocess.Popen([sys.executable, helper])
|
||
return jsonify({'ok': True})
|
||
|
||
@app.route('/api/hbomax_login', methods=['POST'])
|
||
def hbomax_login():
|
||
def _do_login():
|
||
os.makedirs(HBO_MAX_PROFILE_DIR, exist_ok=True)
|
||
try:
|
||
driver = _build_chrome_driver()
|
||
driver.get(HBO_MAX_AUTH_URL)
|
||
# Hold the browser open for 3 minutes for the user to log in
|
||
time.sleep(180)
|
||
with open(HBO_MAX_FIRST_RUN, 'w') as f:
|
||
f.write("ok")
|
||
driver.quit()
|
||
except Exception as e:
|
||
print(f"Max login error: {e}")
|
||
threading.Thread(target=_do_login, daemon=True).start()
|
||
return jsonify({'ok': True})
|
||
|
||
@app.route('/api/hbomax_login_status', methods=['GET'])
|
||
def hbomax_login_status():
|
||
return jsonify({'logged_in': os.path.exists(HBO_MAX_FIRST_RUN)})
|
||
|
||
if not os.environ.get("AETV_NO_RESTORE"): # test scripts that import this file set it, so they don't start a second copy of the queue
|
||
load_queue()
|
||
|
||
if __name__ == '__main__':
|
||
print("\n🎬 AETN Video Downloader")
|
||
print("========================")
|
||
print("Open http://localhost:5000 in your browser\n")
|
||
app.run(debug=False, host=os.environ.get('AETV_HOST', '127.0.0.1'), port=5000, threaded=True)
|