mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-11-11 19:45:06 +00:00
feat(serve): relay service prompts from download jobs to the REST client
A job's input_prompt carries a service's request_input() prompt; the client answers it with POST /api/download/jobs/{job_id}/input.
This commit is contained in:
@@ -251,7 +251,8 @@ Cookies come from files under `directories.cookies`. Credentials come from the
|
||||
use `self.request_input(prompt)`, never a bare `input()`. Under `serve` mode
|
||||
there is no local terminal, so a bare `input()` would hang the server waiting on
|
||||
stdin that never arrives. `request_input` instead relays the prompt to the
|
||||
remote client through the attached `InputBridge`. Locally it routes through the
|
||||
client: a remote-dl session gets it through the attached `InputBridge`, and a
|
||||
REST download job shows it as `input_prompt`. Locally it routes through the
|
||||
shared Rich console (`prompt_user`) so the prompt renders correctly alongside
|
||||
progress and log output.
|
||||
|
||||
|
||||
@@ -551,7 +551,8 @@ Show download jobs, with optional filtering and sorting.
|
||||
"segments_done": 120,
|
||||
"segments_total": 300,
|
||||
"speed": "5.2 MB/s",
|
||||
"skipped_subtitles": []
|
||||
"skipped_subtitles": [],
|
||||
"input_prompt": null
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -709,6 +710,36 @@ Move a queued job to the front of the download queue.
|
||||
| `404` | `JOB_NOT_FOUND` | No such job. |
|
||||
| `409` | `CONFLICT` | Job is not queued. |
|
||||
|
||||
### `POST /api/download/jobs/{job_id}/input`
|
||||
|
||||
Answer the prompt that a running job waits on. A service can ask for input during a job, for
|
||||
example an OTP code, a PIN, or a device-code confirmation. The job then shows the prompt in its
|
||||
`input_prompt` field and in its `progress` events. Show the prompt to the user and send the
|
||||
answer here.
|
||||
|
||||
=== "Request"
|
||||
|
||||
```json
|
||||
{ "response": "123456" }
|
||||
```
|
||||
|
||||
=== "Response `200`"
|
||||
|
||||
```json
|
||||
{ "status": "accepted" }
|
||||
```
|
||||
|
||||
| Status | Error code | Meaning |
|
||||
| --- | --- | --- |
|
||||
| `200` | - | Answer sent to the job. The `input_prompt` field goes back to `null`. |
|
||||
| `400` | `INVALID_INPUT` | The body is not JSON, or it has no `response`. |
|
||||
| `404` | `JOB_NOT_FOUND` | No such job. |
|
||||
| `409` | `CONFLICT` | The job does not wait on a prompt. |
|
||||
|
||||
!!! warning
|
||||
The job keeps its download slot while it waits. It fails with `AUTH_FAILED` if nobody
|
||||
answers the prompt within 10 minutes.
|
||||
|
||||
---
|
||||
|
||||
## Server management
|
||||
|
||||
@@ -166,6 +166,7 @@ request and response in full.
|
||||
| `DELETE` | `/api/download/jobs/{job_id}` | Cancel a running job, or remove a finished one. |
|
||||
| `POST` | `/api/download/jobs/{job_id}/retry` | Re-queue a finished job's parameters as a new job. |
|
||||
| `POST` | `/api/download/jobs/{job_id}/priority` | Move a queued job to the front. |
|
||||
| `POST` | `/api/download/jobs/{job_id}/input` | Answer the prompt that a running job waits on. |
|
||||
|
||||
### History and server info
|
||||
|
||||
|
||||
@@ -162,6 +162,7 @@ full detail:
|
||||
| `segments_total` | number | Total segments for the track in progress. |
|
||||
| `speed` | string | Transfer speed of the track in progress, e.g. `4.20 MB/s`. |
|
||||
| `skipped_subtitles` | list | Subtitles that were skipped as non-fatal failures (when `skip_subtitle_errors` is set). |
|
||||
| `input_prompt` | string or `null` | The prompt the service waits on. Answer it with `POST /api/download/jobs/{job_id}/input`. |
|
||||
|
||||
The full detail adds `started_time`, `completed_time`, `output_files`, the redacted
|
||||
`parameters`, and, for failed jobs, `error_message`, `error_details`, `error_code`,
|
||||
@@ -364,6 +365,10 @@ Three temporary files bridge the parent and the job worker:
|
||||
- **`progress.json`**: the job worker writes it continuously as the download proceeds.
|
||||
- **`result.json`**: the job worker writes it at the end.
|
||||
|
||||
The job worker has no terminal. When a service asks the user for input, the job worker writes
|
||||
the prompt to `progress.json` as `input_prompt` and waits. The answer from
|
||||
`POST /api/download/jobs/{job_id}/input` goes to the job worker on its stdin.
|
||||
|
||||
Running each download out-of-process isolates the server from crashes, memory growth,
|
||||
and blocking work inside the download engine, and makes hard cancellation (killing the
|
||||
process) reliable.
|
||||
|
||||
@@ -87,8 +87,8 @@ These are abstract. A service will not load unless it defines all three:
|
||||
- **DRM license hooks**: `get_widevine_service_certificate`, `get_widevine_license`,
|
||||
`get_playready_license` (delegates to Widevine by default), and
|
||||
`get_clearkey_license` (returns `None` by default). See [DRM hooks](#drm-hooks) below.
|
||||
- **`request_input(prompt)`**: prompts the user. It goes through the serve-mode input
|
||||
bridge or the rich console.
|
||||
- **`request_input(prompt)`**: prompts the user. It goes through the remote-dl input
|
||||
bridge, the REST download job relay, or the rich console.
|
||||
- **Event callbacks**: `on_segment_downloaded`, `on_track_downloaded`,
|
||||
`on_track_decrypted`, `on_track_repacked`, `on_track_multiplex` let a service react to
|
||||
pipeline events (for example, MonaLisa decrypts each segment in `on_segment_downloaded`).
|
||||
|
||||
@@ -8,6 +8,7 @@ from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -224,3 +225,177 @@ def test_perform_download_puts_the_lang_selection_in_the_service_ctx(
|
||||
ctx = exc.value.args[0]
|
||||
p = ctx.params
|
||||
assert (p["lang"], p["v_lang"], p["a_lang"], [c.name for c in p["acodec"]], p["forced_subs"]) == expected
|
||||
|
||||
|
||||
def test_worker_relays_a_prompt_and_reads_the_answer(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
import json
|
||||
import sys
|
||||
|
||||
from unshackle.core.api import download_worker
|
||||
from unshackle.core.console import prompt_user, set_prompt_handler
|
||||
|
||||
answers = tmp_path / "stdin"
|
||||
answers.write_text(json.dumps("12\n34") + "\n", encoding="utf-8")
|
||||
monkeypatch.setattr(sys, "stdin", answers.open("rb"))
|
||||
updates: list[dict] = []
|
||||
download_worker.relay_prompts(updates.append)
|
||||
try:
|
||||
assert prompt_user("Enter PIN") == "12\n34"
|
||||
finally:
|
||||
set_prompt_handler(None)
|
||||
assert updates == [{"input_prompt": "Enter PIN"}, {"input_prompt": None}]
|
||||
|
||||
|
||||
def test_a_prompt_discards_an_answer_sent_before_it(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""A late answer to an earlier prompt must not answer the next one; the timeout is not retryable."""
|
||||
import sys
|
||||
import threading
|
||||
|
||||
from unshackle.core.api import download_worker
|
||||
from unshackle.core.api.errors import APIErrorCode, categorize_exception
|
||||
from unshackle.core.console import prompt_user, set_prompt_handler
|
||||
|
||||
answers = tmp_path / "stdin"
|
||||
answers.write_text('"stale"\n', encoding="utf-8")
|
||||
monkeypatch.setattr(sys, "stdin", answers.open("rb"))
|
||||
monkeypatch.setattr(download_worker, "AUTH_INPUT_TIMEOUT", 0.2)
|
||||
download_worker.relay_prompts(lambda update: None)
|
||||
reader = next(t for t in threading.enumerate() if t.name == "prompt-answers")
|
||||
reader.join(timeout=5)
|
||||
try:
|
||||
with pytest.raises(TimeoutError) as exc_info:
|
||||
prompt_user("Enter PIN")
|
||||
finally:
|
||||
set_prompt_handler(None)
|
||||
assert categorize_exception(exc_info.value).error_code is APIErrorCode.AUTH_FAILED
|
||||
|
||||
|
||||
def test_concurrent_prompts_go_out_one_at_a_time(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from types import SimpleNamespace
|
||||
|
||||
from unshackle.core.api import download_worker
|
||||
from unshackle.core.console import prompt_user, set_prompt_handler
|
||||
|
||||
read_fd, write_fd = os.pipe()
|
||||
monkeypatch.setattr(sys, "stdin", SimpleNamespace(fileno=lambda: read_fd))
|
||||
monkeypatch.setattr(download_worker, "AUTH_INPUT_TIMEOUT", 5.0)
|
||||
updates: list[dict] = []
|
||||
download_worker.relay_prompts(updates.append)
|
||||
results: list[str] = []
|
||||
askers = [threading.Thread(target=lambda p=p: results.append(prompt_user(p)), daemon=True) for p in ("A", "B")]
|
||||
try:
|
||||
for asker in askers:
|
||||
asker.start()
|
||||
time.sleep(0.2)
|
||||
assert len(updates) == 1
|
||||
for answer, prompts_sent in (("1", 1), ("2", 3)):
|
||||
deadline = time.monotonic() + 5
|
||||
while len(updates) < prompts_sent and time.monotonic() < deadline:
|
||||
time.sleep(0.01)
|
||||
os.write(write_fd, (json.dumps(answer) + "\n").encode("utf-8"))
|
||||
for asker in askers:
|
||||
asker.join(timeout=5)
|
||||
finally:
|
||||
set_prompt_handler(None)
|
||||
os.close(write_fd)
|
||||
assert sorted(results) == ["1", "2"]
|
||||
assert [u["input_prompt"] is None for u in updates] == [False, True, False, True]
|
||||
|
||||
|
||||
def test_submit_input_writes_one_json_line_to_the_worker(manager: DownloadQueueManager) -> None:
|
||||
import io
|
||||
from types import SimpleNamespace
|
||||
|
||||
job = manager.create_job("EXAMPLE", "t")
|
||||
stdin = io.BytesIO()
|
||||
manager._download_processes[job.job_id] = SimpleNamespace(stdin=stdin) # type: ignore[assignment]
|
||||
assert manager.submit_input(job, "1234") is False
|
||||
|
||||
job.input_prompt = "Enter PIN"
|
||||
assert manager.submit_input(job, "1234") is True
|
||||
assert stdin.getvalue() == b'"1234"\n'
|
||||
assert job.input_prompt is None
|
||||
assert manager.submit_input(job, "5678") is False
|
||||
|
||||
|
||||
FAKE_PROMPT_WORKER = """
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from unshackle.core.api import download_worker
|
||||
from unshackle.core.console import prompt_user
|
||||
|
||||
result_path, progress_path = Path(sys.argv[-2]), Path(sys.argv[-1])
|
||||
state = {}
|
||||
|
||||
|
||||
def progress(update):
|
||||
state.update(update)
|
||||
download_worker.write_result(progress_path, state)
|
||||
|
||||
|
||||
download_worker.relay_prompts(progress)
|
||||
answers = [prompt_user("Enter OTP"), prompt_user("Enter OTP")]
|
||||
download_worker.write_result(result_path, {"status": "success", "output_files": answers})
|
||||
"""
|
||||
|
||||
|
||||
async def test_a_repeated_prompt_reaches_an_event_listener(
|
||||
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, manager: DownloadQueueManager
|
||||
) -> None:
|
||||
"""A re-prompt with the same text right after an answer must still publish an event."""
|
||||
import asyncio
|
||||
|
||||
from unshackle.core.api import download_manager
|
||||
|
||||
worker = tmp_path / "worker.py"
|
||||
worker.write_text(FAKE_PROMPT_WORKER, encoding="utf-8")
|
||||
spawn = asyncio.create_subprocess_exec
|
||||
|
||||
async def spawn_fake_worker(executable: str, *args: str, **kwargs: Any) -> Any:
|
||||
return await spawn(executable, str(worker), *args[2:], **kwargs)
|
||||
|
||||
monkeypatch.setattr(download_manager.asyncio, "create_subprocess_exec", spawn_fake_worker)
|
||||
job = manager.create_job("EXAMPLE", "t")
|
||||
events = manager.subscribe(job.job_id)
|
||||
run = asyncio.create_task(manager.run_download_async(job))
|
||||
|
||||
async def answer_next_prompt(answer: str) -> None:
|
||||
while True:
|
||||
event = await events.get()
|
||||
if event and event["data"]["input_prompt"]:
|
||||
assert manager.submit_input(job, answer) is True
|
||||
return
|
||||
|
||||
await asyncio.wait_for(answer_next_prompt("111111"), timeout=20)
|
||||
await asyncio.wait_for(answer_next_prompt("222222"), timeout=20)
|
||||
assert await asyncio.wait_for(run, timeout=20) == ["111111", "222222"]
|
||||
assert job.input_prompt is None
|
||||
|
||||
|
||||
async def test_input_handler_rejects_bad_requests(
|
||||
monkeypatch: pytest.MonkeyPatch, manager: DownloadQueueManager
|
||||
) -> None:
|
||||
from unshackle.core.api import download_manager
|
||||
from unshackle.core.api.errors import APIError, APIErrorCode
|
||||
from unshackle.core.api.handlers import download_job_input_handler
|
||||
|
||||
monkeypatch.setattr(download_manager, "get_download_manager", lambda: manager)
|
||||
job = manager.create_job("EXAMPLE", "t")
|
||||
|
||||
async def error_code(data: dict, job_id: str) -> APIErrorCode:
|
||||
with pytest.raises(APIError) as exc_info:
|
||||
await download_job_input_handler(data, job_id)
|
||||
return exc_info.value.error_code
|
||||
|
||||
assert await error_code({"response": "1"}, "missing") is APIErrorCode.JOB_NOT_FOUND
|
||||
assert await error_code({}, job.job_id) is APIErrorCode.INVALID_INPUT
|
||||
assert await error_code({"response": "1"}, job.job_id) is APIErrorCode.CONFLICT
|
||||
job.owner_key = "another-key"
|
||||
assert await error_code({"response": "1"}, job.job_id) is APIErrorCode.JOB_NOT_FOUND
|
||||
|
||||
@@ -73,6 +73,7 @@ def test_setup_routes_full_mode_wires_all_endpoints(make_app) -> None:
|
||||
("GET", "/api/download/jobs"),
|
||||
("GET", "/api/download/jobs/{job_id}"),
|
||||
("DELETE", "/api/download/jobs/{job_id}"),
|
||||
("POST", "/api/download/jobs/{job_id}/input"),
|
||||
("POST", "/api/session/create"),
|
||||
("GET", "/api/session/{session_id}/titles"),
|
||||
("POST", "/api/session/{session_id}/tracks"),
|
||||
|
||||
@@ -124,6 +124,9 @@ class DownloadJob:
|
||||
segments_total: float = 0.0
|
||||
speed: Optional[str] = None
|
||||
|
||||
# Prompt a service raised through prompt_user() and waits on; answered by submit_input().
|
||||
input_prompt: Optional[str] = None
|
||||
|
||||
# Subtitles skipped under skip_subtitle_errors (non-fatal). Each entry is a dl.SkippedSubtitle
|
||||
# dict (id / language / title) so a client can report which weren't available.
|
||||
skipped_subtitles: List[Dict[str, Any]] = field(default_factory=list)
|
||||
@@ -156,6 +159,7 @@ class DownloadJob:
|
||||
"segments_total": self.segments_total,
|
||||
"speed": self.speed,
|
||||
"skipped_subtitles": self.skipped_subtitles,
|
||||
"input_prompt": self.input_prompt,
|
||||
}
|
||||
|
||||
if include_full_details:
|
||||
@@ -805,6 +809,17 @@ class DownloadQueueManager:
|
||||
|
||||
return False
|
||||
|
||||
def submit_input(self, job: DownloadJob, response: str) -> bool:
|
||||
"""Send *response* to the job's worker when it waits on a prompt; ``False`` when no prompt is pending."""
|
||||
process = self._download_processes.get(job.job_id)
|
||||
if not job.input_prompt or process is None or process.stdin is None:
|
||||
return False
|
||||
process.stdin.write((json.dumps(response) + "\n").encode("utf-8"))
|
||||
# Cleared here, not on the worker's next progress write, so a second submit cannot queue a stale answer.
|
||||
job.input_prompt = None
|
||||
self.publish(job, "progress")
|
||||
return True
|
||||
|
||||
def remove_job(self, job_id: str) -> bool:
|
||||
"""Remove a terminal (completed/failed/cancelled) job entirely."""
|
||||
job = self._jobs.get(job_id)
|
||||
@@ -1028,6 +1043,7 @@ class DownloadQueueManager:
|
||||
payload_path,
|
||||
result_path,
|
||||
progress_path,
|
||||
stdin=asyncio.subprocess.PIPE,
|
||||
stdout=asyncio.subprocess.PIPE,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
)
|
||||
@@ -1035,7 +1051,13 @@ class DownloadQueueManager:
|
||||
self._download_processes[job.job_id] = process
|
||||
self._job_temp_files[job.job_id] = {"payload": payload_path, "result": result_path, "progress": progress_path}
|
||||
|
||||
communicate_task = asyncio.create_task(process.communicate())
|
||||
async def drain() -> tuple[bytes, bytes]:
|
||||
"""Collect the worker's output; unlike communicate(), leave stdin open for prompt answers."""
|
||||
assert process.stdout is not None and process.stderr is not None
|
||||
out, err, _ = await asyncio.gather(process.stdout.read(), process.stderr.read(), process.wait())
|
||||
return out, err
|
||||
|
||||
communicate_task = asyncio.create_task(drain())
|
||||
|
||||
stdout_bytes = b""
|
||||
stderr_bytes = b""
|
||||
@@ -1056,6 +1078,7 @@ class DownloadQueueManager:
|
||||
if stat_key != last_progress_stat:
|
||||
with open(progress_path, "r", encoding="utf-8") as handle:
|
||||
progress_data = json.load(handle)
|
||||
prompt_before = job.input_prompt
|
||||
if progress_data.get("phase") and progress_data["phase"] != job.phase:
|
||||
job.phase = progress_data["phase"]
|
||||
if progress_data.get("current_title"):
|
||||
@@ -1080,6 +1103,8 @@ class DownloadQueueManager:
|
||||
pass
|
||||
if progress_data.get("speed"):
|
||||
job.speed = str(progress_data["speed"])
|
||||
if "input_prompt" in progress_data:
|
||||
job.input_prompt = progress_data["input_prompt"]
|
||||
if progress_data.get("skipped_subtitles"):
|
||||
job.skipped_subtitles = progress_data["skipped_subtitles"]
|
||||
if "progress" in progress_data:
|
||||
@@ -1087,7 +1112,9 @@ class DownloadQueueManager:
|
||||
if new_progress != job.progress:
|
||||
job.progress = new_progress
|
||||
log.info(f"Job {job.job_id} progress updated: {job.progress}%")
|
||||
if progress_data != last_published:
|
||||
# submit_input() clears the prompt out of band, so a re-prompt with the same text
|
||||
# leaves the file equal to last_published; publish the prompt change anyway.
|
||||
if progress_data != last_published or job.input_prompt != prompt_before:
|
||||
last_published = progress_data
|
||||
self.publish(job, "progress")
|
||||
last_progress_stat = stat_key
|
||||
@@ -1160,6 +1187,7 @@ class DownloadQueueManager:
|
||||
await communicate_task
|
||||
|
||||
self._download_processes.pop(job.job_id, None)
|
||||
job.input_prompt = None
|
||||
|
||||
temp_paths = self._job_temp_files.pop(job.job_id, {})
|
||||
for path in temp_paths.values():
|
||||
|
||||
@@ -2,17 +2,22 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import queue
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
import traceback
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict
|
||||
from typing import Any, Callable, Dict
|
||||
from uuid import uuid4
|
||||
|
||||
from unshackle.core.api.input_bridge import AUTH_INPUT_TIMEOUT
|
||||
from unshackle.core.console import set_prompt_handler
|
||||
|
||||
from .download_manager import perform_download
|
||||
|
||||
log = logging.getLogger("download_worker")
|
||||
@@ -53,6 +58,42 @@ def write_result(path: Path, payload: Dict[str, Any]) -> None:
|
||||
raise
|
||||
|
||||
|
||||
def relay_prompts(progress_callback: Callable[[Dict[str, Any]], None]) -> None:
|
||||
"""Send each prompt_user() prompt to the parent as ``input_prompt`` progress, and read the answer from stdin.
|
||||
|
||||
The parent writes each answer as one JSON string per line, so an answer can hold a newline. The reader
|
||||
uses unbuffered stdin: a daemon thread blocked in ``sys.stdin.buffer`` aborts the interpreter at shutdown.
|
||||
Prompts go out one at a time, and a prompt discards answers that arrived before it, such as a late
|
||||
answer to a prompt that timed out.
|
||||
"""
|
||||
answers: queue.Queue[str] = queue.Queue()
|
||||
one_prompt = threading.Lock()
|
||||
|
||||
def read_answers() -> None:
|
||||
for line in io.FileIO(sys.stdin.fileno(), "rb", closefd=False):
|
||||
answers.put(json.loads(line.decode("utf-8", errors="replace")))
|
||||
|
||||
threading.Thread(target=read_answers, name="prompt-answers", daemon=True).start()
|
||||
|
||||
def ask(prompt: str) -> str:
|
||||
with one_prompt:
|
||||
while not answers.empty():
|
||||
answers.get_nowait()
|
||||
progress_callback({"input_prompt": prompt})
|
||||
try:
|
||||
return answers.get(timeout=AUTH_INPUT_TIMEOUT)
|
||||
except queue.Empty:
|
||||
# "authentication" makes categorize_exception report AUTH_FAILED, not a retryable network
|
||||
# error: a retry only waits on another prompt.
|
||||
raise TimeoutError(
|
||||
f"No client answer to the authentication prompt within {AUTH_INPUT_TIMEOUT:.0f}s"
|
||||
) from None
|
||||
finally:
|
||||
progress_callback({"input_prompt": None})
|
||||
|
||||
set_prompt_handler(ask)
|
||||
|
||||
|
||||
def main(argv: list[str]) -> int:
|
||||
if len(argv) not in [3, 4]:
|
||||
print(
|
||||
@@ -93,6 +134,7 @@ def main(argv: list[str]) -> int:
|
||||
except Exception as e:
|
||||
log.error(f"Failed to write progress update: {e}")
|
||||
|
||||
relay_prompts(progress_callback)
|
||||
output_files = perform_download(job_id, service, title_id, params, progress_callback=progress_callback)
|
||||
|
||||
result = {"status": "success", "output_files": output_files}
|
||||
|
||||
@@ -2829,6 +2829,26 @@ async def prioritize_download_job_handler(job_id: str, request: Optional[web.Req
|
||||
)
|
||||
|
||||
|
||||
async def download_job_input_handler(
|
||||
data: Dict[str, Any], job_id: str, request: Optional[web.Request] = None
|
||||
) -> web.Response:
|
||||
"""Answer the prompt a running job waits on (its ``input_prompt``), to unblock the service."""
|
||||
from unshackle.core.api.download_manager import get_download_manager
|
||||
|
||||
manager = get_download_manager()
|
||||
job = manager.get_job(job_id)
|
||||
if not job or not owns_job(job, request):
|
||||
raise APIError(APIErrorCode.JOB_NOT_FOUND, "Job not found", details={"job_id": job_id})
|
||||
|
||||
response_text = data.get("response")
|
||||
if response_text is None:
|
||||
raise APIError(APIErrorCode.INVALID_INPUT, "Missing required field: response")
|
||||
|
||||
if not manager.submit_input(job, str(response_text)):
|
||||
raise APIError(APIErrorCode.CONFLICT, "No prompt pending for this job", details={"job_id": job_id})
|
||||
return web.json_response({"status": "accepted"})
|
||||
|
||||
|
||||
CONFIG_SECRET_KEY_RE = re.compile(
|
||||
r"secret|passw|pwd|token|api[_-]?key|credential|auth|cookie|bearer|private", re.IGNORECASE
|
||||
)
|
||||
|
||||
@@ -34,6 +34,7 @@ from unshackle.core.api.handlers import (
|
||||
download_handler,
|
||||
download_history_handler,
|
||||
download_job_events_handler,
|
||||
download_job_input_handler,
|
||||
env_check_handler,
|
||||
get_allowed_services,
|
||||
get_download_job_handler,
|
||||
@@ -1348,6 +1349,60 @@ async def prioritize_download_job(request: web.Request) -> web.Response:
|
||||
return await prioritize_download_job_handler(job_id, request)
|
||||
|
||||
|
||||
@api_handler
|
||||
async def download_job_input(request: web.Request) -> web.Response:
|
||||
"""
|
||||
Answer a download job prompt.
|
||||
---
|
||||
summary: Answer a download job prompt
|
||||
description: >
|
||||
Submit the answer to the prompt a running job waits on (an OTP code, a PIN, a device-code
|
||||
confirmation). The job shows the prompt in its `input_prompt` field and in its SSE `progress`
|
||||
events. The job fails if nobody answers within the prompt timeout.
|
||||
parameters:
|
||||
- name: job_id
|
||||
in: path
|
||||
required: true
|
||||
schema:
|
||||
type: string
|
||||
requestBody:
|
||||
required: true
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
type: object
|
||||
required:
|
||||
- response
|
||||
properties:
|
||||
response:
|
||||
type: string
|
||||
description: The user's answer to the prompt
|
||||
responses:
|
||||
'200':
|
||||
description: Answer accepted
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
type: object
|
||||
properties:
|
||||
status:
|
||||
type: string
|
||||
example: accepted
|
||||
'400':
|
||||
description: Invalid request body
|
||||
'404':
|
||||
description: Job not found
|
||||
'409':
|
||||
description: No prompt pending for this job
|
||||
"""
|
||||
job_id = request.match_info["job_id"]
|
||||
try:
|
||||
data = await request.json()
|
||||
except Exception as e:
|
||||
raise APIError(APIErrorCode.INVALID_INPUT, "Invalid JSON request body", details={"error": str(e)}) from e
|
||||
return await download_job_input_handler(data, job_id, request)
|
||||
|
||||
|
||||
@api_handler
|
||||
async def profiles(request: web.Request) -> web.Response:
|
||||
"""
|
||||
@@ -2838,6 +2893,7 @@ ROUTES: list[tuple[str, str, Handler, bool]] = [
|
||||
("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),
|
||||
("POST", "/api/download/jobs/{job_id}/input", download_job_input, False),
|
||||
("GET", "/api/profiles", profiles, False),
|
||||
("GET", "/api/config", server_config, False),
|
||||
("GET", "/api/history", download_history, False),
|
||||
|
||||
@@ -459,8 +459,19 @@ def print_wide(renderable: RenderableType, pad: PaddingDimensions = (1, 3)) -> N
|
||||
console.width = fixed_width
|
||||
|
||||
|
||||
_prompt_handler: Optional[Callable[[str], str]] = None
|
||||
|
||||
|
||||
def set_prompt_handler(handler: Optional[Callable[[str], str]]) -> None:
|
||||
"""Route every :func:`prompt_user` call to *handler*, for a process with no terminal to prompt on."""
|
||||
global _prompt_handler
|
||||
_prompt_handler = handler
|
||||
|
||||
|
||||
def prompt_user(prompt: str) -> str:
|
||||
"""Ask the user for input on the shared console, themed and indented like the rest of the output."""
|
||||
if _prompt_handler is not None:
|
||||
return _prompt_handler(prompt)
|
||||
indent = " " * 5
|
||||
body = Text(indent + prompt.rstrip("\n ").replace("\n", "\n" + indent), style="text")
|
||||
body.append("\n" + indent + "> ", style="rule.text")
|
||||
@@ -478,4 +489,5 @@ __all__ = (
|
||||
"listing_table",
|
||||
"print_wide",
|
||||
"prompt_user",
|
||||
"set_prompt_handler",
|
||||
)
|
||||
|
||||
@@ -464,10 +464,9 @@ class Service(metaclass=ABCMeta):
|
||||
def request_input(self, prompt: str) -> str:
|
||||
"""Request interactive input from the user.
|
||||
|
||||
When running locally (CLI), prompts through the shared rich console so the
|
||||
prompt renders correctly alongside Live progress / log handlers.
|
||||
When running in serve mode with an :class:`InputBridge` attached,
|
||||
delegates to the bridge which relays the prompt to the remote client.
|
||||
Locally, prompts through the shared rich console so the prompt renders alongside
|
||||
Live progress. A remote-dl session relays it through its :class:`InputBridge`, and a
|
||||
REST download job relays it through ``prompt_user`` to the job's ``input_prompt``.
|
||||
"""
|
||||
if self._input_bridge is not None:
|
||||
return self._input_bridge.request_input(prompt)
|
||||
|
||||
Reference in New Issue
Block a user