add logs page on settings page

app/routers/pages/settings/__init__.py

@@ -6,6 +6,7 @@
     chaptarr,
     download,
     indexers,
+    logs,
     notification,
     prowlarr,
     security,
@@ -19,6 +20,7 @@
 router.include_router(chaptarr.router)
 router.include_router(download.router)
 router.include_router(indexers.router)
+router.include_router(logs.router)
 router.include_router(notification.router)
 router.include_router(prowlarr.router)
 router.include_router(security.router)

app/routers/pages/settings/logs.py

@@ -0,0 +1,72 @@
+from collections.abc import Sequence
+from typing import Annotated
+from urllib.parse import urlencode
+
+from fastapi import APIRouter, Query, Security
+
+from app.internal.auth.authentication import ABRAuth, DetailedUser
+from app.internal.env_settings import Settings
+from app.internal.models import GroupEnum
+from app.util.log import LogEntry, snapshot_logs
+from app.util.templates import catalog_response
+
+router = APIRouter(prefix="/logs")
+
+LOG_LEVELS = ("debug", "info", "warning", "error")
+PAGE_SIZE = 250
+
+
+def _logs_url(levels: Sequence[str], offset: int = 0) -> str:
+    params: dict[str, str] = {}
+    if set(levels) != set(LOG_LEVELS):
+        params["level"] = ",".join(levels)
+    if offset > 0:
+        params["offset"] = str(offset)
+    base = Settings().app.base_url.rstrip("/")
+    return f"{base}/settings/logs" + (f"?{urlencode(params)}" if params else "")
+
+
+@router.get("")
+def read_logs(
+    admin_user: Annotated[DetailedUser, Security(ABRAuth(GroupEnum.admin))],
+    level: Annotated[str | None, Query()] = None,
+    offset: Annotated[int, Query(ge=0)] = 0,
+):
+    if level is None:
+        selected_levels = list(LOG_LEVELS)
+    else:
+        requested = {lvl.strip().lower() for lvl in level.split(",")}
+        selected_levels = [lvl for lvl in LOG_LEVELS if lvl in requested]
+
+    matching: list[LogEntry] = [
+        entry for entry in snapshot_logs() if entry["level"] in selected_levels
+    ]
+    offset = min(offset, len(matching))
+    end = len(matching) - offset
+    start = max(0, end - PAGE_SIZE)
+    logs = matching[start:end]
+
+    toggle_urls: dict[str, str] = {}
+    for lvl in LOG_LEVELS:
+        if lvl in selected_levels:
+            toggled = [other for other in selected_levels if other != lvl]
+        else:
+            toggled = [
+                other
+                for other in LOG_LEVELS
+                if other in selected_levels or other == lvl
+            ]
+        toggle_urls[lvl] = _logs_url(toggled)
+
+    return catalog_response(
+        "Settings.Logs.Index",
+        user=admin_user,
+        logs=logs,
+        log_levels=list(LOG_LEVELS),
+        selected_levels=selected_levels,
+        toggle_urls=toggle_urls,
+        older_url=_logs_url(selected_levels, offset + PAGE_SIZE) if start > 0 else None,
+        newer_url=_logs_url(selected_levels, max(0, offset - PAGE_SIZE))
+        if offset > 0
+        else None,
+    )

app/util/log.py

@@ -1,6 +1,37 @@
 import logging
+import threading
+from collections import deque
+from typing import TypedDict
 
 import structlog
+from structlog.typing import EventDict
+
+
+class LogEntry(TypedDict):
+    level: str
+    message: str
+
+
+# in-memory only, reset on restart. shortcut: capped so a long-running instance cannot grow unbounded, raise if full history is needed
+logs: deque[LogEntry] = deque(maxlen=10_000)
+_logs_lock = threading.Lock()
+
+_plain_renderer = structlog.dev.ConsoleRenderer(colors=False)
+
+
+def snapshot_logs() -> list[LogEntry]:
+    with _logs_lock:
+        return list(logs)
+
+
+def _capture_log_event(
+    logger: logging.Logger, method_name: str, event_dict: EventDict
+) -> EventDict:
+    message = _plain_renderer(logger, method_name, dict(event_dict))
+    with _logs_lock:
+        logs.append({"level": method_name, "message": message})
+    return event_dict
+
 
 structlog.configure(
     processors=[
@@ -9,6 +40,7 @@
         structlog.processors.StackInfoRenderer(),
         structlog.dev.set_exc_info,
         structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S", utc=False),
+        _capture_log_event,
         structlog.dev.ConsoleRenderer(),
     ],
     wrapper_class=structlog.make_filtering_bound_logger(logging.NOTSET),

templates/layouts/BaseLayout.jinja

@@ -2,6 +2,7 @@
     user: DetailedUser | None = None,
     title="AudiobookRequest",
     hide_navbar=False,
+    full_height: bool = False,
 #}
 
 <!DOCTYPE html>
@@ -78,7 +79,9 @@
         <js.Toast />
 
     </head>
-    <body class="w-screen min-h-screen overflow-x-hidden" hx-ext="preload">
+    <body class="w-screen flex flex-col overflow-x-hidden {% if full_height %}overflow-hidden{% else %}min-h-screen{% endif %}"
+          {% if full_height %}style="height: 100vh; height: 100dvh; overflow: hidden"{% endif %}
+          hx-ext="preload">
 
         <ToastBlock init />
 

templates/layouts/SettingsLayout.jinja

@@ -2,9 +2,10 @@
     user: DetailedUser,
     title: str,
     page: str,
+    fill: bool = False,
 #}
 
-<BaseLayout title={{ title }} user={{ user }}>
+<BaseLayout title={{ title }} user={{ user }} full_height={{ fill }}>
 
 {% if version %}
     <button class="fixed bottom-4 right-4 font-mono text-xs font-semibold text-neutral/80 dark:text-neutral-content/80 underline cursor-pointer z-50"
@@ -23,8 +24,10 @@
         </form>
     </dialog>
 {% endif %}
-<div class="flex items-start justify-center p-2 md:p-4 relative">
-    <main class="flex flex-col w-[90%] md:w-3/4 max-w-160 gap-4 gap-y-8 pb-40">
+<div class="flex justify-center p-2 md:p-4 relative {% if fill %}flex-1 min-h-0{% else %}items-start{% endif %}"
+     {% if fill %}style="flex: 1 1 0%; min-height: 0"{% endif %}>
+    <main class="flex flex-col w-[90%] md:w-3/4 max-w-160 gap-4 {% if fill %}min-h-0 pb-2{% else %}gap-y-8 pb-40{% endif %}"
+          {% if fill %}style="min-height: 0"{% endif %}>
         {% if user.is_admin() %}
             <div role="tablist" class="tabs tabs-box">
                 <a preload
@@ -59,6 +62,10 @@
                    href="{{ base_url }}/settings/indexers"
                    role="tab"
                    class="tab {% if page=='indexers' %}tab-active{% endif %}">Indexers</a>
+                <a preload
+                   href="{{ base_url }}/settings/logs"
+                   role="tab"
+                   class="tab {% if page=='logs' %}tab-active{% endif %}">Logs</a>
             </div>
         {% endif %}
 

templates/pages/Settings/Logs/Index.jinja

@@ -0,0 +1,60 @@
+{#def
+    user: DetailedUser,
+    logs: list[LogEntry],
+    log_levels: list[str],
+    selected_levels: list[str],
+    toggle_urls: dict[str, str],
+    older_url: str | None,
+    newer_url: str | None,
+#}
+
+{% set level_colors = {"debug": "text-base-content/50", "info": "text-info", "warning": "text-warning", "error": "text-error"} %}
+
+<SettingsLayout user={{ user }} page="logs" title="Settings - Logs" fill>
+
+<div class="flex flex-1 min-h-0 flex-col gap-2"
+     style="flex: 1 1 0%;
+            min-height: 0">
+    <div class="flex flex-wrap items-center justify-between gap-2">
+        <div class="flex flex-col">
+            <h2 class="text-lg">
+                Logs
+            </h2>
+            <p class="text-xs opacity-60">
+                Logs since the last restart.
+            </p>
+        </div>
+        <div class="join">
+            {% for level in log_levels %}
+                <a href="{{ toggle_urls[level] }}"
+                   class="btn btn-sm join-item {% if level in selected_levels %}btn-active{% else %}btn-ghost opacity-60{% endif %}"
+                   aria-pressed="{{ 'true' if level in selected_levels else 'false' }}">
+                    {{ level | upper }}
+                </a>
+            {% endfor %}
+        </div>
+    </div>
+    <div class="flex flex-1 min-h-0 flex-col overflow-hidden rounded-box bg-base-200"
+         style="flex: 1 1 0%;
+                min-height: 0">
+        {% if older_url %}
+            <a href="{{ older_url }}"
+               class="shrink-0 border-b border-base-300 py-1 text-center text-xs hover:bg-base-300">
+                Load more (older logs)
+            </a>
+        {% endif %}
+        <pre id="log-lines" class="flex-1 min-h-0 overflow-auto p-4 text-xs font-mono whitespace-pre-wrap" style="flex: 1 1 0%; min-height: 0">{% if logs %}{% if not older_url %}<span class="block text-center opacity-60">Top of the 10,000 log history reached</span>
+{% endif %}{% for entry in logs %}<span class="block {{ level_colors.get(entry.level, '') }}">{{ entry.message | e }}</span>
+{% endfor %}{% else %}<span class="opacity-60">No logs.</span>{% endif %}</pre>
+        {% if newer_url %}
+            <a href="{{ newer_url }}"
+               class="shrink-0 border-t border-base-300 py-1 text-center text-xs hover:bg-base-300">
+                Show newer logs
+            </a>
+        {% endif %}
+    </div>
+</div>
+<script>
+    document.getElementById("log-lines").scrollTop = document.getElementById("log-lines").scrollHeight;
+</script>
+</SettingsLayout>

tests/test_page_endpoints.py

@@ -8,6 +8,7 @@
 from app.internal.auth.config import auth_config
 from app.internal.auth.login_types import LoginTypeEnum
 from app.internal.models import APIKey
+from app.util.log import logger
 from tests.constants import ASIN, MANUAL_ID
 from tests.http import as_dict, json_list
 from tests.mock_services import MockServer
@@ -372,4 +373,38 @@ def test_users_page(admin_client: TestClient) -> None:
     assert deleted.status_code == 200, deleted.text
 
 
+# -- settings: logs --------------------------------------------------------
+def test_logs_page(admin_client: TestClient) -> None:
+    logger.info("ui-logs-marker")
+    logger.error("ui-logs-<b>marker</b>")
+
+    response = admin_client.get("/settings/logs")
+    assert response.status_code == 200
+    assert "ui-logs-marker" in response.text
+    assert "ui-logs-&lt;b&gt;marker&lt;/b&gt;" in response.text
+    assert response.text.count('aria-pressed="true"') == 4
+
+    filtered = admin_client.get("/settings/logs?level=error")
+    assert "ui-logs-&lt;b&gt;marker&lt;/b&gt;" in filtered.text
+    assert "ui-logs-marker" not in filtered.text
+
+    for i in range(260):
+        logger.error(f"log-page-marker-{i}")
+
+    newest = admin_client.get("/settings/logs?level=error")
+    assert "log-page-marker-259" in newest.text
+    assert "log-page-marker-0" not in newest.text
+    assert "Load more (older logs)" in newest.text
+
+    older = admin_client.get("/settings/logs?level=error&offset=250")
+    assert "log-page-marker-0" in older.text
+    assert "log-page-marker-259" not in older.text
+    assert "Show newer logs" in older.text
+    assert "Top of the 10,000 log history reached" in older.text
+    assert older.text.index('id="log-lines"') < older.text.index(
+        "Top of the 10,000 log history reached"
+    )
+    assert "Load more (older logs)" not in older.text
+
+
 # -- static ----------------------------------------------------------------