mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-11-11 19:45:06 +00:00
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
@@ -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)
|
||||
|
||||
Enables comprehensive debug logging. Default: `false`
|
||||
|
||||
@@ -7,6 +7,10 @@ queued -> 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.
|
||||
|
||||
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
@@ -242,6 +242,7 @@ List all download jobs with optional filtering and sorting. Disabled in `--remot
|
||||
| `service` | string | all | Filter by service tag |
|
||||
| `sort_by` | string | `created_time` | Sort field: `created_time`, `started_time`, `completed_time`, `progress`, `status`, `service` |
|
||||
| `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
|
||||
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
|
||||
{
|
||||
"job_id": "504db959-80b0-446c-a764-7924b761d613",
|
||||
"status": "completed",
|
||||
"status": "downloading",
|
||||
"created_time": "2026-02-27T18:00:00.000000",
|
||||
"service": "EXAMPLE1",
|
||||
"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": { },
|
||||
"started_time": "2026-02-27T18:00:01.000000",
|
||||
"completed_time": "2026-02-27T18:00:15.000000",
|
||||
"output_files": [],
|
||||
"completed_time": null,
|
||||
"output_files": [
|
||||
"/downloads/Example.Show.S01E01.mkv",
|
||||
"/downloads/Example.Show.S01E02.mkv",
|
||||
"/downloads/Example.Show.S01E03.mkv"
|
||||
],
|
||||
"error_message": 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}
|
||||
|
||||
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
@@ -17,16 +17,9 @@ from unshackle.core.constants import context_settings
|
||||
from unshackle.core.services import Services
|
||||
|
||||
|
||||
@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."""
|
||||
# Define all dependencies
|
||||
all_deps = [
|
||||
def get_dependencies() -> list[dict]:
|
||||
"""Binary dependency inventory shared by `env check` and the API env check."""
|
||||
return [
|
||||
# Core Media Tools
|
||||
{"name": "FFmpeg", "binary": binaries.FFMPEG, "required": True, "desc": "Media processing", "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"},
|
||||
]
|
||||
|
||||
|
||||
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
|
||||
all_required_installed = True
|
||||
total_installed = 0
|
||||
@@ -214,12 +235,11 @@ def cache(service: Optional[str]) -> None:
|
||||
if service:
|
||||
cache_dir = cache_dir / Services.get_tag(service)
|
||||
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:
|
||||
log.info("No files to delete")
|
||||
else:
|
||||
log.info(f"Deleting {files_count} files...")
|
||||
shutil.rmtree(cache_dir)
|
||||
log.info(f"Deleted {files_count} files")
|
||||
log.info("Cleared")
|
||||
|
||||
|
||||
@@ -228,10 +248,9 @@ def temp() -> None:
|
||||
"""Clear the environment temp directory."""
|
||||
log = logging.getLogger("env")
|
||||
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:
|
||||
log.info("No files to delete")
|
||||
else:
|
||||
log.info(f"Deleting {files_count} files...")
|
||||
shutil.rmtree(config.directories.temp)
|
||||
log.info(f"Deleted {files_count} files")
|
||||
log.info("Cleared")
|
||||
|
||||
@@ -7,6 +7,7 @@ import sys
|
||||
import tempfile
|
||||
import threading
|
||||
import uuid
|
||||
from collections import deque
|
||||
from contextlib import suppress
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timedelta
|
||||
@@ -75,6 +76,10 @@ class JobStatus(Enum):
|
||||
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
|
||||
class DownloadJob:
|
||||
"""Represents a download job with all its parameters and status."""
|
||||
@@ -119,6 +124,9 @@ class DownloadJob:
|
||||
# Cancellation support
|
||||
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]:
|
||||
"""Convert job to dictionary for JSON response."""
|
||||
result = {
|
||||
@@ -161,6 +169,125 @@ class DownloadJob:
|
||||
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]:
|
||||
"""Map case-insensitive strings (by member name or value) to enum members, dropping unknowns."""
|
||||
lookup = {m.name.upper(): m for m in enum_cls}
|
||||
@@ -471,6 +598,9 @@ class DownloadQueueManager:
|
||||
|
||||
self._jobs: Dict[str, DownloadJob] = {}
|
||||
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._download_processes: Dict[str, asyncio.subprocess.Process] = {}
|
||||
self._job_temp_files: Dict[str, Dict[str, str]] = {}
|
||||
@@ -516,6 +646,8 @@ class DownloadQueueManager:
|
||||
if job.status == JobStatus.QUEUED:
|
||||
job.status = JobStatus.CANCELLED
|
||||
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)}")
|
||||
return True
|
||||
elif job.status == JobStatus.DOWNLOADING:
|
||||
@@ -542,6 +674,36 @@ class DownloadQueueManager:
|
||||
|
||||
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:
|
||||
"""Remove jobs older than retention period."""
|
||||
cutoff_time = datetime.now() - timedelta(hours=self.job_retention_hours)
|
||||
@@ -623,8 +785,15 @@ class DownloadQueueManager:
|
||||
|
||||
while not self._shutdown_event.is_set():
|
||||
try:
|
||||
# Wait for a job or shutdown signal
|
||||
job = await asyncio.wait_for(self._job_queue.get(), timeout=1.0)
|
||||
# Prioritized jobs run first; their stale copies still in the FIFO queue are skipped.
|
||||
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:
|
||||
continue
|
||||
@@ -650,6 +819,7 @@ class DownloadQueueManager:
|
||||
log.error(f"Job {job.job_id} failed: {e}")
|
||||
finally:
|
||||
job.completed_time = datetime.now()
|
||||
record_job_history(job)
|
||||
if job.job_id in self._active_downloads:
|
||||
del self._active_downloads[job.job_id]
|
||||
|
||||
|
||||
@@ -34,6 +34,8 @@ class APIErrorCode(str, Enum):
|
||||
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
|
||||
|
||||
CONFLICT = "CONFLICT" # Job/resource is in a state that disallows the action
|
||||
|
||||
RATE_LIMITED = "RATE_LIMITED" # Service rate limiting
|
||||
|
||||
# Server errors (5xx)
|
||||
@@ -93,6 +95,8 @@ class APIError(Exception):
|
||||
APIErrorCode.JOB_NOT_FOUND: 404,
|
||||
APIErrorCode.SESSION_NOT_FOUND: 404,
|
||||
APIErrorCode.TRACK_NOT_FOUND: 404,
|
||||
# 409 Conflict
|
||||
APIErrorCode.CONFLICT: 409,
|
||||
# 429 Too Many Requests
|
||||
APIErrorCode.RATE_LIMITED: 429,
|
||||
# 500 Internal Server Error
|
||||
|
||||
+423
-29
@@ -1,6 +1,8 @@
|
||||
import asyncio
|
||||
import enum
|
||||
import logging
|
||||
import re
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
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.tracks import Audio, Subtitle, Video
|
||||
from unshackle.core.utils.collections import ci_get
|
||||
from unshackle.core.utils.redact import REDACTED, URL_USERINFO_RE
|
||||
|
||||
log = logging.getLogger("api")
|
||||
|
||||
@@ -893,6 +896,37 @@ def validate_download_parameters(data: Dict[str, Any]) -> Optional[str]:
|
||||
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:
|
||||
"""Handle download request - create and queue a download job."""
|
||||
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},
|
||||
)
|
||||
|
||||
# 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.
|
||||
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.",
|
||||
)
|
||||
enforce_download_gates(data)
|
||||
|
||||
try:
|
||||
# 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)
|
||||
|
||||
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})
|
||||
|
||||
@@ -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:
|
||||
"""Handle cancel download job request."""
|
||||
from unshackle.core.api.download_manager import get_download_manager
|
||||
"""Handle cancel/remove download job request."""
|
||||
from unshackle.core.api.download_manager import TERMINAL_STATUSES, get_download_manager
|
||||
|
||||
try:
|
||||
manager = get_download_manager()
|
||||
|
||||
if not manager.get_job(job_id):
|
||||
job = manager.get_job(job_id)
|
||||
if not job:
|
||||
raise APIError(
|
||||
APIErrorCode.JOB_NOT_FOUND,
|
||||
"Job not found",
|
||||
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)
|
||||
|
||||
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
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -10,12 +10,17 @@ from aiohttp_swagger3 import SwaggerDocs, SwaggerInfo, SwaggerUiSettings
|
||||
|
||||
from unshackle.core import __version__
|
||||
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,
|
||||
get_download_job_handler, list_download_jobs_handler, list_titles_handler,
|
||||
list_tracks_handler, search_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.api.handlers import (cancel_download_job_handler, clear_cache_handler,
|
||||
clear_finished_download_jobs_handler, clear_temp_handler,
|
||||
delete_history_handler, download_handler, download_history_handler,
|
||||
env_check_handler, get_allowed_services, get_download_job_handler,
|
||||
list_download_jobs_handler, list_titles_handler, list_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.update_checker import UpdateChecker
|
||||
|
||||
@@ -832,6 +837,14 @@ async def download_jobs(request: web.Request) -> web.Response:
|
||||
enum: [asc, desc]
|
||||
default: desc
|
||||
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:
|
||||
'200':
|
||||
description: List of download jobs
|
||||
@@ -868,6 +881,7 @@ async def download_jobs(request: web.Request) -> web.Response:
|
||||
"service": request.query.get("service"),
|
||||
"sort_by": request.query.get("sort_by", "created_time"),
|
||||
"sort_order": request.query.get("sort_order", "desc"),
|
||||
"full": request.query.get("full"),
|
||||
}
|
||||
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
|
||||
async def cancel_download_job(request: web.Request) -> web.Response:
|
||||
"""
|
||||
Cancel download job.
|
||||
Cancel or remove download job.
|
||||
---
|
||||
summary: Cancel download job
|
||||
description: Cancel a queued or running download job
|
||||
summary: Cancel or remove download job
|
||||
description: Cancel a queued or running download job, or remove a completed/failed/cancelled job entirely
|
||||
parameters:
|
||||
- name: job_id
|
||||
in: path
|
||||
@@ -913,6 +927,8 @@ async def cancel_download_job(request: web.Request) -> web.Response:
|
||||
responses:
|
||||
'200':
|
||||
description: Job cancelled successfully
|
||||
'204':
|
||||
description: Terminal job removed from the manager
|
||||
'400':
|
||||
description: Job cannot be cancelled
|
||||
'404':
|
||||
@@ -924,6 +940,419 @@ async def cancel_download_job(request: web.Request) -> web.Response:
|
||||
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
|
||||
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/download", download, 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),
|
||||
("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),
|
||||
("GET", "/api/session/{session_id}/titles", session_titles, True),
|
||||
("POST", "/api/session/{session_id}/tracks", session_tracks, True),
|
||||
|
||||
Reference in New Issue
Block a user