diff --git a/astrbot/dashboard/api/static_files.py b/astrbot/dashboard/api/static_files.py index 2a7e5d84d..0329cfeab 100644 --- a/astrbot/dashboard/api/static_files.py +++ b/astrbot/dashboard/api/static_files.py @@ -1,10 +1,20 @@ from __future__ import annotations +import asyncio +import mimetypes +from pathlib import Path + from fastapi import APIRouter, HTTPException, Request -from fastapi.responses import FileResponse, HTMLResponse +from fastapi.responses import FileResponse, HTMLResponse, Response from astrbot.core.desktop_runtime import is_desktop_managed_backend from astrbot.dashboard.services.static_file_service import StaticFileService +from astrbot.dashboard.static_encoding import ( + accepts_gzip, + gzip_static_body, + is_compressible_media_type, + with_gzip_headers, +) router = APIRouter(include_in_schema=False) service = StaticFileService() @@ -14,6 +24,42 @@ def _static_folder(request: Request) -> str | None: return getattr(request.app.state, "dashboard_static_folder", None) +def _media_type(file_path: Path) -> str | None: + guessed, _encoding = mimetypes.guess_type(file_path.name) + return guessed + + +def _read_gzip_body(file_path: Path) -> bytes | None: + return gzip_static_body(file_path.read_bytes()) + + +async def _file_response(request: Request, file_path: Path, headers: dict[str, str]): + """Return a static file, gzip-encoded when the client and type allow it. + + Range requests stay on FileResponse. A gzip body cannot satisfy a byte range + of the original file. Reading and compression run off the event loop. + """ + media_type = _media_type(file_path) + accept_encoding = request.headers.get("accept-encoding") + if request.headers.get("range") or not accepts_gzip(accept_encoding): + return FileResponse(file_path, headers=headers) + if not is_compressible_media_type(media_type): + return FileResponse(file_path, headers=headers) + + compressed = await asyncio.to_thread(_read_gzip_body, file_path) + if compressed is None: + return FileResponse(file_path, headers=headers) + + content_type = media_type or "application/octet-stream" + if content_type.startswith("text/") and "charset=" not in content_type: + content_type += "; charset=utf-8" + return Response( + content=compressed, + media_type=content_type, + headers=with_gzip_headers(headers), + ) + + def _not_found_response() -> HTMLResponse: return HTMLResponse( service.get_not_found_message(), @@ -32,7 +78,7 @@ async def serve_index(request: Request): # resources change. That makes this request bypass an old URL cache # entry and evicts legacy subresources without clearing cookies or storage. headers["Clear-Site-Data"] = '"cache"' - return FileResponse(index_file, headers=headers) + return await _file_response(request, index_file, headers) async def serve_static_file(request: Request, static_path: str): @@ -58,7 +104,7 @@ async def serve_static_file(request: Request, static_path: str): and request.query_params.get("astrbot_bundle") ): headers["Clear-Site-Data"] = '"cache"' - return FileResponse(file_path, headers=headers) + return await _file_response(request, file_path, headers) for index_route in service.list_index_routes(): diff --git a/astrbot/dashboard/static_encoding.py b/astrbot/dashboard/static_encoding.py new file mode 100644 index 000000000..cccf4a970 --- /dev/null +++ b/astrbot/dashboard/static_encoding.py @@ -0,0 +1,76 @@ +"""Gzip encoding for dashboard static files. + +Only JS, CSS, and HTML are compressed, and only when the client accepts gzip. +Images, fonts, and event streams are not handled here. Cache-Control is left +to the caller. +""" + +from __future__ import annotations + +import gzip +from collections.abc import Mapping + +# text/javascript is what the stdlib reports for .js on current CPython. +# application/javascript is kept for older mimetypes maps. +COMPRESSIBLE_MEDIA_TYPES = frozenset( + { + "text/html", + "text/css", + "text/javascript", + "application/javascript", + "application/x-javascript", + } +) + +# Gzip of a few hundred bytes is often larger than the original. +MIN_GZIP_BYTES = 1024 + + +def accepts_gzip(header: str | None) -> bool: + """Return whether Accept-Encoding explicitly allows gzip with q > 0.""" + if not header: + return False + for part in header.split(","): + bits = [item.strip() for item in part.split(";") if item.strip()] + if not bits or bits[0].lower() != "gzip": + continue + quality = 1.0 + for param in bits[1:]: + if not param.lower().startswith("q="): + continue + try: + quality = float(param[2:]) + except ValueError: + quality = 0.0 + return quality > 0 + return False + + +def is_compressible_media_type(media_type: str | None) -> bool: + """Return whether this Content-Type is JS, CSS, or HTML.""" + if not media_type: + return False + base = media_type.split(";", 1)[0].strip().lower() + return base in COMPRESSIBLE_MEDIA_TYPES + + +def gzip_static_body(data: bytes) -> bytes | None: + """Return a smaller gzip body, or None when compression should be skipped.""" + if len(data) < MIN_GZIP_BYTES: + return None + compressed = gzip.compress(data, compresslevel=6) + if len(compressed) >= len(data): + return None + return compressed + + +def with_gzip_headers(headers: Mapping[str, str]) -> dict[str, str]: + """Add gzip headers without changing Cache-Control.""" + merged = dict(headers) + merged["Content-Encoding"] = "gzip" + vary = merged.get("Vary", "") + parts = [part.strip() for part in vary.split(",") if part.strip()] + if not any(part.lower() == "accept-encoding" for part in parts): + parts.append("Accept-Encoding") + merged["Vary"] = ", ".join(parts) + return merged diff --git a/tests/unit/test_dashboard_static_encoding.py b/tests/unit/test_dashboard_static_encoding.py new file mode 100644 index 000000000..5b443acc6 --- /dev/null +++ b/tests/unit/test_dashboard_static_encoding.py @@ -0,0 +1,51 @@ +"""Tests for dashboard static gzip encoding.""" + +import gzip + +from astrbot.dashboard.static_encoding import ( + accepts_gzip, + gzip_static_body, + is_compressible_media_type, + with_gzip_headers, +) + + +def test_accepts_gzip_only_when_explicitly_allowed(): + assert accepts_gzip("gzip") + assert accepts_gzip("br, gzip, deflate") + assert accepts_gzip("gzip;q=0.8") + assert not accepts_gzip(None) + assert not accepts_gzip("") + assert not accepts_gzip("br") + assert not accepts_gzip("gzip;q=0") + assert not accepts_gzip("identity") + + +def test_only_js_css_and_html_are_compressible(): + assert is_compressible_media_type("text/javascript") + assert is_compressible_media_type("text/javascript; charset=utf-8") + assert is_compressible_media_type("application/javascript") + assert is_compressible_media_type("text/css") + assert is_compressible_media_type("text/html") + assert not is_compressible_media_type("font/woff2") + assert not is_compressible_media_type("image/png") + assert not is_compressible_media_type("image/svg+xml") + assert not is_compressible_media_type("text/event-stream") + assert not is_compressible_media_type("application/json") + assert not is_compressible_media_type(None) + + +def test_gzip_static_body_shrinks_text_and_skips_tiny_payloads(): + raw = b"function boot(){return 'dashboard';}" * 80 + compressed = gzip_static_body(raw) + assert compressed is not None + assert len(compressed) < len(raw) + assert gzip.decompress(compressed) == raw + assert gzip_static_body(b"tiny") is None + + +def test_gzip_headers_keep_cache_control(): + headers = with_gzip_headers({"Cache-Control": "no-cache", "Vary": "Origin"}) + assert headers["Cache-Control"] == "no-cache" + assert headers["Content-Encoding"] == "gzip" + assert headers["Vary"] == "Origin, Accept-Encoding"