| … | … | @@ -619,7 +619,7 @@ TV_GENRES = [ |
| 619 | 619 | |
| 620 | 620 | MANIFEST = { |
| 621 | 621 | "id": "org.tabertron.vsembed", |
| 622 | | "version": "0.20.0", |
| 622 | "version": "0.21.0", |
| 623 | 623 | "name": "Tabertron VsEmbed", |
| 624 | 624 | "description": "Self-hosted extractor with TMDB browse + search, played via vidcore and HdHub", |
| 625 | 625 | "types": ["movie", "series"], |
| … | … | @@ -1226,6 +1226,127 @@ def _dest_lock(dest: str) -> threading.Lock: |
| 1226 | 1226 | with _dest_locks_guard: |
| 1227 | 1227 | return _dest_locks.setdefault(os.path.abspath(dest), threading.Lock()) |
| 1228 | 1228 | |
| 1229 | |
| 1230 | # What a live job is doing, for the download manager (labs#13). In memory only, |
| 1231 | # deliberately: the persisted queue keeps its running/done/failed semantics, |
| 1232 | # which the resume path depends on, and a phase is only meaningful while the |
| 1233 | # process that owns the job is alive. Entries exist for exactly the life of |
| 1234 | # _run_job. |
| 1235 | # phase: waiting -> resolving -> downloading |
| 1236 | # source: vidcore | hdhub | None (None = nothing fetched, e.g. already held) |
| 1237 | _job_info: dict[str, dict] = {} |
| 1238 | _cancel_events: dict[str, threading.Event] = {} |
| 1239 | |
| 1240 | |
| 1241 | def _job_update(key: str, **fields) -> None: |
| 1242 | """Annotate a live job. A no-op for a key with no live job, so |
| 1243 | _run_ingest stays callable on its own (as its tests call it).""" |
| 1244 | info = _job_info.get(key) |
| 1245 | if info is not None: |
| 1246 | info.update(fields) |
| 1247 | |
| 1248 | |
| 1249 | def _acquire_or_cancel(lock, cancel: threading.Event, |
| 1250 | timeout: Optional[float] = None) -> bool: |
| 1251 | """Acquire `lock` (a Lock or Semaphore) unless `cancel` fires first. |
| 1252 | |
| 1253 | Polls in short slices because neither primitive can wait on an Event too. |
| 1254 | Returns False if cancelled or timed out, and then holds nothing. |
| 1255 | """ |
| 1256 | deadline = None if timeout is None else time.time() + timeout |
| 1257 | while not cancel.is_set(): |
| 1258 | if lock.acquire(timeout=0.25): |
| 1259 | if cancel.is_set(): |
| 1260 | lock.release() |
| 1261 | return False |
| 1262 | return True |
| 1263 | if deadline is not None and time.time() > deadline: |
| 1264 | return False |
| 1265 | return False |
| 1266 | |
| 1267 | |
| 1268 | def _cancel_job(key: str) -> bool: |
| 1269 | """Ask a live job to stop. True if there was one to ask. |
| 1270 | |
| 1271 | It stops at the next point it can: immediately while waiting, when the |
| 1272 | resolve returns, or within one MP4_STALL_POLL while ffmpeg is copying |
| 1273 | (_run_remux kills it). The .part is left where it is -- ffmpeg's -y reuses |
| 1274 | it on the next attempt, and nothing in this module deletes to recover. |
| 1275 | """ |
| 1276 | with _mp4_jobs_guard: |
| 1277 | event = _cancel_events.get(key) |
| 1278 | if event is None: |
| 1279 | return False |
| 1280 | event.set() |
| 1281 | log.info("ingest: cancel requested for %s", key) |
| 1282 | return True |
| 1283 | |
| 1284 | |
| 1285 | # Every finished attempt, one JSON object per line, appended and never |
| 1286 | # rewritten. The queue file cannot serve as history: it holds one LATEST state |
| 1287 | # per key, so a failure followed by a success leaves no trace of the failure. |
| 1288 | # |
| 1289 | # Appends are serialised under a lock and each is a single write of one line. |
| 1290 | # The reader skips a line it cannot parse rather than failing: this file is |
| 1291 | # display-only and nothing ever rebuilds it from what was read, so the shape of |
| 1292 | # the 2026-08-31 queue loss (an unreadable read persisted as empty) cannot occur. |
| 1293 | INGEST_HISTORY_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), |
| 1294 | "ingest_history.jsonl") |
| 1295 | _history_guard = threading.Lock() |
| 1296 | |
| 1297 | |
| 1298 | def _append_history(entry: dict) -> None: |
| 1299 | """Never raises: losing a history line must not fail the job it records.""" |
| 1300 | try: |
| 1301 | line = json.dumps(entry, ensure_ascii=False) + "\n" |
| 1302 | with _history_guard: |
| 1303 | with open(INGEST_HISTORY_FILE, "a", encoding="utf-8") as f: |
| 1304 | f.write(line) |
| 1305 | except Exception as exc: |
| 1306 | log.warning("history append failed: %s", exc) |
| 1307 | |
| 1308 | |
| 1309 | def _read_history(limit: int = 500) -> list[dict]: |
| 1310 | """Newest first. A missing file is an empty history.""" |
| 1311 | try: |
| 1312 | with open(INGEST_HISTORY_FILE, encoding="utf-8") as f: |
| 1313 | lines = f.readlines() |
| 1314 | except FileNotFoundError: |
| 1315 | return [] |
| 1316 | except OSError as exc: |
| 1317 | log.warning("history unreadable: %s", exc) |
| 1318 | return [] |
| 1319 | out = [] |
| 1320 | for line in reversed(lines): |
| 1321 | try: |
| 1322 | entry = json.loads(line) |
| 1323 | except ValueError: |
| 1324 | continue |
| 1325 | if isinstance(entry, dict): |
| 1326 | out.append(entry) |
| 1327 | if len(out) >= limit: |
| 1328 | break |
| 1329 | return out |
| 1330 | |
| 1331 | |
| 1332 | def _record_attempt(key: str, info: dict, state: str, started: float) -> None: |
| 1333 | dest = info.get("dest") |
| 1334 | size = None |
| 1335 | if dest and state == "done": |
| 1336 | try: |
| 1337 | size = os.path.getsize(dest) |
| 1338 | except OSError: |
| 1339 | pass |
| 1340 | _append_history({ |
| 1341 | "ts": time.time(), "started": started, "key": key, |
| 1342 | "kind": info.get("kind"), "ident": info.get("ident"), |
| 1343 | "season": info.get("season"), "episode": info.get("episode"), |
| 1344 | "height": info.get("height"), "title": info.get("title"), |
| 1345 | "label": info.get("label"), "outcome": state, |
| 1346 | "source": info.get("source"), "bytes": size, |
| 1347 | "took": round(time.time() - started, 1), "dest": dest, |
| 1348 | }) |
| 1349 | |
| 1229 | 1350 | # Job state was in-memory only, so a restart lost every intent: a copy that was |
| 1230 | 1351 | # mid-flight never resumed, and since a partial writes to `<dest>.part` while the |
| 1231 | 1352 | # row keys off the finished file, nothing surfaced the loss. Rolling forward |
| … | … | @@ -1718,6 +1839,24 @@ class RemuxTimedOut(RuntimeError): |
| 1718 | 1839 | """The copy exceeded its wall-clock budget while still making progress.""" |
| 1719 | 1840 | |
| 1720 | 1841 | |
| 1842 | class RemuxCancelled(RuntimeError): |
| 1843 | """Someone pressed Cancel. Not a failure of the source, so it must never be |
| 1844 | treated as one -- in particular it must not send _run_ingest on to the |
| 1845 | HdHub fallback, which would fetch the very title the user just stopped.""" |
| 1846 | |
| 1847 | |
| 1848 | # The job a thread is running, so _run_remux can see its cancel event without |
| 1849 | # the event being threaded through _remux_to_mp4 and _copy_progressive (whose |
| 1850 | # two-argument shape the test fakes also rely on). Set by _run_job for the life |
| 1851 | # of one ingest; absent on any thread that is not running a job. |
| 1852 | _job_ctx = threading.local() |
| 1853 | |
| 1854 | |
| 1855 | def _cancel_requested() -> bool: |
| 1856 | event = getattr(_job_ctx, "cancel", None) |
| 1857 | return bool(event and event.is_set()) |
| 1858 | |
| 1859 | |
| 1721 | 1860 | def _read_back(f) -> str: |
| 1722 | 1861 | """Everything written to an open temp file, from the start. Never raises: |
| 1723 | 1862 | this only ever supplies a diagnostic, and a diagnostic that can fail the |
| … | … | @@ -1781,6 +1920,8 @@ def _run_remux(cmd: list[str], out_path: str, stall_timeout: float, |
| 1781 | 1920 | break |
| 1782 | 1921 | except subprocess.TimeoutExpired: |
| 1783 | 1922 | pass |
| 1923 | if _cancel_requested(): |
| 1924 | raise RemuxCancelled("cancelled") |
| 1784 | 1925 | now = time.time() |
| 1785 | 1926 | size = _size() |
| 1786 | 1927 | if size > last_size: |
| … | … | @@ -2101,20 +2242,39 @@ def _run_ingest(kind: str, ident: str, season: int, episode: int, |
| 2101 | 2242 | # a CDN that refuses the bytes, as on 2026-09-28) falls through. |
| 2102 | 2243 | started = time.time() |
| 2103 | 2244 | failure = None |
| 2245 | _job_update(key, phase="resolving", dest=dest) |
| 2104 | 2246 | url = _resolve_vidcore(kind, ident, season, episode) |
| 2247 | # The resolve is a browser subprocess and is let run to its end; Cancel is |
| 2248 | # honoured the moment it returns, before any byte is fetched. |
| 2249 | if _cancel_requested(): |
| 2250 | return "cancelled" |
| 2105 | 2251 | if url: |
| 2106 | 2252 | variant = _cap_variant(url, height) |
| 2107 | 2253 | log.info("ingest start: %s -> %s", key, dest) |
| 2254 | _job_update(key, phase="downloading", source="vidcore", |
| 2255 | downloading_since=time.time()) |
| 2108 | 2256 | try: |
| 2109 | 2257 | _remux_to_mp4(variant, dest) |
| 2258 | except RemuxCancelled: |
| 2259 | log.info("ingest cancelled: %s", key) |
| 2260 | return "cancelled" |
| 2110 | 2261 | except Exception as exc: |
| 2111 | 2262 | log.warning("ingest failed for %s: %s", key, exc) |
| 2112 | 2263 | failure = f"failed: {exc}" |
| 2113 | 2264 | else: |
| 2114 | 2265 | failure = "failed: could not resolve title" |
| 2115 | 2266 | if failure: |
| 2267 | if _cancel_requested(): |
| 2268 | return "cancelled" |
| 2116 | 2269 | started = time.time() |
| 2270 | _job_update(key, phase="downloading", source="hdhub", |
| 2271 | downloading_since=started) |
| 2117 | 2272 | why = _ingest_from_hdhub(kind, ident, season, episode, height, dest, meta) |
| 2273 | # _ingest_from_hdhub turns every exception into a reason string, a |
| 2274 | # cancel included, so the event is the authority here, not `why`. |
| 2275 | if _cancel_requested(): |
| 2276 | log.info("ingest cancelled: %s", key) |
| 2277 | return "cancelled" |
| 2118 | 2278 | if why: |
| 2119 | 2279 | log.info("ingest: hdhub fallback for %s declined: %s", key, why) |
| 2120 | 2280 | return f"{failure}; hdhub: {why}" |
| … | … | @@ -2148,7 +2308,8 @@ def _start_mp4_job(kind: str, ident: str, season: int, episode: int, |
| 2148 | 2308 | |
| 2149 | 2309 | |
| 2150 | 2310 | def _run_job(kind: str, ident: str, season: int, episode: int, |
| 2151 | | height: int, force: bool = False) -> str: |
| 2311 | height: int, force: bool = False, |
| 2312 | cancel: Optional[threading.Event] = None) -> str: |
| 2152 | 2313 | """One ingest, through the remux semaphore, with its state recorded. |
| 2153 | 2314 | |
| 2154 | 2315 | Split out of the thread so a season can drive episodes SEQUENTIALLY through |
| … | … | @@ -2158,30 +2319,72 @@ def _run_job(kind: str, ident: str, season: int, episode: int, |
| 2158 | 2319 | actually wrong with. |
| 2159 | 2320 | """ |
| 2160 | 2321 | key = _mp4_key(kind, ident, season, episode, height) |
| 2161 | | # The file is claimed BEFORE a remux slot, so a job waiting on another |
| 2162 | | # ident's copy of the same file does not sit on one of the two slots while |
| 2163 | | # it waits. Once it gets the file, _run_ingest's _library_has check finds |
| 2164 | | # the copy the first job made and answers "done" without fetching again. |
| 2165 | | # No identity (TMDB down) means no file to claim; _run_ingest then declines |
| 2166 | | # on its own, before writing anything. |
| 2167 | | meta = _library_meta(kind, ident, season, episode) |
| 2168 | | claim = (_dest_lock(_library_path(kind, meta["title"], meta["year"], |
| 2169 | | meta["tmdb_id"], season, episode, |
| 2170 | | meta["episode_title"])) |
| 2171 | | if meta else contextlib.nullcontext()) |
| 2172 | | with claim: |
| 2173 | | acquired = _mp4_sem.acquire(timeout=MP4_REMUX_TIMEOUT) |
| 2174 | | try: |
| 2175 | | state = (_run_ingest(kind, ident, season, episode, height, force) |
| 2176 | | if acquired else "failed: remux slots busy") |
| 2177 | | except Exception as exc: # never leave a job "running" |
| 2178 | | log.warning("ingest crashed for %s: %s", key, exc) |
| 2179 | | state = f"failed: {exc}" |
| 2180 | | finally: |
| 2181 | | if acquired: |
| 2182 | | _mp4_sem.release() |
| 2322 | started = time.time() |
| 2323 | # A season passes in an event it registered BEFORE checking its own cancel |
| 2324 | # flag, so a season Cancel cannot fall between choosing this episode and |
| 2325 | # this job becoming cancellable. |
| 2326 | cancel = cancel or threading.Event() |
| 2327 | label = f"S{season:02d}E{episode:02d}" if kind == "series" else "movie" |
| 2328 | with _mp4_jobs_guard: |
| 2329 | _job_info[key] = {"phase": "waiting", "source": None, "dest": None, |
| 2330 | "started": started, "kind": kind, "ident": ident, |
| 2331 | "season": season, "episode": episode, |
| 2332 | "height": height, "label": label, "title": ident} |
| 2333 | _cancel_events[key] = cancel |
| 2334 | # A season drives episodes through here directly, never through |
| 2335 | # _start_mp4_job, so without this an episode being fetched by a season |
| 2336 | # read as not-running -- and the web page offered to fetch it again. |
| 2337 | if _mp4_jobs.get(key) != "running": |
| 2338 | _set_job_state(key, "running", |
| 2339 | args=(kind, ident, season, episode, height)) |
| 2340 | _job_ctx.cancel = cancel |
| 2341 | meta = None |
| 2342 | acquired = False |
| 2343 | try: |
| 2344 | # The file is claimed BEFORE a remux slot, so a job waiting on another |
| 2345 | # ident's copy of the same file does not sit on one of the two slots |
| 2346 | # while it waits. Once it gets the file, _run_ingest's _library_has |
| 2347 | # check finds the copy the first job made and answers "done" without |
| 2348 | # fetching again. No identity (TMDB down) means no file to claim; |
| 2349 | # _run_ingest then declines on its own, before writing anything. |
| 2350 | meta = _library_meta(kind, ident, season, episode) |
| 2351 | if meta: |
| 2352 | _job_update(key, title=meta["title"]) |
| 2353 | claim = (_dest_lock(_library_path(kind, meta["title"], meta["year"], |
| 2354 | meta["tmdb_id"], season, episode, |
| 2355 | meta["episode_title"])) |
| 2356 | if meta else None) |
| 2357 | # Both waits poll, so a job still queued behind another can be |
| 2358 | # cancelled without first waiting out the one in front of it. |
| 2359 | if claim is not None and not _acquire_or_cancel(claim, cancel): |
| 2360 | state = "cancelled" |
| 2361 | else: |
| 2362 | try: |
| 2363 | acquired = _acquire_or_cancel(_mp4_sem, cancel, |
| 2364 | timeout=MP4_REMUX_TIMEOUT) |
| 2365 | if cancel.is_set(): |
| 2366 | state = "cancelled" |
| 2367 | elif not acquired: |
| 2368 | state = "failed: remux slots busy" |
| 2369 | else: |
| 2370 | state = _run_ingest(kind, ident, season, episode, height, |
| 2371 | force) |
| 2372 | finally: |
| 2373 | if acquired: |
| 2374 | _mp4_sem.release() |
| 2375 | if claim is not None: |
| 2376 | claim.release() |
| 2377 | except Exception as exc: # never leave a job "running" |
| 2378 | log.warning("ingest crashed for %s: %s", key, exc) |
| 2379 | state = f"failed: {exc}" |
| 2380 | finally: |
| 2381 | _job_ctx.cancel = None |
| 2382 | info = _job_info.get(key) or {} |
| 2383 | _record_attempt(key, info, state, started) |
| 2183 | 2384 | with _mp4_jobs_guard: |
| 2184 | 2385 | _set_job_state(key, state) |
| 2386 | _job_info.pop(key, None) |
| 2387 | _cancel_events.pop(key, None) |
| 2185 | 2388 | return state |
| 2186 | 2389 | |
| 2187 | 2390 | |
| … | … | @@ -2211,6 +2414,26 @@ def _released_episodes(tmdb_id: str, season: int) -> list[int]: |
| 2211 | 2414 | |
| 2212 | 2415 | _season_jobs: dict[str, str] = {} |
| 2213 | 2416 | _season_guard = threading.Lock() |
| 2417 | # Live season jobs only, like _job_info: position in the episode list and the |
| 2418 | # episode being fetched, for the download manager. |
| 2419 | _season_info: dict[str, dict] = {} |
| 2420 | _season_cancel: dict[str, threading.Event] = {} |
| 2421 | |
| 2422 | |
| 2423 | def _cancel_season(key: str) -> bool: |
| 2424 | """Stop a season job: no further episodes, and the current one cancelled.""" |
| 2425 | with _season_guard: |
| 2426 | event = _season_cancel.get(key) |
| 2427 | if event is None: |
| 2428 | return False |
| 2429 | # Both set under the season lock -- the same lock the loop holds while |
| 2430 | # it registers the next episode -- so no episode can start unseen. |
| 2431 | event.set() |
| 2432 | current = (_season_info.get(key) or {}).get("current_cancel") |
| 2433 | if current is not None: |
| 2434 | current.set() |
| 2435 | log.info("season: cancel requested for %s", key) |
| 2436 | return True |
| 2214 | 2437 | |
| 2215 | 2438 | |
| 2216 | 2439 | def _start_season_job(ident: str, season: int, height: int, |
| … | … | @@ -2223,28 +2446,47 @@ def _start_season_job(ident: str, season: int, height: int, |
| 2223 | 2446 | as it already did for a single episode. |
| 2224 | 2447 | """ |
| 2225 | 2448 | key = f"season:{ident}:{season}:{height}" |
| 2449 | cancel = threading.Event() |
| 2226 | 2450 | with _season_guard: |
| 2227 | 2451 | if _season_jobs.get(key) == "running": |
| 2228 | 2452 | return "running" |
| 2229 | 2453 | _season_jobs[key] = "running" |
| 2454 | _season_cancel[key] = cancel |
| 2455 | _season_info[key] = {"ident": ident, "season": season, |
| 2456 | "height": height, "episodes": [], "index": 0, |
| 2457 | "current": None, "started": time.time()} |
| 2230 | 2458 | |
| 2231 | 2459 | def run() -> None: |
| 2232 | 2460 | done = 0 |
| 2233 | 2461 | try: |
| 2234 | 2462 | meta = _library_meta("series", ident, season, 1) |
| 2235 | 2463 | episodes = _released_episodes(meta["tmdb_id"], season) if meta else [] |
| 2464 | _season_info[key].update(episodes=episodes, |
| 2465 | title=meta["title"] if meta else ident) |
| 2236 | 2466 | log.info("season %s S%02d: %d released episode(s)", |
| 2237 | 2467 | ident, season, len(episodes)) |
| 2238 | | for episode in episodes: |
| 2468 | for i, episode in enumerate(episodes, 1): |
| 2239 | 2469 | if not force and _library_has("series", ident, season, episode): |
| 2240 | 2470 | continue |
| 2241 | | _run_job("series", ident, season, episode, height, force) |
| 2471 | episode_cancel = threading.Event() |
| 2472 | with _season_guard: |
| 2473 | if cancel.is_set(): |
| 2474 | break |
| 2475 | _season_info[key].update( |
| 2476 | index=i, current_cancel=episode_cancel, |
| 2477 | current=_mp4_key("series", ident, season, episode, |
| 2478 | height)) |
| 2479 | _run_job("series", ident, season, episode, height, force, |
| 2480 | cancel=episode_cancel) |
| 2242 | 2481 | done += 1 |
| 2243 | 2482 | except Exception as exc: |
| 2244 | 2483 | log.warning("season job crashed for %s: %s", key, exc) |
| 2245 | 2484 | finally: |
| 2246 | 2485 | with _season_guard: |
| 2247 | | _season_jobs[key] = f"done ({done})" |
| 2486 | _season_jobs[key] = (f"cancelled ({done})" if cancel.is_set() |
| 2487 | else f"done ({done})") |
| 2488 | _season_cancel.pop(key, None) |
| 2489 | _season_info.pop(key, None) |
| 2248 | 2490 | |
| 2249 | 2491 | threading.Thread(target=run, name=f"season-{key}", daemon=True).start() |
| 2250 | 2492 | return "running" |