mirror of
https://github.com/unshackle-dl/unshackle.git
synced 2026-11-14 13:05:09 +00:00
feat(serve): forward per-session service logs to remote-dl clients
Mirror each remote session's self.log into a bounded buffer, drained by the client via GET /api/session/{id}/logs and re-emitted locally. Server-account sessions get no relay. Also namespace service CLI options under service_params so a transport key (profile) no longer collides with a service's own option, on both the session and download-job paths.
This commit is contained in:
@@ -836,6 +836,8 @@ DELETE /api/session/{id} → tear down, harvest updated cache
|
||||
|
||||
Make a remote session for a service and title. Authentication runs asynchronously in the background. This call returns immediately with `status: "authenticating"`, and you then poll the prompt endpoint. The body accepts `service` and `title_id` (both required). It also accepts a broad set of optional keys, because the body allows `additionalProperties`. These are `credentials` (`{username, password, extra?}`), `cookies` (base64 of zlib-compressed Netscape cookie file), `proxy`, `no_proxy`, `profile`, `cache` (map of forwarded cache files), `client_region`, `proxy_region`, `cdm_type`, `range_`, `vcodec`, `quality`, `best_available`, and any service CLI options. `proxy_region` is the country code the client resolved `proxy` from; the server matches it against its own accounts.
|
||||
|
||||
Service CLI options also travel in a nested `service_params` object, which wins over a flat key with the same name. `profile` at the top level always means the credentials profile, never a service's own `--profile` option.
|
||||
|
||||
The `proxy` value must be a full proxy URI, unless the operator gives your API key `server_proxy` in the `serve.users` config. Without it, the server does not resolve country codes with its own proxy providers. It rejects the request with `INVALID_PROXY` when no `proxy` is set and the reported `client_region` differs from the server's region. A request that reports no `client_region` is not blocked. Pass `proxy` with your own proxy, or `no_proxy` to accept the server's own connection. With `server_proxy`, the server resolves a country code and picks a proxy for your `client_region` itself.
|
||||
|
||||
=== "Response `200`"
|
||||
@@ -907,6 +909,30 @@ Submit a response to a pending prompt. Body: `{ "response": "..." }` (required).
|
||||
| `400` | `INVALID_INPUT` | Missing `response`, or no prompt is pending. |
|
||||
| `404` | `SESSION_NOT_FOUND` | No such session. |
|
||||
|
||||
### `GET /api/session/{session_id}/logs`
|
||||
|
||||
Drain the service's server-side log output for this remote session. The server mirrors every log call the service makes (`self.log.info`, warnings, errors) into a bounded per-session buffer, so a remote client can see why an auth, title, or license step failed. Pass `since=<seq>` to get only records newer than the last drained sequence number. Works in every auth state, including `failed`.
|
||||
|
||||
=== "Response `200`"
|
||||
|
||||
```json
|
||||
{
|
||||
"session_id": "a1b2c3",
|
||||
"logs": [
|
||||
{ "seq": 1, "level": "INFO", "message": "Signed in as profile 'Main'", "ts": 1724990000.1 },
|
||||
{ "seq": 2, "level": "WARNING", "message": " - ThePlatform returned no streams", "ts": 1724990002.7 }
|
||||
],
|
||||
"last_seq": 2
|
||||
}
|
||||
```
|
||||
|
||||
| Status | Error code | Meaning |
|
||||
| --- | --- | --- |
|
||||
| `200` | - | Records returned (empty list when nothing new). |
|
||||
| `404` | `SESSION_NOT_FOUND` | No such session. |
|
||||
|
||||
The buffer keeps the newest 500 records. Only the service instance's own logger is mirrored; core pipeline loggers stay on the server. A server-account session gets no log relay at all - auth-time service output can carry the operator account's identity, so it stays server-side.
|
||||
|
||||
### `GET /api/session/{session_id}/titles`
|
||||
|
||||
Fetch the titles for an authenticated remote session.
|
||||
|
||||
@@ -195,7 +195,9 @@ request and response in full.
|
||||
| `GET` | `/api/session/{session_id}/titles` | Titles for the session's title ID. |
|
||||
| `POST` | `/api/session/{session_id}/tracks` | Tracks (with URLs) for a chosen title. |
|
||||
| `POST` | `/api/session/{session_id}/segments` | Per-track segment/download details. |
|
||||
| `POST` | `/api/session/{session_id}/segment_filter` | Server-side HLS segment filter results for one track. |
|
||||
| `POST` | `/api/session/{session_id}/license` | Proxy or server-side DRM licensing. |
|
||||
| `GET` | `/api/session/{session_id}/logs` | Drain the service's server-side log output. |
|
||||
| `GET` | `/api/session/{session_id}/prompt` | Poll for interactive auth status/prompt. |
|
||||
| `POST` | `/api/session/{session_id}/prompt` | Submit an answer to an auth prompt. |
|
||||
|
||||
|
||||
@@ -152,6 +152,18 @@ hood the client walks a remote session through its lifecycle.
|
||||
Enter the 6-digit code sent to your email: 483920
|
||||
```
|
||||
|
||||
A client older than the `service_params` change sends service options as
|
||||
flat body keys only. A flat key that shares a name with a transport field
|
||||
(such as `profile`, which at the top level always means the credentials
|
||||
profile) no longer reaches the service; the server logs a warning naming
|
||||
the dropped key. Update the client to send service options under
|
||||
`service_params`.
|
||||
|
||||
The server also mirrors the service's own log output (`self.log`) into a
|
||||
per-session buffer. The client drains it through
|
||||
`GET /api/session/{id}/logs` after each step and prints the records
|
||||
locally, so you see the server-side reason when a step fails.
|
||||
|
||||
=== "3. Titles + tracks"
|
||||
|
||||
Once authenticated, the client fetches the title list, then the tracks for the
|
||||
@@ -294,6 +306,7 @@ The server mounts all these routes, even in `--remote-only` mode. Paths use the
|
||||
| `POST` | `/api/session/{id}/segments` | Resolve per-segment/track download descriptors |
|
||||
| `POST` | `/api/session/{id}/segment_filter` | Unwanted HLS segment URIs for one track (ads, bumpers) |
|
||||
| `POST` | `/api/session/{id}/license` | DRM licensing (proxy or server CDM) |
|
||||
| `GET` | `/api/session/{id}/logs` | Drain the service's server-side log output |
|
||||
| `GET` | `/api/session/{id}/prompt` | Poll interactive auth status / pending prompt |
|
||||
| `POST` | `/api/session/{id}/prompt` | Submit an answer to a pending prompt |
|
||||
| `GET` | `/api/session/{id}` | Session info (validity, TTL, counts) |
|
||||
|
||||
@@ -74,6 +74,7 @@ def test_setup_routes_full_mode_wires_all_endpoints(make_app) -> None:
|
||||
("POST", "/api/session/{session_id}/tracks"),
|
||||
("POST", "/api/session/{session_id}/segments"),
|
||||
("POST", "/api/session/{session_id}/license"),
|
||||
("GET", "/api/session/{session_id}/logs"),
|
||||
("GET", "/api/session/{session_id}/prompt"),
|
||||
("POST", "/api/session/{session_id}/prompt"),
|
||||
("GET", "/api/session/{session_id}"),
|
||||
@@ -99,6 +100,7 @@ def test_setup_routes_remote_only_excludes_list_and_download(make_app) -> None:
|
||||
("POST", "/api/session/{session_id}/tracks"),
|
||||
("POST", "/api/session/{session_id}/segments"),
|
||||
("POST", "/api/session/{session_id}/license"),
|
||||
("GET", "/api/session/{session_id}/logs"),
|
||||
("GET", "/api/session/{session_id}/prompt"),
|
||||
("POST", "/api/session/{session_id}/prompt"),
|
||||
("GET", "/api/session/{session_id}"),
|
||||
|
||||
@@ -0,0 +1,182 @@
|
||||
"""Session log forwarding: buffer semantics, mirror wrapping, handler drain."""
|
||||
|
||||
import logging
|
||||
import threading
|
||||
|
||||
import pytest
|
||||
|
||||
from unshackle.core.api.handlers import session_logs_handler
|
||||
from unshackle.core.api.session_log import SessionLogBuffer, SessionLogMirror, capture_service_logs
|
||||
from unshackle.core.api.session_store import get_session_store
|
||||
|
||||
|
||||
def test_buffer_sequences_and_since() -> None:
|
||||
buf = SessionLogBuffer(maxlen=3)
|
||||
for i in range(5):
|
||||
buf.append(logging.INFO, f"msg {i}")
|
||||
records = buf.since(0)
|
||||
# bounded: only the newest 3 survive, sequence numbers keep counting
|
||||
assert [r["seq"] for r in records] == [3, 4, 5]
|
||||
assert buf.since(4) == [records[-1]]
|
||||
assert buf.last_seq == 5
|
||||
|
||||
|
||||
def test_buffer_sanitizes_messages() -> None:
|
||||
buf = SessionLogBuffer()
|
||||
buf.append(logging.WARNING, "line1\nline2")
|
||||
record = buf.since(0)[0]
|
||||
assert "\n" not in record["message"]
|
||||
assert record["level"] == "WARNING"
|
||||
|
||||
|
||||
def test_mirror_copies_all_levels_and_lazy_formatting() -> None:
|
||||
buf = SessionLogBuffer()
|
||||
mirror = SessionLogMirror(logging.getLogger("test-session-log"), buf)
|
||||
mirror.info("hello %s", "world")
|
||||
mirror.warning("warn")
|
||||
mirror.error("bad")
|
||||
messages = [(r["level"], r["message"]) for r in buf.since(0)]
|
||||
assert messages == [("INFO", "hello world"), ("WARNING", "warn"), ("ERROR", "bad")]
|
||||
|
||||
|
||||
def test_mirror_buffers_below_logger_level() -> None:
|
||||
logger = logging.getLogger("test-session-log-level")
|
||||
logger.setLevel(logging.WARNING)
|
||||
buf = SessionLogBuffer()
|
||||
SessionLogMirror(logger, buf).info("still captured")
|
||||
assert [r["message"] for r in buf.since(0)] == ["still captured"]
|
||||
|
||||
|
||||
def test_capture_service_logs_attaches_and_detaches() -> None:
|
||||
logger = logging.getLogger("test-capture-tag")
|
||||
buf = SessionLogBuffer()
|
||||
with capture_service_logs("test-capture-tag", buf):
|
||||
logger.warning("inside")
|
||||
logger.warning("outside")
|
||||
assert [r["message"] for r in buf.since(0)] == ["inside"]
|
||||
assert not logger.handlers
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_session_logs_handler_drains_since() -> None:
|
||||
store = get_session_store()
|
||||
entry = await store.create("EXAMPLE", object(), session_id="log-test-session")
|
||||
try:
|
||||
entry.log_buffer = SessionLogBuffer()
|
||||
entry.log_buffer.append(logging.INFO, "first")
|
||||
entry.log_buffer.append(logging.ERROR, "second")
|
||||
|
||||
resp = await session_logs_handler("log-test-session", since=0)
|
||||
import json
|
||||
|
||||
body = json.loads(resp.body)
|
||||
assert [r["message"] for r in body["logs"]] == ["first", "second"]
|
||||
assert body["last_seq"] == 2
|
||||
|
||||
resp = await session_logs_handler("log-test-session", since=2)
|
||||
body = json.loads(resp.body)
|
||||
assert body["logs"] == []
|
||||
assert body["last_seq"] == 2
|
||||
finally:
|
||||
await store.delete("log-test-session")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_session_logs_handler_no_buffer() -> None:
|
||||
store = get_session_store()
|
||||
await store.create("EXAMPLE", object(), session_id="log-test-nobuf")
|
||||
try:
|
||||
resp = await session_logs_handler("log-test-nobuf", since=0)
|
||||
import json
|
||||
|
||||
assert json.loads(resp.body)["logs"] == []
|
||||
finally:
|
||||
await store.delete("log-test-nobuf")
|
||||
|
||||
|
||||
def test_capture_service_logs_none_buffer_noop() -> None:
|
||||
logger = logging.getLogger("test-capture-none")
|
||||
with capture_service_logs("test-capture-none", None):
|
||||
logger.warning("nothing captured")
|
||||
assert not logger.handlers
|
||||
|
||||
|
||||
def test_capture_service_logs_lowers_level_for_window() -> None:
|
||||
logger = logging.getLogger("test-capture-level")
|
||||
logger.setLevel(logging.WARNING)
|
||||
buf = SessionLogBuffer()
|
||||
with capture_service_logs("test-capture-level", buf):
|
||||
assert logger.getEffectiveLevel() == logging.INFO
|
||||
logger.info("info inside window")
|
||||
assert logger.level == logging.WARNING
|
||||
assert [r["message"] for r in buf.since(0)] == ["info inside window"]
|
||||
|
||||
|
||||
def test_transport_keys_block_profile_and_service_params() -> None:
|
||||
from unshackle.core.api.handlers import LIST_HANDLER_TRANSPORT_KEYS, SESSION_TRANSPORT_KEYS
|
||||
|
||||
assert "profile" in SESSION_TRANSPORT_KEYS
|
||||
assert "service_params" in SESSION_TRANSPORT_KEYS
|
||||
assert "profile" in LIST_HANDLER_TRANSPORT_KEYS
|
||||
assert "service_params" in LIST_HANDLER_TRANSPORT_KEYS
|
||||
|
||||
|
||||
def test_drain_server_logs_survives_client_systemexit() -> None:
|
||||
"""RemoteClient raises SystemExit on transport errors; the drain must swallow it."""
|
||||
from unshackle.core.remote_service import RemoteService
|
||||
|
||||
svc = RemoteService.__new__(RemoteService)
|
||||
svc._session_id = "abc"
|
||||
svc._log_seq = 0
|
||||
svc._log_drain_lock = threading.Lock()
|
||||
|
||||
class ExitingClient:
|
||||
def get_optional(self, endpoint: str) -> dict:
|
||||
raise SystemExit(1)
|
||||
|
||||
svc.client = ExitingClient()
|
||||
svc.drain_server_logs() # must not raise
|
||||
assert svc._log_seq == 0
|
||||
|
||||
|
||||
def test_drain_server_logs_reemits_and_advances_seq() -> None:
|
||||
from unshackle.core.remote_service import RemoteService
|
||||
|
||||
svc = RemoteService.__new__(RemoteService)
|
||||
svc._session_id = "abc"
|
||||
svc._log_seq = 0
|
||||
svc._log_drain_lock = threading.Lock()
|
||||
svc.log = logging.getLogger("test-drain-reemit")
|
||||
|
||||
class StubClient:
|
||||
def get_optional(self, endpoint: str) -> dict:
|
||||
assert endpoint.endswith("since=0")
|
||||
return {"logs": [{"seq": 1, "level": "WARNING", "message": "server says"}], "last_seq": 1}
|
||||
|
||||
svc.client = StubClient()
|
||||
records: list[logging.LogRecord] = []
|
||||
|
||||
class Capture(logging.Handler):
|
||||
def emit(self, record: logging.LogRecord) -> None:
|
||||
records.append(record)
|
||||
|
||||
handler = Capture()
|
||||
svc.log.addHandler(handler)
|
||||
try:
|
||||
svc.drain_server_logs()
|
||||
finally:
|
||||
svc.log.removeHandler(handler)
|
||||
assert svc._log_seq == 1
|
||||
assert [(r.levelno, r.getMessage()) for r in records] == [(logging.WARNING, "server says")]
|
||||
|
||||
|
||||
def test_mirror_buffers_exception_cause() -> None:
|
||||
buf = SessionLogBuffer()
|
||||
mirror = SessionLogMirror(logging.getLogger("test-mirror-exc"), buf)
|
||||
try:
|
||||
raise ValueError("boom")
|
||||
except ValueError:
|
||||
mirror.exception("step failed")
|
||||
record = buf.since(0)[0]
|
||||
assert record["level"] == "ERROR"
|
||||
assert record["message"] == "step failed | ValueError: boom"
|
||||
@@ -514,8 +514,13 @@ def perform_download(
|
||||
service_kwargs["title"] = title_id
|
||||
|
||||
for key, value in params.items():
|
||||
if key in service_init_params and key not in ["service", "title_id"]:
|
||||
if key in service_init_params and key not in ["service", "title_id", "profile", "service_params"]:
|
||||
service_kwargs[key] = value
|
||||
service_params = params.get("service_params")
|
||||
if isinstance(service_params, dict):
|
||||
for key, value in service_params.items():
|
||||
if key in service_init_params and key != "title":
|
||||
service_kwargs[key] = value
|
||||
|
||||
for param_name, param_info in service_init_params.items():
|
||||
if param_name not in service_kwargs and param_name not in ["self", "ctx"]:
|
||||
|
||||
@@ -178,8 +178,17 @@ def categorize_exception(
|
||||
APIError with appropriate error code and details
|
||||
"""
|
||||
context = context or {}
|
||||
exc_str = str(exc).lower()
|
||||
exc_type = type(exc).__name__
|
||||
root = exc.__cause__ or exc
|
||||
exc_str = f"{exc} {root}".lower()
|
||||
exc_type = type(root).__name__
|
||||
|
||||
if "proxy" in str(root).lower():
|
||||
return APIError(
|
||||
error_code=APIErrorCode.INVALID_PROXY,
|
||||
message=f"The proxy for this session failed: {root}",
|
||||
details={**context, "reason": "proxy_error"},
|
||||
retryable=False,
|
||||
)
|
||||
|
||||
if any(keyword in exc_str for keyword in ["auth", "login", "credential", "unauthorized", "forbidden", "token"]):
|
||||
return APIError(
|
||||
|
||||
@@ -15,6 +15,7 @@ from unshackle.core.api.compression import safe_inflate
|
||||
from unshackle.core.api.errors import APIError, APIErrorCode, handle_api_exception
|
||||
from unshackle.core.api.input_bridge import AuthStatus, InputBridge
|
||||
from unshackle.core.api.sanitize import safe_cache_key, sanitize_log
|
||||
from unshackle.core.api.session_log import SessionLogBuffer, SessionLogMirror, capture_service_logs
|
||||
from unshackle.core.api.session_store import SessionStore
|
||||
from unshackle.core.cacher import Cacher
|
||||
from unshackle.core.config import config
|
||||
@@ -111,6 +112,7 @@ LIST_HANDLER_TRANSPORT_KEYS = {
|
||||
"proxy",
|
||||
"no_proxy",
|
||||
"query",
|
||||
"service_params",
|
||||
}
|
||||
|
||||
|
||||
@@ -220,6 +222,18 @@ def instantiate_service(
|
||||
for k, v in data.items():
|
||||
if k in cli_param_names and k not in transport_keys and k != "title":
|
||||
extras[k] = v
|
||||
service_params = data.get("service_params")
|
||||
if isinstance(service_params, dict):
|
||||
for k, v in service_params.items():
|
||||
if k in cli_param_names and k != "title":
|
||||
extras[k] = v
|
||||
elif service_params is None:
|
||||
for k in cli_param_names & transport_keys & set(data):
|
||||
if k not in ("service", "title_id", "proxy", "no_proxy") and data.get(k) is not None:
|
||||
log.warning(
|
||||
f"Ignoring flat '{sanitize_log(k)}' as a service option (it names a transport field); "
|
||||
"update the client to send service options under 'service_params'"
|
||||
)
|
||||
return parent_ctx.invoke(service_module.cli, title=title, **extras)
|
||||
|
||||
|
||||
@@ -2212,6 +2226,8 @@ async def env_check_handler(request: Optional[web.Request] = None) -> web.Respon
|
||||
SESSION_TRANSPORT_KEYS = {
|
||||
"service",
|
||||
"title_id",
|
||||
"profile",
|
||||
"service_params",
|
||||
"season",
|
||||
"episode",
|
||||
"part",
|
||||
@@ -2388,15 +2404,20 @@ async def session_create_handler(data: Dict[str, Any], request: Optional[web.Req
|
||||
profile = next_server_profile(normalized_service, region)
|
||||
log.info(f"Using server account '{sanitize_log(profile or 'default')}' for {normalized_service}")
|
||||
|
||||
service_instance, cookies, credential = create_service_instance(
|
||||
normalized_service,
|
||||
title_id,
|
||||
data,
|
||||
proxy_param,
|
||||
proxy_providers,
|
||||
profile,
|
||||
server_account=server_account,
|
||||
)
|
||||
log_buffer = None if server_account else SessionLogBuffer()
|
||||
service_class_name = getattr(Services.load(normalized_service), "__name__", normalized_service)
|
||||
with capture_service_logs(service_class_name, log_buffer):
|
||||
service_instance, cookies, credential = create_service_instance(
|
||||
normalized_service,
|
||||
title_id,
|
||||
data,
|
||||
proxy_param,
|
||||
proxy_providers,
|
||||
profile,
|
||||
server_account=server_account,
|
||||
)
|
||||
if log_buffer:
|
||||
service_instance.log = SessionLogMirror(service_instance.log, log_buffer)
|
||||
|
||||
if server_account:
|
||||
session_cache_tag = None
|
||||
@@ -2435,6 +2456,7 @@ async def session_create_handler(data: Dict[str, Any], request: Optional[web.Req
|
||||
if isinstance(data.get("client"), dict) and len(json.dumps(data["client"], default=str)) <= 4096:
|
||||
session.client = data["client"]
|
||||
session.input_bridge = bridge
|
||||
session.log_buffer = log_buffer
|
||||
session.auth_status = AuthStatus.AUTHENTICATING
|
||||
|
||||
async def run_auth() -> None:
|
||||
@@ -2853,6 +2875,26 @@ async def session_prompt_post_handler(
|
||||
return web.json_response({"status": "accepted"})
|
||||
|
||||
|
||||
async def session_logs_handler(session_id: str, since: int = 0, request: Optional[web.Request] = None) -> web.Response:
|
||||
"""Drain the remote session's service log records newer than *since*.
|
||||
|
||||
Gives a remote client the service's server-side ``self.log`` output for
|
||||
its own remote session - the reason an auth, title, or manifest step
|
||||
failed. Works in every auth state, because a failed session is exactly
|
||||
when the client needs the logs.
|
||||
"""
|
||||
session = await get_validated_session(session_id, request)
|
||||
buffer = session.log_buffer
|
||||
logs = buffer.since(since) if buffer else []
|
||||
return web.json_response(
|
||||
{
|
||||
"session_id": session_id,
|
||||
"logs": logs,
|
||||
"last_seq": logs[-1]["seq"] if logs else since,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
async def get_validated_session(session_id: str, request: Optional[web.Request]) -> Any:
|
||||
"""Fetch a remote session and make sure that the caller owns it.
|
||||
|
||||
|
||||
@@ -47,6 +47,7 @@ from unshackle.core.api.handlers import (
|
||||
session_delete_handler,
|
||||
session_info_handler,
|
||||
session_license_handler,
|
||||
session_logs_handler,
|
||||
session_prompt_get_handler,
|
||||
session_prompt_post_handler,
|
||||
session_segment_filter_handler,
|
||||
@@ -1913,6 +1914,66 @@ async def session_prompt_get(request: web.Request) -> web.Response:
|
||||
)
|
||||
|
||||
|
||||
@api_handler
|
||||
async def session_logs(request: web.Request) -> web.Response:
|
||||
"""
|
||||
Drain the remote session's service log records.
|
||||
---
|
||||
summary: Get session logs
|
||||
description: The service's server-side log output for this remote session, newer than the given sequence number
|
||||
parameters:
|
||||
- name: session_id
|
||||
in: path
|
||||
required: true
|
||||
schema:
|
||||
type: string
|
||||
- name: since
|
||||
in: query
|
||||
required: false
|
||||
schema:
|
||||
type: integer
|
||||
default: 0
|
||||
description: Return only records with a sequence number greater than this
|
||||
responses:
|
||||
'200':
|
||||
description: Log records, oldest first
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
type: object
|
||||
properties:
|
||||
logs:
|
||||
type: array
|
||||
items:
|
||||
type: object
|
||||
properties:
|
||||
seq:
|
||||
type: integer
|
||||
level:
|
||||
type: string
|
||||
message:
|
||||
type: string
|
||||
ts:
|
||||
type: number
|
||||
last_seq:
|
||||
type: integer
|
||||
'404':
|
||||
description: Remote session not found
|
||||
"""
|
||||
session_id = request.match_info["session_id"]
|
||||
try:
|
||||
since = int(request.query.get("since", 0))
|
||||
except ValueError:
|
||||
since = 0
|
||||
try:
|
||||
return await session_logs_handler(session_id, since, request)
|
||||
except Exception as e:
|
||||
log.exception("Error in session logs")
|
||||
return handle_api_exception(
|
||||
e, context={"operation": "session_logs"}, debug_mode=request.app.get("debug_api", False)
|
||||
)
|
||||
|
||||
|
||||
@api_handler
|
||||
async def session_prompt_submit(request: web.Request) -> web.Response:
|
||||
"""
|
||||
@@ -2128,6 +2189,7 @@ ROUTES: list[tuple[str, str, Handler, bool]] = [
|
||||
("POST", "/api/session/{session_id}/segments", session_segments, True),
|
||||
("POST", "/api/session/{session_id}/segment_filter", session_segment_filter, True),
|
||||
("POST", "/api/session/{session_id}/license", session_license, True),
|
||||
("GET", "/api/session/{session_id}/logs", session_logs, True),
|
||||
("GET", "/api/session/{session_id}/prompt", session_prompt_get, True),
|
||||
("POST", "/api/session/{session_id}/prompt", session_prompt_submit, True),
|
||||
("GET", "/api/session/{session_id}", session_info, True),
|
||||
|
||||
@@ -0,0 +1,150 @@
|
||||
"""Per-session capture of service log output for remote-dl clients.
|
||||
|
||||
A service running under ``serve`` logs to the server's console, so a remote
|
||||
client never sees why an auth or manifest step failed. Each remote session
|
||||
gets a bounded :class:`SessionLogBuffer`; the service instance's ``self.log``
|
||||
is wrapped in a :class:`SessionLogMirror` that copies every call into the
|
||||
buffer, and the client drains it through ``GET /api/session/{id}/logs``.
|
||||
|
||||
Only ``self.log`` calls are mirrored. Output from module-level loggers (the
|
||||
manifest parsers, the downloader) stays server-side.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
import traceback
|
||||
from collections import deque
|
||||
from typing import Any, Deque, Dict, List, MutableMapping, Optional, Tuple
|
||||
|
||||
from unshackle.core.api.sanitize import sanitize_log
|
||||
|
||||
MAX_RECORDS = 500
|
||||
MAX_MESSAGE_LEN = 2000
|
||||
|
||||
|
||||
class SessionLogBuffer:
|
||||
"""Thread-safe bounded buffer of log records with a monotonic sequence."""
|
||||
|
||||
def __init__(self, maxlen: int = MAX_RECORDS) -> None:
|
||||
self._records: Deque[Dict[str, Any]] = deque(maxlen=maxlen)
|
||||
self._seq = 0
|
||||
self._lock = threading.Lock()
|
||||
|
||||
def append(self, level: int, message: str) -> None:
|
||||
with self._lock:
|
||||
self._seq += 1
|
||||
self._records.append(
|
||||
{
|
||||
"seq": self._seq,
|
||||
"level": logging.getLevelName(level),
|
||||
"message": sanitize_log(message)[:MAX_MESSAGE_LEN],
|
||||
"ts": time.time(),
|
||||
}
|
||||
)
|
||||
|
||||
def since(self, seq: int) -> List[Dict[str, Any]]:
|
||||
"""Records with a sequence number greater than *seq*, oldest first."""
|
||||
with self._lock:
|
||||
return [dict(r) for r in self._records if r["seq"] > seq]
|
||||
|
||||
@property
|
||||
def last_seq(self) -> int:
|
||||
with self._lock:
|
||||
return self._seq
|
||||
|
||||
|
||||
def format_message(msg: Any, args: Tuple[Any, ...]) -> str:
|
||||
"""Render a logging call's message the way ``LogRecord`` would."""
|
||||
try:
|
||||
return str(msg) % args if args else str(msg)
|
||||
except (TypeError, ValueError):
|
||||
return str(msg)
|
||||
|
||||
|
||||
class SessionLogMirror(logging.LoggerAdapter):
|
||||
"""Wraps a service's ``self.log`` so every call also lands in the remote session's buffer.
|
||||
|
||||
The mirror fills the buffer regardless of the server's log level, so a
|
||||
client still gets INFO detail from a server that only prints warnings.
|
||||
"""
|
||||
|
||||
def __init__(self, logger: logging.Logger, buffer: SessionLogBuffer) -> None:
|
||||
super().__init__(logger, {})
|
||||
self.buffer = buffer
|
||||
|
||||
def log(self, level: int, msg: Any, *args: Any, **kwargs: Any) -> None:
|
||||
message = format_message(msg, args)
|
||||
# self.log.exception(...) carries its cause in exc_info; buffer the
|
||||
# exception line only - a full traceback would ship server paths.
|
||||
exc_info = kwargs.get("exc_info")
|
||||
if exc_info:
|
||||
if exc_info is True:
|
||||
exc_info = sys.exc_info()
|
||||
elif isinstance(exc_info, BaseException):
|
||||
exc_info = (type(exc_info), exc_info, exc_info.__traceback__)
|
||||
if isinstance(exc_info, tuple) and exc_info[1] is not None:
|
||||
cause = "".join(traceback.format_exception_only(exc_info[0], exc_info[1])).strip()
|
||||
message = f"{message} | {cause}"
|
||||
self.buffer.append(level, message)
|
||||
super().log(level, msg, *args, **kwargs)
|
||||
|
||||
def process(self, msg: Any, kwargs: MutableMapping[str, Any]) -> tuple[Any, MutableMapping[str, Any]]:
|
||||
return msg, kwargs
|
||||
|
||||
|
||||
class SessionLogHandler(logging.Handler):
|
||||
"""Copies records from a real logger into a remote session's buffer.
|
||||
|
||||
Attached to the service tag's logger only for the short window before the
|
||||
instance's ``self.log`` can be swapped for a :class:`SessionLogMirror`
|
||||
(service ``__init__``: geofence checks, proxy selection).
|
||||
"""
|
||||
|
||||
def __init__(self, buffer: SessionLogBuffer) -> None:
|
||||
super().__init__()
|
||||
self.buffer = buffer
|
||||
|
||||
def emit(self, record: logging.LogRecord) -> None:
|
||||
try:
|
||||
self.buffer.append(record.levelno, record.getMessage())
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
class capture_service_logs:
|
||||
"""Context manager: mirror a class logger's records into *buffer* while active.
|
||||
|
||||
*logger_name* must be the service CLASS name (``Service.__init__`` logs to
|
||||
``logging.getLogger(self.__class__.__name__)``), not the directory tag.
|
||||
The context manager lowers the logger's level to INFO for the window so a
|
||||
quiet server still fills the buffer; a no-op when *buffer* is None.
|
||||
|
||||
The logger is process-global, but the capture window only wraps the
|
||||
synchronous ``create_service_instance`` call on the event loop thread, so
|
||||
two session creates cannot interleave inside it. If an ``await`` ever moves
|
||||
inside the window, add a thread/context filter or records cross tenants.
|
||||
"""
|
||||
|
||||
def __init__(self, logger_name: str, buffer: Optional[SessionLogBuffer]) -> None:
|
||||
self._logger = logging.getLogger(logger_name)
|
||||
self._handler: Optional[SessionLogHandler] = SessionLogHandler(buffer) if buffer else None
|
||||
self._prev_level: Optional[int] = None
|
||||
|
||||
def __enter__(self) -> None:
|
||||
if self._handler:
|
||||
if self._logger.getEffectiveLevel() > logging.INFO:
|
||||
self._prev_level = self._logger.level
|
||||
self._logger.setLevel(logging.INFO)
|
||||
self._logger.addHandler(self._handler)
|
||||
|
||||
def __exit__(self, *exc: Any) -> None:
|
||||
if self._handler:
|
||||
self._logger.removeHandler(self._handler)
|
||||
if self._prev_level is not None:
|
||||
self._logger.setLevel(self._prev_level)
|
||||
self._prev_level = None
|
||||
self._handler = None
|
||||
@@ -42,6 +42,7 @@ class SessionEntry:
|
||||
cache_tag: Optional[str] = None
|
||||
server_account: Optional[str] = None # profile name when the server lent its own account
|
||||
input_bridge: Optional[InputBridge] = None
|
||||
log_buffer: Optional[Any] = None # SessionLogBuffer mirroring the service's self.log
|
||||
auth_status: AuthStatus = AuthStatus.AUTHENTICATED
|
||||
auth_error: Optional[str] = None
|
||||
client: Dict[str, Any] = field(default_factory=dict) # version/argv/platform the client reported
|
||||
|
||||
@@ -65,7 +65,7 @@ async def stats_middleware(request: web.Request, handler: Any) -> web.StreamResp
|
||||
else:
|
||||
stats.requests_by_key[mask_key(request_secret_key(request))] += 1
|
||||
session_id = request.match_info.get("session_id")
|
||||
if session_id and request.method != "OPTIONS":
|
||||
if session_id and request.method != "OPTIONS" and not request.path.endswith("/logs"):
|
||||
from unshackle.core.api.session_store import get_session_store
|
||||
|
||||
get_session_store().record_action(
|
||||
|
||||
@@ -15,6 +15,7 @@ from datetime import date as date_
|
||||
from enum import Enum
|
||||
from http.cookiejar import CookieJar
|
||||
from pathlib import Path
|
||||
from threading import Lock
|
||||
from typing import Any, Callable, Dict, Optional, Union
|
||||
|
||||
import click
|
||||
@@ -123,6 +124,10 @@ class RemoteClient:
|
||||
def get(self, endpoint: str) -> Dict[str, Any]:
|
||||
return self.request("get", endpoint)
|
||||
|
||||
def get_optional(self, endpoint: str) -> Dict[str, Any]:
|
||||
"""GET an endpoint that an older server may not have. Empty dict when the route is missing."""
|
||||
return self.request("get", endpoint, optional=True)
|
||||
|
||||
def delete(self, endpoint: str) -> Dict[str, Any]:
|
||||
return self.request("delete", endpoint)
|
||||
|
||||
@@ -646,6 +651,8 @@ class RemoteService:
|
||||
self._session_id: Optional[str] = None
|
||||
self._server_cdm_type: str = "widevine"
|
||||
self._segment_filters: Dict[str, tuple[set[str], set[str]]] = {}
|
||||
self._log_seq = 0
|
||||
self._log_drain_lock = Lock()
|
||||
|
||||
self._session = requests.Session()
|
||||
self._session.headers.update(config.headers)
|
||||
@@ -779,6 +786,7 @@ class RemoteService:
|
||||
|
||||
if self._service_params:
|
||||
create_data.update(self._service_params)
|
||||
create_data["service_params"] = self._service_params
|
||||
|
||||
cdm = self.ctx.obj.cdm if self.ctx.obj else None
|
||||
if cdm is not None:
|
||||
@@ -796,6 +804,30 @@ class RemoteService:
|
||||
status = result.get("status", "authenticated")
|
||||
if status == "authenticating":
|
||||
self.poll_auth_completion()
|
||||
self.drain_server_logs()
|
||||
|
||||
def drain_server_logs(self) -> None:
|
||||
"""Fetch the service's server-side log records for this remote session and re-emit them locally.
|
||||
|
||||
The server mirrors every ``self.log`` call the service makes into a
|
||||
per-session buffer. Draining after each remote call shows the client
|
||||
why a step failed. A relay failure must never break the operation
|
||||
itself, so this method swallows every error.
|
||||
"""
|
||||
if not self._session_id:
|
||||
return
|
||||
with self._log_drain_lock:
|
||||
try:
|
||||
resp = self.client.get_optional(f"/api/session/{self._session_id}/logs?since={self._log_seq}")
|
||||
except (Exception, SystemExit):
|
||||
return
|
||||
if not resp:
|
||||
return
|
||||
level_names = logging.getLevelNamesMapping()
|
||||
for record in resp.get("logs", []):
|
||||
level = level_names.get(str(record.get("level", "INFO")), logging.INFO)
|
||||
self.log.log(level, str(record.get("message", "")))
|
||||
self._log_seq = int(resp.get("last_seq", self._log_seq))
|
||||
|
||||
def poll_auth_completion(self, poll_interval: float = 2.0, timeout: float = 600.0) -> None:
|
||||
"""Poll the server until authentication completes, handling interactive prompts.
|
||||
@@ -809,6 +841,7 @@ class RemoteService:
|
||||
while time.monotonic() < deadline:
|
||||
resp = self.client.get(f"/api/session/{self._session_id}/prompt")
|
||||
status = resp.get("status")
|
||||
self.drain_server_logs()
|
||||
|
||||
if status == "authenticated":
|
||||
return
|
||||
@@ -839,7 +872,10 @@ class RemoteService:
|
||||
def get_titles(self) -> Titles_T:
|
||||
if self._titles is not None:
|
||||
return self._titles
|
||||
result = self.client.get(f"/api/session/{self._session_id}/titles")
|
||||
try:
|
||||
result = self.client.get(f"/api/session/{self._session_id}/titles")
|
||||
finally:
|
||||
self.drain_server_logs()
|
||||
titles_list = [build_title(t, self.service_tag, self.title_id) for t in result.get("titles", [])]
|
||||
first = titles_list[0] if titles_list else None
|
||||
if isinstance(first, Episode):
|
||||
@@ -863,7 +899,10 @@ class RemoteService:
|
||||
title_id = str(title.id)
|
||||
if title_id in self._tracks_by_title:
|
||||
return self._tracks_by_title[title_id]
|
||||
result = self.client.post(f"/api/session/{self._session_id}/tracks", {"title_id": title_id})
|
||||
try:
|
||||
result = self.client.post(f"/api/session/{self._session_id}/tracks", {"title_id": title_id})
|
||||
finally:
|
||||
self.drain_server_logs()
|
||||
tracks = build_tracks(result)
|
||||
|
||||
remote_title = result.get("title") or {}
|
||||
@@ -953,6 +992,7 @@ class RemoteService:
|
||||
"drm_type": drm_type,
|
||||
},
|
||||
)
|
||||
self.drain_server_logs()
|
||||
keys_by_track = resp.get("keys", {})
|
||||
server_drm_type = resp.get("drm_type", drm_type)
|
||||
drm_types_by_track = resp.get("drm_types", {})
|
||||
@@ -1116,7 +1156,10 @@ class RemoteService:
|
||||
if pssh_b64:
|
||||
payload["pssh"] = pssh_b64
|
||||
|
||||
resp = self.client.post(f"/api/session/{self._session_id}/license", payload)
|
||||
try:
|
||||
resp = self.client.post(f"/api/session/{self._session_id}/license", payload)
|
||||
finally:
|
||||
self.drain_server_logs()
|
||||
return base64.b64decode(resp["license"])
|
||||
|
||||
def on_segment_downloaded(self, track: AnyTrack, segment: Any) -> None:
|
||||
|
||||
Reference in New Issue
Block a user