mirror of
https://github.com/maziggy/bambuddy.git
synced 2026-09-30 03:01:21 +02:00
maziggy
7c10412f99
When a print's own 3MF cannot be fetched the usage tracker borrows one from the library or a previous archive, matching on the filename stem. That is far weaker evidence than it looks: Bambu Studio writes the printer-side filename from the project's Title metadata, so every plate of a project reaches the printer under one name however the file was renamed on disk. The reporter's single-filament job was handed a previous archive's three-filament plate. Three spools were debited for material that was never extruded, and nothing on the archive said the numbers were someone else's. A candidate is now rejected when it positively contradicts the print - a different plate, or a filament count the slicer's ams_mapping disagrees with - and the accepted one is logged with both expectations. Only on a contradiction: the plate needs firmware that echoes it and the count needs a print command Bambuddy saw, and refusing everything uncorroborated would retire the fallback recovery this same issue asked for. The count is scoped to one plate or not compared at all. Unscoped, the filament reader collects every <filament> in the file, and that sum against one plate's count would reject every multi-plate library upload on exactly the firmwares that cannot tell us the plate. ----- Give a download the time the file needs, and one printer at a time (issue #2957) ftp_timeout is handed to every download as both the socket inactivity timeout and the whole-transfer deadline, which makes its 30s default a cap on how big a file a printer may serve. The reporter measured one 5.4 MB 3MF at 45s off a worn P1S SD card and 25s off a new one, and a 15.15 MB 3MF at 105s. None of those links were broken - they were slow, which is what the inactivity timeout exists to tell apart. download_to_file already asks for SIZE. It now reports it, and the total deadline follows the file at the 25 KB/s floor _upload_deadline has used since do, so #2572's cap on the executor queue wait is untouched. Capped at 300s for a reason that is not about FTP: on_print_start holds a pooled DB connection across its whole 3MF hunt. Downloads also take turns per printer now. He watched Bambu Studio lose its own connection while Bambuddy pulled a 12 MB 3MF, and a later log caught two Bambuddy downloads of the same file overlapping at print start. The gate is soft - whoever cannot have it within 30s goes anyway, because a print losing its 3MF to queueing is worse than the contention, and a soft gate cannot deadlock. It is also meaningful for the first time. The 90s cap on a multi-path lookup returned while its worker kept walking the remaining paths, still on the printer's socket; that walk is now cancelled and waited out before the printer is handed on. ----- Look in the shared 3MF cache again before each cover retry (issue #2957) The cover endpoint and the print-start archive flow share a cache so whichever fetches the 3MF first hands it to the other (#972). The cover consulted it once on the way in, then retried for up to two and a half minutes without looking again. In the reporter's log the archive flow published the file 42 seconds into that sequence and the cover's third attempt still pulled its own 5,250,969-byte copy, off a printer that was mid-print on the same SD card. A file picked up that way is left alone rather than re-registered under this endpoint's own name or deleted on the way out. It is the archive flow's.
729 lines
28 KiB
Python
729 lines
28 KiB
Python
"""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)
|