Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f1e16b7abe | ||
|
|
0ebff0ec17 | ||
|
|
c8ab06d864 | ||
|
|
d5ff7757c8 | ||
|
|
cd604eb10a | ||
|
|
abcf5d9940 |
+9
-6
@@ -19,8 +19,9 @@ from stream_zip import ZIP_AUTO, stream_zip
|
|||||||
from zstandard import ZstdCompressor
|
from zstandard import ZstdCompressor
|
||||||
|
|
||||||
from cista import auth, config, preview, session, sso, watching
|
from cista import auth, config, preview, session, sso, watching
|
||||||
|
from cista.preview import shutdown_preview_workers, start_preview_workers
|
||||||
from cista.api import bp
|
from cista.api import bp
|
||||||
from cista.sanic_logging import configure_access_logging, format_access_log
|
from cista.sanic_logging import configure_access_logging, configure_main_logging, format_access_log
|
||||||
from cista.sanic_logging import logger as access_logger
|
from cista.sanic_logging import logger as access_logger
|
||||||
from cista.util.apphelpers import handle_sanic_exception
|
from cista.util.apphelpers import handle_sanic_exception
|
||||||
|
|
||||||
@@ -30,6 +31,7 @@ sanic.helpers._ENTITY_HEADERS = frozenset()
|
|||||||
configure_access_logging()
|
configure_access_logging()
|
||||||
|
|
||||||
app = Sanic("cista", strict_slashes=True)
|
app = Sanic("cista", strict_slashes=True)
|
||||||
|
configure_main_logging()
|
||||||
# Register either SSO proxy or built-in auth routes based on PASKIA_BACKEND_URL
|
# Register either SSO proxy or built-in auth routes based on PASKIA_BACKEND_URL
|
||||||
if sso.paskia_enabled():
|
if sso.paskia_enabled():
|
||||||
app.blueprint(sso.bp) # SSO proxy for /auth/* routes
|
app.blueprint(sso.bp) # SSO proxy for /auth/* routes
|
||||||
@@ -47,13 +49,12 @@ setproctitle("cista-main")
|
|||||||
async def main_start(app):
|
async def main_start(app):
|
||||||
config.load_config()
|
config.load_config()
|
||||||
setproctitle(f"cista {config.config.path.name}")
|
setproctitle(f"cista {config.config.path.name}")
|
||||||
# Small pool for memory-intensive preview generation
|
|
||||||
preview_workers = max(2, min(8, cpu_count()))
|
|
||||||
app.ctx.threadexec = ThreadPoolExecutor(
|
app.ctx.threadexec = ThreadPoolExecutor(
|
||||||
max_workers=preview_workers, thread_name_prefix="cista-preview"
|
max_workers=4, thread_name_prefix="cista-worker"
|
||||||
)
|
)
|
||||||
# Larger pool for long-running but low-memory zip operations
|
# Larger pool for long-running but low-memory zip operations
|
||||||
app.ctx.zipexec = ThreadPoolExecutor(max_workers=32, thread_name_prefix="cista-zip")
|
app.ctx.zipexec = ThreadPoolExecutor(max_workers=32, thread_name_prefix="cista-zip")
|
||||||
|
await start_preview_workers()
|
||||||
watching.start(app)
|
watching.start(app)
|
||||||
|
|
||||||
|
|
||||||
@@ -61,6 +62,7 @@ async def main_start(app):
|
|||||||
@app.before_server_stop
|
@app.before_server_stop
|
||||||
async def main_stop(app):
|
async def main_stop(app):
|
||||||
watching.stop(app)
|
watching.stop(app)
|
||||||
|
await shutdown_preview_workers()
|
||||||
app.ctx.threadexec.shutdown()
|
app.ctx.threadexec.shutdown()
|
||||||
app.ctx.zipexec.shutdown(cancel_futures=True)
|
app.ctx.zipexec.shutdown(cancel_futures=True)
|
||||||
await sso.close_client()
|
await sso.close_client()
|
||||||
@@ -94,7 +96,7 @@ async def log_access(req, res):
|
|||||||
return res
|
return res
|
||||||
start = getattr(req.ctx, "_log_start", None)
|
start = getattr(req.ctx, "_log_start", None)
|
||||||
duration_ms = (time.perf_counter() - start) * 1000 if start is not None else 0.0
|
duration_ms = (time.perf_counter() - start) * 1000 if start is not None else 0.0
|
||||||
client = req.ip or "-"
|
client = req.client_ip or "-"
|
||||||
host = req.host or "-"
|
host = req.host or "-"
|
||||||
path = req.path
|
path = req.path
|
||||||
if req.query_string:
|
if req.query_string:
|
||||||
@@ -102,7 +104,8 @@ async def log_access(req, res):
|
|||||||
if isinstance(qs, bytes):
|
if isinstance(qs, bytes):
|
||||||
qs = qs.decode(errors="replace")
|
qs = qs.decode(errors="replace")
|
||||||
path = f"{path}?{qs}"
|
path = f"{path}?{qs}"
|
||||||
line = format_access_log(client, res.status, req.method, host, path, duration_ms)
|
extra = getattr(req.ctx, "_log_extra", None)
|
||||||
|
line = format_access_log(client, res.status, req.method, host, path, duration_ms, extra=extra)
|
||||||
access_logger.info(line)
|
access_logger.info(line)
|
||||||
return res
|
return res
|
||||||
|
|
||||||
|
|||||||
+334
-56
@@ -2,25 +2,34 @@ import asyncio
|
|||||||
import gc
|
import gc
|
||||||
import io
|
import io
|
||||||
import mimetypes
|
import mimetypes
|
||||||
|
import os
|
||||||
|
import struct
|
||||||
|
import sys
|
||||||
import threading
|
import threading
|
||||||
import urllib.parse
|
import urllib.parse
|
||||||
from collections import OrderedDict
|
from collections import OrderedDict
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
|
from multiprocessing import cpu_count
|
||||||
from pathlib import PurePosixPath
|
from pathlib import PurePosixPath
|
||||||
from time import perf_counter
|
from time import perf_counter
|
||||||
from urllib.parse import unquote
|
from urllib.parse import unquote
|
||||||
from wsgiref.handlers import format_date_time
|
from wsgiref.handlers import format_date_time
|
||||||
|
|
||||||
|
import msgspec
|
||||||
|
|
||||||
import av
|
import av
|
||||||
import fitz # PyMuPDF
|
import fitz # PyMuPDF
|
||||||
import numpy as np
|
import numpy as np
|
||||||
import pillow_heif
|
import pillow_heif
|
||||||
|
import pyvips
|
||||||
|
from blake3 import blake3
|
||||||
from PIL import Image
|
from PIL import Image
|
||||||
from sanic import Blueprint, empty, raw, redirect
|
from sanic import Blueprint, empty, raw, redirect
|
||||||
from sanic.exceptions import NotFound
|
from sanic.exceptions import NotFound
|
||||||
from sanic.log import logger
|
from sanic.log import logger
|
||||||
|
|
||||||
from cista import auth, config
|
from cista import auth, config
|
||||||
|
from cista.preview_worker import PreviewRequest, PreviewResponse
|
||||||
from cista.util.filename import sanitize
|
from cista.util.filename import sanitize
|
||||||
|
|
||||||
pillow_heif.register_heif_opener()
|
pillow_heif.register_heif_opener()
|
||||||
@@ -70,6 +79,208 @@ class PreviewCache:
|
|||||||
# Global preview cache instance
|
# Global preview cache instance
|
||||||
_preview_cache = PreviewCache(capacity=500)
|
_preview_cache = PreviewCache(capacity=500)
|
||||||
|
|
||||||
|
PREVIEW_TIMEOUT = 3.0 # seconds until preview subprocess is killed
|
||||||
|
PREVIEW_WORKERS = max(2, min(8, cpu_count()))
|
||||||
|
_active_procs: set[asyncio.subprocess.Process] = set()
|
||||||
|
_preview_pool = None
|
||||||
|
_preview_pool_lock = asyncio.Lock()
|
||||||
|
AVIF_FAST_EFFORT = 0
|
||||||
|
FORCE_PIL = os.environ.get("CISTA_PIL") == "1"
|
||||||
|
WORKER_CHECKSUM_BYTES = 32
|
||||||
|
WORKER_MAX_JSON_BYTES = 1_000_000
|
||||||
|
|
||||||
|
|
||||||
|
class WorkerChecksumError(Exception):
|
||||||
|
"""Raised when worker response checksum does not match the packet."""
|
||||||
|
|
||||||
|
|
||||||
|
class WorkerProtocolError(Exception):
|
||||||
|
"""Raised when worker response packet is malformed."""
|
||||||
|
|
||||||
|
|
||||||
|
class _PreviewWorker:
|
||||||
|
def __init__(self, proc: asyncio.subprocess.Process):
|
||||||
|
self.proc = proc
|
||||||
|
|
||||||
|
async def request(self, filepath, quality: int, maxsize: int, maxzoom: float):
|
||||||
|
if self.proc.returncode is not None:
|
||||||
|
raise WorkerProtocolError("worker already exited")
|
||||||
|
if self.proc.stdin is None or self.proc.stdout is None:
|
||||||
|
raise WorkerProtocolError("worker streams not available")
|
||||||
|
|
||||||
|
line = (
|
||||||
|
msgspec.json.encode(
|
||||||
|
PreviewRequest(
|
||||||
|
path=str(filepath),
|
||||||
|
quality=quality,
|
||||||
|
maxsize=maxsize,
|
||||||
|
maxzoom=maxzoom,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
+ b"\n"
|
||||||
|
)
|
||||||
|
self.proc.stdin.write(line)
|
||||||
|
await self.proc.stdin.drain()
|
||||||
|
|
||||||
|
checksum = await self.proc.stdout.readexactly(WORKER_CHECKSUM_BYTES)
|
||||||
|
header = await self.proc.stdout.readexactly(8)
|
||||||
|
json_size, data_size = struct.unpack("<II", header)
|
||||||
|
if json_size > WORKER_MAX_JSON_BYTES:
|
||||||
|
raise WorkerProtocolError(f"worker JSON too large: {json_size}")
|
||||||
|
meta_raw = await self.proc.stdout.readexactly(json_size)
|
||||||
|
payload = await self.proc.stdout.readexactly(data_size)
|
||||||
|
packet = header + meta_raw + payload
|
||||||
|
if blake3(packet).digest() != checksum:
|
||||||
|
raise WorkerChecksumError("worker checksum mismatch")
|
||||||
|
|
||||||
|
resp = msgspec.json.decode(meta_raw, type=PreviewResponse)
|
||||||
|
if not resp.ok:
|
||||||
|
raise PreviewError(resp.error or "preview worker error")
|
||||||
|
return payload or None, resp
|
||||||
|
|
||||||
|
async def kill(self) -> None:
|
||||||
|
if self.proc.returncode is None:
|
||||||
|
try:
|
||||||
|
self.proc.kill()
|
||||||
|
except ProcessLookupError:
|
||||||
|
pass
|
||||||
|
await self.proc.wait()
|
||||||
|
_active_procs.discard(self.proc)
|
||||||
|
|
||||||
|
|
||||||
|
class _PreviewWorkerPool:
|
||||||
|
def __init__(self, size: int):
|
||||||
|
self.size = size
|
||||||
|
self._idle: asyncio.Queue[_PreviewWorker] = asyncio.Queue()
|
||||||
|
self._workers: set[_PreviewWorker] = set()
|
||||||
|
self._closed = False
|
||||||
|
|
||||||
|
async def _spawn_worker(self) -> _PreviewWorker:
|
||||||
|
proc = await asyncio.create_subprocess_exec(
|
||||||
|
sys.executable,
|
||||||
|
"-m",
|
||||||
|
"cista.preview_worker",
|
||||||
|
stdin=asyncio.subprocess.PIPE,
|
||||||
|
stdout=asyncio.subprocess.PIPE,
|
||||||
|
stderr=asyncio.subprocess.DEVNULL,
|
||||||
|
start_new_session=True,
|
||||||
|
)
|
||||||
|
_active_procs.add(proc)
|
||||||
|
return _PreviewWorker(proc)
|
||||||
|
|
||||||
|
async def _add_worker(self) -> None:
|
||||||
|
worker = await self._spawn_worker()
|
||||||
|
self._workers.add(worker)
|
||||||
|
await self._idle.put(worker)
|
||||||
|
|
||||||
|
async def _replace_worker(self, worker: _PreviewWorker) -> None:
|
||||||
|
self._workers.discard(worker)
|
||||||
|
await worker.kill()
|
||||||
|
if self._closed:
|
||||||
|
return
|
||||||
|
try:
|
||||||
|
await self._add_worker()
|
||||||
|
except Exception:
|
||||||
|
logger.exception("Failed to replace preview worker")
|
||||||
|
|
||||||
|
async def start(self) -> None:
|
||||||
|
for _ in range(self.size):
|
||||||
|
await self._add_worker()
|
||||||
|
|
||||||
|
async def run(self, filepath, quality: int, maxsize: int, maxzoom: float):
|
||||||
|
if self._closed:
|
||||||
|
raise PreviewError("preview worker pool closed")
|
||||||
|
worker = await self._idle.get()
|
||||||
|
replace = False
|
||||||
|
try:
|
||||||
|
out, resp = await asyncio.wait_for(
|
||||||
|
worker.request(filepath, quality, maxsize, maxzoom),
|
||||||
|
timeout=PREVIEW_TIMEOUT,
|
||||||
|
)
|
||||||
|
return out, resp
|
||||||
|
except asyncio.TimeoutError:
|
||||||
|
replace = True
|
||||||
|
logger.warning(
|
||||||
|
"Preview timeout (%ds) for %s", int(PREVIEW_TIMEOUT), filepath.name
|
||||||
|
)
|
||||||
|
raise PreviewTimeout(filepath.name)
|
||||||
|
except WorkerChecksumError:
|
||||||
|
replace = True
|
||||||
|
logger.error("Preview checksum mismatch for %s", filepath.name)
|
||||||
|
raise PreviewError(filepath.name)
|
||||||
|
except PreviewError:
|
||||||
|
raise
|
||||||
|
except (
|
||||||
|
WorkerProtocolError,
|
||||||
|
asyncio.IncompleteReadError,
|
||||||
|
BrokenPipeError,
|
||||||
|
ConnectionResetError,
|
||||||
|
OSError,
|
||||||
|
ValueError,
|
||||||
|
msgspec.json.DecodeError,
|
||||||
|
) as e:
|
||||||
|
replace = True
|
||||||
|
logger.warning(
|
||||||
|
"Preview worker protocol failure for %s: %s", filepath.name, e
|
||||||
|
)
|
||||||
|
raise PreviewError(filepath.name)
|
||||||
|
finally:
|
||||||
|
if replace:
|
||||||
|
await self._replace_worker(worker)
|
||||||
|
else:
|
||||||
|
if worker.proc.returncode is None:
|
||||||
|
await self._idle.put(worker)
|
||||||
|
else:
|
||||||
|
await self._replace_worker(worker)
|
||||||
|
|
||||||
|
async def close(self) -> None:
|
||||||
|
self._closed = True
|
||||||
|
workers = list(self._workers)
|
||||||
|
self._workers.clear()
|
||||||
|
while not self._idle.empty():
|
||||||
|
try:
|
||||||
|
self._idle.get_nowait()
|
||||||
|
except asyncio.QueueEmpty:
|
||||||
|
break
|
||||||
|
await asyncio.gather(
|
||||||
|
*(worker.kill() for worker in workers), return_exceptions=True
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def start_preview_workers() -> None:
|
||||||
|
"""Warm up persistent preview workers during server startup."""
|
||||||
|
global _preview_pool
|
||||||
|
if _preview_pool is not None:
|
||||||
|
return
|
||||||
|
async with _preview_pool_lock:
|
||||||
|
if _preview_pool is not None:
|
||||||
|
return
|
||||||
|
pool = _PreviewWorkerPool(PREVIEW_WORKERS)
|
||||||
|
await pool.start()
|
||||||
|
_preview_pool = pool
|
||||||
|
logger.info("Started %d persistent preview workers", PREVIEW_WORKERS)
|
||||||
|
|
||||||
|
|
||||||
|
async def shutdown_preview_workers() -> None:
|
||||||
|
"""Kill persistent preview workers (called during server shutdown)."""
|
||||||
|
global _preview_pool
|
||||||
|
async with _preview_pool_lock:
|
||||||
|
pool = _preview_pool
|
||||||
|
_preview_pool = None
|
||||||
|
if pool is not None:
|
||||||
|
await pool.close()
|
||||||
|
if not _active_procs:
|
||||||
|
return
|
||||||
|
for proc in list(_active_procs):
|
||||||
|
try:
|
||||||
|
proc.kill()
|
||||||
|
except ProcessLookupError:
|
||||||
|
pass
|
||||||
|
await asyncio.gather(
|
||||||
|
*(proc.wait() for proc in list(_active_procs)), return_exceptions=True
|
||||||
|
)
|
||||||
|
_active_procs.clear()
|
||||||
|
|
||||||
|
|
||||||
@bp.on_request
|
@bp.on_request
|
||||||
async def verify_preview(request):
|
async def verify_preview(request):
|
||||||
@@ -77,6 +288,24 @@ async def verify_preview(request):
|
|||||||
await auth.verify(request)
|
await auth.verify(request)
|
||||||
|
|
||||||
|
|
||||||
|
class PreviewTimeout(Exception):
|
||||||
|
"""Raised when the preview subprocess exceeds PREVIEW_TIMEOUT."""
|
||||||
|
|
||||||
|
|
||||||
|
class PreviewError(Exception):
|
||||||
|
"""Raised when the preview subprocess exits with a non-zero status."""
|
||||||
|
|
||||||
|
|
||||||
|
async def _run_preview_process(
|
||||||
|
filepath, quality: int, maxsize: int, maxzoom: float
|
||||||
|
) -> tuple[bytes | None, PreviewResponse | None]:
|
||||||
|
"""Run preview request in a persistent worker process."""
|
||||||
|
await start_preview_workers()
|
||||||
|
if _preview_pool is None:
|
||||||
|
raise PreviewError(filepath.name)
|
||||||
|
return await _preview_pool.run(filepath, quality, maxsize, maxzoom)
|
||||||
|
|
||||||
|
|
||||||
# Map EXIF Orientation value to a corresponding PIL transpose
|
# Map EXIF Orientation value to a corresponding PIL transpose
|
||||||
EXIF_ORI = {
|
EXIF_ORI = {
|
||||||
2: Image.Transpose.FLIP_LEFT_RIGHT,
|
2: Image.Transpose.FLIP_LEFT_RIGHT,
|
||||||
@@ -89,6 +318,19 @@ EXIF_ORI = {
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
DOC_PREVIEW_SUFFIXES = {".pdf", ".xps", ".epub", ".mobi"}
|
||||||
|
|
||||||
|
|
||||||
|
def is_previewable_path(path) -> bool:
|
||||||
|
suffix = path.suffix.lower()
|
||||||
|
if suffix in DOC_PREVIEW_SUFFIXES:
|
||||||
|
return True
|
||||||
|
mime_type, _ = mimetypes.guess_type(path.name)
|
||||||
|
if not mime_type:
|
||||||
|
return False
|
||||||
|
return mime_type.startswith("image/") or mime_type.startswith("video/")
|
||||||
|
|
||||||
|
|
||||||
@bp.get("/<path:path>")
|
@bp.get("/<path:path>")
|
||||||
async def preview(req, path):
|
async def preview(req, path):
|
||||||
"""Preview a file"""
|
"""Preview a file"""
|
||||||
@@ -102,6 +344,9 @@ async def preview(req, path):
|
|||||||
except FileNotFoundError:
|
except FileNotFoundError:
|
||||||
raise NotFound() from None
|
raise NotFound() from None
|
||||||
|
|
||||||
|
if not is_previewable_path(filepath):
|
||||||
|
return empty(415)
|
||||||
|
|
||||||
etag = config.derived_secret(
|
etag = config.derived_secret(
|
||||||
"preview", rel, stat.st_mtime_ns, quality, maxsize, maxzoom
|
"preview", rel, stat.st_mtime_ns, quality, maxsize, maxzoom
|
||||||
).hex()
|
).hex()
|
||||||
@@ -117,21 +362,39 @@ async def preview(req, path):
|
|||||||
return raw(cached.body, headers=cached.headers)
|
return raw(cached.body, headers=cached.headers)
|
||||||
|
|
||||||
# Generate preview
|
# Generate preview
|
||||||
img = await asyncio.get_event_loop().run_in_executor(
|
try:
|
||||||
req.app.ctx.threadexec, dispatch, filepath, quality, maxsize, maxzoom
|
img, preview_resp = await _run_preview_process(
|
||||||
)
|
filepath, quality, maxsize, maxzoom
|
||||||
|
)
|
||||||
|
except PreviewTimeout:
|
||||||
|
return empty(504)
|
||||||
|
except PreviewError:
|
||||||
|
return empty(422)
|
||||||
|
if preview_resp and preview_resp.backend:
|
||||||
|
if preview_resp.timings:
|
||||||
|
timing_detail = "/".join(
|
||||||
|
str(int(round(value))) for value in preview_resp.timings
|
||||||
|
)
|
||||||
|
req.ctx._log_extra = f"{preview_resp.backend} {timing_detail} ➛"
|
||||||
|
else:
|
||||||
|
req.ctx._log_extra = preview_resp.backend
|
||||||
if not img:
|
if not img:
|
||||||
# Preview generation failed, redirect to the file itself
|
# Preview generation failed, redirect to the file itself
|
||||||
return redirect(f"/files/{path}", status=303)
|
return redirect(f"/files/{path}", status=303)
|
||||||
|
|
||||||
# Build headers and cache the full response
|
# Build headers and cache the full response
|
||||||
|
preview_mime = (
|
||||||
|
preview_resp.mime
|
||||||
|
if preview_resp is not None and preview_resp.mime is not None
|
||||||
|
else "image/avif"
|
||||||
|
)
|
||||||
savename = PurePosixPath(filepath.name).with_suffix(".avif")
|
savename = PurePosixPath(filepath.name).with_suffix(".avif")
|
||||||
headers = {
|
headers = {
|
||||||
"etag": etag,
|
"etag": etag,
|
||||||
"last-modified": format_date_time(stat.st_mtime),
|
"last-modified": format_date_time(stat.st_mtime),
|
||||||
"cache-control": "max-age=604800, immutable"
|
"cache-control": "max-age=604800, immutable"
|
||||||
+ ("" if config.config.public else ", private"),
|
+ ("" if config.config.public else ", private"),
|
||||||
"content-type": "image/avif",
|
"content-type": preview_mime,
|
||||||
"content-disposition": f"inline; filename*=UTF-8''{urllib.parse.quote(savename.as_posix())}",
|
"content-disposition": f"inline; filename*=UTF-8''{urllib.parse.quote(savename.as_posix())}",
|
||||||
}
|
}
|
||||||
_preview_cache.set(etag, CachedPreview(headers=headers, body=img))
|
_preview_cache.set(etag, CachedPreview(headers=headers, body=img))
|
||||||
@@ -141,55 +404,52 @@ async def preview(req, path):
|
|||||||
|
|
||||||
def dispatch(path, quality, maxsize, maxzoom):
|
def dispatch(path, quality, maxsize, maxzoom):
|
||||||
try:
|
try:
|
||||||
if path.suffix.lower() in (".pdf", ".xps", ".epub", ".mobi"):
|
if path.suffix.lower() in DOC_PREVIEW_SUFFIXES:
|
||||||
return process_pdf(path, quality=quality, maxsize=maxsize, maxzoom=maxzoom)
|
return process_pdf(path, quality=quality, maxsize=maxsize, maxzoom=maxzoom)
|
||||||
type, _ = mimetypes.guess_type(path.name)
|
mime_type, _ = mimetypes.guess_type(path.name)
|
||||||
if type and type.startswith("video/"):
|
if mime_type and mime_type.startswith("video/"):
|
||||||
return process_video(path, quality=quality, maxsize=maxsize)
|
return process_video(path, quality=quality, maxsize=maxsize)
|
||||||
return process_image(path, quality=quality, maxsize=maxsize)
|
if mime_type and mime_type.startswith("image/"):
|
||||||
|
return process_image(path, quality=quality, maxsize=maxsize)
|
||||||
except ValueError as e:
|
except ValueError as e:
|
||||||
logger.warning(f"Cannot generate preview for {path}: {e}")
|
logger.warning(f"Cannot generate preview for {path}: {e}")
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.exception(f"Error generating preview for {path}: {e}")
|
logger.exception(f"Error generating preview for {path}: {e}")
|
||||||
|
return None, PreviewResponse(ok=False)
|
||||||
|
|
||||||
|
|
||||||
def process_image(path, *, maxsize, quality):
|
def process_image(path, *, maxsize, quality):
|
||||||
try:
|
return process_image_with_timing(path, maxsize=maxsize, quality=quality)
|
||||||
return process_image_pyvips(path, maxsize=maxsize, quality=quality)
|
|
||||||
except Exception as e:
|
|
||||||
logger.debug("Falling back to Pillow preview for %s: %s", path.name, e)
|
def process_image_with_timing(path, *, maxsize, quality):
|
||||||
|
if FORCE_PIL:
|
||||||
return process_image_pillow(path, maxsize=maxsize, quality=quality)
|
return process_image_pillow(path, maxsize=maxsize, quality=quality)
|
||||||
|
return process_image_pyvips(path, maxsize=maxsize, quality=quality)
|
||||||
|
|
||||||
|
|
||||||
def process_image_pyvips(path, *, maxsize, quality):
|
def process_image_pyvips(path, *, maxsize, quality):
|
||||||
import pyvips
|
t_start = perf_counter()
|
||||||
|
|
||||||
t_load = perf_counter()
|
|
||||||
img = pyvips.Image.new_from_file(str(path), access="sequential")
|
img = pyvips.Image.new_from_file(str(path), access="sequential")
|
||||||
t_proc = perf_counter()
|
|
||||||
|
|
||||||
img = img.autorot()
|
img = img.autorot()
|
||||||
scale = min(maxsize / img.width, maxsize / img.height, 1.0)
|
scale = min(maxsize / img.width, maxsize / img.height, 1.0)
|
||||||
if scale < 1.0:
|
if scale < 1.0:
|
||||||
img = img.resize(scale)
|
img = img.resize(scale)
|
||||||
|
ret = img.write_to_buffer(
|
||||||
t_save = perf_counter()
|
".avif",
|
||||||
ret = img.write_to_buffer(".avif", Q=quality)
|
Q=quality,
|
||||||
|
effort=AVIF_FAST_EFFORT,
|
||||||
|
strip=True,
|
||||||
|
)
|
||||||
t_end = perf_counter()
|
t_end = perf_counter()
|
||||||
|
|
||||||
load_ms = (t_proc - t_load) * 1000
|
return ret, PreviewResponse(
|
||||||
proc_ms = (t_save - t_proc) * 1000
|
ok=True,
|
||||||
save_ms = (t_end - t_save) * 1000
|
mime="image/avif",
|
||||||
logger.debug(
|
backend="pyvips",
|
||||||
"Preview image %s via pyvips: load=%.1fms process=%.1fms save=%.1fms",
|
timings=[round((t_end - t_start) * 1000, 1)],
|
||||||
path.name,
|
|
||||||
load_ms,
|
|
||||||
proc_ms,
|
|
||||||
save_ms,
|
|
||||||
)
|
)
|
||||||
|
|
||||||
return ret
|
|
||||||
|
|
||||||
|
|
||||||
def process_image_pillow(path, *, maxsize, quality):
|
def process_image_pillow(path, *, maxsize, quality):
|
||||||
t_load = perf_counter()
|
t_load = perf_counter()
|
||||||
@@ -222,16 +482,13 @@ def process_image_pillow(path, *, maxsize, quality):
|
|||||||
load_ms = (t_proc - t_load) * 1000
|
load_ms = (t_proc - t_load) * 1000
|
||||||
proc_ms = (t_save - t_proc) * 1000
|
proc_ms = (t_save - t_proc) * 1000
|
||||||
save_ms = (t_end - t_save) * 1000
|
save_ms = (t_end - t_save) * 1000
|
||||||
logger.debug(
|
return ret, PreviewResponse(
|
||||||
"Preview image %s via Pillow: load=%.1fms process=%.1fms save=%.1fms",
|
ok=True,
|
||||||
path.name,
|
mime="image/avif",
|
||||||
load_ms,
|
backend="pillow",
|
||||||
proc_ms,
|
timings=[round(load_ms, 1), round(proc_ms, 1), round(save_ms, 1)],
|
||||||
save_ms,
|
|
||||||
)
|
)
|
||||||
|
|
||||||
return ret
|
|
||||||
|
|
||||||
|
|
||||||
def process_pdf(path, *, maxsize, maxzoom, quality, page_number=0):
|
def process_pdf(path, *, maxsize, maxzoom, quality, page_number=0):
|
||||||
t_load_start = perf_counter()
|
t_load_start = perf_counter()
|
||||||
@@ -244,18 +501,30 @@ def process_pdf(path, *, maxsize, maxzoom, quality, page_number=0):
|
|||||||
t_load_end = perf_counter()
|
t_load_end = perf_counter()
|
||||||
|
|
||||||
t_save_start = perf_counter()
|
t_save_start = perf_counter()
|
||||||
ret = pix.pil_tobytes(
|
if FORCE_PIL:
|
||||||
format="avif", quality=quality, speed=10, max_threads=1, avif=1
|
ret = pix.pil_tobytes(
|
||||||
)
|
format="avif", quality=quality, speed=10, max_threads=1, avif=1
|
||||||
|
)
|
||||||
|
backend = "pdf"
|
||||||
|
else:
|
||||||
|
img = pyvips.Image.new_from_memory(
|
||||||
|
pix.samples_mv, pix.width, pix.height, pix.n, "uchar"
|
||||||
|
)
|
||||||
|
ret = img.write_to_buffer(
|
||||||
|
".avif", Q=quality, effort=AVIF_FAST_EFFORT, strip=True
|
||||||
|
)
|
||||||
|
backend = "pdf+pyvips"
|
||||||
t_save_end = perf_counter()
|
t_save_end = perf_counter()
|
||||||
|
|
||||||
logger.debug(
|
return ret, PreviewResponse(
|
||||||
"Preview pdf %s: load+render=%.1fms save=%.1fms",
|
ok=True,
|
||||||
path.name,
|
mime="image/avif",
|
||||||
(t_load_end - t_load_start) * 1000,
|
backend=backend,
|
||||||
(t_save_end - t_save_start) * 1000,
|
timings=[
|
||||||
|
round((t_load_end - t_load_start) * 1000, 1),
|
||||||
|
round((t_save_end - t_save_start) * 1000, 1),
|
||||||
|
],
|
||||||
)
|
)
|
||||||
return ret
|
|
||||||
|
|
||||||
|
|
||||||
def process_video(path, *, maxsize, quality):
|
def process_video(path, *, maxsize, quality):
|
||||||
@@ -268,7 +537,13 @@ def process_video(path, *, maxsize, quality):
|
|||||||
t_save_start = t_load_start
|
t_save_start = t_load_start
|
||||||
t_save_end = t_load_start
|
t_save_end = t_load_start
|
||||||
with (
|
with (
|
||||||
av.open(str(path)) as icontainer,
|
av.open(
|
||||||
|
str(path),
|
||||||
|
options={
|
||||||
|
"analyzeduration": "1000000", # 1 second (in microseconds)
|
||||||
|
"fflags": "fastseek",
|
||||||
|
},
|
||||||
|
) as icontainer,
|
||||||
av.open(imgdata, "w", format="avif") as ocontainer,
|
av.open(imgdata, "w", format="avif") as ocontainer,
|
||||||
):
|
):
|
||||||
istream = icontainer.streams.video[0]
|
istream = icontainer.streams.video[0]
|
||||||
@@ -360,14 +635,17 @@ def process_video(path, *, maxsize, quality):
|
|||||||
ocontainer.mux(ostream.encode(None)) # Flush the stream
|
ocontainer.mux(ostream.encode(None)) # Flush the stream
|
||||||
t_save_end = perf_counter()
|
t_save_end = perf_counter()
|
||||||
|
|
||||||
# Capture frame dimensions before cleanup
|
# Capture result before cleanup
|
||||||
ret = imgdata.getvalue()
|
ret = imgdata.getvalue()
|
||||||
logger.debug(
|
resp = PreviewResponse(
|
||||||
"Preview video %s: load+decode=%.1fms save=%.1fms",
|
ok=True,
|
||||||
path.name,
|
mime="image/avif",
|
||||||
(t_load_end - t_load_start) * 1000,
|
backend="video",
|
||||||
(t_save_end - t_save_start) * 1000,
|
timings=[
|
||||||
|
round((t_load_end - t_load_start) * 1000, 1),
|
||||||
|
round((t_save_end - t_save_start) * 1000, 1),
|
||||||
|
],
|
||||||
)
|
)
|
||||||
del imgdata, istream, ostream, icc, occ, frame
|
del imgdata, istream, ostream, icc, occ, frame
|
||||||
gc.collect()
|
gc.collect()
|
||||||
return ret
|
return ret, resp
|
||||||
|
|||||||
@@ -0,0 +1,93 @@
|
|||||||
|
"""Preview generation worker subprocess.
|
||||||
|
|
||||||
|
Two modes are supported:
|
||||||
|
1) Legacy one-shot mode: argv has path/quality/maxsize/maxzoom.
|
||||||
|
2) Long-lived mode: read JSONL commands from stdin and write framed responses.
|
||||||
|
|
||||||
|
Framed response format:
|
||||||
|
(blake3(packet))(uint32 json size)(uint32 payload size)(json)(binary payload)
|
||||||
|
where packet = (uint32 json size)(uint32 payload size)(json)(binary payload).
|
||||||
|
"""
|
||||||
|
|
||||||
|
import logging
|
||||||
|
import struct
|
||||||
|
import sys
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import msgspec
|
||||||
|
from blake3 import blake3
|
||||||
|
|
||||||
|
|
||||||
|
class PreviewRequest(msgspec.Struct, omit_defaults=True):
|
||||||
|
path: str
|
||||||
|
quality: int
|
||||||
|
maxsize: int
|
||||||
|
maxzoom: float
|
||||||
|
|
||||||
|
|
||||||
|
class PreviewResponse(msgspec.Struct, omit_defaults=True):
|
||||||
|
ok: bool
|
||||||
|
mime: str | None = None
|
||||||
|
backend: str | None = None
|
||||||
|
timings: list[float] | None = None
|
||||||
|
error: str | None = None
|
||||||
|
|
||||||
|
|
||||||
|
_enc = msgspec.json.Encoder()
|
||||||
|
_dec_req = msgspec.json.Decoder(PreviewRequest)
|
||||||
|
|
||||||
|
|
||||||
|
def _write_response(resp: PreviewResponse, payload: bytes) -> None:
|
||||||
|
meta_bytes = _enc.encode(resp)
|
||||||
|
packet = struct.pack("<II", len(meta_bytes), len(payload)) + meta_bytes + payload
|
||||||
|
checksum = blake3(packet).digest()
|
||||||
|
sys.stdout.buffer.write(checksum)
|
||||||
|
sys.stdout.buffer.write(packet)
|
||||||
|
sys.stdout.buffer.flush()
|
||||||
|
|
||||||
|
|
||||||
|
def _run_once() -> None:
|
||||||
|
if len(sys.argv) != 5:
|
||||||
|
sys.stderr.write(f"Usage: {sys.argv[0]} <path> <quality> <maxsize> <maxzoom>\n")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
from cista.preview import dispatch
|
||||||
|
|
||||||
|
path = Path(sys.argv[1])
|
||||||
|
quality = int(sys.argv[2])
|
||||||
|
maxsize = int(sys.argv[3])
|
||||||
|
maxzoom = float(sys.argv[4])
|
||||||
|
result, _ = dispatch(path, quality, maxsize, maxzoom)
|
||||||
|
if result:
|
||||||
|
sys.stdout.buffer.write(result)
|
||||||
|
sys.stdout.buffer.flush()
|
||||||
|
|
||||||
|
|
||||||
|
def _run_loop() -> None:
|
||||||
|
from cista.preview import dispatch
|
||||||
|
|
||||||
|
while True:
|
||||||
|
line = sys.stdin.buffer.readline()
|
||||||
|
if not line:
|
||||||
|
return
|
||||||
|
try:
|
||||||
|
req = _dec_req.decode(line)
|
||||||
|
result, resp = dispatch(
|
||||||
|
Path(req.path), req.quality, req.maxsize, req.maxzoom
|
||||||
|
)
|
||||||
|
_write_response(resp, result or b"")
|
||||||
|
except Exception as e:
|
||||||
|
_write_response(PreviewResponse(ok=False, error=str(e)), b"")
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> None:
|
||||||
|
# Configure all log output to stderr before any imports that may emit logs.
|
||||||
|
logging.basicConfig(stream=sys.stderr, level=logging.INFO)
|
||||||
|
if len(sys.argv) > 1:
|
||||||
|
_run_once()
|
||||||
|
return
|
||||||
|
_run_loop()
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
+41
-5
@@ -55,8 +55,8 @@ def format_client_ip(ip: str) -> str:
|
|||||||
return "-"
|
return "-"
|
||||||
stripped = ip.strip("[]")
|
stripped = ip.strip("[]")
|
||||||
if ":" in stripped:
|
if ":" in stripped:
|
||||||
return format_ipv6_network(ip)
|
return format_ipv6_network(stripped)
|
||||||
return ip
|
return stripped
|
||||||
|
|
||||||
|
|
||||||
def status_color(status: int) -> str:
|
def status_color(status: int) -> str:
|
||||||
@@ -113,7 +113,8 @@ def _format_method_label(label: str, *, color: str | None = None) -> str:
|
|||||||
|
|
||||||
|
|
||||||
def format_access_log(
|
def format_access_log(
|
||||||
client: str, status: int, method: str, host: str, path: str, duration_ms: float
|
client: str, status: int, method: str, host: str, path: str, duration_ms: float,
|
||||||
|
extra: str | None = None,
|
||||||
) -> str:
|
) -> str:
|
||||||
ip = _format_left(format_client_ip(client))
|
ip = _format_left(format_client_ip(client))
|
||||||
status_str = f"{status_color(status)}{str(status).rjust(3)}{_RESET}"
|
status_str = f"{status_color(status)}{str(status).rjust(3)}{_RESET}"
|
||||||
@@ -121,7 +122,8 @@ def format_access_log(
|
|||||||
host_str = f"{_HOST}{host}{_RESET}"
|
host_str = f"{_HOST}{host}{_RESET}"
|
||||||
path_str = f"{_PATH}{path}{_RESET}"
|
path_str = f"{_PATH}{path}{_RESET}"
|
||||||
timing_str = f"{_TIMING}{format_duration_ms(duration_ms)}{_RESET}"
|
timing_str = f"{_TIMING}{format_duration_ms(duration_ms)}{_RESET}"
|
||||||
return f"{ip} {status_str} {method_str} {host_str}{path_str} {timing_str}"
|
extra_str = f" {_TIMING}{extra}{_RESET}" if extra else ""
|
||||||
|
return f"{ip} {status_str} {method_str} {host_str}{path_str}{extra_str} {timing_str}"
|
||||||
|
|
||||||
|
|
||||||
_ws_counter = 1
|
_ws_counter = 1
|
||||||
@@ -144,7 +146,7 @@ def log_ws_open(request, extra: str | None = None) -> int:
|
|||||||
"""Log WebSocket connection open. Returns connection ID for use in log_ws_close."""
|
"""Log WebSocket connection open. Returns connection ID for use in log_ws_close."""
|
||||||
ws_id = _next_ws_id()
|
ws_id = _next_ws_id()
|
||||||
|
|
||||||
client = request.ip or "-"
|
client = request.client_ip or "-"
|
||||||
host = request.host or "-"
|
host = request.host or "-"
|
||||||
path = request.path
|
path = request.path
|
||||||
origin = request.headers.get("origin")
|
origin = request.headers.get("origin")
|
||||||
@@ -218,3 +220,37 @@ def configure_access_logging() -> None:
|
|||||||
logger.addHandler(handler)
|
logger.addHandler(handler)
|
||||||
logger.setLevel(logging.INFO)
|
logger.setLevel(logging.INFO)
|
||||||
logger.propagate = False
|
logger.propagate = False
|
||||||
|
|
||||||
|
|
||||||
|
_LEVEL_EMOJI = {
|
||||||
|
logging.DEBUG: "🔍",
|
||||||
|
logging.INFO: "ℹ️",
|
||||||
|
logging.WARNING: "⚠️",
|
||||||
|
logging.ERROR: "🛑",
|
||||||
|
logging.CRITICAL: "🛑",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
class _EmojiFormatter(logging.Formatter):
|
||||||
|
"""Compact formatter: emoji + message, no timestamp/level text/logger name."""
|
||||||
|
|
||||||
|
def format(self, record: logging.LogRecord) -> str:
|
||||||
|
emoji = _LEVEL_EMOJI.get(record.levelno, "▪️")
|
||||||
|
return f"{emoji} {record.getMessage()}"
|
||||||
|
|
||||||
|
|
||||||
|
def configure_main_logging() -> None:
|
||||||
|
"""Replace Sanic's verbose 'Main yyyy-mm-dd INFO:' prefix with emoji-only format.
|
||||||
|
|
||||||
|
Patches LOGGING_CONFIG_DEFAULTS so the formatter survives every dictConfig
|
||||||
|
call Sanic makes during serve_single() / serve().
|
||||||
|
"""
|
||||||
|
from sanic.log import LOGGING_CONFIG_DEFAULTS
|
||||||
|
|
||||||
|
LOGGING_CONFIG_DEFAULTS["formatters"]["generic"] = {
|
||||||
|
"class": "cista.sanic_logging._EmojiFormatter",
|
||||||
|
}
|
||||||
|
# Also reformat any handlers already attached (covers the initial Sanic() call)
|
||||||
|
for name in ("sanic.root", "sanic.error", "sanic.server", "sanic.websockets"):
|
||||||
|
for handler in logging.getLogger(name).handlers:
|
||||||
|
handler.setFormatter(_EmojiFormatter())
|
||||||
|
|||||||
@@ -2,7 +2,7 @@
|
|||||||
<div v-if=showProgress() class="preview-progress" aria-label="Preview pending">
|
<div v-if=showProgress() class="preview-progress" aria-label="Preview pending">
|
||||||
<SpinnerIcon />
|
<SpinnerIcon />
|
||||||
</div>
|
</div>
|
||||||
<img v-else-if=previewSrc :src="previewSrc" alt="">
|
<img v-else-if="previewSrc && !video() && !audio()" :src="previewSrc" alt="">
|
||||||
<img v-else-if=doc.img :src=doc.url alt="">
|
<img v-else-if=doc.img :src=doc.url alt="">
|
||||||
<span v-else-if=doc.dir class="folder icon"></span>
|
<span v-else-if=doc.dir class="folder icon"></span>
|
||||||
<div v-else-if=video() class="video-container" :class="{ pending: !doc.complete }">
|
<div v-else-if=video() class="video-container" :class="{ pending: !doc.complete }">
|
||||||
|
|||||||
@@ -70,7 +70,7 @@ export class Doc {
|
|||||||
return ['mp4', 'mkv', 'webm', 'ogg', 'mp3', 'flac', 'aac', 'pdf'].includes(this.ext)
|
return ['mp4', 'mkv', 'webm', 'ogg', 'mp3', 'flac', 'aac', 'pdf'].includes(this.ext)
|
||||||
}
|
}
|
||||||
get previewurl(): string {
|
get previewurl(): string {
|
||||||
if (!this.complete || this.dir) return ''
|
if (!this.complete || !this.previewable) return ''
|
||||||
return this.url.replace(/^\/files/, '/preview')
|
return this.url.replace(/^\/files/, '/preview')
|
||||||
}
|
}
|
||||||
get ext(): string {
|
get ext(): string {
|
||||||
|
|||||||
@@ -0,0 +1,102 @@
|
|||||||
|
import argparse
|
||||||
|
import mimetypes
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from cista.preview import process_image_with_timing
|
||||||
|
|
||||||
|
|
||||||
|
def parse_args() -> argparse.Namespace:
|
||||||
|
parser = argparse.ArgumentParser(
|
||||||
|
description="Generate image previews for all files in a folder, one at a time.",
|
||||||
|
)
|
||||||
|
parser.add_argument("folder", type=Path, help="Folder to scan recursively")
|
||||||
|
parser.add_argument(
|
||||||
|
"--px",
|
||||||
|
type=int,
|
||||||
|
default=1024,
|
||||||
|
help="Maximum preview dimension in pixels (default: 1024)",
|
||||||
|
)
|
||||||
|
parser.add_argument(
|
||||||
|
"--quality",
|
||||||
|
type=int,
|
||||||
|
default=60,
|
||||||
|
help="AVIF quality passed to preview generation (default: 60)",
|
||||||
|
)
|
||||||
|
return parser.parse_args()
|
||||||
|
|
||||||
|
|
||||||
|
def is_image_file(path: Path) -> bool:
|
||||||
|
mime_type, _ = mimetypes.guess_type(path.name)
|
||||||
|
return bool(mime_type and mime_type.startswith("image/"))
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> int:
|
||||||
|
args = parse_args()
|
||||||
|
folder = args.folder.resolve()
|
||||||
|
if not folder.is_dir():
|
||||||
|
raise SystemExit(f"Not a directory: {folder}")
|
||||||
|
|
||||||
|
files = sorted(
|
||||||
|
path for path in folder.rglob("*") if path.is_file() and is_image_file(path)
|
||||||
|
)
|
||||||
|
if not files:
|
||||||
|
print(f"No image files found under {folder}")
|
||||||
|
return 0
|
||||||
|
|
||||||
|
total_files = 0
|
||||||
|
total_bytes = 0
|
||||||
|
total_timing_slots: list[float] = []
|
||||||
|
total_preview_ms: float = 0.0
|
||||||
|
failures = 0
|
||||||
|
|
||||||
|
print(f"Scanning {folder}")
|
||||||
|
print(f"Generating previews for {len(files)} image files")
|
||||||
|
|
||||||
|
for path in files:
|
||||||
|
total_files += 1
|
||||||
|
rel = path.relative_to(folder)
|
||||||
|
try:
|
||||||
|
preview, timing = process_image_with_timing(
|
||||||
|
path,
|
||||||
|
maxsize=args.px,
|
||||||
|
quality=args.quality,
|
||||||
|
)
|
||||||
|
except Exception as exc:
|
||||||
|
failures += 1
|
||||||
|
print(f"FAIL {rel} error={exc}")
|
||||||
|
continue
|
||||||
|
|
||||||
|
total_bytes += len(preview)
|
||||||
|
timings = timing.timings or []
|
||||||
|
if len(total_timing_slots) < len(timings):
|
||||||
|
total_timing_slots.extend([0.0] * (len(timings) - len(total_timing_slots)))
|
||||||
|
for i, value in enumerate(timings):
|
||||||
|
total_timing_slots[i] += value
|
||||||
|
total_ms = sum(timings)
|
||||||
|
total_preview_ms += total_ms
|
||||||
|
|
||||||
|
detail = " / ".join(f"{value:.1f}ms" for value in timings)
|
||||||
|
if detail:
|
||||||
|
detail = f"timings={detail} total={total_ms:.1f}ms"
|
||||||
|
else:
|
||||||
|
detail = f"total={total_ms:.1f}ms"
|
||||||
|
print(f"OK {rel} backend={timing.backend} bytes={len(preview)} {detail}")
|
||||||
|
|
||||||
|
completed = total_files - failures
|
||||||
|
print()
|
||||||
|
print("Summary")
|
||||||
|
print(f" files={total_files}")
|
||||||
|
print(f" completed={completed}")
|
||||||
|
print(f" failed={failures}")
|
||||||
|
print(f" preview_bytes={total_bytes}")
|
||||||
|
if completed:
|
||||||
|
if total_timing_slots:
|
||||||
|
for i, value in enumerate(total_timing_slots, start=1):
|
||||||
|
print(f" timing{i}_total_ms={value:.1f}")
|
||||||
|
print(f" preview_total_ms={total_preview_ms:.1f}")
|
||||||
|
print(f" preview_avg_ms={total_preview_ms / completed:.1f}")
|
||||||
|
return 0 if failures == 0 else 1
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
raise SystemExit(main())
|
||||||
Reference in New Issue
Block a user