Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d52b24372a | ||
|
|
742b3006c9 | ||
|
|
01a979c00a | ||
|
|
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 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.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.util.apphelpers import handle_sanic_exception
|
||||
|
||||
@@ -30,6 +31,7 @@ sanic.helpers._ENTITY_HEADERS = frozenset()
|
||||
configure_access_logging()
|
||||
|
||||
app = Sanic("cista", strict_slashes=True)
|
||||
configure_main_logging()
|
||||
# Register either SSO proxy or built-in auth routes based on PASKIA_BACKEND_URL
|
||||
if sso.paskia_enabled():
|
||||
app.blueprint(sso.bp) # SSO proxy for /auth/* routes
|
||||
@@ -47,13 +49,12 @@ setproctitle("cista-main")
|
||||
async def main_start(app):
|
||||
config.load_config()
|
||||
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(
|
||||
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
|
||||
app.ctx.zipexec = ThreadPoolExecutor(max_workers=32, thread_name_prefix="cista-zip")
|
||||
await start_preview_workers()
|
||||
watching.start(app)
|
||||
|
||||
|
||||
@@ -61,6 +62,7 @@ async def main_start(app):
|
||||
@app.before_server_stop
|
||||
async def main_stop(app):
|
||||
watching.stop(app)
|
||||
await shutdown_preview_workers()
|
||||
app.ctx.threadexec.shutdown()
|
||||
app.ctx.zipexec.shutdown(cancel_futures=True)
|
||||
await sso.close_client()
|
||||
@@ -94,7 +96,7 @@ async def log_access(req, res):
|
||||
return res
|
||||
start = getattr(req.ctx, "_log_start", None)
|
||||
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 "-"
|
||||
path = req.path
|
||||
if req.query_string:
|
||||
@@ -102,7 +104,8 @@ async def log_access(req, res):
|
||||
if isinstance(qs, bytes):
|
||||
qs = qs.decode(errors="replace")
|
||||
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)
|
||||
return res
|
||||
|
||||
|
||||
+344
-106
@@ -2,29 +2,33 @@ import asyncio
|
||||
import gc
|
||||
import io
|
||||
import mimetypes
|
||||
import struct
|
||||
import sys
|
||||
import threading
|
||||
import urllib.parse
|
||||
from collections import OrderedDict
|
||||
from dataclasses import dataclass
|
||||
from multiprocessing import cpu_count
|
||||
from pathlib import PurePosixPath
|
||||
from time import perf_counter
|
||||
from urllib.parse import unquote
|
||||
from wsgiref.handlers import format_date_time
|
||||
|
||||
import msgspec
|
||||
|
||||
import av
|
||||
import fitz # PyMuPDF
|
||||
import numpy as np
|
||||
import pillow_heif
|
||||
from PIL import Image
|
||||
import pyvips
|
||||
from blake3 import blake3
|
||||
from sanic import Blueprint, empty, raw, redirect
|
||||
from sanic.exceptions import NotFound
|
||||
from sanic.log import logger
|
||||
|
||||
from cista import auth, config
|
||||
from cista.preview_worker import PreviewRequest, PreviewResponse
|
||||
from cista.util.filename import sanitize
|
||||
|
||||
pillow_heif.register_heif_opener()
|
||||
|
||||
bp = Blueprint("preview", url_prefix="/preview")
|
||||
|
||||
|
||||
@@ -70,6 +74,213 @@ class PreviewCache:
|
||||
# Global preview cache instance
|
||||
_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
|
||||
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",
|
||||
stderr=resp.stderr,
|
||||
backend=resp.backend,
|
||||
)
|
||||
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(f"worker checksum mismatch for {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(
|
||||
f"worker protocol failure for {filepath.name}: {e}"
|
||||
)
|
||||
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
|
||||
async def verify_preview(request):
|
||||
@@ -77,16 +288,46 @@ async def verify_preview(request):
|
||||
await auth.verify(request)
|
||||
|
||||
|
||||
# Map EXIF Orientation value to a corresponding PIL transpose
|
||||
EXIF_ORI = {
|
||||
2: Image.Transpose.FLIP_LEFT_RIGHT,
|
||||
3: Image.Transpose.ROTATE_180,
|
||||
4: Image.Transpose.FLIP_TOP_BOTTOM,
|
||||
5: Image.Transpose.TRANSPOSE,
|
||||
6: Image.Transpose.ROTATE_270,
|
||||
7: Image.Transpose.TRANSVERSE,
|
||||
8: Image.Transpose.ROTATE_90,
|
||||
}
|
||||
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."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
message: str,
|
||||
*,
|
||||
stderr: str | None = None,
|
||||
backend: str | None = None,
|
||||
):
|
||||
super().__init__(message)
|
||||
self.stderr = stderr
|
||||
self.backend = backend
|
||||
|
||||
|
||||
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(f"preview worker pool unavailable for {filepath.name}")
|
||||
return await _preview_pool.run(filepath, quality, maxsize, maxzoom)
|
||||
|
||||
|
||||
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>")
|
||||
@@ -102,6 +343,9 @@ async def preview(req, path):
|
||||
except FileNotFoundError:
|
||||
raise NotFound() from None
|
||||
|
||||
if not is_previewable_path(filepath):
|
||||
return empty(415)
|
||||
|
||||
etag = config.derived_secret(
|
||||
"preview", rel, stat.st_mtime_ns, quality, maxsize, maxzoom
|
||||
).hex()
|
||||
@@ -117,21 +361,47 @@ async def preview(req, path):
|
||||
return raw(cached.body, headers=cached.headers)
|
||||
|
||||
# Generate preview
|
||||
img = await asyncio.get_event_loop().run_in_executor(
|
||||
req.app.ctx.threadexec, dispatch, filepath, quality, maxsize, maxzoom
|
||||
)
|
||||
try:
|
||||
img, preview_resp = await _run_preview_process(
|
||||
filepath, quality, maxsize, maxzoom
|
||||
)
|
||||
except PreviewTimeout:
|
||||
return empty(504)
|
||||
except PreviewError as e:
|
||||
if e.backend:
|
||||
req.ctx._log_extra = e.backend
|
||||
detail = str(e)
|
||||
if detail == "preview worker error" and e.stderr:
|
||||
captured = e.stderr.strip()
|
||||
if captured:
|
||||
detail = captured.splitlines()[0]
|
||||
logger.error("%s preview: %s", filepath, detail)
|
||||
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:
|
||||
# Preview generation failed, redirect to the file itself
|
||||
return redirect(f"/files/{path}", status=303)
|
||||
|
||||
# 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")
|
||||
headers = {
|
||||
"etag": etag,
|
||||
"last-modified": format_date_time(stat.st_mtime),
|
||||
"cache-control": "max-age=604800, immutable"
|
||||
+ ("" 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())}",
|
||||
}
|
||||
_preview_cache.set(etag, CachedPreview(headers=headers, body=img))
|
||||
@@ -140,98 +410,51 @@ async def preview(req, path):
|
||||
|
||||
|
||||
def dispatch(path, quality, maxsize, maxzoom):
|
||||
backend = "unknown"
|
||||
try:
|
||||
if path.suffix.lower() in (".pdf", ".xps", ".epub", ".mobi"):
|
||||
if path.suffix.lower() in DOC_PREVIEW_SUFFIXES:
|
||||
backend = "pdf"
|
||||
return process_pdf(path, quality=quality, maxsize=maxsize, maxzoom=maxzoom)
|
||||
type, _ = mimetypes.guess_type(path.name)
|
||||
if type and type.startswith("video/"):
|
||||
mime_type, _ = mimetypes.guess_type(path.name)
|
||||
if mime_type and mime_type.startswith("video/"):
|
||||
backend = "video"
|
||||
return process_video(path, quality=quality, maxsize=maxsize)
|
||||
return process_image(path, quality=quality, maxsize=maxsize)
|
||||
if mime_type and mime_type.startswith("image/"):
|
||||
backend = "pyvips"
|
||||
return process_image(path, quality=quality, maxsize=maxsize)
|
||||
except ValueError as e:
|
||||
logger.warning(f"Cannot generate preview for {path}: {e}")
|
||||
return None, PreviewResponse(ok=False, backend=backend, error=str(e))
|
||||
except Exception as e:
|
||||
logger.exception(f"Error generating preview for {path}: {e}")
|
||||
return None, PreviewResponse(ok=False, backend=backend, error=str(e))
|
||||
return None, PreviewResponse(ok=False, backend=backend, error="preview unsupported")
|
||||
|
||||
|
||||
def process_image(path, *, maxsize, quality):
|
||||
try:
|
||||
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)
|
||||
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):
|
||||
import pyvips
|
||||
|
||||
t_load = perf_counter()
|
||||
t_start = perf_counter()
|
||||
img = pyvips.Image.new_from_file(str(path), access="sequential")
|
||||
t_proc = perf_counter()
|
||||
|
||||
img = img.autorot()
|
||||
scale = min(maxsize / img.width, maxsize / img.height, 1.0)
|
||||
if scale < 1.0:
|
||||
img = img.resize(scale)
|
||||
|
||||
t_save = perf_counter()
|
||||
ret = img.write_to_buffer(".avif", Q=quality)
|
||||
ret = img.write_to_buffer(
|
||||
".avif",
|
||||
Q=quality,
|
||||
effort=AVIF_FAST_EFFORT,
|
||||
strip=True,
|
||||
)
|
||||
t_end = perf_counter()
|
||||
|
||||
load_ms = (t_proc - t_load) * 1000
|
||||
proc_ms = (t_save - t_proc) * 1000
|
||||
save_ms = (t_end - t_save) * 1000
|
||||
logger.debug(
|
||||
"Preview image %s via pyvips: load=%.1fms process=%.1fms save=%.1fms",
|
||||
path.name,
|
||||
load_ms,
|
||||
proc_ms,
|
||||
save_ms,
|
||||
return ret, PreviewResponse(
|
||||
ok=True,
|
||||
mime="image/avif",
|
||||
backend="pyvips",
|
||||
timings=[round((t_end - t_start) * 1000, 1)],
|
||||
)
|
||||
|
||||
return ret
|
||||
|
||||
|
||||
def process_image_pillow(path, *, maxsize, quality):
|
||||
t_load = perf_counter()
|
||||
with Image.open(path) as img:
|
||||
# Force decode to include I/O in load timing
|
||||
img.load()
|
||||
t_proc = perf_counter()
|
||||
# Resize
|
||||
w, h = img.size
|
||||
img.thumbnail((min(w, maxsize), min(h, maxsize)))
|
||||
# Transpose pixels according to EXIF Orientation
|
||||
orientation = img.getexif().get(274, 1)
|
||||
if orientation in EXIF_ORI:
|
||||
img = img.transpose(EXIF_ORI[orientation])
|
||||
# Save as AVIF
|
||||
imgdata = io.BytesIO()
|
||||
t_save = perf_counter()
|
||||
img.save(
|
||||
imgdata,
|
||||
format="avif",
|
||||
quality=quality,
|
||||
speed=10,
|
||||
max_threads=1,
|
||||
avif=1,
|
||||
)
|
||||
|
||||
t_end = perf_counter()
|
||||
ret = imgdata.getvalue()
|
||||
|
||||
load_ms = (t_proc - t_load) * 1000
|
||||
proc_ms = (t_save - t_proc) * 1000
|
||||
save_ms = (t_end - t_save) * 1000
|
||||
logger.debug(
|
||||
"Preview image %s via Pillow: load=%.1fms process=%.1fms save=%.1fms",
|
||||
path.name,
|
||||
load_ms,
|
||||
proc_ms,
|
||||
save_ms,
|
||||
)
|
||||
|
||||
return ret
|
||||
|
||||
|
||||
def process_pdf(path, *, maxsize, maxzoom, quality, page_number=0):
|
||||
t_load_start = perf_counter()
|
||||
@@ -244,18 +467,24 @@ def process_pdf(path, *, maxsize, maxzoom, quality, page_number=0):
|
||||
t_load_end = perf_counter()
|
||||
|
||||
t_save_start = perf_counter()
|
||||
ret = pix.pil_tobytes(
|
||||
format="avif", quality=quality, speed=10, max_threads=1, avif=1
|
||||
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()
|
||||
|
||||
logger.debug(
|
||||
"Preview pdf %s: load+render=%.1fms save=%.1fms",
|
||||
path.name,
|
||||
(t_load_end - t_load_start) * 1000,
|
||||
(t_save_end - t_save_start) * 1000,
|
||||
return ret, PreviewResponse(
|
||||
ok=True,
|
||||
mime="image/avif",
|
||||
backend=backend,
|
||||
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):
|
||||
@@ -268,7 +497,13 @@ def process_video(path, *, maxsize, quality):
|
||||
t_save_start = t_load_start
|
||||
t_save_end = t_load_start
|
||||
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,
|
||||
):
|
||||
istream = icontainer.streams.video[0]
|
||||
@@ -360,14 +595,17 @@ def process_video(path, *, maxsize, quality):
|
||||
ocontainer.mux(ostream.encode(None)) # Flush the stream
|
||||
t_save_end = perf_counter()
|
||||
|
||||
# Capture frame dimensions before cleanup
|
||||
# Capture result before cleanup
|
||||
ret = imgdata.getvalue()
|
||||
logger.debug(
|
||||
"Preview video %s: load+decode=%.1fms save=%.1fms",
|
||||
path.name,
|
||||
(t_load_end - t_load_start) * 1000,
|
||||
(t_save_end - t_save_start) * 1000,
|
||||
resp = PreviewResponse(
|
||||
ok=True,
|
||||
mime="image/avif",
|
||||
backend="video",
|
||||
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
|
||||
gc.collect()
|
||||
return ret
|
||||
return ret, resp
|
||||
|
||||
@@ -0,0 +1,116 @@
|
||||
"""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 contextlib
|
||||
import io
|
||||
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
|
||||
stderr: 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
|
||||
stderr_capture = io.StringIO()
|
||||
handler = logging.StreamHandler(stderr_capture)
|
||||
root_logger = logging.getLogger()
|
||||
root_logger.addHandler(handler)
|
||||
try:
|
||||
with contextlib.redirect_stderr(stderr_capture):
|
||||
req = _dec_req.decode(line)
|
||||
result, resp = dispatch(
|
||||
Path(req.path), req.quality, req.maxsize, req.maxzoom
|
||||
)
|
||||
if not resp.ok:
|
||||
captured = stderr_capture.getvalue().strip()
|
||||
if captured:
|
||||
resp = PreviewResponse(
|
||||
ok=False,
|
||||
backend=resp.backend,
|
||||
error=resp.error,
|
||||
stderr=captured,
|
||||
)
|
||||
_write_response(resp, result or b"")
|
||||
except Exception as e:
|
||||
captured = stderr_capture.getvalue().strip()
|
||||
_write_response(
|
||||
PreviewResponse(ok=False, error=str(e), stderr=captured or None), b""
|
||||
)
|
||||
finally:
|
||||
root_logger.removeHandler(handler)
|
||||
handler.close()
|
||||
|
||||
|
||||
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()
|
||||
+42
-5
@@ -55,8 +55,8 @@ def format_client_ip(ip: str) -> str:
|
||||
return "-"
|
||||
stripped = ip.strip("[]")
|
||||
if ":" in stripped:
|
||||
return format_ipv6_network(ip)
|
||||
return ip
|
||||
return format_ipv6_network(stripped)
|
||||
return stripped
|
||||
|
||||
|
||||
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(
|
||||
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:
|
||||
ip = _format_left(format_client_ip(client))
|
||||
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}"
|
||||
path_str = f"{_PATH}{path}{_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
|
||||
@@ -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."""
|
||||
ws_id = _next_ws_id()
|
||||
|
||||
client = request.ip or "-"
|
||||
client = request.client_ip or "-"
|
||||
host = request.host or "-"
|
||||
path = request.path
|
||||
origin = request.headers.get("origin")
|
||||
@@ -218,3 +220,38 @@ def configure_access_logging() -> None:
|
||||
logger.addHandler(handler)
|
||||
logger.setLevel(logging.INFO)
|
||||
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, "▪️")
|
||||
sep = " " if record.levelno in (logging.INFO, logging.WARNING) else " "
|
||||
return f"{emoji}{sep}{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">
|
||||
<SpinnerIcon />
|
||||
</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="">
|
||||
<span v-else-if=doc.dir class="folder icon"></span>
|
||||
<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)
|
||||
}
|
||||
get previewurl(): string {
|
||||
if (!this.complete || this.dir) return ''
|
||||
if (!this.complete || !this.previewable) return ''
|
||||
return this.url.replace(/^\/files/, '/preview')
|
||||
}
|
||||
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