'
+ f'
",
+ subtype="html",
+ )
+ with smtplib.SMTP_SSL(SMTP_HOST, SMTP_PORT, timeout=30) as s:
+ s.login(SMTP_USER, SMTP_PASS)
+ s.send_message(msg)
+
+
+def poll_once():
+ with open(FEED_CONFIG) as f:
+ channels = (yaml.safe_load(f) or {}).get("channels") or {}
+ with state_lock:
+ ids = load_json("channel-ids.json", {})
+ seen = load_json("seen.json", {})
+ pending = load_json("pending.json", {})
+ for name, url in channels.items():
+ try:
+ cid = channel_id_for(url, ids)
+ entries = feed_entries(cid)
+ except Exception as e:
+ log(f"{name}: fetch failed: {e}")
+ continue
+ if cid not in seen:
+ seen[cid] = [e["id"] for e in entries]
+ log(f"{name}: now tracking ({len(entries)} existing videos skipped)")
+ continue
+ known = set(seen[cid])
+ for entry in reversed(entries):
+ if entry["id"] in known:
+ continue
+ entry["channel"] = name
+ pending[entry["id"]] = entry
+ seen[cid].append(entry["id"])
+ known.add(entry["id"])
+ log(f"{name}: queued {entry['title']!r}")
+ try:
+ send_notification(name, entry)
+ except Exception as e:
+ log(f"{name}: notification failed: {e}")
+ seen[cid] = seen[cid][-SEEN_CAP:]
+ with state_lock:
+ save_json("channel-ids.json", ids)
+ save_json("seen.json", seen)
+ save_json("pending.json", pending)
+
+
+def poller():
+ while True:
+ try:
+ poll_once()
+ except Exception as e:
+ log(f"poll failed: {e}")
+ time.sleep(INTERVAL)
+
+
+# ------------------------------------------------------------ download worker
+
+def update_job(job_id, **fields):
+ with state_lock:
+ jobs = load_json("jobs.json", [])
+ for j in jobs:
+ if j["id"] == job_id:
+ j.update(fields)
+ save_json("jobs.json", jobs)
+
+
+def resolve_url(url):
+ out = subprocess.run(
+ ["yt-dlp", "--no-playlist", "--print",
+ "%(id)s\t%(title)s\t%(channel)s\t%(upload_date>%Y-%m-%d)s\t%(thumbnail)s", url],
+ capture_output=True, text=True, timeout=90)
+ if out.returncode != 0:
+ if WALL_RE.search(out.stderr):
+ raise WallError(url)
+ raise ValueError(out.stderr[-300:])
+ vid, title, channel, published, thumb = out.stdout.strip().split("\t")
+ return {"id": vid, "title": title, "channel": channel, "published": published,
+ "thumb": thumb, "link": f"https://www.youtube.com/watch?v={vid}"}
+
+
+def resolve_and_update(stub_id, url):
+ try:
+ entry = resolve_url(url)
+ except WallError:
+ set_wall(True)
+ return
+ except Exception as e:
+ log(f"resolve failed for {url}: {e}")
+ return
+ with state_lock:
+ pending = load_json("pending.json", {})
+ # if the stub is gone the user already ingested or skipped it
+ if stub_id in pending:
+ del pending[stub_id]
+ entry["stub_id"] = stub_id
+ pending[entry["id"]] = entry
+ save_json("pending.json", pending)
+
+
+def write_nfo(filepath, root, tags):
+ base, _ = os.path.splitext(filepath)
+ body = "\n".join(f" <{k}>{escape(str(v))}" for k, v in tags.items() if v != "")
+ with open(base + ".nfo", "w") as f:
+ f.write(f"\n<{root}>\n{body}\n\n")
+
+
+def run_job(job):
+ # youtube intermittently bot-walls the home ip after heavy traffic, so
+ # transient failures get retried with backoff before declaring an error
+ for delay in (0, 30, 90):
+ if delay:
+ update_job(job["id"], status="retrying", progress="")
+ time.sleep(delay)
+ try:
+ if run_job_once(job):
+ return
+ except WallError:
+ raise
+ except Exception as e:
+ log(f"job {job['id']} attempt failed: {e}")
+ update_job(job["id"], status="error", progress="")
+
+
+def run_job_once(job):
+ if job.get("unresolved"):
+ update_job(job["id"], status="resolving")
+ meta = resolve_url(job["url"])
+ job.update(title=meta["title"], channel=meta["channel"],
+ published=meta["published"], url=meta["link"], unresolved=False)
+ update_job(job["id"], title=meta["title"])
+ dest, url = job["dest"], job["url"]
+ if dest == "indie":
+ outdir = os.path.join(INDIE_DIR, safe_name(job["show"]), f"Season {job['season']}")
+ prefix = f"S{job['season']:02d}E{job['episode']:02d}"
+ ep_name = safe_name(job.get("ep_title") or job["title"])
+ out = f"{outdir}/{prefix} - {ep_name}.%(ext)s"
+ thumb_out = f"thumbnail:{outdir}/{prefix} - {ep_name}-thumb.%(ext)s"
+ elif dest == "independent":
+ outdir = os.path.join(INDEP_DIR, safe_name(job["channel"]))
+ out = f"{outdir}/%(upload_date>%Y-%m-%d)s - %(title)s.%(ext)s"
+ thumb_out = f"thumbnail:{outdir}/%(upload_date>%Y-%m-%d)s - %(title)s.%(ext)s"
+ else:
+ outdir = os.path.join(MUSIC_DIR, safe_name(job["channel"]))
+ out = f"{outdir}/%(title)s.%(ext)s"
+ thumb_out = None
+ os.makedirs(outdir, exist_ok=True)
+ cmd = ["yt-dlp", "--newline", "--no-playlist", "--embed-chapters",
+ "--sleep-requests", "0.75",
+ "--print", "after_move:filepath", "--no-simulate", "-o", out]
+ if dest == "music":
+ cmd += ["-x"]
+ else:
+ cmd += ["-f", "bv*+ba/b", "--write-thumbnail", "--convert-thumbnails", "jpg",
+ "-o", thumb_out]
+ cmd.append(url)
+ update_job(job["id"], status="downloading")
+ filepath = None
+ tail = []
+ proc = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
+ for line in proc.stdout:
+ line = line.rstrip()
+ tail = (tail + [line])[-30:]
+ m = re.search(r"\[download\]\s+([\d.]+%)", line)
+ if m:
+ update_job(job["id"], progress=m.group(1))
+ elif line.startswith("/"):
+ filepath = line
+ proc.wait()
+ if proc.returncode != 0 or (dest != "music" and not filepath):
+ if WALL_RE.search("\n".join(tail)):
+ raise WallError(url)
+ log(f"job {job['id']} attempt failed (exit {proc.returncode})")
+ return False
+ if dest == "indie":
+ write_nfo(filepath, "episodedetails", {
+ "title": job.get("ep_title") or job["title"],
+ "season": job["season"], "episode": job["episode"],
+ "aired": job.get("published", ""), "plot": url,
+ })
+ elif dest == "independent":
+ write_nfo(filepath, "movie", {
+ "title": job["title"], "premiered": job.get("published", ""), "plot": url,
+ })
+ update_job(job["id"], status="done", progress="")
+ log(f"job {job['id']} done: {filepath or job['title']}")
+ return True
+
+
+def worker():
+ while True:
+ job = job_queue.get()
+ if wall_active():
+ update_job(job["id"], status="waiting", progress="")
+ while wall_active():
+ time.sleep(30)
+ try:
+ run_job(job)
+ except WallError:
+ set_wall(True)
+ update_job(job["id"], status="queued", progress="")
+ job_queue.put(job)
+ except Exception as e:
+ update_job(job["id"], status="error")
+ log(f"job {job['id']} crashed: {e}")
+
+
+# -------------------------------------------------------------------- web app
+
+app = Flask(__name__)
+
+
+def indie_shows():
+ shows = {}
+ try:
+ names = sorted(os.listdir(INDIE_DIR))
+ except FileNotFoundError:
+ names = []
+ for name in names:
+ path = os.path.join(INDIE_DIR, name)
+ if not os.path.isdir(path):
+ continue
+ seasons = {}
+ for sub in os.listdir(path):
+ m = re.fullmatch(r"Season (\d+)", sub)
+ if m and os.path.isdir(os.path.join(path, sub)):
+ count = sum(1 for f in os.listdir(os.path.join(path, sub))
+ if f.lower().endswith(VIDEO_EXTS))
+ seasons[int(m.group(1))] = count
+ shows[name] = seasons
+ return shows
+
+
+PAGE = """
+
+
+{escape(channel)} uploaded:
' + f'' + f'{h} · {entry["published"]}
' + f'' + f"yt triage · {npending} pending
+{cards} + +{jobs}
+
+"""
+
+
+def card_html(v, shows):
+ opts = ['',
+ '']
+ for s in shows:
+ opts.append(f'')
+ opts.append('')
+ img = (f'recent jobs
"] if jobs else [] + for j in jobs[:15]: + cls = j["status"] if j["status"] in ("done", "error") else "" + prog = f" {j.get('progress', '')}" if j["status"] == "downloading" else "" + retry = ("" if j["status"] != "error" else + f' ') + out.append(f"{icon.get(j['status'], '·')} " + f"{escape(j['title'])} → {escape(j['dest_label'])}{prog}{retry}
") + return "\n".join(out) + + +@app.get("/") +def index(): + with state_lock: + pending = load_json("pending.json", {}) + shows = indie_shows() + cards = "\n".join(card_html(v, shows) for v in reversed(list(pending.values()))) + if not pending: + cards = "nothing pending. enjoy the silence.
" + if wall_active(): + cards = ("youtube has bot-walled this ip — downloads "
+ "are paused and will resume automatically once the wall lifts "
+ "(probed every few hours). queueing still works.
" + cards)
+ return PAGE.format(npending=len(pending), cards=cards,
+ jobs=jobs_html(), shows_json=json.dumps(indie_shows()))
+
+
+@app.get("/jobs.html")
+def jobs_partial():
+ return Response(jobs_html(), mimetype="text/html")
+
+
+@app.post("/retry")
+def retry():
+ jid = request.form["job"]
+ with state_lock:
+ jobs = load_json("jobs.json", [])
+ job = next((j for j in jobs if j["id"] == jid), None)
+ if job:
+ job.update(status="queued", progress="")
+ save_json("jobs.json", jobs)
+ if job:
+ job_queue.put(job)
+ return redirect("/")
+
+
+@app.post("/skip")
+def skip():
+ with state_lock:
+ pending = load_json("pending.json", {})
+ pending.pop(request.form["vid"], None)
+ save_json("pending.json", pending)
+ return redirect("/")
+
+
+@app.post("/add")
+def add():
+ # accepts one url or a whole batch separated by whitespace/commas. cards
+ # appear instantly as stubs and resolve in the background — they can be
+ # ingested before the title/thumbnail has arrived.
+ urls = [u for u in re.split(r"[\s,]+", request.form["url"].strip()) if u]
+ with state_lock:
+ pending = load_json("pending.json", {})
+ for i, url in enumerate(urls):
+ stub_id = f"u{int(time.time() * 1000)}{i}"
+ pending[stub_id] = {"id": stub_id, "title": url, "channel": "",
+ "published": "", "thumb": "", "link": url,
+ "unresolved": True}
+ threading.Thread(target=resolve_and_update, args=(stub_id, url),
+ daemon=True).start()
+ save_json("pending.json", pending)
+ return redirect("/")
+
+
+@app.post("/ingest")
+def ingest():
+ vid = request.form["vid"]
+ with state_lock:
+ pending = load_json("pending.json", {})
+ v = pending.get(vid)
+ if not v:
+ # a stub may have resolved (new key) between page load and submit
+ v = next((p for p in pending.values() if p.get("stub_id") == vid), None)
+ if not v:
+ return Response("video not in pending queue", 404, mimetype="text/plain")
+ dest = request.form["dest"]
+ job = {"id": f"j{int(time.time() * 1000)}", "url": v["link"], "title": v["title"],
+ "channel": v.get("channel", "unknown"), "published": v.get("published", ""),
+ "unresolved": v.get("unresolved", False), "status": "queued", "progress": ""}
+ if dest.startswith("indie|"):
+ show = request.form.get("new_show", "").strip() if dest == "indie|__new__" else dest[6:]
+ if not show:
+ return Response("missing show name", 400, mimetype="text/plain")
+ # a custom episode title; left equal to the card title means "use the
+ # video title" (which, for a still-resolving stub, arrives later)
+ ep_title = request.form.get("ep_title", "").strip()
+ job.update(dest="indie", show=show,
+ ep_title=ep_title if ep_title and ep_title != v["title"] else "",
+ season=int(request.form["season"]), episode=int(request.form["episode"]))
+ job["dest_label"] = f"{show} S{job['season']:02d}E{job['episode']:02d}"
+ elif dest == "independent":
+ job.update(dest="independent", dest_label="Independent")
+ else:
+ job.update(dest="music", dest_label="music_intake")
+ with state_lock:
+ pending.pop(v["id"], None)
+ save_json("pending.json", pending)
+ jobs = load_json("jobs.json", [])
+ jobs.insert(0, job)
+ save_json("jobs.json", jobs[:50])
+ job_queue.put(job)
+ return redirect("/")
+
+
+if __name__ == "__main__":
+ # jobs interrupted by a container restart pick up where they left off
+ with state_lock:
+ for j in reversed(load_json("jobs.json", [])):
+ if j.get("status") in ("queued", "resolving", "downloading", "retrying"):
+ job_queue.put(j)
+ threading.Thread(target=poller, daemon=True).start()
+ threading.Thread(target=worker, daemon=True).start()
+ threading.Thread(target=wall_prober, daemon=True).start()
+ app.run(host="0.0.0.0", port=80, threaded=True)
diff --git a/config/yt/archive-loop.sh b/config/yt/archive-loop.sh
new file mode 100644
index 0000000000000000000000000000000000000000..cce5ccf2815b829cd77d033d45f91bd75b1fb7fc
--- /dev/null
+++ b/config/yt/archive-loop.sh
@@ -0,0 +1,13 @@
+#!/bin/sh
+# subscription pass on a 6h cycle — but if youtube has bot-walled the ip
+# ("sign in to confirm you're not a bot"), back off for a whole day instead
+# of hammering it every cycle, which prolongs the wall.
+while true; do
+ ytdl-sub --config /config-yt/config.yaml sub /config-yt/subscriptions.yaml 2>&1 | tee /tmp/last-pass.log
+ if grep -q "confirm you.re not a bot" /tmp/last-pass.log; then
+ echo "[archive-loop] bot wall detected; sleeping 24h"
+ sleep 86400
+ else
+ sleep 21600
+ fi
+done
diff --git a/config/yt/config.yaml b/config/yt/config.yaml
new file mode 100644
index 0000000000000000000000000000000000000000..76009bbb1ff6a88756fd558da9c9d898ced7ef58
--- /dev/null
+++ b/config/yt/config.yaml
@@ -0,0 +1,28 @@
+# ytdl-sub tool configuration — the channel list lives in subscriptions.yaml
+configuration:
+ working_directory: "/config/work"
+
+presets:
+ # per-channel backlog control: only download uploads on/after download_after
+ # (yyyymmdd). subscriptions.yaml sets the default and per-channel values.
+ only-after:
+ date_range:
+ after: "{download_after}"
+ overrides:
+ download_after: "19700101"
+
+ # output for the Independent library, which is a jellyfin "home videos"
+ # library: flat files (no Season