urieljareth
0682d2d806
Videos view: multi-select (checkboxes) + select-all/deselect-all + bulk action
bar (Download .md batch, Download thumbnails, Download audio). Per-video buttons:
.md download (done) or Process to .md (pending), clip.
Channels: per-channel pending count + 'Download pending .md' (batches pending
video_ids). Each channel keeps Videos/.md folder/Remove actions.
Backend (jobs.py + api.py):
- POST /api/scrape/batch {video_ids} + POST /api/scrape/video/{id} (on-demand
.md for any video incl. pending) as SSE-tracked jobs
- GET /api/videos/{id}/markdown (file download)
- POST /api/tools/thumbnails + GET /api/thumbnails/{id} (local cache, offline)
- GET /api/clip/{id}?from=&to= (transcript segment)
- GET /api/stats (aggregate totals)
- POST /api/tools/audio (mp3 job)
Tools view rebuilt as full Knowledge Base grid surfacing every OPPORTUNITIES.md
feature as electable buttons (md download, re-render, export, search, analysis,
thumbnails, audio, clip, stats, cookies, channels, watch, open folders).
Floating job widget (SSE) for any running job. Thumbnails served via
/api/thumbnails/{id} everywhere (works offline once cached).
Launcher: robust Open-Browser helper (3 fallback methods) auto-opens
http://127.0.0.1:<port> on start + on idempotent re-open. data/ gitignored.
276 lines
12 KiB
Python
276 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from collections import deque
|
|
from typing import Any
|
|
|
|
from ..config import Config, load_config
|
|
from ..cookies import resolve_active_path
|
|
from ..discover import discover_channel
|
|
from ..pipeline import process_video
|
|
from ..ratelimit import polite_sleep
|
|
from ..store import Store
|
|
|
|
|
|
class JobManager:
|
|
"""Single-worker scrape job runner with an in-memory event log per job (SSE-polled)."""
|
|
|
|
def __init__(self, store: Store, cfg: Config):
|
|
self.store = store
|
|
self.cfg = cfg
|
|
self._lock = threading.Lock()
|
|
self._events: dict[str, list[dict]] = {}
|
|
self._cancel: set[str] = set()
|
|
self._queue: deque[str] = deque()
|
|
self._thread = threading.Thread(target=self._worker, daemon=True, name="scrape-worker")
|
|
self._running = False
|
|
|
|
def start(self) -> None:
|
|
if self._running:
|
|
return
|
|
self._running = True
|
|
self._thread.start()
|
|
|
|
def enqueue(self, channel_id: str | None, opts: dict[str, Any]) -> str:
|
|
job_id = uuid.uuid4().hex[:12]
|
|
self.store.create_job(job_id, channel_id, opts)
|
|
with self._lock:
|
|
self._events[job_id] = []
|
|
self._emit(job_id, "queued", {"job_id": job_id, "channel_id": channel_id, "opts": opts})
|
|
self._queue.append(job_id)
|
|
return job_id
|
|
|
|
def cancel(self, job_id: str) -> bool:
|
|
job = self.store.get_job(job_id)
|
|
if not job or job.status not in ("queued", "running"):
|
|
return False
|
|
self._cancel.add(job_id)
|
|
self._emit(job_id, "log", {"msg": "cancel requested"})
|
|
return True
|
|
|
|
def events_since(self, job_id: str, cursor: int) -> list[dict]:
|
|
with self._lock:
|
|
evs = self._events.get(job_id, [])
|
|
return evs[cursor:]
|
|
|
|
def is_terminal(self, job_id: str) -> str | None:
|
|
job = self.store.get_job(job_id)
|
|
if not job:
|
|
return None
|
|
if job.status in ("done", "error", "cancelled"):
|
|
return job.status
|
|
return None
|
|
|
|
# ---------------------------------------------------------- internals
|
|
|
|
def _emit(self, job_id: str, event: str, data: dict) -> None:
|
|
with self._lock:
|
|
self._events.setdefault(job_id, []).append({"event": event, "data": data})
|
|
|
|
def _worker(self) -> None:
|
|
while True:
|
|
try:
|
|
job_id = self._queue.popleft()
|
|
except IndexError:
|
|
time.sleep(0.2)
|
|
continue
|
|
try:
|
|
self._run_job(job_id)
|
|
except Exception as exc:
|
|
self.store.update_job(job_id, status="error", last_error=str(exc), finished=True)
|
|
self._emit(job_id, "error", {"message": str(exc)})
|
|
|
|
def _run_job(self, job_id: str) -> None:
|
|
job = self.store.get_job(job_id)
|
|
if not job:
|
|
return
|
|
opts = json.loads(job.opts_json) if job.opts_json else {}
|
|
if opts.get("video_ids"):
|
|
self._run_batch(job_id, opts)
|
|
return
|
|
if opts.get("mode") == "audio":
|
|
self._run_audio(job_id, opts)
|
|
return
|
|
self._run_channel(job_id, opts)
|
|
|
|
def _resolve_channel_for_video(self, video_id: str) -> tuple[str, str, str]:
|
|
row = self.store.get_video(video_id)
|
|
if not row:
|
|
return ("unknown", "", f"https://www.youtube.com/watch?v={video_id}")
|
|
ch = self.store.get_channel(row.channel_id) or {}
|
|
name = ch.get("name") or "unknown"
|
|
handle = (ch.get("handle") or "").lstrip("@")
|
|
url = f"https://www.youtube.com/@{handle}/videos" if handle else row.url
|
|
return (name, row.channel_id, url)
|
|
|
|
def _run_batch(self, job_id: str, opts: dict[str, Any]) -> None:
|
|
video_ids = opts.get("video_ids") or []
|
|
cookie_path = opts.get("cookies_file") or resolve_active_path(self.store)
|
|
total = len(video_ids)
|
|
self.store.update_job(job_id, status="running", total=total, completed=0)
|
|
self._emit(job_id, "progress", {"completed": 0, "total": total})
|
|
self._emit(job_id, "log", {"msg": f"processing {total} videos"})
|
|
cfg = _clone_config(self.cfg)
|
|
if opts.get("languages"):
|
|
cfg.languages = opts["languages"]
|
|
completed = 0
|
|
for vid in video_ids:
|
|
if job_id in self._cancel:
|
|
self.store.update_job(job_id, status="cancelled", completed=completed, finished=True)
|
|
self._emit(job_id, "cancelled", {"completed": completed, "total": total})
|
|
return
|
|
row = self.store.get_video(vid)
|
|
if not row:
|
|
self._emit(job_id, "log", {"msg": f"skip unknown {vid}"})
|
|
completed += 1
|
|
self.store.update_job(job_id, completed=completed)
|
|
continue
|
|
channel_name, channel_id, channel_url = self._resolve_channel_for_video(vid)
|
|
status = process_video(
|
|
row, cfg, self.store, channel_name, channel_id, channel_url,
|
|
cookies_file=cookie_path,
|
|
on_log=lambda m: self._emit(job_id, "log", {"msg": m}),
|
|
)
|
|
completed += 1
|
|
self.store.update_job(job_id, completed=completed)
|
|
self._emit(job_id, "progress", {"completed": completed, "total": total, "video_id": vid, "status": status})
|
|
self.store.update_job(job_id, status="done", completed=completed, finished=True)
|
|
self._emit(job_id, "done", {"completed": completed, "total": total})
|
|
|
|
def _run_audio(self, job_id: str, opts: dict[str, Any]) -> None:
|
|
import yt_dlp
|
|
video_ids = opts.get("video_ids") or []
|
|
channel_id = opts.get("channel_id")
|
|
cookie_path = opts.get("cookies_file") or resolve_active_path(self.store)
|
|
if video_ids:
|
|
videos = [self.store.get_video(v) for v in video_ids if self.store.get_video(v)]
|
|
else:
|
|
videos = [v for v in self.store.get_all(channel_id) if v.status == "done"]
|
|
out_dir = Path(self.cfg.output_dir_resolved).parent / "audio"
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
total = len(videos)
|
|
self.store.update_job(job_id, status="running", total=total, completed=0)
|
|
self._emit(job_id, "log", {"msg": f"downloading {total} audio tracks -> {out_dir}"})
|
|
ydl_opts = {
|
|
"format": "bestaudio/best",
|
|
"outtmpl": str(out_dir / "%(title)s.%(ext)s"),
|
|
"postprocessors": [{"key": "FFmpegExtractAudio", "preferredcodec": "mp3", "preferredquality": "128"}],
|
|
"quiet": True, "no_warnings": True, "noprogress": True,
|
|
}
|
|
if cookie_path:
|
|
ydl_opts["cookiefile"] = cookie_path
|
|
completed = 0
|
|
with yt_dlp.YoutubeDL(ydl_opts) as ydl:
|
|
for v in videos:
|
|
if job_id in self._cancel:
|
|
self.store.update_job(job_id, status="cancelled", completed=completed, finished=True)
|
|
self._emit(job_id, "cancelled", {"completed": completed, "total": total})
|
|
return
|
|
try:
|
|
ydl.download([v.url])
|
|
self._emit(job_id, "log", {"msg": f"OK {v.video_id}"})
|
|
except Exception as exc:
|
|
self._emit(job_id, "log", {"msg": f"FAIL {v.video_id}: {exc}"})
|
|
completed += 1
|
|
self.store.update_job(job_id, completed=completed)
|
|
self._emit(job_id, "progress", {"completed": completed, "total": total, "video_id": v.video_id})
|
|
self.store.update_job(job_id, status="done", completed=completed, finished=True)
|
|
self._emit(job_id, "done", {"completed": completed, "total": total})
|
|
|
|
def _run_channel(self, job_id: str, opts: dict[str, Any]) -> None:
|
|
job = self.store.get_job(job_id)
|
|
if not job:
|
|
return
|
|
channel_id = job.channel_id
|
|
cfg = _clone_config(self.cfg)
|
|
channel_url = _resolve_channel_url(self.store, cfg, channel_id)
|
|
if not channel_url:
|
|
self.store.update_job(job_id, status="error", last_error="no channel url", finished=True)
|
|
self._emit(job_id, "error", {"message": "no channel url"})
|
|
return
|
|
cfg.channel_url = channel_url
|
|
|
|
self.store.update_job(job_id, status="running")
|
|
self._emit(job_id, "log", {"msg": f"discovering {channel_url}"})
|
|
try:
|
|
_channel_id, channel_name, refs = discover_channel(channel_url, sleep_subrequests=cfg.yt_dlp.sleep_subrequests)
|
|
except Exception as exc:
|
|
self.store.update_job(job_id, status="error", last_error=str(exc), finished=True)
|
|
self._emit(job_id, "error", {"message": f"discovery failed: {exc}"})
|
|
return
|
|
|
|
limit = opts.get("limit")
|
|
since = opts.get("since")
|
|
languages = opts.get("languages")
|
|
include_shorts = opts.get("include_shorts", cfg.include_shorts)
|
|
no_live = opts.get("no_live", not cfg.include_live)
|
|
cookie_override = opts.get("cookies_file")
|
|
|
|
if languages:
|
|
cfg.languages = languages
|
|
if since:
|
|
refs = [r for r in refs if (r.upload_date or "") >= since.replace("-", "")]
|
|
if not include_shorts:
|
|
refs = [r for r in refs if "/shorts/" not in (r.url or "")]
|
|
if no_live:
|
|
refs = [r for r in refs if not (r.url or "").startswith("https://www.youtube.com/live/")]
|
|
if limit:
|
|
refs = refs[: int(limit)]
|
|
|
|
self.store.upsert_channel(_channel_id, _handle(channel_url), channel_name, len(refs))
|
|
self.store.upsert_videos(refs)
|
|
|
|
pending = [r for r in self.store.get_pending(_channel_id) if r.video_id in {x.video_id for x in refs}]
|
|
total = len(pending)
|
|
self.store.update_job(job_id, total=total)
|
|
self._emit(job_id, "progress", {"completed": 0, "total": total})
|
|
self._emit(job_id, "log", {"msg": f"{total} videos to process"})
|
|
|
|
cookie_path = cookie_override or resolve_active_path(self.store)
|
|
completed = 0
|
|
for row in pending:
|
|
if job_id in self._cancel:
|
|
self.store.update_job(job_id, status="cancelled", completed=completed, finished=True)
|
|
self._emit(job_id, "cancelled", {"completed": completed, "total": total})
|
|
return
|
|
status = process_video(
|
|
row, cfg, self.store, channel_name, _channel_id, channel_url,
|
|
cookies_file=cookie_path,
|
|
on_log=lambda m: self._emit(job_id, "log", {"msg": m}),
|
|
)
|
|
completed += 1
|
|
self.store.update_job(job_id, completed=completed)
|
|
self._emit(job_id, "progress", {"completed": completed, "total": total, "video_id": row.video_id, "status": status})
|
|
if completed < total:
|
|
polite_sleep(cfg.delay.min_seconds, cfg.delay.max_seconds)
|
|
|
|
self.store.update_job(job_id, status="done", completed=completed, finished=True)
|
|
self._emit(job_id, "done", {"completed": completed, "total": total})
|
|
|
|
|
|
def _clone_config(cfg: Config) -> Config:
|
|
import copy
|
|
return copy.deepcopy(cfg)
|
|
|
|
|
|
def _resolve_channel_url(store: Store, cfg: Config, channel_id: str | None) -> str | None:
|
|
if not channel_id:
|
|
return cfg.channel_url or None
|
|
ch = store.get_channel(channel_id)
|
|
if not ch:
|
|
return None
|
|
handle = (ch.get("handle") or "").lstrip("@")
|
|
if handle:
|
|
return f"https://www.youtube.com/@{handle}/videos"
|
|
return f"https://www.youtube.com/channel/{channel_id}"
|
|
|
|
|
|
def _handle(url: str) -> str:
|
|
if "@" in url:
|
|
return "@" + url.split("@", 1)[1].split("/", 1)[0]
|
|
return ""
|