printer_media.py 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753
  1. """Helpers for matching and downloading printer-side video files."""
  2. from __future__ import annotations
  3. import asyncio
  4. import json
  5. import logging
  6. import re
  7. import secrets
  8. import shutil
  9. import tempfile
  10. import time
  11. import zipfile
  12. from collections.abc import Awaitable, Callable
  13. from dataclasses import dataclass, replace
  14. from datetime import datetime, timedelta, timezone
  15. from pathlib import Path, PurePosixPath
  16. from backend.app.core.config import settings
  17. from backend.app.core.tasks import spawn_background_task
  18. from backend.app.services.bambu_ftp import (
  19. DownloadCancelled,
  20. DownloadDeadlineExceeded,
  21. DownloadInsufficientSpace,
  22. DownloadLimitExceeded,
  23. download_file_async,
  24. )
  25. logger = logging.getLogger(__name__)
  26. VIDEO_SUFFIXES = (".mp4", ".avi", ".mkv")
  27. MAX_PRINTER_ZIP_BYTES = 10 * 1024**3
  28. PRINTER_ZIP_FREE_SPACE_RESERVE = 256 * 1024**2
  29. _STALE_BUNDLE_SECONDS = 60 * 60
  30. MAX_PRINTER_ZIP_PREPARE_SECONDS = 30 * 60
  31. MAX_OPEN_ARCHIVE_IPCAM_SECONDS = 24 * 60 * 60
  32. _BUNDLE_KEY_RE = re.compile(r"^[A-Za-z0-9_-]{1,200}$")
  33. _JOB_KEY_RE = re.compile(r"^[A-Za-z0-9_-]{20,200}$")
  34. _LOCAL_JOB_TASKS: dict[str, asyncio.Task] = {}
  35. _cleanup_task: asyncio.Task | None = None
  36. _CLEANUP_INTERVAL_SECONDS = 15 * 60
  37. class PrinterFileDownloadError(Exception):
  38. """The file is on the printer but could not be downloaded from it.
  39. Not FileNotFoundError: the routes answered that with 404 "File not found"
  40. for a file the browser had just listed.
  41. """
  42. class PrinterFilesZipTooLargeError(ValueError):
  43. """The selected printer files exceed the bounded ZIP staging limit."""
  44. class PrinterFilesZipInsufficientSpaceError(OSError):
  45. """The app data volume cannot safely stage the selected files."""
  46. @dataclass(frozen=True)
  47. class PrinterFilesZipResult:
  48. """Result of staging one printer ZIP."""
  49. path: Path
  50. requested: int
  51. successful: int
  52. failed_paths: tuple[str, ...]
  53. total_bytes: int
  54. @dataclass(frozen=True)
  55. class PrinterFilesJobStatus:
  56. """Serializable state for an asynchronous browser preparation job."""
  57. job_id: str
  58. printer_id: int
  59. state: str
  60. requested: int
  61. successful: int = 0
  62. failed: int = 0
  63. token: str | None = None
  64. filename: str | None = None
  65. message: str | None = None
  66. class _FileCancelSignal:
  67. """Cross-worker cancellation signal checked by the FTP callback thread."""
  68. def __init__(self, path: Path):
  69. self.path = path
  70. self._last_check = 0.0
  71. self._cached = False
  72. def is_set(self) -> bool:
  73. if self._cached:
  74. return True
  75. now = time.monotonic()
  76. if now - self._last_check >= 0.25:
  77. self._last_check = now
  78. self._cached = self.path.exists()
  79. return self._cached
  80. def _job_status_path(job_id: str) -> Path:
  81. if not _JOB_KEY_RE.fullmatch(job_id):
  82. raise ValueError("Invalid printer download job id")
  83. return _printer_zip_root() / f"job-{job_id}.json"
  84. def _job_cancel_path(job_id: str) -> Path:
  85. if not _JOB_KEY_RE.fullmatch(job_id):
  86. raise ValueError("Invalid printer download job id")
  87. return _printer_zip_root() / f"job-{job_id}.cancel"
  88. def _write_job_status(status: PrinterFilesJobStatus) -> None:
  89. """Atomically publish job state for polling from any app worker."""
  90. path = _job_status_path(status.job_id)
  91. temp_path = path.with_suffix(".tmp")
  92. temp_path.write_text(json.dumps(status.__dict__, separators=(",", ":")), encoding="utf-8")
  93. temp_path.replace(path)
  94. def _read_job_status(job_id: str) -> PrinterFilesJobStatus | None:
  95. try:
  96. data = json.loads(_job_status_path(job_id).read_text(encoding="utf-8"))
  97. return PrinterFilesJobStatus(**data)
  98. except (FileNotFoundError, OSError, ValueError, TypeError, json.JSONDecodeError):
  99. return None
  100. def _naive_utc(value: datetime | None) -> datetime | None:
  101. if value is None:
  102. return None
  103. if value.tzinfo is not None:
  104. return value.astimezone(timezone.utc).replace(tzinfo=None)
  105. return value
  106. def match_ipcam_chunks(
  107. files: list[dict],
  108. started_at: datetime | None,
  109. completed_at: datetime | None,
  110. *,
  111. now: datetime | None = None,
  112. ) -> list[dict]:
  113. """Return `/ipcam` chunks whose completion time overlaps a print.
  114. Bambu's `ipcam-record.*.mp4` files are fixed-size chunks. On the tested X1C
  115. and H2D firmware, their FTP mtime is the chunk completion time in the same
  116. UTC-naive basis used by archive timestamps. Some firmware reports FTP LIST
  117. mtimes in printer-local time instead; LIST carries no timezone with which
  118. to correct those values reliably. A ten-minute tail includes the final
  119. chunk, whose mtime lands after the print-complete event.
  120. """
  121. start = _naive_utc(started_at)
  122. if start is None:
  123. return []
  124. live_end = _naive_utc(now) or datetime.now(timezone.utc).replace(tzinfo=None)
  125. # A crash can leave an archive in ``printing`` indefinitely. Do not turn
  126. # that stale row into a window covering every chunk created since then.
  127. end = _naive_utc(completed_at) or min(live_end, start + timedelta(seconds=MAX_OPEN_ARCHIVE_IPCAM_SECONDS))
  128. lower = start - timedelta(minutes=1)
  129. upper = max(start, end) + timedelta(minutes=10)
  130. matches: list[dict] = []
  131. for file in files:
  132. name = str(file.get("name") or "")
  133. mtime = file.get("mtime")
  134. if file.get("is_directory") or not name.lower().startswith("ipcam-record."):
  135. continue
  136. if not name.lower().endswith(VIDEO_SUFFIXES) or not isinstance(mtime, datetime):
  137. continue
  138. timestamp = _naive_utc(mtime)
  139. if timestamp is not None and lower <= timestamp <= upper:
  140. matches.append(file)
  141. matches.sort(key=lambda item: _naive_utc(item.get("mtime")) or datetime.min)
  142. return matches
  143. def _zip_arcname(remote_path: str, used: set[str]) -> str:
  144. """Return a safe, unique relative archive name for a printer path."""
  145. parts = [part for part in PurePosixPath(remote_path).parts if part not in ("/", "", ".", "..")]
  146. candidate = "/".join(parts) or "printer-file"
  147. stem = candidate
  148. suffix = ""
  149. if "." in PurePosixPath(candidate).name:
  150. suffix = "".join(PurePosixPath(candidate).suffixes)
  151. stem = candidate[: -len(suffix)] if suffix else candidate
  152. counter = 2
  153. while candidate in used:
  154. candidate = f"{stem}-{counter}{suffix}"
  155. counter += 1
  156. used.add(candidate)
  157. return candidate
  158. def _printer_zip_root() -> Path:
  159. """Return the dedicated staging root without doing event-loop I/O."""
  160. return settings.archive_dir / "temp" / "printer-file-downloads"
  161. def _ensure_printer_zip_root() -> Path:
  162. """Create and return the staging root on the persistent data volume."""
  163. root = _printer_zip_root()
  164. root.mkdir(parents=True, exist_ok=True)
  165. return root
  166. def _prune_stale_bundles(root: Path) -> None:
  167. """Remove abandoned bundles after token expiry, without touching archives."""
  168. cutoff = time.time() - _STALE_BUNDLE_SECONDS
  169. if not root.exists():
  170. return
  171. for child in root.iterdir():
  172. try:
  173. if child.is_dir() and child.stat().st_mtime < cutoff:
  174. shutil.rmtree(child, ignore_errors=True)
  175. elif child.is_file() and child.name.startswith("job-") and child.stat().st_mtime < cutoff:
  176. child.unlink(missing_ok=True)
  177. except OSError:
  178. continue
  179. async def prune_stale_printer_file_bundles() -> None:
  180. """Prune abandoned printer ZIPs without blocking the event loop."""
  181. root = await asyncio.to_thread(_ensure_printer_zip_root)
  182. await asyncio.to_thread(_prune_stale_bundles, root)
  183. async def _printer_download_cleanup_loop() -> None:
  184. while True:
  185. try:
  186. await asyncio.sleep(_CLEANUP_INTERVAL_SECONDS)
  187. await prune_stale_printer_file_bundles()
  188. except asyncio.CancelledError:
  189. break
  190. except Exception:
  191. logger.exception("Periodic printer-download cleanup failed")
  192. def start_printer_download_cleanup() -> None:
  193. global _cleanup_task
  194. if _cleanup_task is None:
  195. _cleanup_task = spawn_background_task(_printer_download_cleanup_loop(), name="printer-download-cleanup")
  196. async def stop_printer_download_cleanup() -> None:
  197. """Stop cleanup and cancel every in-process preparation before shutdown."""
  198. global _cleanup_task
  199. tasks: list[asyncio.Task] = []
  200. cleanup_task = _cleanup_task
  201. _cleanup_task = None
  202. if cleanup_task is not None:
  203. cleanup_task.cancel()
  204. tasks.append(cleanup_task)
  205. # Jobs can be inside an FTP worker thread. Publish the same cooperative
  206. # cancellation marker used by the DELETE endpoint before cancelling the
  207. # asyncio wrapper, then await every wrapper so no executor work is left
  208. # behind when the application event loop closes.
  209. for job_id, task in list(_LOCAL_JOB_TASKS.items()):
  210. if not task.done():
  211. await asyncio.to_thread(_job_cancel_path(job_id).touch)
  212. task.cancel()
  213. tasks.append(task)
  214. if tasks:
  215. await asyncio.gather(*tasks, return_exceptions=True)
  216. _LOCAL_JOB_TASKS.clear()
  217. def printer_files_zip_path(printer_id: int, token: str) -> Path | None:
  218. """Resolve the staged ZIP for a resource-bound browser token."""
  219. bundle_key = f"{printer_id}-{token}"
  220. if not _BUNDLE_KEY_RE.fullmatch(bundle_key):
  221. return None
  222. return _printer_zip_root() / bundle_key / "printer-files.zip"
  223. def bind_printer_files_zip_to_token(
  224. result: PrinterFilesZipResult,
  225. printer_id: int,
  226. token: str,
  227. ) -> PrinterFilesZipResult:
  228. """Move a prepared bundle to the path derived from its persisted token."""
  229. target = printer_files_zip_path(printer_id, token)
  230. if target is None:
  231. raise ValueError("Invalid printer ZIP token")
  232. result.path.parent.rename(target.parent)
  233. return replace(result, path=target)
  234. def _check_initial_space(root: Path, sizes: dict[str, int]) -> None:
  235. # These sizes are client-reported hints used only for an early rejection,
  236. # so this is a courtesy, not the bound. The real one is enforced per write
  237. # and per FTP callback below, against actual bytes and the live free space,
  238. # which is the only thing that can hold when several preparations run at
  239. # once -- and they do: nothing serializes them. Two concurrent jobs that
  240. # both pass here stop independently at the reserve, and the one that gets
  241. # there second fails with a message saying so.
  242. expected_total = sum(sizes.values())
  243. if expected_total > MAX_PRINTER_ZIP_BYTES:
  244. raise PrinterFilesZipTooLargeError(
  245. f"Selected files total {expected_total} bytes; the limit is {MAX_PRINTER_ZIP_BYTES} bytes"
  246. )
  247. largest_file = max(sizes.values(), default=0)
  248. # In the worst case the ZIP is as large as the inputs while the largest
  249. # source is still staged beside it. Keep a reserve for the database/logs.
  250. required = expected_total + largest_file + PRINTER_ZIP_FREE_SPACE_RESERVE
  251. free = shutil.disk_usage(root).free
  252. if free < required:
  253. raise PrinterFilesZipInsufficientSpaceError(
  254. f"The app data volume needs {required} bytes free to stage this selection; {free} bytes are available"
  255. )
  256. async def build_printer_files_zip(
  257. printer,
  258. paths: list[str],
  259. sizes: dict[str, int],
  260. *,
  261. bundle_key: str | None = None,
  262. preserve_paths: bool = True,
  263. allow_empty: bool = False,
  264. cancel_signal: _FileCancelSignal | None = None,
  265. progress_callback: Callable[[int, int], Awaitable[None]] | None = None,
  266. ) -> PrinterFilesZipResult:
  267. """Download printer files one at a time into a disk-backed ZIP.
  268. The previous implementation held every source file and the final ZIP in
  269. memory. Continuous `/ipcam` chunks are commonly ~250 MB each, so selecting
  270. only a few could exhaust both server and browser memory.
  271. """
  272. root = await asyncio.to_thread(_ensure_printer_zip_root)
  273. await asyncio.to_thread(_prune_stale_bundles, root)
  274. await asyncio.to_thread(_check_initial_space, root, sizes)
  275. bundle_dir: Path | None = None
  276. try:
  277. if bundle_key is None:
  278. bundle_dir = Path(await asyncio.to_thread(tempfile.mkdtemp, prefix="bundle-", dir=root))
  279. else:
  280. if not _BUNDLE_KEY_RE.fullmatch(bundle_key):
  281. raise ValueError("Invalid printer ZIP bundle key")
  282. bundle_dir = root / bundle_key
  283. await asyncio.to_thread(bundle_dir.mkdir, mode=0o700)
  284. zip_path = bundle_dir / "printer-files.zip"
  285. successful = 0
  286. total_bytes = 0
  287. failed_paths: list[str] = []
  288. used_names: set[str] = set()
  289. archive = await asyncio.to_thread(zipfile.ZipFile, zip_path, "w", allowZip64=True)
  290. try:
  291. for index, remote_path in enumerate(paths):
  292. if cancel_signal is not None and cancel_signal.is_set():
  293. raise asyncio.CancelledError
  294. if not isinstance(remote_path, str) or not remote_path.startswith("/") or "\x00" in remote_path:
  295. logger.warning("Skipping invalid printer file path: %r", remote_path)
  296. failed_paths.append(remote_path)
  297. continue
  298. staged_path = bundle_dir / f"download-{index}"
  299. try:
  300. expected_size = sizes.get(remote_path)
  301. if expected_size is not None:
  302. free = (await asyncio.to_thread(shutil.disk_usage, root)).free
  303. if free < expected_size + PRINTER_ZIP_FREE_SPACE_RESERVE:
  304. raise PrinterFilesZipInsufficientSpaceError(
  305. "The app data volume lacks space for the next selected file"
  306. )
  307. downloaded = await download_file_async(
  308. printer.ip_address,
  309. printer.access_code,
  310. remote_path,
  311. staged_path,
  312. timeout=600,
  313. socket_timeout=60,
  314. # The job's budget, not print start's 300 s: a 250 MB
  315. # /ipcam chunk at a few hundred KB/s needs ~15 minutes,
  316. # and a dead link still ends at the socket timeout.
  317. max_total_timeout=MAX_PRINTER_ZIP_PREPARE_SECONDS,
  318. printer_model=printer.model,
  319. expected_size=expected_size,
  320. max_bytes=MAX_PRINTER_ZIP_BYTES - total_bytes,
  321. cancel_event=cancel_signal,
  322. min_free_bytes=PRINTER_ZIP_FREE_SPACE_RESERVE,
  323. # Outside the per-printer download gate (#2957), in both
  324. # directions. A selection of ~250 MB /ipcam chunks holds
  325. # the printer for as long as it legitimately takes, and
  326. # nothing else should be made to wait that out; equally,
  327. # each file here must not stall behind a thumbnail.
  328. serialize=False,
  329. )
  330. if not downloaded:
  331. failed_paths.append(remote_path)
  332. continue
  333. # Deliberately no second size comparison here. The transfer
  334. # was already checked against the printer's own SIZE, which
  335. # download_to_file treats as the authority precisely because
  336. # it beats a hint the browser round-tripped; re-judging the
  337. # result against that hint would overrule the better number
  338. # with the worse one. The hint goes stale in exactly the case
  339. # this feature exists for -- an /ipcam chunk or a timelapse
  340. # still being written when the listing was taken -- and a
  341. # complete file would then be dropped as "truncated".
  342. file_size = (await asyncio.to_thread(staged_path.stat)).st_size
  343. if total_bytes + file_size > MAX_PRINTER_ZIP_BYTES:
  344. raise PrinterFilesZipTooLargeError(
  345. f"Downloaded files exceed the {MAX_PRINTER_ZIP_BYTES}-byte limit"
  346. )
  347. free = (await asyncio.to_thread(shutil.disk_usage, root)).free
  348. if free < file_size + PRINTER_ZIP_FREE_SPACE_RESERVE:
  349. raise PrinterFilesZipInsufficientSpaceError(
  350. "The app data volume ran out of safe staging space while building the ZIP"
  351. )
  352. compression = (
  353. zipfile.ZIP_STORED if remote_path.lower().endswith(VIDEO_SUFFIXES) else zipfile.ZIP_DEFLATED
  354. )
  355. arc_source = remote_path if preserve_paths else PurePosixPath(remote_path).name
  356. await asyncio.to_thread(
  357. archive.write,
  358. staged_path,
  359. _zip_arcname(arc_source, used_names),
  360. compress_type=compression,
  361. )
  362. successful += 1
  363. total_bytes += file_size
  364. except DownloadLimitExceeded as exc:
  365. raise PrinterFilesZipTooLargeError(
  366. f"Downloaded files exceed the {MAX_PRINTER_ZIP_BYTES}-byte limit"
  367. ) from exc
  368. except DownloadInsufficientSpace as exc:
  369. raise PrinterFilesZipInsufficientSpaceError(
  370. "The app data volume ran out of safe staging space during transfer"
  371. ) from exc
  372. except DownloadCancelled as exc:
  373. raise asyncio.CancelledError from exc
  374. except (PrinterFilesZipTooLargeError, PrinterFilesZipInsufficientSpaceError):
  375. raise
  376. except Exception as exc:
  377. logger.warning("Failed to add %s to printer ZIP: %s", remote_path, exc)
  378. failed_paths.append(remote_path)
  379. finally:
  380. await asyncio.to_thread(staged_path.unlink, missing_ok=True)
  381. if progress_callback is not None:
  382. await progress_callback(successful, len(failed_paths))
  383. finally:
  384. await asyncio.shield(asyncio.to_thread(archive.close))
  385. except BaseException:
  386. if bundle_dir is not None:
  387. await asyncio.shield(asyncio.to_thread(shutil.rmtree, bundle_dir, ignore_errors=True))
  388. raise
  389. if successful == 0 and not allow_empty:
  390. await asyncio.to_thread(shutil.rmtree, bundle_dir, ignore_errors=True)
  391. raise FileNotFoundError("No files could be downloaded")
  392. return PrinterFilesZipResult(
  393. path=zip_path,
  394. requested=len(paths),
  395. successful=successful,
  396. failed_paths=tuple(failed_paths),
  397. total_bytes=total_bytes,
  398. )
  399. def printer_file_path(printer_id: int, token: str) -> Path | None:
  400. """Resolve a prepared native single-file download."""
  401. bundle_key = f"{printer_id}-{token}"
  402. if not _BUNDLE_KEY_RE.fullmatch(bundle_key):
  403. return None
  404. return _printer_zip_root() / bundle_key / "printer-file"
  405. def bind_printer_file_to_token(result: PrinterFilesZipResult, printer_id: int, token: str) -> PrinterFilesZipResult:
  406. target = printer_file_path(printer_id, token)
  407. if target is None:
  408. raise ValueError("Invalid printer file token")
  409. result.path.parent.rename(target.parent)
  410. return replace(result, path=target)
  411. async def build_printer_file(
  412. printer,
  413. remote_path: str,
  414. expected_size: int | None,
  415. *,
  416. bundle_key: str,
  417. cancel_signal: _FileCancelSignal | None = None,
  418. ) -> PrinterFilesZipResult:
  419. """Stage one printer file on disk for a browser-native download.
  420. Also the read path for the 3MF preview in the file browser, which is why
  421. nothing here waits on a shared lock: a preview must not queue behind
  422. somebody else's ten-gigabyte selection for as long as that takes.
  423. """
  424. if not remote_path.startswith("/") or "\x00" in remote_path:
  425. raise FileNotFoundError("Invalid printer file path")
  426. root = await asyncio.to_thread(_ensure_printer_zip_root)
  427. size_hints = {remote_path: expected_size} if expected_size is not None else {}
  428. await asyncio.to_thread(_check_initial_space, root, size_hints)
  429. bundle_dir = root / bundle_key
  430. try:
  431. await asyncio.to_thread(bundle_dir.mkdir, mode=0o700)
  432. local_path = bundle_dir / "printer-file"
  433. downloaded = await download_file_async(
  434. printer.ip_address,
  435. printer.access_code,
  436. remote_path,
  437. local_path,
  438. timeout=600,
  439. socket_timeout=60,
  440. max_total_timeout=MAX_PRINTER_ZIP_PREPARE_SECONDS,
  441. printer_model=printer.model,
  442. expected_size=expected_size,
  443. max_bytes=MAX_PRINTER_ZIP_BYTES,
  444. cancel_event=cancel_signal,
  445. min_free_bytes=PRINTER_ZIP_FREE_SPACE_RESERVE,
  446. # The lock-free promise in this function's docstring, kept: a preview
  447. # must not queue behind somebody else's selection (#2957).
  448. serialize=False,
  449. )
  450. if not downloaded:
  451. raise PrinterFileDownloadError("The selected printer file could not be downloaded")
  452. file_size = (await asyncio.to_thread(local_path.stat)).st_size
  453. return PrinterFilesZipResult(
  454. path=local_path,
  455. requested=1,
  456. successful=1,
  457. failed_paths=(),
  458. total_bytes=file_size,
  459. )
  460. except DownloadDeadlineExceeded as exc:
  461. await asyncio.shield(asyncio.to_thread(shutil.rmtree, bundle_dir, ignore_errors=True))
  462. raise TimeoutError(f"Downloading {remote_path} from the printer took longer than its size allows") from exc
  463. except PrinterFileDownloadError:
  464. await asyncio.shield(asyncio.to_thread(shutil.rmtree, bundle_dir, ignore_errors=True))
  465. raise
  466. except DownloadLimitExceeded as exc:
  467. await asyncio.shield(asyncio.to_thread(shutil.rmtree, bundle_dir, ignore_errors=True))
  468. raise PrinterFilesZipTooLargeError(f"Downloaded file exceeds the {MAX_PRINTER_ZIP_BYTES}-byte limit") from exc
  469. except DownloadInsufficientSpace as exc:
  470. await asyncio.shield(asyncio.to_thread(shutil.rmtree, bundle_dir, ignore_errors=True))
  471. raise PrinterFilesZipInsufficientSpaceError(
  472. "The app data volume ran out of safe staging space during transfer"
  473. ) from exc
  474. except DownloadCancelled as exc:
  475. await asyncio.shield(asyncio.to_thread(shutil.rmtree, bundle_dir, ignore_errors=True))
  476. raise asyncio.CancelledError from exc
  477. except BaseException:
  478. await asyncio.shield(asyncio.to_thread(shutil.rmtree, bundle_dir, ignore_errors=True))
  479. raise
  480. async def _run_printer_files_job(
  481. printer,
  482. job_id: str,
  483. paths: list[str],
  484. sizes: dict[str, int],
  485. filename: str,
  486. as_zip: bool,
  487. ) -> None:
  488. from backend.app.core.auth import create_slicer_download_token
  489. cancel_signal = _FileCancelSignal(_job_cancel_path(job_id))
  490. status = PrinterFilesJobStatus(job_id, printer.id, "preparing", len(paths), filename=filename)
  491. await asyncio.to_thread(_write_job_status, status)
  492. async def report_progress(successful: int, failed: int) -> None:
  493. await asyncio.to_thread(
  494. _write_job_status,
  495. PrinterFilesJobStatus(
  496. job_id,
  497. printer.id,
  498. "preparing",
  499. len(paths),
  500. successful=successful,
  501. failed=failed,
  502. filename=filename,
  503. ),
  504. )
  505. try:
  506. async with asyncio.timeout(MAX_PRINTER_ZIP_PREPARE_SECONDS):
  507. if as_zip:
  508. result = await build_printer_files_zip(
  509. printer,
  510. paths,
  511. sizes,
  512. bundle_key=f"job-{job_id}",
  513. cancel_signal=cancel_signal,
  514. progress_callback=report_progress,
  515. )
  516. else:
  517. result = await build_printer_file(
  518. printer,
  519. paths[0],
  520. sizes.get(paths[0]),
  521. bundle_key=f"job-{job_id}",
  522. cancel_signal=cancel_signal,
  523. )
  524. if cancel_signal.is_set():
  525. await asyncio.to_thread(remove_printer_files_zip, result.path)
  526. raise asyncio.CancelledError
  527. token = await create_slicer_download_token("printer-files", printer.id)
  528. if as_zip:
  529. result = await asyncio.to_thread(bind_printer_files_zip_to_token, result, printer.id, token)
  530. else:
  531. result = await asyncio.to_thread(bind_printer_file_to_token, result, printer.id, token)
  532. await asyncio.to_thread(
  533. _write_job_status,
  534. PrinterFilesJobStatus(
  535. job_id,
  536. printer.id,
  537. "ready",
  538. len(paths),
  539. successful=result.successful,
  540. failed=len(result.failed_paths),
  541. token=token,
  542. filename=filename,
  543. ),
  544. )
  545. except asyncio.CancelledError:
  546. await asyncio.shield(
  547. asyncio.to_thread(
  548. _write_job_status,
  549. PrinterFilesJobStatus(job_id, printer.id, "cancelled", len(paths), filename=filename),
  550. )
  551. )
  552. except PrinterFilesZipTooLargeError as exc:
  553. await asyncio.to_thread(
  554. _write_job_status,
  555. PrinterFilesJobStatus(job_id, printer.id, "failed", len(paths), filename=filename, message=str(exc)),
  556. )
  557. except PrinterFilesZipInsufficientSpaceError as exc:
  558. await asyncio.to_thread(
  559. _write_job_status,
  560. PrinterFilesJobStatus(job_id, printer.id, "failed", len(paths), filename=filename, message=str(exc)),
  561. )
  562. except TimeoutError as exc:
  563. await asyncio.to_thread(
  564. _write_job_status,
  565. PrinterFilesJobStatus(
  566. job_id,
  567. printer.id,
  568. "failed",
  569. len(paths),
  570. filename=filename,
  571. message=str(exc) or "Printer download preparation exceeded the 30-minute limit",
  572. ),
  573. )
  574. except PrinterFileDownloadError as exc:
  575. await asyncio.to_thread(
  576. _write_job_status,
  577. PrinterFilesJobStatus(job_id, printer.id, "failed", len(paths), filename=filename, message=str(exc)),
  578. )
  579. except FileNotFoundError as exc:
  580. await asyncio.to_thread(
  581. _write_job_status,
  582. PrinterFilesJobStatus(job_id, printer.id, "failed", len(paths), filename=filename, message=str(exc)),
  583. )
  584. except Exception:
  585. logger.exception("Printer download job %s failed", job_id)
  586. await asyncio.to_thread(
  587. _write_job_status,
  588. PrinterFilesJobStatus(
  589. job_id,
  590. printer.id,
  591. "failed",
  592. len(paths),
  593. filename=filename,
  594. message="Printer download preparation failed",
  595. ),
  596. )
  597. finally:
  598. await asyncio.to_thread(_job_cancel_path(job_id).unlink, missing_ok=True)
  599. async def start_printer_files_job(
  600. printer,
  601. paths: list[str],
  602. sizes: dict[str, int],
  603. filename: str,
  604. *,
  605. as_zip: bool,
  606. ) -> PrinterFilesJobStatus:
  607. """Start a bounded background preparation and return immediately."""
  608. if not paths:
  609. raise ValueError("No files specified")
  610. root = await asyncio.to_thread(_ensure_printer_zip_root)
  611. await asyncio.to_thread(_prune_stale_bundles, root)
  612. await asyncio.to_thread(_check_initial_space, root, sizes)
  613. job_id = secrets.token_urlsafe(24)
  614. status = PrinterFilesJobStatus(job_id, printer.id, "queued", len(paths), filename=filename)
  615. await asyncio.to_thread(_write_job_status, status)
  616. task = spawn_background_task(
  617. _run_printer_files_job(printer, job_id, paths, sizes, filename, as_zip),
  618. name=f"printer-download-{printer.id}-{job_id}",
  619. )
  620. _LOCAL_JOB_TASKS[job_id] = task
  621. task.add_done_callback(lambda _task: _LOCAL_JOB_TASKS.pop(job_id, None))
  622. return status
  623. async def get_printer_files_job(job_id: str, printer_id: int) -> PrinterFilesJobStatus | None:
  624. status = await asyncio.to_thread(_read_job_status, job_id)
  625. if status is None or status.printer_id != printer_id:
  626. return None
  627. return status
  628. async def cancel_printer_files_job(job_id: str, printer_id: int) -> bool:
  629. status = await get_printer_files_job(job_id, printer_id)
  630. if status is None:
  631. return False
  632. await asyncio.to_thread(_job_cancel_path(job_id).touch)
  633. task = _LOCAL_JOB_TASKS.get(job_id)
  634. if task is not None and not task.done():
  635. task.cancel()
  636. if status.state == "ready" and status.token:
  637. zip_path = printer_files_zip_path(printer_id, status.token)
  638. prepared = (
  639. zip_path
  640. if zip_path is not None and await asyncio.to_thread(zip_path.is_file)
  641. else printer_file_path(printer_id, status.token)
  642. )
  643. if prepared is not None:
  644. await asyncio.to_thread(remove_printer_files_zip, prepared)
  645. await asyncio.to_thread(
  646. _write_job_status,
  647. replace(status, state="cancelled", token=None),
  648. )
  649. return True
  650. def remove_printer_files_zip(zip_path: Path) -> None:
  651. """Remove a completed download bundle after FileResponse finishes."""
  652. shutil.rmtree(zip_path.parent, ignore_errors=True)