"""Helpers for matching and downloading printer-side video files.""" from __future__ import annotations import asyncio import json import logging import re import secrets import shutil import tempfile import time import zipfile from collections.abc import Awaitable, Callable from dataclasses import dataclass, replace from datetime import datetime, timedelta, timezone from pathlib import Path, PurePosixPath from backend.app.core.config import settings from backend.app.core.tasks import spawn_background_task from backend.app.services.bambu_ftp import ( DownloadCancelled, DownloadInsufficientSpace, DownloadLimitExceeded, download_file_async, ) logger = logging.getLogger(__name__) VIDEO_SUFFIXES = (".mp4", ".avi", ".mkv") MAX_PRINTER_ZIP_BYTES = 10 * 1024**3 PRINTER_ZIP_FREE_SPACE_RESERVE = 256 * 1024**2 _STALE_BUNDLE_SECONDS = 60 * 60 MAX_PRINTER_ZIP_PREPARE_SECONDS = 30 * 60 MAX_OPEN_ARCHIVE_IPCAM_SECONDS = 24 * 60 * 60 _BUNDLE_KEY_RE = re.compile(r"^[A-Za-z0-9_-]{1,200}$") _JOB_KEY_RE = re.compile(r"^[A-Za-z0-9_-]{20,200}$") _LOCAL_JOB_TASKS: dict[str, asyncio.Task] = {} _cleanup_task: asyncio.Task | None = None _CLEANUP_INTERVAL_SECONDS = 15 * 60 class PrinterFilesZipTooLargeError(ValueError): """The selected printer files exceed the bounded ZIP staging limit.""" class PrinterFilesZipInsufficientSpaceError(OSError): """The app data volume cannot safely stage the selected files.""" @dataclass(frozen=True) class PrinterFilesZipResult: """Result of staging one printer ZIP.""" path: Path requested: int successful: int failed_paths: tuple[str, ...] total_bytes: int @dataclass(frozen=True) class PrinterFilesJobStatus: """Serializable state for an asynchronous browser preparation job.""" job_id: str printer_id: int state: str requested: int successful: int = 0 failed: int = 0 token: str | None = None filename: str | None = None message: str | None = None class _FileCancelSignal: """Cross-worker cancellation signal checked by the FTP callback thread.""" def __init__(self, path: Path): self.path = path self._last_check = 0.0 self._cached = False def is_set(self) -> bool: if self._cached: return True now = time.monotonic() if now - self._last_check >= 0.25: self._last_check = now self._cached = self.path.exists() return self._cached def _job_status_path(job_id: str) -> Path: if not _JOB_KEY_RE.fullmatch(job_id): raise ValueError("Invalid printer download job id") return _printer_zip_root() / f"job-{job_id}.json" def _job_cancel_path(job_id: str) -> Path: if not _JOB_KEY_RE.fullmatch(job_id): raise ValueError("Invalid printer download job id") return _printer_zip_root() / f"job-{job_id}.cancel" def _write_job_status(status: PrinterFilesJobStatus) -> None: """Atomically publish job state for polling from any app worker.""" path = _job_status_path(status.job_id) temp_path = path.with_suffix(".tmp") temp_path.write_text(json.dumps(status.__dict__, separators=(",", ":")), encoding="utf-8") temp_path.replace(path) def _read_job_status(job_id: str) -> PrinterFilesJobStatus | None: try: data = json.loads(_job_status_path(job_id).read_text(encoding="utf-8")) return PrinterFilesJobStatus(**data) except (FileNotFoundError, OSError, ValueError, TypeError, json.JSONDecodeError): return None def _naive_utc(value: datetime | None) -> datetime | None: if value is None: return None if value.tzinfo is not None: return value.astimezone(timezone.utc).replace(tzinfo=None) return value def match_ipcam_chunks( files: list[dict], started_at: datetime | None, completed_at: datetime | None, *, now: datetime | None = None, ) -> list[dict]: """Return `/ipcam` chunks whose completion time overlaps a print. Bambu's `ipcam-record.*.mp4` files are fixed-size chunks. On the tested X1C and H2D firmware, their FTP mtime is the chunk completion time in the same UTC-naive basis used by archive timestamps. Some firmware reports FTP LIST mtimes in printer-local time instead; LIST carries no timezone with which to correct those values reliably. A ten-minute tail includes the final chunk, whose mtime lands after the print-complete event. """ start = _naive_utc(started_at) if start is None: return [] live_end = _naive_utc(now) or datetime.now(timezone.utc).replace(tzinfo=None) # A crash can leave an archive in ``printing`` indefinitely. Do not turn # that stale row into a window covering every chunk created since then. end = _naive_utc(completed_at) or min(live_end, start + timedelta(seconds=MAX_OPEN_ARCHIVE_IPCAM_SECONDS)) lower = start - timedelta(minutes=1) upper = max(start, end) + timedelta(minutes=10) matches: list[dict] = [] for file in files: name = str(file.get("name") or "") mtime = file.get("mtime") if file.get("is_directory") or not name.lower().startswith("ipcam-record."): continue if not name.lower().endswith(VIDEO_SUFFIXES) or not isinstance(mtime, datetime): continue timestamp = _naive_utc(mtime) if timestamp is not None and lower <= timestamp <= upper: matches.append(file) matches.sort(key=lambda item: _naive_utc(item.get("mtime")) or datetime.min) return matches def _zip_arcname(remote_path: str, used: set[str]) -> str: """Return a safe, unique relative archive name for a printer path.""" parts = [part for part in PurePosixPath(remote_path).parts if part not in ("/", "", ".", "..")] candidate = "/".join(parts) or "printer-file" stem = candidate suffix = "" if "." in PurePosixPath(candidate).name: suffix = "".join(PurePosixPath(candidate).suffixes) stem = candidate[: -len(suffix)] if suffix else candidate counter = 2 while candidate in used: candidate = f"{stem}-{counter}{suffix}" counter += 1 used.add(candidate) return candidate def _printer_zip_root() -> Path: """Return the dedicated staging root without doing event-loop I/O.""" return settings.archive_dir / "temp" / "printer-file-downloads" def _ensure_printer_zip_root() -> Path: """Create and return the staging root on the persistent data volume.""" root = _printer_zip_root() root.mkdir(parents=True, exist_ok=True) return root def _prune_stale_bundles(root: Path) -> None: """Remove abandoned bundles after token expiry, without touching archives.""" cutoff = time.time() - _STALE_BUNDLE_SECONDS if not root.exists(): return for child in root.iterdir(): try: if child.is_dir() and child.stat().st_mtime < cutoff: shutil.rmtree(child, ignore_errors=True) elif child.is_file() and child.name.startswith("job-") and child.stat().st_mtime < cutoff: child.unlink(missing_ok=True) except OSError: continue async def prune_stale_printer_file_bundles() -> None: """Prune abandoned printer ZIPs without blocking the event loop.""" root = await asyncio.to_thread(_ensure_printer_zip_root) await asyncio.to_thread(_prune_stale_bundles, root) async def _printer_download_cleanup_loop() -> None: while True: try: await asyncio.sleep(_CLEANUP_INTERVAL_SECONDS) await prune_stale_printer_file_bundles() except asyncio.CancelledError: break except Exception: logger.exception("Periodic printer-download cleanup failed") def start_printer_download_cleanup() -> None: global _cleanup_task if _cleanup_task is None: _cleanup_task = spawn_background_task(_printer_download_cleanup_loop(), name="printer-download-cleanup") async def stop_printer_download_cleanup() -> None: """Stop cleanup and cancel every in-process preparation before shutdown.""" global _cleanup_task tasks: list[asyncio.Task] = [] cleanup_task = _cleanup_task _cleanup_task = None if cleanup_task is not None: cleanup_task.cancel() tasks.append(cleanup_task) # Jobs can be inside an FTP worker thread. Publish the same cooperative # cancellation marker used by the DELETE endpoint before cancelling the # asyncio wrapper, then await every wrapper so no executor work is left # behind when the application event loop closes. for job_id, task in list(_LOCAL_JOB_TASKS.items()): if not task.done(): await asyncio.to_thread(_job_cancel_path(job_id).touch) task.cancel() tasks.append(task) if tasks: await asyncio.gather(*tasks, return_exceptions=True) _LOCAL_JOB_TASKS.clear() def printer_files_zip_path(printer_id: int, token: str) -> Path | None: """Resolve the staged ZIP for a resource-bound browser token.""" bundle_key = f"{printer_id}-{token}" if not _BUNDLE_KEY_RE.fullmatch(bundle_key): return None return _printer_zip_root() / bundle_key / "printer-files.zip" def bind_printer_files_zip_to_token( result: PrinterFilesZipResult, printer_id: int, token: str, ) -> PrinterFilesZipResult: """Move a prepared bundle to the path derived from its persisted token.""" target = printer_files_zip_path(printer_id, token) if target is None: raise ValueError("Invalid printer ZIP token") result.path.parent.rename(target.parent) return replace(result, path=target) def _check_initial_space(root: Path, sizes: dict[str, int]) -> None: # These sizes are client-reported hints used only for an early rejection, # so this is a courtesy, not the bound. The real one is enforced per write # and per FTP callback below, against actual bytes and the live free space, # which is the only thing that can hold when several preparations run at # once -- and they do: nothing serializes them. Two concurrent jobs that # both pass here stop independently at the reserve, and the one that gets # there second fails with a message saying so. expected_total = sum(sizes.values()) if expected_total > MAX_PRINTER_ZIP_BYTES: raise PrinterFilesZipTooLargeError( f"Selected files total {expected_total} bytes; the limit is {MAX_PRINTER_ZIP_BYTES} bytes" ) largest_file = max(sizes.values(), default=0) # In the worst case the ZIP is as large as the inputs while the largest # source is still staged beside it. Keep a reserve for the database/logs. required = expected_total + largest_file + PRINTER_ZIP_FREE_SPACE_RESERVE free = shutil.disk_usage(root).free if free < required: raise PrinterFilesZipInsufficientSpaceError( f"The app data volume needs {required} bytes free to stage this selection; {free} bytes are available" ) async def build_printer_files_zip( printer, paths: list[str], sizes: dict[str, int], *, bundle_key: str | None = None, preserve_paths: bool = True, allow_empty: bool = False, cancel_signal: _FileCancelSignal | None = None, progress_callback: Callable[[int, int], Awaitable[None]] | None = None, ) -> PrinterFilesZipResult: """Download printer files one at a time into a disk-backed ZIP. The previous implementation held every source file and the final ZIP in memory. Continuous `/ipcam` chunks are commonly ~250 MB each, so selecting only a few could exhaust both server and browser memory. """ root = await asyncio.to_thread(_ensure_printer_zip_root) await asyncio.to_thread(_prune_stale_bundles, root) await asyncio.to_thread(_check_initial_space, root, sizes) bundle_dir: Path | None = None try: if bundle_key is None: bundle_dir = Path(await asyncio.to_thread(tempfile.mkdtemp, prefix="bundle-", dir=root)) else: if not _BUNDLE_KEY_RE.fullmatch(bundle_key): raise ValueError("Invalid printer ZIP bundle key") bundle_dir = root / bundle_key await asyncio.to_thread(bundle_dir.mkdir, mode=0o700) zip_path = bundle_dir / "printer-files.zip" successful = 0 total_bytes = 0 failed_paths: list[str] = [] used_names: set[str] = set() archive = await asyncio.to_thread(zipfile.ZipFile, zip_path, "w", allowZip64=True) try: for index, remote_path in enumerate(paths): if cancel_signal is not None and cancel_signal.is_set(): raise asyncio.CancelledError if not isinstance(remote_path, str) or not remote_path.startswith("/") or "\x00" in remote_path: logger.warning("Skipping invalid printer file path: %r", remote_path) failed_paths.append(remote_path) continue staged_path = bundle_dir / f"download-{index}" try: expected_size = sizes.get(remote_path) if expected_size is not None: free = (await asyncio.to_thread(shutil.disk_usage, root)).free if free < expected_size + PRINTER_ZIP_FREE_SPACE_RESERVE: raise PrinterFilesZipInsufficientSpaceError( "The app data volume lacks space for the next selected file" ) downloaded = await download_file_async( printer.ip_address, printer.access_code, remote_path, staged_path, timeout=600, socket_timeout=60, printer_model=printer.model, expected_size=expected_size, max_bytes=MAX_PRINTER_ZIP_BYTES - total_bytes, cancel_event=cancel_signal, min_free_bytes=PRINTER_ZIP_FREE_SPACE_RESERVE, # Outside the per-printer download gate (#2957), in both # directions. A selection of ~250 MB /ipcam chunks holds # the printer for as long as it legitimately takes, and # nothing else should be made to wait that out; equally, # each file here must not stall behind a thumbnail. serialize=False, ) if not downloaded: failed_paths.append(remote_path) continue # Deliberately no second size comparison here. The transfer # was already checked against the printer's own SIZE, which # download_to_file treats as the authority precisely because # it beats a hint the browser round-tripped; re-judging the # result against that hint would overrule the better number # with the worse one. The hint goes stale in exactly the case # this feature exists for -- an /ipcam chunk or a timelapse # still being written when the listing was taken -- and a # complete file would then be dropped as "truncated". file_size = (await asyncio.to_thread(staged_path.stat)).st_size if total_bytes + file_size > MAX_PRINTER_ZIP_BYTES: raise PrinterFilesZipTooLargeError( f"Downloaded files exceed the {MAX_PRINTER_ZIP_BYTES}-byte limit" ) free = (await asyncio.to_thread(shutil.disk_usage, root)).free if free < file_size + PRINTER_ZIP_FREE_SPACE_RESERVE: raise PrinterFilesZipInsufficientSpaceError( "The app data volume ran out of safe staging space while building the ZIP" ) compression = ( zipfile.ZIP_STORED if remote_path.lower().endswith(VIDEO_SUFFIXES) else zipfile.ZIP_DEFLATED ) arc_source = remote_path if preserve_paths else PurePosixPath(remote_path).name await asyncio.to_thread( archive.write, staged_path, _zip_arcname(arc_source, used_names), compress_type=compression, ) successful += 1 total_bytes += file_size except DownloadLimitExceeded as exc: raise PrinterFilesZipTooLargeError( f"Downloaded files exceed the {MAX_PRINTER_ZIP_BYTES}-byte limit" ) from exc except DownloadInsufficientSpace as exc: raise PrinterFilesZipInsufficientSpaceError( "The app data volume ran out of safe staging space during transfer" ) from exc except DownloadCancelled as exc: raise asyncio.CancelledError from exc except (PrinterFilesZipTooLargeError, PrinterFilesZipInsufficientSpaceError): raise except Exception as exc: logger.warning("Failed to add %s to printer ZIP: %s", remote_path, exc) failed_paths.append(remote_path) finally: await asyncio.to_thread(staged_path.unlink, missing_ok=True) if progress_callback is not None: await progress_callback(successful, len(failed_paths)) finally: await asyncio.shield(asyncio.to_thread(archive.close)) except BaseException: if bundle_dir is not None: await asyncio.shield(asyncio.to_thread(shutil.rmtree, bundle_dir, ignore_errors=True)) raise if successful == 0 and not allow_empty: await asyncio.to_thread(shutil.rmtree, bundle_dir, ignore_errors=True) raise FileNotFoundError("No files could be downloaded") return PrinterFilesZipResult( path=zip_path, requested=len(paths), successful=successful, failed_paths=tuple(failed_paths), total_bytes=total_bytes, ) def printer_file_path(printer_id: int, token: str) -> Path | None: """Resolve a prepared native single-file download.""" bundle_key = f"{printer_id}-{token}" if not _BUNDLE_KEY_RE.fullmatch(bundle_key): return None return _printer_zip_root() / bundle_key / "printer-file" def bind_printer_file_to_token(result: PrinterFilesZipResult, printer_id: int, token: str) -> PrinterFilesZipResult: target = printer_file_path(printer_id, token) if target is None: raise ValueError("Invalid printer file token") result.path.parent.rename(target.parent) return replace(result, path=target) async def build_printer_file( printer, remote_path: str, expected_size: int | None, *, bundle_key: str, cancel_signal: _FileCancelSignal | None = None, ) -> PrinterFilesZipResult: """Stage one printer file on disk for a browser-native download. Also the read path for the 3MF preview in the file browser, which is why nothing here waits on a shared lock: a preview must not queue behind somebody else's ten-gigabyte selection for as long as that takes. """ if not remote_path.startswith("/") or "\x00" in remote_path: raise FileNotFoundError("Invalid printer file path") root = await asyncio.to_thread(_ensure_printer_zip_root) size_hints = {remote_path: expected_size} if expected_size is not None else {} await asyncio.to_thread(_check_initial_space, root, size_hints) bundle_dir = root / bundle_key try: await asyncio.to_thread(bundle_dir.mkdir, mode=0o700) local_path = bundle_dir / "printer-file" downloaded = await download_file_async( printer.ip_address, printer.access_code, remote_path, local_path, timeout=600, socket_timeout=60, printer_model=printer.model, expected_size=expected_size, max_bytes=MAX_PRINTER_ZIP_BYTES, cancel_event=cancel_signal, min_free_bytes=PRINTER_ZIP_FREE_SPACE_RESERVE, # The lock-free promise in this function's docstring, kept: a preview # must not queue behind somebody else's selection (#2957). serialize=False, ) if not downloaded: raise FileNotFoundError("The selected printer file could not be downloaded") file_size = (await asyncio.to_thread(local_path.stat)).st_size return PrinterFilesZipResult( path=local_path, requested=1, successful=1, failed_paths=(), total_bytes=file_size, ) except DownloadLimitExceeded as exc: await asyncio.shield(asyncio.to_thread(shutil.rmtree, bundle_dir, ignore_errors=True)) raise PrinterFilesZipTooLargeError(f"Downloaded file exceeds the {MAX_PRINTER_ZIP_BYTES}-byte limit") from exc except DownloadInsufficientSpace as exc: await asyncio.shield(asyncio.to_thread(shutil.rmtree, bundle_dir, ignore_errors=True)) raise PrinterFilesZipInsufficientSpaceError( "The app data volume ran out of safe staging space during transfer" ) from exc except DownloadCancelled as exc: await asyncio.shield(asyncio.to_thread(shutil.rmtree, bundle_dir, ignore_errors=True)) raise asyncio.CancelledError from exc except BaseException: await asyncio.shield(asyncio.to_thread(shutil.rmtree, bundle_dir, ignore_errors=True)) raise async def _run_printer_files_job( printer, job_id: str, paths: list[str], sizes: dict[str, int], filename: str, as_zip: bool, ) -> None: from backend.app.core.auth import create_slicer_download_token cancel_signal = _FileCancelSignal(_job_cancel_path(job_id)) status = PrinterFilesJobStatus(job_id, printer.id, "preparing", len(paths), filename=filename) await asyncio.to_thread(_write_job_status, status) async def report_progress(successful: int, failed: int) -> None: await asyncio.to_thread( _write_job_status, PrinterFilesJobStatus( job_id, printer.id, "preparing", len(paths), successful=successful, failed=failed, filename=filename, ), ) try: async with asyncio.timeout(MAX_PRINTER_ZIP_PREPARE_SECONDS): if as_zip: result = await build_printer_files_zip( printer, paths, sizes, bundle_key=f"job-{job_id}", cancel_signal=cancel_signal, progress_callback=report_progress, ) else: result = await build_printer_file( printer, paths[0], sizes.get(paths[0]), bundle_key=f"job-{job_id}", cancel_signal=cancel_signal, ) if cancel_signal.is_set(): await asyncio.to_thread(remove_printer_files_zip, result.path) raise asyncio.CancelledError token = await create_slicer_download_token("printer-files", printer.id) if as_zip: result = await asyncio.to_thread(bind_printer_files_zip_to_token, result, printer.id, token) else: result = await asyncio.to_thread(bind_printer_file_to_token, result, printer.id, token) await asyncio.to_thread( _write_job_status, PrinterFilesJobStatus( job_id, printer.id, "ready", len(paths), successful=result.successful, failed=len(result.failed_paths), token=token, filename=filename, ), ) except asyncio.CancelledError: await asyncio.shield( asyncio.to_thread( _write_job_status, PrinterFilesJobStatus(job_id, printer.id, "cancelled", len(paths), filename=filename), ) ) except PrinterFilesZipTooLargeError as exc: await asyncio.to_thread( _write_job_status, PrinterFilesJobStatus(job_id, printer.id, "failed", len(paths), filename=filename, message=str(exc)), ) except PrinterFilesZipInsufficientSpaceError as exc: await asyncio.to_thread( _write_job_status, PrinterFilesJobStatus(job_id, printer.id, "failed", len(paths), filename=filename, message=str(exc)), ) except TimeoutError: await asyncio.to_thread( _write_job_status, PrinterFilesJobStatus( job_id, printer.id, "failed", len(paths), filename=filename, message="Printer download preparation exceeded the 30-minute limit", ), ) except FileNotFoundError as exc: await asyncio.to_thread( _write_job_status, PrinterFilesJobStatus(job_id, printer.id, "failed", len(paths), filename=filename, message=str(exc)), ) except Exception: logger.exception("Printer download job %s failed", job_id) await asyncio.to_thread( _write_job_status, PrinterFilesJobStatus( job_id, printer.id, "failed", len(paths), filename=filename, message="Printer download preparation failed", ), ) finally: await asyncio.to_thread(_job_cancel_path(job_id).unlink, missing_ok=True) async def start_printer_files_job( printer, paths: list[str], sizes: dict[str, int], filename: str, *, as_zip: bool, ) -> PrinterFilesJobStatus: """Start a bounded background preparation and return immediately.""" if not paths: raise ValueError("No files specified") root = await asyncio.to_thread(_ensure_printer_zip_root) await asyncio.to_thread(_prune_stale_bundles, root) await asyncio.to_thread(_check_initial_space, root, sizes) job_id = secrets.token_urlsafe(24) status = PrinterFilesJobStatus(job_id, printer.id, "queued", len(paths), filename=filename) await asyncio.to_thread(_write_job_status, status) task = spawn_background_task( _run_printer_files_job(printer, job_id, paths, sizes, filename, as_zip), name=f"printer-download-{printer.id}-{job_id}", ) _LOCAL_JOB_TASKS[job_id] = task task.add_done_callback(lambda _task: _LOCAL_JOB_TASKS.pop(job_id, None)) return status async def get_printer_files_job(job_id: str, printer_id: int) -> PrinterFilesJobStatus | None: status = await asyncio.to_thread(_read_job_status, job_id) if status is None or status.printer_id != printer_id: return None return status async def cancel_printer_files_job(job_id: str, printer_id: int) -> bool: status = await get_printer_files_job(job_id, printer_id) if status is None: return False await asyncio.to_thread(_job_cancel_path(job_id).touch) task = _LOCAL_JOB_TASKS.get(job_id) if task is not None and not task.done(): task.cancel() if status.state == "ready" and status.token: zip_path = printer_files_zip_path(printer_id, status.token) prepared = ( zip_path if zip_path is not None and await asyncio.to_thread(zip_path.is_file) else printer_file_path(printer_id, status.token) ) if prepared is not None: await asyncio.to_thread(remove_printer_files_zip, prepared) await asyncio.to_thread( _write_job_status, replace(status, state="cancelled", token=None), ) return True def remove_printer_files_zip(zip_path: Path) -> None: """Remove a completed download bundle after FileResponse finishes.""" shutil.rmtree(zip_path.parent, ignore_errors=True)