feat(api): add job retry/priority/clear, history, config and maintenance endpoints

Persistent job history (capped by serve.history_limit, stored under cache), profiles/config reads, maintenance + env-check endpoints, and 409 conflict code.
This commit is contained in:
imSp4rky
2026-07-01 22:42:48 -06:00
parent f69ebb4fba
commit 7fdd514e18
8 changed files with 1345 additions and 61 deletions
+11
View File
@@ -124,6 +124,17 @@ download_job_retention_hours: 48
--- ---
## serve.history_limit (int)
How many terminal jobs to keep in the persistent history file (`api_history.jsonl`, stored under the cache directory so it stays out of the config/data tree). Older entries are dropped as new jobs finish. Set to `0` for unlimited. Default: `100`.
```yaml
serve:
history_limit: 250
```
---
## debug (bool) ## debug (bool)
Enables comprehensive debug logging. Default: `false` Enables comprehensive debug logging. Default: `false`
+4
View File
@@ -7,6 +7,10 @@ queued -> cancelled
downloading -> cancelled downloading -> cancelled
``` ```
## Progress reporting
A multi-episode request (`wanted: ["S01E01", ...]`) is a single job that processes episodes sequentially. During a run the job status updates live: `current_title` names the episode downloading now, `output_files` grows as each episode finishes muxing, and `track_progress` lists per-track progress when `downloads` > 1 runs tracks concurrently. `progress` and track counts reset as each episode starts; use `output_files` against `parameters.wanted` to derive overall completion.
Jobs are retained for 24 hours after completion (override via top-level `download_job_retention_hours` in `unshackle.yaml`). The server runs up to 2 concurrent download jobs by default; override via top-level `max_concurrent_downloads`. This is independent of `serve.downloads`, which controls parallel tracks **within** a single job. Jobs are retained for 24 hours after completion (override via top-level `download_job_retention_hours` in `unshackle.yaml`). The server runs up to 2 concurrent download jobs by default; override via top-level `max_concurrent_downloads`. This is independent of `serve.downloads`, which controls parallel tracks **within** a single job.
Remote sessions are managed by `SessionStore` (`unshackle/core/api/session_store.py`); idle sessions and their `InputBridge` instances are cleaned up by a background loop started/stopped with the app lifecycle. Remote sessions are managed by `SessionStore` (`unshackle/core/api/session_store.py`); idle sessions and their `InputBridge` instances are cleaned up by a background loop started/stopped with the app lifecycle.
+247 -5
View File
@@ -242,6 +242,7 @@ List all download jobs with optional filtering and sorting. Disabled in `--remot
| `service` | string | all | Filter by service tag | | `service` | string | all | Filter by service tag |
| `sort_by` | string | `created_time` | Sort field: `created_time`, `started_time`, `completed_time`, `progress`, `status`, `service` | | `sort_by` | string | `created_time` | Sort field: `created_time`, `started_time`, `completed_time`, `progress`, `status`, `service` |
| `sort_order` | string | `desc` | Sort order: `asc`, `desc` | | `sort_order` | string | `desc` | Sort order: `asc`, `desc` |
| `full` | string | `false` | `true` to include full job details per job (parameters, timestamps, `output_files`, error fields) |
```bash ```bash
curl -H "X-Secret-Key: $KEY" "http://localhost:8786/api/download/jobs?status=completed" curl -H "X-Secret-Key: $KEY" "http://localhost:8786/api/download/jobs?status=completed"
@@ -256,22 +257,263 @@ Get detailed information about a specific download job including progress, param
```json ```json
{ {
"job_id": "504db959-80b0-446c-a764-7924b761d613", "job_id": "504db959-80b0-446c-a764-7924b761d613",
"status": "completed", "status": "downloading",
"created_time": "2026-02-27T18:00:00.000000", "created_time": "2026-02-27T18:00:00.000000",
"service": "EXAMPLE1", "service": "EXAMPLE1",
"title_id": "abc123def456", "title_id": "abc123def456",
"progress": 100.0, "title": "Example Show",
"progress": 74.9,
"phase": "downloading video 1080p SDR",
"current_title": "S01E04",
"completed_tracks": 4,
"total_tracks": 13,
"active_tracks": ["video 1080p SDR"],
"track_progress": [
{ "label": "video 1080p SDR", "progress": 74.9, "speed": "DASH 265.5 MB/s" }
],
"segments_done": 75.0,
"segments_total": 100.0,
"speed": "DASH 265.5 MB/s",
"parameters": { }, "parameters": { },
"started_time": "2026-02-27T18:00:01.000000", "started_time": "2026-02-27T18:00:01.000000",
"completed_time": "2026-02-27T18:00:15.000000", "completed_time": null,
"output_files": [], "output_files": [
"/downloads/Example.Show.S01E01.mkv",
"/downloads/Example.Show.S01E02.mkv",
"/downloads/Example.Show.S01E03.mkv"
],
"error_message": null, "error_message": null,
"error_details": null "error_details": null
} }
``` ```
**Progress fields:**
| Field | Type | Description |
| --- | --- | --- |
| `progress` | number | Overall progress for the title currently downloading, 0–100. Resets per episode on multi-episode jobs |
| `phase` | string | Human-readable current step, e.g. `downloading video 1080p SDR`, `muxing` |
| `title` | string | Human-readable show/movie name, available once the download starts |
| `current_title` | string | `SxxEyy` code of the episode downloading now (episodic jobs only) |
| `completed_tracks` / `total_tracks` | int | Track counts for the current title |
| `active_tracks` | array[string] | Labels of tracks downloading now |
| `track_progress` | array | Per-track `{label, progress, speed}` for each concurrently active track |
| `segments_done` / `segments_total` | number | Segment counts of the track downloading now |
| `speed` | string | Transfer speed of the track downloading now |
| `output_files` | array[string] | Final file paths, appended as each title finishes muxing |
The list endpoint (`GET /api/download/jobs`) returns the same progress fields per job but omits `parameters`, timestamps, `output_files`, and error details.
--- ---
## DELETE /api/download/jobs/{job_id} ## DELETE /api/download/jobs/{job_id}
Cancel a queued or running download job. Returns 400 if the job has already terminated. Cancel a queued or running download job, or remove a finished one.
- `queued` / `downloading` job: cancelled, returns `200` with `{"status": "success", "message": "Job cancelled"}`.
- `completed` / `failed` / `cancelled` job: removed from the manager entirely, returns `204` (no body).
- Unknown job: `404`.
---
## POST /api/download/jobs/clear-finished
Remove all `completed`, `failed`, and `cancelled` jobs from the manager. Disabled in `--remote-only` mode.
```json
{ "removed": 3 }
```
---
## POST /api/download/jobs/{job_id}/retry
Enqueue a new job reusing a finished job's service, title, and original (unredacted) parameters. Only allowed on `completed`, `failed`, or `cancelled` jobs — returns `409` otherwise, `404` if unknown. The original job is left in place.
Creation-time authorization is re-applied at retry time: the job's service must pass the caller's service allowlist (otherwise `400` invalid service), and jobs whose original parameters carried a `cdm` override or per-job credentials return `403` if `serve.cdm_overrides` / `serve.allow_job_credentials` no longer permit them.
Returns `202` with the same shape as `POST /api/download`:
```json
{
"job_id": "b7e9c1f2-...",
"status": "queued",
"created_time": "2026-02-27T18:00:00.000000"
}
```
---
## POST /api/download/jobs/{job_id}/priority
Move a `queued` job to the front of the download queue so a worker picks it up next. Returns `409` if the job is not queued, `404` if unknown.
```json
{ "job_id": "504db959-...", "position": "front" }
```
---
## GET /api/profiles
List named credential profiles per service, usable as the `profile` parameter on search/list/download. Only services whose credentials are configured as a `profile-name: credential` mapping appear (a `default` key is included if present); a service with a single plain (unnamed) credential is omitted entirely. Filtered by the caller's service allowlist. Disabled in `--remote-only` mode.
```bash
curl -H "X-Secret-Key: $KEY" http://localhost:8786/api/profiles
```
```json
{ "profiles": { "EXAMPLE1": ["default", "uk", "us"], "EXAMPLE2": ["alice"] } }
```
---
## GET /api/config
Read-only, redacted view of the effective server configuration for a UI settings page. Secrets (`api_secret`, `users`, credentials, tokens) are never included; secret-looking keys inside `dl` are masked with `***`. Disabled in `--remote-only` mode.
```bash
curl -H "X-Secret-Key: $KEY" http://localhost:8786/api/config
```
```json
{
"config": {
"dl": { "lang": "orig", "sub_format": "srt", "downloads": 12 },
"serve": {
"max_concurrent_downloads": 2,
"job_retention_hours": 24,
"history_limit": 100,
"services": null,
"remote_only": false,
"cdm_overrides": null,
"allow_job_credentials": false
},
"directories": {
"downloads": "/srv/downloads",
"temp": "/srv/temp",
"cache": "/srv/cache"
},
"services": ["EXAMPLE1", "EXAMPLE2"]
}
}
```
| Field | Description |
| --- | --- |
| `dl` | Default `dl` parameters from the config file (redacted) |
| `serve.services` | Global service allowlist, or `null` when all services are allowed |
| `serve.cdm_overrides` | Permitted per-request CDM device names (list), `true`, or `null` |
| `services` | Available service tags, filtered by the caller's allowlist |
---
## GET /api/history
Persistent download history. Every job that reaches a terminal state (`completed`, `failed`, `cancelled`) is appended as one JSON line to `api_history.jsonl` in the cache directory, surviving restarts and job cleanup. Only the newest `serve.history_limit` entries are retained (default `100`, `0` for unlimited; see Advanced Config). Newest first. Corrupt lines are skipped; a missing file yields an empty list. Entries are filtered by the caller's service allowlist. Disabled in `--remote-only` mode.
**Query parameters:**
| Parameter | Type | Default | Description |
| --- | --- | --- | --- |
| `limit` | int | `100` | Maximum entries returned (must be >= 1) |
| `service` | string | all | Filter by service tag (case-insensitive) |
```bash
curl -H "X-Secret-Key: $KEY" "http://localhost:8786/api/history?limit=50&service=EXAMPLE1"
```
```json
{
"history": [
{
"job_id": "504db959-80b0-446c-a764-7924b761d613",
"service": "EXAMPLE1",
"title_id": "abc123def456",
"title": "Example Show",
"status": "completed",
"created_time": "2026-02-27T18:00:00.000000",
"completed_time": "2026-02-27T18:12:34.000000",
"output_files": ["/downloads/Example.Show.S01E01.mkv"],
"parameters": { "wanted": ["S01E01"], "quality": [1080], "skip_dl": false },
"error_message": null
}
],
"count": 1
}
```
`error_message` is redacted with the same secret-scrubbing applied to job responses. `parameters` is the redacted download selection the job ran with (quality/vcodec/range/wanted/skip_dl, etc.). `completed_time` is `null` for jobs cancelled while still queued.
---
## DELETE /api/history/{job_id}
Remove a single history entry by `job_id` (the file is rewritten without it). Returns `204` on success, `404` if no matching entry exists or it falls outside the caller's service allowlist. Corrupt lines in the history file are preserved.
```bash
curl -X DELETE -H "X-Secret-Key: $KEY" "http://localhost:8786/api/history/504db959-80b0-446c-a764-7924b761d613"
```
---
## POST /api/maintenance/clear-cache
Delete the contents of the cache directory (the directory is recreated empty). `freed_bytes` is best effort. This also removes `api_history.jsonl`, so persistent download history is cleared. Returns `409` with the active job ids if any job is currently downloading, since in-flight jobs may be reading the cache. Disabled in `--remote-only` mode.
```bash
curl -X POST -H "X-Secret-Key: $KEY" http://localhost:8786/api/maintenance/clear-cache
```
```json
{ "cleared": true, "freed_bytes": 1234567 }
```
---
## POST /api/maintenance/clear-temp
Same as clear-cache, for the temp directory. Also refuses with `409` while any job is downloading — active workers write segment files under temp.
```json
{ "cleared": true, "freed_bytes": 0 }
```
---
## POST /api/maintenance/refresh-services
Force-sync (git pull, hard reset to upstream) every service repo configured in `directories.services`. `refreshed` is `true` when all repos synced (or none are configured). Disabled in `--remote-only` mode.
```bash
curl -X POST -H "X-Secret-Key: $KEY" http://localhost:8786/api/maintenance/refresh-services
```
```json
{
"refreshed": true,
"repos": [
{ "spec": "https://github.com/example/services", "updated": true, "changes": ["~ EXAMPLE1 (modified)"] }
]
}
```
An empty `changes` array means the repo was already up to date. A repo that failed to sync has `updated: false` (and flips `refreshed` to `false`).
---
## GET /api/env/check
Install status of the binaries `unshackle env check` inspects (FFmpeg, MKVToolNix, shaka-packager, etc.). `version` is a best-effort probe and may be `null` even when installed. Disabled in `--remote-only` mode.
```bash
curl -H "X-Secret-Key: $KEY" http://localhost:8786/api/env/check
```
```json
{
"checks": [
{ "name": "FFmpeg", "installed": true, "version": "7.1.0", "required": true },
{ "name": "mp4decrypt", "installed": false, "version": null, "required": false }
]
}
```
+35 -16
View File
@@ -17,16 +17,9 @@ from unshackle.core.constants import context_settings
from unshackle.core.services import Services from unshackle.core.services import Services
@click.group(short_help="Manage and configure the project environment.", context_settings=context_settings) def get_dependencies() -> list[dict]:
def env() -> None: """Binary dependency inventory shared by `env check` and the API env check."""
"""Manage and configure the project environment.""" return [
@env.command()
def check() -> None:
"""Checks environment for the required dependencies."""
# Define all dependencies
all_deps = [
# Core Media Tools # Core Media Tools
{"name": "FFmpeg", "binary": binaries.FFMPEG, "required": True, "desc": "Media processing", "cat": "Core"}, {"name": "FFmpeg", "binary": binaries.FFMPEG, "required": True, "desc": "Media processing", "cat": "Core"},
{"name": "FFprobe", "binary": binaries.FFProbe, "required": True, "desc": "Media analysis", "cat": "Core"}, {"name": "FFprobe", "binary": binaries.FFProbe, "required": True, "desc": "Media analysis", "cat": "Core"},
@@ -99,6 +92,34 @@ def check() -> None:
{"name": "git", "binary": binaries.Git, "required": False, "desc": "Service repos", "cat": "Network"}, {"name": "git", "binary": binaries.Git, "required": False, "desc": "Service repos", "cat": "Network"},
] ]
def clear_directory(path: Path) -> tuple[int, int]:
"""Delete a directory's contents, returning (files_removed, freed_bytes); recreates the dir."""
files_count = 0
freed_bytes = 0
if path.exists():
for entry in path.glob("**/*"):
if entry.is_file():
files_count += 1
try:
freed_bytes += entry.stat().st_size
except OSError:
pass
shutil.rmtree(path, ignore_errors=True)
path.mkdir(parents=True, exist_ok=True)
return files_count, freed_bytes
@click.group(short_help="Manage and configure the project environment.", context_settings=context_settings)
def env() -> None:
"""Manage and configure the project environment."""
@env.command()
def check() -> None:
"""Checks environment for the required dependencies."""
all_deps = get_dependencies()
# Track overall status # Track overall status
all_required_installed = True all_required_installed = True
total_installed = 0 total_installed = 0
@@ -214,12 +235,11 @@ def cache(service: Optional[str]) -> None:
if service: if service:
cache_dir = cache_dir / Services.get_tag(service) cache_dir = cache_dir / Services.get_tag(service)
log.info(f"Clearing cache directory: {cache_dir}") log.info(f"Clearing cache directory: {cache_dir}")
files_count = len(list(cache_dir.glob("**/*"))) files_count, _ = clear_directory(cache_dir)
if not files_count: if not files_count:
log.info("No files to delete") log.info("No files to delete")
else: else:
log.info(f"Deleting {files_count} files...") log.info(f"Deleted {files_count} files")
shutil.rmtree(cache_dir)
log.info("Cleared") log.info("Cleared")
@@ -228,10 +248,9 @@ def temp() -> None:
"""Clear the environment temp directory.""" """Clear the environment temp directory."""
log = logging.getLogger("env") log = logging.getLogger("env")
log.info(f"Clearing temp directory: {config.directories.temp}") log.info(f"Clearing temp directory: {config.directories.temp}")
files_count = len(list(config.directories.temp.glob("**/*"))) files_count, _ = clear_directory(config.directories.temp)
if not files_count: if not files_count:
log.info("No files to delete") log.info("No files to delete")
else: else:
log.info(f"Deleting {files_count} files...") log.info(f"Deleted {files_count} files")
shutil.rmtree(config.directories.temp)
log.info("Cleared") log.info("Cleared")
+172 -2
View File
@@ -7,6 +7,7 @@ import sys
import tempfile import tempfile
import threading import threading
import uuid import uuid
from collections import deque
from contextlib import suppress from contextlib import suppress
from dataclasses import dataclass, field from dataclasses import dataclass, field
from datetime import datetime, timedelta from datetime import datetime, timedelta
@@ -75,6 +76,10 @@ class JobStatus(Enum):
CANCELLED = "cancelled" CANCELLED = "cancelled"
# Statuses a job can never leave; such jobs are safe to remove or retry.
TERMINAL_STATUSES = (JobStatus.COMPLETED, JobStatus.FAILED, JobStatus.CANCELLED)
@dataclass @dataclass
class DownloadJob: class DownloadJob:
"""Represents a download job with all its parameters and status.""" """Represents a download job with all its parameters and status."""
@@ -119,6 +124,9 @@ class DownloadJob:
# Cancellation support # Cancellation support
cancel_event: threading.Event = field(default_factory=threading.Event) cancel_event: threading.Event = field(default_factory=threading.Event)
# Guards against writing the same job to the persistent history file twice.
history_recorded: bool = False
def to_dict(self, include_full_details: bool = False) -> Dict[str, Any]: def to_dict(self, include_full_details: bool = False) -> Dict[str, Any]:
"""Convert job to dictionary for JSON response.""" """Convert job to dictionary for JSON response."""
result = { result = {
@@ -161,6 +169,125 @@ class DownloadJob:
return result return result
def _history_path() -> Path:
"""Path of the persistent job history file (under the cache dir, kept out of the config/data tree)."""
from unshackle.core.config import config
return config.directories.cache / "api_history.jsonl"
def _history_limit() -> int:
"""Max history entries to retain (serve.history_limit, default 100; <= 0 means unlimited)."""
from unshackle.core.config import config
try:
return int((config.serve or {}).get("history_limit", 100))
except (TypeError, ValueError):
return 100
def record_job_history(job: DownloadJob) -> None:
"""Append a terminal job's summary as one JSON line to the history file (best effort)."""
if job.history_recorded or job.status not in TERMINAL_STATUSES:
return
job.history_recorded = True
entry = {
"job_id": job.job_id,
"service": job.service,
"title_id": job.title_id,
"title": job.title,
"status": job.status.value,
"created_time": job.created_time.isoformat(),
"completed_time": job.completed_time.isoformat() if job.completed_time else None,
"output_files": job.output_files,
"parameters": _redact_parameters(job.parameters),
"error_message": _redact_text(job.error_message, job.parameters),
}
try:
path = _history_path()
path.parent.mkdir(parents=True, exist_ok=True)
line = json.dumps(entry)
limit = _history_limit()
# Serialized on the manager's event loop, so read-append-trim doesn't race.
if limit > 0 and path.exists():
existing = [ln for ln in path.read_text(encoding="utf-8").splitlines() if ln.strip()]
existing.append(line)
path.write_text("\n".join(existing[-limit:]) + "\n", encoding="utf-8") # keep newest N
else:
with open(path, "a", encoding="utf-8") as handle:
handle.write(line + "\n")
except OSError as e:
log.warning(f"Could not write job history: {e}")
def read_job_history(limit: int = 100, service: Optional[str] = None) -> List[Dict[str, Any]]:
"""Read persisted job history, newest first; corrupt lines are skipped, missing file = empty."""
path = _history_path()
if not path.exists():
return []
try:
lines = path.read_text(encoding="utf-8").splitlines()
except OSError as e:
log.warning(f"Could not read job history: {e}")
return []
entries: List[Dict[str, Any]] = []
for line in lines:
line = line.strip()
if not line:
continue
try:
entry = json.loads(line)
except json.JSONDecodeError:
continue
if not isinstance(entry, dict):
continue
if service and str(entry.get("service") or "").upper() != service.upper():
continue
entries.append(entry)
entries.reverse() # newest first
return entries[:limit] if limit and limit > 0 else entries
def delete_job_history(job_id: str, allowed: Optional[set] = None) -> bool:
"""Remove a history entry by job_id (rewriting the file). Returns True if one was deleted.
When `allowed` is given, an entry outside the caller's service allowlist is treated as
absent (returns False) so it can't be deleted by a restricted key.
"""
path = _history_path()
if not path.exists():
return False
try:
lines = path.read_text(encoding="utf-8").splitlines()
except OSError as e:
log.warning(f"Could not read job history: {e}")
return False
kept: List[str] = []
deleted = False
for line in lines:
stripped = line.strip()
if not stripped:
continue
try:
entry = json.loads(stripped)
except json.JSONDecodeError:
kept.append(stripped) # preserve corrupt lines untouched
continue
match = isinstance(entry, dict) and entry.get("job_id") == job_id
if match and (allowed is None or str(entry.get("service") or "").upper() in allowed):
deleted = True
continue
kept.append(stripped)
if not deleted:
return False
try:
path.write_text("\n".join(kept) + ("\n" if kept else ""), encoding="utf-8")
except OSError as e:
log.warning(f"Could not write job history: {e}")
return False
return True
def to_enum(values: List[str], enum_cls: type[Enum]) -> List[Enum]: def to_enum(values: List[str], enum_cls: type[Enum]) -> List[Enum]:
"""Map case-insensitive strings (by member name or value) to enum members, dropping unknowns.""" """Map case-insensitive strings (by member name or value) to enum members, dropping unknowns."""
lookup = {m.name.upper(): m for m in enum_cls} lookup = {m.name.upper(): m for m in enum_cls}
@@ -471,6 +598,9 @@ class DownloadQueueManager:
self._jobs: Dict[str, DownloadJob] = {} self._jobs: Dict[str, DownloadJob] = {}
self._job_queue: asyncio.Queue = asyncio.Queue() self._job_queue: asyncio.Queue = asyncio.Queue()
# asyncio.Queue has no reorder, so prioritized jobs go in a deque workers drain first.
self._priority_jobs: deque = deque()
self._promoted_ids: set = set() # their stale FIFO-queue copies get skipped via this set
self._active_downloads: Dict[str, asyncio.Task] = {} self._active_downloads: Dict[str, asyncio.Task] = {}
self._download_processes: Dict[str, asyncio.subprocess.Process] = {} self._download_processes: Dict[str, asyncio.subprocess.Process] = {}
self._job_temp_files: Dict[str, Dict[str, str]] = {} self._job_temp_files: Dict[str, Dict[str, str]] = {}
@@ -516,6 +646,8 @@ class DownloadQueueManager:
if job.status == JobStatus.QUEUED: if job.status == JobStatus.QUEUED:
job.status = JobStatus.CANCELLED job.status = JobStatus.CANCELLED
job.cancel_event.set() # Signal cancellation job.cancel_event.set() # Signal cancellation
job.completed_time = datetime.now()
record_job_history(job) # queued jobs never reach a worker, so persist here
log.info(f"Cancelled queued job {sanitize_log(job_id)}") log.info(f"Cancelled queued job {sanitize_log(job_id)}")
return True return True
elif job.status == JobStatus.DOWNLOADING: elif job.status == JobStatus.DOWNLOADING:
@@ -542,6 +674,36 @@ class DownloadQueueManager:
return False return False
def remove_job(self, job_id: str) -> bool:
"""Remove a terminal (completed/failed/cancelled) job entirely."""
job = self._jobs.get(job_id)
if not job or job.status not in TERMINAL_STATUSES:
return False
del self._jobs[job_id]
log.info(f"Removed job {sanitize_log(job_id)}")
return True
def clear_finished_jobs(self) -> int:
"""Remove all terminal (completed/failed/cancelled) jobs, returning the count removed."""
finished = [job_id for job_id, job in self._jobs.items() if job.status in TERMINAL_STATUSES]
for job_id in finished:
del self._jobs[job_id]
if finished:
log.info(f"Cleared {len(finished)} finished jobs")
return len(finished)
def prioritize_job(self, job_id: str) -> bool:
"""Move a queued job to the front of the line."""
job = self._jobs.get(job_id)
if not job or job.status != JobStatus.QUEUED:
return False
if job in self._priority_jobs:
return True
self._priority_jobs.append(job)
self._promoted_ids.add(job_id)
log.info(f"Moved job {sanitize_log(job_id)} to front of queue")
return True
def cleanup_old_jobs(self) -> int: def cleanup_old_jobs(self) -> int:
"""Remove jobs older than retention period.""" """Remove jobs older than retention period."""
cutoff_time = datetime.now() - timedelta(hours=self.job_retention_hours) cutoff_time = datetime.now() - timedelta(hours=self.job_retention_hours)
@@ -623,8 +785,15 @@ class DownloadQueueManager:
while not self._shutdown_event.is_set(): while not self._shutdown_event.is_set():
try: try:
# Wait for a job or shutdown signal # Prioritized jobs run first; their stale copies still in the FIFO queue are skipped.
job = await asyncio.wait_for(self._job_queue.get(), timeout=1.0) if self._priority_jobs:
job = self._priority_jobs.popleft()
else:
# Wait for a job or shutdown signal
job = await asyncio.wait_for(self._job_queue.get(), timeout=1.0)
if job.job_id in self._promoted_ids:
self._promoted_ids.discard(job.job_id)
continue
if job.status == JobStatus.CANCELLED: if job.status == JobStatus.CANCELLED:
continue continue
@@ -650,6 +819,7 @@ class DownloadQueueManager:
log.error(f"Job {job.job_id} failed: {e}") log.error(f"Job {job.job_id} failed: {e}")
finally: finally:
job.completed_time = datetime.now() job.completed_time = datetime.now()
record_job_history(job)
if job.job_id in self._active_downloads: if job.job_id in self._active_downloads:
del self._active_downloads[job.job_id] del self._active_downloads[job.job_id]
+4
View File
@@ -34,6 +34,8 @@ class APIErrorCode(str, Enum):
SESSION_NOT_FOUND = "SESSION_NOT_FOUND" # Remote-dl session doesn't exist or expired SESSION_NOT_FOUND = "SESSION_NOT_FOUND" # Remote-dl session doesn't exist or expired
TRACK_NOT_FOUND = "TRACK_NOT_FOUND" # Track ID not found in session TRACK_NOT_FOUND = "TRACK_NOT_FOUND" # Track ID not found in session
CONFLICT = "CONFLICT" # Job/resource is in a state that disallows the action
RATE_LIMITED = "RATE_LIMITED" # Service rate limiting RATE_LIMITED = "RATE_LIMITED" # Service rate limiting
# Server errors (5xx) # Server errors (5xx)
@@ -93,6 +95,8 @@ class APIError(Exception):
APIErrorCode.JOB_NOT_FOUND: 404, APIErrorCode.JOB_NOT_FOUND: 404,
APIErrorCode.SESSION_NOT_FOUND: 404, APIErrorCode.SESSION_NOT_FOUND: 404,
APIErrorCode.TRACK_NOT_FOUND: 404, APIErrorCode.TRACK_NOT_FOUND: 404,
# 409 Conflict
APIErrorCode.CONFLICT: 409,
# 429 Too Many Requests # 429 Too Many Requests
APIErrorCode.RATE_LIMITED: 429, APIErrorCode.RATE_LIMITED: 429,
# 500 Internal Server Error # 500 Internal Server Error
+423 -29
View File
@@ -1,6 +1,8 @@
import asyncio import asyncio
import enum import enum
import logging import logging
import re
from pathlib import Path
from types import SimpleNamespace from types import SimpleNamespace
from typing import Any, Dict, List, Optional from typing import Any, Dict, List, Optional
@@ -16,6 +18,7 @@ from unshackle.core.services import Services
from unshackle.core.titles import Episode, Movie, Title_T from unshackle.core.titles import Episode, Movie, Title_T
from unshackle.core.tracks import Audio, Subtitle, Video from unshackle.core.tracks import Audio, Subtitle, Video
from unshackle.core.utils.collections import ci_get from unshackle.core.utils.collections import ci_get
from unshackle.core.utils.redact import REDACTED, URL_USERINFO_RE
log = logging.getLogger("api") log = logging.getLogger("api")
@@ -893,6 +896,37 @@ def validate_download_parameters(data: Dict[str, Any]) -> Optional[str]:
return None return None
def enforce_download_gates(params: Dict[str, Any]) -> None:
"""Enforce serve-config gates on per-job cdm overrides and client-supplied credentials.
A per-request `cdm` selects a server-side device, so it is gated here rather than honoured
blindly. `serve.cdm_overrides` opts in: a list permits only those device names, or `true`
permits any (for a single trusted client). Unset/false rejects every override.
A per-request `credential` (or `credentials` map) authenticates the job with client-supplied
secrets instead of the server-side credentials. Gate it behind `serve.allow_job_credentials`
(default off) so a default deployment stays locked to its own credentials; mirrors the CDM gate.
"""
requested_cdm = params.get("cdm")
if requested_cdm:
allowed = (config.serve or {}).get("cdm_overrides")
permitted = allowed is True or (
isinstance(allowed, (list, tuple, set)) and requested_cdm in allowed
)
if not permitted:
raise APIError(
APIErrorCode.FORBIDDEN,
"The requested CDM is not permitted for API downloads.",
details={"cdm": requested_cdm},
)
if params.get("credential") or params.get("credentials"):
if not (config.serve or {}).get("allow_job_credentials"):
raise APIError(
APIErrorCode.FORBIDDEN,
"Per-request credentials are not permitted for API downloads.",
)
async def download_handler(data: Dict[str, Any], request: Optional[web.Request] = None) -> web.Response: async def download_handler(data: Dict[str, Any], request: Optional[web.Request] = None) -> web.Response:
"""Handle download request - create and queue a download job.""" """Handle download request - create and queue a download job."""
from unshackle.core.api.download_manager import get_download_manager from unshackle.core.api.download_manager import get_download_manager
@@ -917,31 +951,7 @@ async def download_handler(data: Dict[str, Any], request: Optional[web.Request]
details={"service": normalized_service, "title_id": title_id}, details={"service": normalized_service, "title_id": title_id},
) )
# A per-request `cdm` selects a server-side device, so it is gated here rather than honoured enforce_download_gates(data)
# blindly. `serve.cdm_overrides` opts in: a list permits only those device names, or `true`
# permits any (for a single trusted client). Unset/false rejects every override.
requested_cdm = data.get("cdm")
if requested_cdm:
allowed = (config.serve or {}).get("cdm_overrides")
permitted = allowed is True or (
isinstance(allowed, (list, tuple, set)) and requested_cdm in allowed
)
if not permitted:
raise APIError(
APIErrorCode.FORBIDDEN,
"The requested CDM is not permitted for API downloads.",
details={"cdm": requested_cdm},
)
# A per-request `credential` (or `credentials` map) authenticates the job with client-supplied
# secrets instead of the server-side credentials. Gate it behind `serve.allow_job_credentials`
# (default off) so a default deployment stays locked to its own credentials; mirrors the CDM gate.
if data.get("credential") or data.get("credentials"):
if not (config.serve or {}).get("allow_job_credentials"):
raise APIError(
APIErrorCode.FORBIDDEN,
"Per-request credentials are not permitted for API downloads.",
)
try: try:
# Load service module to extract service-specific parameter defaults # Load service module to extract service-specific parameter defaults
@@ -1045,7 +1055,8 @@ async def list_download_jobs_handler(data: Dict[str, Any], request: Optional[web
jobs = sorted(jobs, key=get_sort_key, reverse=reverse) jobs = sorted(jobs, key=get_sort_key, reverse=reverse)
job_list = [job.to_dict(include_full_details=False) for job in jobs] include_full = str(data.get("full") or "").lower() == "true"
job_list = [job.to_dict(include_full_details=include_full) for job in jobs]
return web.json_response({"jobs": job_list}) return web.json_response({"jobs": job_list})
@@ -1091,19 +1102,25 @@ async def get_download_job_handler(job_id: str, request: Optional[web.Request] =
async def cancel_download_job_handler(job_id: str, request: Optional[web.Request] = None) -> web.Response: async def cancel_download_job_handler(job_id: str, request: Optional[web.Request] = None) -> web.Response:
"""Handle cancel download job request.""" """Handle cancel/remove download job request."""
from unshackle.core.api.download_manager import get_download_manager from unshackle.core.api.download_manager import TERMINAL_STATUSES, get_download_manager
try: try:
manager = get_download_manager() manager = get_download_manager()
if not manager.get_job(job_id): job = manager.get_job(job_id)
if not job:
raise APIError( raise APIError(
APIErrorCode.JOB_NOT_FOUND, APIErrorCode.JOB_NOT_FOUND,
"Job not found", "Job not found",
details={"job_id": job_id}, details={"job_id": job_id},
) )
# Terminal jobs can't be cancelled; DELETE removes them from the manager instead.
if job.status in TERMINAL_STATUSES:
manager.remove_job(job_id)
return web.Response(status=204)
success = manager.cancel_job(job_id) success = manager.cancel_job(job_id)
if success: if success:
@@ -1127,6 +1144,383 @@ async def cancel_download_job_handler(job_id: str, request: Optional[web.Request
) )
async def clear_finished_download_jobs_handler(request: Optional[web.Request] = None) -> web.Response:
"""Handle clear finished download jobs request."""
from unshackle.core.api.download_manager import get_download_manager
try:
manager = get_download_manager()
removed = manager.clear_finished_jobs()
return web.json_response({"removed": removed})
except APIError:
raise
except (Exception, SystemExit) as e:
log.exception("Error clearing finished download jobs")
debug_mode = request.app.get("debug_api", False) if request else False
return handle_api_exception(
e,
context={"operation": "clear_finished_download_jobs"},
debug_mode=debug_mode,
)
async def retry_download_job_handler(job_id: str, request: Optional[web.Request] = None) -> web.Response:
"""Handle retry download job request - enqueue a new job with the original's parameters."""
from unshackle.core.api.download_manager import TERMINAL_STATUSES, get_download_manager
try:
manager = get_download_manager()
job = manager.get_job(job_id)
if not job:
raise APIError(
APIErrorCode.JOB_NOT_FOUND,
"Job not found",
details={"job_id": job_id},
)
if job.status not in TERMINAL_STATUSES:
raise APIError(
APIErrorCode.CONFLICT,
"Only completed, failed, or cancelled jobs can be retried",
details={"job_id": job_id, "status": job.status.value},
)
# Re-apply creation-time gates so retry cannot bypass the caller's service allowlist
# or currently-disabled cdm_overrides / allow_job_credentials config.
if not validate_service(job.service, request):
raise APIError(
APIErrorCode.INVALID_SERVICE,
f"Invalid or unavailable service: {job.service}",
details={"service": job.service},
)
enforce_download_gates(job.parameters)
await manager.start_workers()
# Reuse the raw in-memory parameters; redaction only ever applies to serialized copies.
new_job = manager.create_job(job.service, job.title_id, **job.parameters)
return web.json_response(
{"job_id": new_job.job_id, "status": new_job.status.value, "created_time": new_job.created_time.isoformat()},
status=202,
)
except APIError:
raise
except (Exception, SystemExit) as e:
log.exception(f"Error retrying download job {sanitize_log(job_id)}")
debug_mode = request.app.get("debug_api", False) if request else False
return handle_api_exception(
e,
context={"operation": "retry_download_job", "job_id": job_id},
debug_mode=debug_mode,
)
async def prioritize_download_job_handler(job_id: str, request: Optional[web.Request] = None) -> web.Response:
"""Handle prioritize download job request - move a queued job to the front of the queue."""
from unshackle.core.api.download_manager import JobStatus, get_download_manager
try:
manager = get_download_manager()
job = manager.get_job(job_id)
if not job:
raise APIError(
APIErrorCode.JOB_NOT_FOUND,
"Job not found",
details={"job_id": job_id},
)
if job.status != JobStatus.QUEUED:
raise APIError(
APIErrorCode.CONFLICT,
"Only queued jobs can be prioritized",
details={"job_id": job_id, "status": job.status.value},
)
manager.prioritize_job(job_id)
return web.json_response({"job_id": job_id, "position": "front"})
except APIError:
raise
except (Exception, SystemExit) as e:
log.exception(f"Error prioritizing download job {sanitize_log(job_id)}")
debug_mode = request.app.get("debug_api", False) if request else False
return handle_api_exception(
e,
context={"operation": "prioritize_download_job", "job_id": job_id},
debug_mode=debug_mode,
)
# ---------------------------------------------------------------------------
# Platform Handlers (profiles, config, history, maintenance)
# ---------------------------------------------------------------------------
_CONFIG_SECRET_KEY_RE = re.compile(r"secret|password|token|api_key|credential", re.IGNORECASE)
def _redact_config(value: Any) -> Any:
"""Recursively mask secret-looking keys and URL userinfo; stringify paths."""
if isinstance(value, dict):
return {
str(k): (REDACTED if v and _CONFIG_SECRET_KEY_RE.search(str(k)) else _redact_config(v))
for k, v in value.items()
}
if isinstance(value, (list, tuple)):
return [_redact_config(v) for v in value]
if isinstance(value, Path):
return str(value)
if isinstance(value, str) and "@" in value:
return URL_USERINFO_RE.sub(f"{REDACTED}@", value)
return value
async def profiles_handler(request: Optional[web.Request] = None) -> web.Response:
"""Handle list credential profiles request."""
try:
allowed = get_allowed_services(request)
profiles: Dict[str, List[str]] = {}
for service, creds in (config.credentials or {}).items():
# a plain (non-dict) credential is unnamed; that service is omitted entirely
if not isinstance(creds, dict):
continue
try:
tag = Services.get_tag(service)
except Exception:
tag = service
if allowed is not None and tag not in allowed:
continue
profiles[tag] = sorted(str(name) for name in creds)
return web.json_response({"profiles": profiles})
except APIError:
raise
except (Exception, SystemExit) as e:
log.exception("Error listing profiles")
debug_mode = request.app.get("debug_api", False) if request else False
return handle_api_exception(e, context={"operation": "profiles"}, debug_mode=debug_mode)
async def server_config_handler(request: Optional[web.Request] = None) -> web.Response:
"""Handle read-only effective server config request (secrets redacted)."""
from unshackle.core.api.download_manager import get_download_manager
try:
manager = get_download_manager()
serve_cfg = config.serve or {}
allowed = get_allowed_services(request)
service_tags = Services.get_tags()
if allowed is not None:
service_tags = [t for t in service_tags if t in allowed]
payload = {
"dl": _redact_config(config.dl),
"serve": {
"max_concurrent_downloads": manager.max_concurrent_downloads,
"job_retention_hours": manager.job_retention_hours,
"history_limit": int(serve_cfg.get("history_limit", 100)),
"services": serve_cfg.get("services") or None,
"remote_only": bool(serve_cfg.get("remote_only", False)),
"cdm_overrides": _redact_config(serve_cfg.get("cdm_overrides")),
"allow_job_credentials": bool(serve_cfg.get("allow_job_credentials", False)),
},
"directories": {
"downloads": str(config.directories.downloads),
"temp": str(config.directories.temp),
"cache": str(config.directories.cache),
},
"services": service_tags,
}
return web.json_response({"config": payload})
except APIError:
raise
except (Exception, SystemExit) as e:
log.exception("Error building server config")
debug_mode = request.app.get("debug_api", False) if request else False
return handle_api_exception(e, context={"operation": "server_config"}, debug_mode=debug_mode)
async def download_history_handler(data: Dict[str, Any], request: Optional[web.Request] = None) -> web.Response:
"""Handle persisted download history request."""
from unshackle.core.api.download_manager import read_job_history
try:
limit = 100
limit_raw = data.get("limit")
if limit_raw is not None:
try:
limit = int(limit_raw)
except (TypeError, ValueError):
raise APIError(
APIErrorCode.INVALID_PARAMETERS, "limit must be an integer", details={"limit": limit_raw}
)
if limit < 1:
raise APIError(APIErrorCode.INVALID_PARAMETERS, "limit must be >= 1", details={"limit": limit_raw})
allowed = get_allowed_services(request)
if allowed is None:
history = read_job_history(limit=limit, service=data.get("service"))
else:
# Read unbounded, drop entries outside the caller's allowlist, then apply limit.
allowed_upper = {a.upper() for a in allowed}
entries = read_job_history(limit=0, service=data.get("service"))
history = [e for e in entries if str(e.get("service") or "").upper() in allowed_upper][:limit]
return web.json_response({"history": history, "count": len(history)})
except APIError:
raise
except (Exception, SystemExit) as e:
log.exception("Error reading download history")
debug_mode = request.app.get("debug_api", False) if request else False
return handle_api_exception(e, context={"operation": "download_history"}, debug_mode=debug_mode)
async def delete_history_handler(job_id: str, request: Optional[web.Request] = None) -> web.Response:
"""Delete one persisted history entry by job_id."""
from unshackle.core.api.download_manager import delete_job_history
try:
allowed = get_allowed_services(request)
allowed_upper = {a.upper() for a in allowed} if allowed is not None else None
if not delete_job_history(job_id, allowed=allowed_upper):
raise APIError(APIErrorCode.NOT_FOUND, "History entry not found", details={"job_id": job_id})
return web.Response(status=204)
except APIError:
raise
except (Exception, SystemExit) as e:
log.exception("Error deleting download history")
debug_mode = request.app.get("debug_api", False) if request else False
return handle_api_exception(e, context={"operation": "delete_history"}, debug_mode=debug_mode)
def _require_no_active_downloads(operation: str) -> None:
"""Raise 409 CONFLICT if any job is currently downloading."""
from unshackle.core.api.download_manager import JobStatus, get_download_manager
active = [j.job_id for j in get_download_manager().list_jobs() if j.status == JobStatus.DOWNLOADING]
if active:
raise APIError(
APIErrorCode.CONFLICT,
f"Cannot {operation} while downloads are active",
details={"active_jobs": active},
)
async def clear_cache_handler(request: Optional[web.Request] = None) -> web.Response:
"""Handle clear cache directory request."""
from unshackle.commands.env import clear_directory
try:
_require_no_active_downloads("clear cache")
_, freed_bytes = await asyncio.to_thread(clear_directory, config.directories.cache)
return web.json_response({"cleared": True, "freed_bytes": freed_bytes})
except APIError:
raise
except (Exception, SystemExit) as e:
log.exception("Error clearing cache")
debug_mode = request.app.get("debug_api", False) if request else False
return handle_api_exception(e, context={"operation": "clear_cache"}, debug_mode=debug_mode)
async def clear_temp_handler(request: Optional[web.Request] = None) -> web.Response:
"""Handle clear temp directory request."""
from unshackle.commands.env import clear_directory
try:
_require_no_active_downloads("clear temp")
_, freed_bytes = await asyncio.to_thread(clear_directory, config.directories.temp)
return web.json_response({"cleared": True, "freed_bytes": freed_bytes})
except APIError:
raise
except (Exception, SystemExit) as e:
log.exception("Error clearing temp")
debug_mode = request.app.get("debug_api", False) if request else False
return handle_api_exception(e, context={"operation": "clear_temp"}, debug_mode=debug_mode)
async def refresh_services_handler(request: Optional[web.Request] = None) -> web.Response:
"""Handle refresh of service repos configured in directories.services."""
from unshackle.core.service_repo import is_repo_spec, refresh_repo
try:
entries = config.directories.services
if not isinstance(entries, list):
entries = [entries]
specs = [e for e in entries if isinstance(e, str) and is_repo_spec(e)]
repos = []
for spec in specs:
dest, changes = await asyncio.to_thread(refresh_repo, spec)
repos.append({"spec": spec, "updated": dest is not None, "changes": list(changes or [])})
return web.json_response({"refreshed": all(r["updated"] for r in repos), "repos": repos})
except APIError:
raise
except (Exception, SystemExit) as e:
log.exception("Error refreshing service repos")
debug_mode = request.app.get("debug_api", False) if request else False
return handle_api_exception(e, context={"operation": "refresh_services"}, debug_mode=debug_mode)
_VERSION_RE = re.compile(r"\d+\.\d+(?:\.\d+)*")
def _binary_version(path: Any) -> Optional[str]:
"""Best-effort version probe of a binary; None when nothing parseable."""
import subprocess
for flag in ("--version", "-version"):
try:
proc = subprocess.run([str(path), flag], capture_output=True, text=True, timeout=5)
except (OSError, subprocess.SubprocessError, ValueError):
return None
match = _VERSION_RE.search(proc.stdout or "") or _VERSION_RE.search(proc.stderr or "")
if match:
return match.group(0)
return None
async def env_check_handler(request: Optional[web.Request] = None) -> web.Response:
"""Handle environment dependency check request."""
from unshackle.commands.env import get_dependencies
def _run_checks() -> List[Dict[str, Any]]:
checks = []
for dep in get_dependencies():
binary = dep["binary"]
checks.append(
{
"name": dep["name"],
"installed": binary is not None,
"version": _binary_version(binary) if binary else None,
"required": dep["required"],
}
)
return checks
try:
checks = await asyncio.to_thread(_run_checks)
return web.json_response({"checks": checks})
except APIError:
raise
except (Exception, SystemExit) as e:
log.exception("Error running env check")
debug_mode = request.app.get("debug_api", False) if request else False
return handle_api_exception(e, context={"operation": "env_check"}, debug_mode=debug_mode)
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# Remote-DL Session Handlers # Remote-DL Session Handlers
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
+449 -9
View File
@@ -10,12 +10,17 @@ from aiohttp_swagger3 import SwaggerDocs, SwaggerInfo, SwaggerUiSettings
from unshackle.core import __version__ from unshackle.core import __version__
from unshackle.core.api.errors import APIError, APIErrorCode, build_error_response, handle_api_exception from unshackle.core.api.errors import APIError, APIErrorCode, build_error_response, handle_api_exception
from unshackle.core.api.handlers import (cancel_download_job_handler, download_handler, get_allowed_services, from unshackle.core.api.handlers import (cancel_download_job_handler, clear_cache_handler,
get_download_job_handler, list_download_jobs_handler, list_titles_handler, clear_finished_download_jobs_handler, clear_temp_handler,
list_tracks_handler, search_handler, session_create_handler, delete_history_handler, download_handler, download_history_handler,
session_delete_handler, session_info_handler, session_license_handler, env_check_handler, get_allowed_services, get_download_job_handler,
session_prompt_get_handler, session_prompt_post_handler, list_download_jobs_handler, list_titles_handler, list_tracks_handler,
session_segments_handler, session_titles_handler, session_tracks_handler) prioritize_download_job_handler, profiles_handler, refresh_services_handler,
retry_download_job_handler, search_handler, server_config_handler,
session_create_handler, session_delete_handler, session_info_handler,
session_license_handler, session_prompt_get_handler,
session_prompt_post_handler, session_segments_handler,
session_titles_handler, session_tracks_handler)
from unshackle.core.services import Services from unshackle.core.services import Services
from unshackle.core.update_checker import UpdateChecker from unshackle.core.update_checker import UpdateChecker
@@ -832,6 +837,14 @@ async def download_jobs(request: web.Request) -> web.Response:
enum: [asc, desc] enum: [asc, desc]
default: desc default: desc
description: Sort order (ascending or descending) description: Sort order (ascending or descending)
- name: full
in: query
required: false
schema:
type: string
enum: ["true", "false"]
default: "false"
description: When "true", include full job details (parameters, timestamps, output files, errors) per job
responses: responses:
'200': '200':
description: List of download jobs description: List of download jobs
@@ -868,6 +881,7 @@ async def download_jobs(request: web.Request) -> web.Response:
"service": request.query.get("service"), "service": request.query.get("service"),
"sort_by": request.query.get("sort_by", "created_time"), "sort_by": request.query.get("sort_by", "created_time"),
"sort_order": request.query.get("sort_order", "desc"), "sort_order": request.query.get("sort_order", "desc"),
"full": request.query.get("full"),
} }
return await list_download_jobs_handler(query_params, request) return await list_download_jobs_handler(query_params, request)
@@ -900,10 +914,10 @@ async def download_job_detail(request: web.Request) -> web.Response:
@api_handler @api_handler
async def cancel_download_job(request: web.Request) -> web.Response: async def cancel_download_job(request: web.Request) -> web.Response:
""" """
Cancel download job. Cancel or remove download job.
--- ---
summary: Cancel download job summary: Cancel or remove download job
description: Cancel a queued or running download job description: Cancel a queued or running download job, or remove a completed/failed/cancelled job entirely
parameters: parameters:
- name: job_id - name: job_id
in: path in: path
@@ -913,6 +927,8 @@ async def cancel_download_job(request: web.Request) -> web.Response:
responses: responses:
'200': '200':
description: Job cancelled successfully description: Job cancelled successfully
'204':
description: Terminal job removed from the manager
'400': '400':
description: Job cannot be cancelled description: Job cannot be cancelled
'404': '404':
@@ -924,6 +940,419 @@ async def cancel_download_job(request: web.Request) -> web.Response:
return await cancel_download_job_handler(job_id, request) return await cancel_download_job_handler(job_id, request)
@api_handler
async def clear_finished_download_jobs(request: web.Request) -> web.Response:
"""
Clear finished download jobs.
---
summary: Clear finished download jobs
description: Remove all completed, failed, and cancelled jobs from the manager
responses:
'200':
description: Finished jobs removed
content:
application/json:
schema:
type: object
properties:
removed:
type: integer
description: Number of jobs removed
'500':
description: Server error
"""
return await clear_finished_download_jobs_handler(request)
@api_handler
async def retry_download_job(request: web.Request) -> web.Response:
"""
Retry download job.
---
summary: Retry download job
description: Enqueue a new job reusing a completed, failed, or cancelled job's service, title, and parameters
parameters:
- name: job_id
in: path
required: true
schema:
type: string
responses:
'202':
description: New job queued
content:
application/json:
schema:
type: object
properties:
job_id:
type: string
status:
type: string
created_time:
type: string
'404':
description: Job not found
'409':
description: Job is not in a terminal state
'500':
description: Server error
"""
job_id = request.match_info["job_id"]
return await retry_download_job_handler(job_id, request)
@api_handler
async def prioritize_download_job(request: web.Request) -> web.Response:
"""
Prioritize download job.
---
summary: Prioritize download job
description: Move a queued job to the front of the download queue
parameters:
- name: job_id
in: path
required: true
schema:
type: string
responses:
'200':
description: Job moved to front of queue
content:
application/json:
schema:
type: object
properties:
job_id:
type: string
position:
type: string
example: front
'404':
description: Job not found
'409':
description: Job is not queued
'500':
description: Server error
"""
job_id = request.match_info["job_id"]
return await prioritize_download_job_handler(job_id, request)
@api_handler
async def profiles(request: web.Request) -> web.Response:
"""
List configured credential profiles per service.
---
summary: List credential profiles
description: >
Enumerate named credential profiles configured per service (usable as the `profile`
parameter). Only services whose credentials are a mapping of profile-name to credential
are listed (including a `default` key if present); a service configured with a single
plain (unnamed) credential is omitted entirely. Filtered by the caller's service allowlist.
responses:
'200':
description: Profiles per service
content:
application/json:
schema:
type: object
properties:
profiles:
type: object
additionalProperties:
type: array
items:
type: string
'500':
description: Server error
"""
return await profiles_handler(request)
@api_handler
async def server_config(request: web.Request) -> web.Response:
"""
Get the redacted effective server configuration.
---
summary: Get server config
description: >
Read-only, redacted view of the effective server configuration for display in a UI
settings page. Secrets (api_secret, users, credentials, tokens) are never included;
secret-looking keys inside `dl` are masked.
responses:
'200':
description: Redacted server configuration
content:
application/json:
schema:
type: object
properties:
config:
type: object
properties:
dl:
type: object
description: Default dl parameters from config (secret-looking keys masked)
serve:
type: object
properties:
max_concurrent_downloads:
type: integer
job_retention_hours:
type: integer
services:
type: array
items:
type: string
nullable: true
remote_only:
type: boolean
cdm_overrides:
nullable: true
description: List of permitted CDM device names, true, or null
allow_job_credentials:
type: boolean
directories:
type: object
properties:
downloads:
type: string
temp:
type: string
cache:
type: string
services:
type: array
items:
type: string
description: Available service tags (allowlist-filtered)
'500':
description: Server error
"""
return await server_config_handler(request)
@api_handler
async def download_history(request: web.Request) -> web.Response:
"""
Get persistent download history.
---
summary: Get download history
description: >
Read the persisted job history (jobs that reached a terminal state), newest first.
Corrupt lines in the history file are skipped; a missing file yields an empty list.
parameters:
- name: limit
in: query
required: false
schema:
type: integer
minimum: 1
default: 100
description: Maximum number of entries to return
- name: service
in: query
required: false
schema:
type: string
description: Filter entries by service tag (case-insensitive)
responses:
'200':
description: History entries, newest first
content:
application/json:
schema:
type: object
properties:
history:
type: array
items:
type: object
properties:
job_id:
type: string
service:
type: string
title_id:
type: string
title:
type: string
nullable: true
status:
type: string
enum: [completed, failed, cancelled]
created_time:
type: string
completed_time:
type: string
nullable: true
output_files:
type: array
items:
type: string
error_message:
type: string
nullable: true
count:
type: integer
'400':
description: Invalid query parameters
'500':
description: Server error
"""
query_params = {"limit": request.query.get("limit"), "service": request.query.get("service")}
return await download_history_handler(query_params, request)
@api_handler
async def delete_history(request: web.Request) -> web.Response:
"""
Delete a download history entry.
---
summary: Delete download history entry
description: Remove a single persisted history entry by job_id.
parameters:
- name: job_id
in: path
required: true
schema:
type: string
responses:
'204':
description: History entry removed
'404':
description: History entry not found
'500':
description: Server error
"""
return await delete_history_handler(request.match_info["job_id"], request)
@api_handler
async def maintenance_clear_cache(request: web.Request) -> web.Response:
"""
Clear the cache directory.
---
summary: Clear cache
description: Delete the contents of the cache directory (recreated empty). Freed size is best effort.
responses:
'200':
description: Cache cleared
content:
application/json:
schema:
type: object
properties:
cleared:
type: boolean
freed_bytes:
type: integer
'500':
description: Server error
"""
return await clear_cache_handler(request)
@api_handler
async def maintenance_clear_temp(request: web.Request) -> web.Response:
"""
Clear the temp directory.
---
summary: Clear temp
description: Delete the contents of the temp directory (recreated empty). Freed size is best effort.
responses:
'200':
description: Temp cleared
content:
application/json:
schema:
type: object
properties:
cleared:
type: boolean
freed_bytes:
type: integer
'500':
description: Server error
"""
return await clear_temp_handler(request)
@api_handler
async def maintenance_refresh_services(request: web.Request) -> web.Response:
"""
Refresh configured service repos.
---
summary: Refresh service repos
description: >
Force-sync (git pull) every service repo configured in directories.services.
`refreshed` is true when all repos synced (or none are configured); per-repo
results are listed under `repos`.
responses:
'200':
description: Refresh results
content:
application/json:
schema:
type: object
properties:
refreshed:
type: boolean
repos:
type: array
items:
type: object
properties:
spec:
type: string
updated:
type: boolean
changes:
type: array
items:
type: string
'500':
description: Server error
"""
return await refresh_services_handler(request)
@api_handler
async def env_check(request: web.Request) -> web.Response:
"""
Check environment dependencies.
---
summary: Environment check
description: Report install status of the binaries `env check` inspects, with best-effort versions.
responses:
'200':
description: Dependency check results
content:
application/json:
schema:
type: object
properties:
checks:
type: array
items:
type: object
properties:
name:
type: string
installed:
type: boolean
version:
type: string
nullable: true
required:
type: boolean
'500':
description: Server error
"""
return await env_check_handler(request)
@api_handler @api_handler
async def session_create(request: web.Request) -> web.Response: async def session_create(request: web.Request) -> web.Response:
""" """
@@ -1325,8 +1754,19 @@ ROUTES: list[tuple[str, str, Handler, bool]] = [
("POST", "/api/list-tracks", list_tracks, False), ("POST", "/api/list-tracks", list_tracks, False),
("POST", "/api/download", download, False), ("POST", "/api/download", download, False),
("GET", "/api/download/jobs", download_jobs, False), ("GET", "/api/download/jobs", download_jobs, False),
("POST", "/api/download/jobs/clear-finished", clear_finished_download_jobs, False),
("GET", "/api/download/jobs/{job_id}", download_job_detail, False), ("GET", "/api/download/jobs/{job_id}", download_job_detail, False),
("DELETE", "/api/download/jobs/{job_id}", cancel_download_job, False), ("DELETE", "/api/download/jobs/{job_id}", cancel_download_job, False),
("POST", "/api/download/jobs/{job_id}/retry", retry_download_job, False),
("POST", "/api/download/jobs/{job_id}/priority", prioritize_download_job, False),
("GET", "/api/profiles", profiles, False),
("GET", "/api/config", server_config, False),
("GET", "/api/history", download_history, False),
("DELETE", "/api/history/{job_id}", delete_history, False),
("POST", "/api/maintenance/clear-cache", maintenance_clear_cache, False),
("POST", "/api/maintenance/clear-temp", maintenance_clear_temp, False),
("POST", "/api/maintenance/refresh-services", maintenance_refresh_services, False),
("GET", "/api/env/check", env_check, False),
("POST", "/api/session/create", session_create, True), ("POST", "/api/session/create", session_create, True),
("GET", "/api/session/{session_id}/titles", session_titles, True), ("GET", "/api/session/{session_id}/titles", session_titles, True),
("POST", "/api/session/{session_id}/tracks", session_tracks, True), ("POST", "/api/session/{session_id}/tracks", session_tracks, True),