archive_merge.py 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670
  1. """Merge duplicate archives of one file into one (archive redesign P2).
  2. An archive is a file; each print of it is a run. Before runs, every dispatch
  3. of a library or project file archived it anew, so a file printed plate by
  4. plate has several archives. They are one when they hold the same file
  5. (content hash), have the same owner and are in the same project.
  6. Merging keeps the oldest archive and moves everything the others hold to it:
  7. their runs (one is written for an archive printed before runs were logged),
  8. queue items, batches, BOM items, pipeline runs, library folders, spool usage
  9. and pending Telegram verdicts, and their photos and videos, into its folder.
  10. The others stay, soft-deleted, each with an alias to the one it went into: the
  11. money booked against them (wallet charges, budget reservations) is never
  12. rewritten, and their ids keep finding the archive (``resolve_alias``).
  13. An archive at work is left out: printing, queued, tracked by Spoolman, being
  14. dispatched or converted in this process (services/archive_work), sliced by a
  15. pipeline, edited, or printed within the last half hour (its timelapse and
  16. finish photo may still be on their way, by its own id). So is a group whose
  17. archives carry different source or design files: only one can stay.
  18. """
  19. from __future__ import annotations
  20. import asyncio
  21. import hashlib
  22. import logging
  23. import os
  24. import shutil
  25. from dataclasses import dataclass, field
  26. from datetime import datetime, timedelta, timezone
  27. from pathlib import Path
  28. from sqlalchemy import func, or_, select, update
  29. from sqlalchemy.ext.asyncio import AsyncSession
  30. from backend.app.core.config import settings
  31. from backend.app.core.printer_scope import PrinterScope
  32. from backend.app.models.active_print_spoolman import ActivePrintSpoolman
  33. from backend.app.models.archive import PrintArchive
  34. from backend.app.models.archive_alias import ArchiveAlias
  35. from backend.app.models.print_log import INCLUDE_OPEN_RUNS, RUN_OPEN, PrintLogEntry
  36. from backend.app.models.print_queue import PrintQueueItem
  37. from backend.app.services import archive_work
  38. from backend.app.services.print_cost_estimate import refresh_plate_estimates
  39. from backend.app.utils.archive_paths import archive_dir, archive_photos_dir, find_archive_photo
  40. from backend.app.utils.local_time import utcnow_naive
  41. from backend.app.utils.safe_path import PathTraversalError, safe_join_under
  42. logger = logging.getLogger(__name__)
  43. # Long enough for the completion's timelapse scan and finish-photo upgrade,
  44. # which look for their print by its archive's id.
  45. _JUST_PRINTED = timedelta(minutes=30)
  46. # What an archive printed before runs were logged can say of its print.
  47. _ENDED = frozenset({"completed", "failed", "cancelled", "aborted", "stopped"})
  48. _PIPELINE_ACTIVE = ("queued", "slicing", "dispatching", "in_progress")
  49. # The file's own extras, of which the archive keeps one each.
  50. _FILE_EXTRAS = ("source_3mf_path", "f3d_path")
  51. SKIPPED_BUSY = "busy"
  52. SKIPPED_CONFLICTING_FILES = "conflicting_files"
  53. SKIPPED_FAILED = "failed"
  54. @dataclass
  55. class MergeGroup:
  56. survivor: PrintArchive
  57. merged: list[PrintArchive]
  58. run_counts: dict[int, int] = field(default_factory=dict)
  59. skipped: str | None = None
  60. async def resolve_alias(db: AsyncSession, archive_id: int) -> int | None:
  61. """The archive a merged one went into, or None."""
  62. return await db.scalar(select(ArchiveAlias.archive_id).where(ArchiveAlias.merged_archive_id == archive_id))
  63. def _key(archive: PrintArchive) -> tuple:
  64. return (archive.content_hash, archive.created_by_id, archive.project_id)
  65. def _group_keys(visible: list):
  66. return (
  67. select(PrintArchive.content_hash, PrintArchive.created_by_id, PrintArchive.project_id)
  68. .where(PrintArchive.content_hash.is_not(None), PrintArchive.deleted_at.is_(None), *visible)
  69. .group_by(PrintArchive.content_hash, PrintArchive.created_by_id, PrintArchive.project_id)
  70. .having(func.count(PrintArchive.id) > 1)
  71. )
  72. async def count_groups(db: AsyncSession, visible: list) -> int:
  73. """How many duplicate groups there are, whether or not one can merge now:
  74. cheap, for offering the merge (no files read, no work checked)."""
  75. return await db.scalar(select(func.count()).select_from(_group_keys(visible).subquery())) or 0
  76. async def find_groups(db: AsyncSession, visible: list) -> list[MergeGroup]:
  77. """The duplicate groups among the archives ``visible`` (WHERE clauses)
  78. admits, oldest first, each with why it is left out (``skipped``)."""
  79. hashes = {row.content_hash for row in (await db.execute(_group_keys(visible))).all()}
  80. if not hashes:
  81. return []
  82. archives = (
  83. await db.scalars(
  84. select(PrintArchive)
  85. .where(PrintArchive.content_hash.in_(hashes), PrintArchive.deleted_at.is_(None), *visible)
  86. .order_by(PrintArchive.id)
  87. )
  88. ).all()
  89. by_key: dict[tuple, list[PrintArchive]] = {}
  90. for archive in archives:
  91. by_key.setdefault(_key(archive), []).append(archive)
  92. members = [a for group in by_key.values() if len(group) > 1 for a in group]
  93. busy = await _busy(db, [a.id for a in members])
  94. run_counts = dict(
  95. (
  96. await db.execute(
  97. select(PrintLogEntry.archive_id, func.count(PrintLogEntry.id))
  98. .where(PrintLogEntry.archive_id.in_([a.id for a in members]))
  99. .group_by(PrintLogEntry.archive_id)
  100. )
  101. ).all()
  102. )
  103. member_runs: dict[int, list[PrintLogEntry]] = {}
  104. for run in await db.scalars(
  105. select(PrintLogEntry)
  106. .where(PrintLogEntry.archive_id.in_([a.id for a in members]))
  107. .execution_options(**{INCLUDE_OPEN_RUNS: True})
  108. ):
  109. if run.archive_id is not None:
  110. member_runs.setdefault(run.archive_id, []).append(run)
  111. groups = []
  112. for group in by_key.values():
  113. if len(group) < 2:
  114. continue
  115. skipped = None
  116. if any(a.id in busy for a in group):
  117. skipped = SKIPPED_BUSY
  118. elif await asyncio.to_thread(_conflicting_extras, group) or _videos_conflict(group, member_runs):
  119. skipped = SKIPPED_CONFLICTING_FILES
  120. groups.append(
  121. MergeGroup(
  122. survivor=group[0],
  123. merged=group[1:],
  124. run_counts={a.id: run_counts.get(a.id, 0) for a in group},
  125. skipped=skipped,
  126. )
  127. )
  128. return sorted(groups, key=lambda g: g.survivor.id)
  129. def _digest(relpath: str) -> str | None:
  130. path = _under_base(relpath)
  131. if path is None or not path.is_file():
  132. return None
  133. digest = hashlib.sha256()
  134. with path.open("rb") as handle:
  135. for chunk in iter(lambda: handle.read(1 << 20), b""):
  136. digest.update(chunk)
  137. return digest.hexdigest()
  138. def _conflicting_extras(group: list[PrintArchive]) -> bool:
  139. """Whether two archives of the group carry different source 3MFs or
  140. design files: the archive keeps one, and the other is the user's."""
  141. for attr in _FILE_EXTRAS:
  142. digests = {_digest(getattr(a, attr)) for a in group if getattr(a, attr)}
  143. if len(digests - {None}) > 1:
  144. return True
  145. return False
  146. async def _busy(db: AsyncSession, archive_ids: list[int]) -> set[int]:
  147. """The archives at work (see the module docstring)."""
  148. from backend.app.main import _active_prints, _expected_prints
  149. from backend.app.models.pipeline_run import PipelineRun
  150. from backend.app.services import timelapse_jobs
  151. if not archive_ids:
  152. return set()
  153. recent = utcnow_naive() - _JUST_PRINTED
  154. queries = [
  155. select(PrintArchive.id).where(
  156. PrintArchive.id.in_(archive_ids),
  157. or_(PrintArchive.status == "printing", PrintArchive.completed_at >= recent),
  158. ),
  159. select(PrintLogEntry.archive_id)
  160. .where(
  161. PrintLogEntry.archive_id.in_(archive_ids),
  162. or_(PrintLogEntry.status == RUN_OPEN, PrintLogEntry.completed_at >= recent),
  163. )
  164. .execution_options(**{INCLUDE_OPEN_RUNS: True}),
  165. select(PrintQueueItem.archive_id).where(
  166. PrintQueueItem.archive_id.in_(archive_ids), PrintQueueItem.status.in_(("pending", "printing"))
  167. ),
  168. select(ActivePrintSpoolman.archive_id).where(ActivePrintSpoolman.archive_id.in_(archive_ids)),
  169. select(PipelineRun.source_archive_id).where(
  170. PipelineRun.source_archive_id.in_(archive_ids), PipelineRun.status.in_(_PIPELINE_ACTIVE)
  171. ),
  172. ]
  173. busy: set[int | None] = set()
  174. for query in queries:
  175. busy |= set((await db.scalars(query)).all())
  176. # Held in memory by id: a print waiting for its start or end, work in
  177. # this process (a dispatch, a conversion), an editor job.
  178. in_memory = set(_active_prints.values()) | set(_expected_prints.values())
  179. for archive_id in archive_ids:
  180. job = timelapse_jobs.get_job(archive_id)
  181. if (
  182. archive_id in in_memory
  183. or archive_work.is_held(archive_id)
  184. or (job is not None and job.status == timelapse_jobs.RUNNING)
  185. ):
  186. busy.add(archive_id)
  187. return {archive_id for archive_id in busy if archive_id is not None}
  188. # Rows that point at an archive and move with it. Wallet charges and budget
  189. # reservations stay on the merged one (soft-deleted, not removed): they are
  190. # money already booked, and the old charges are unique per archive.
  191. def _moved_references():
  192. from backend.app.models.library import LibraryFolder
  193. from backend.app.models.notification import TelegramPendingVerdict
  194. from backend.app.models.pending_upload import PendingUpload
  195. from backend.app.models.pipeline_run import PipelineRun
  196. from backend.app.models.print_batch import PrintBatch
  197. from backend.app.models.project_bom import ProjectBOMItem
  198. from backend.app.models.spool_usage_history import SpoolUsageHistory
  199. return (
  200. (PrintLogEntry, "archive_id"),
  201. (PrintQueueItem, "archive_id"),
  202. (PrintBatch, "archive_id"),
  203. (ProjectBOMItem, "archive_id"),
  204. (SpoolUsageHistory, "archive_id"),
  205. (TelegramPendingVerdict, "archive_id"),
  206. (LibraryFolder, "archive_id"),
  207. (PendingUpload, "archived_id"),
  208. (PipelineRun, "source_archive_id"),
  209. )
  210. class _Files:
  211. """Files placed in the survivor's folder, removed again if the merge fails."""
  212. def __init__(self) -> None:
  213. self.created: list[Path] = []
  214. def place(self, source: Path | None, target_dir: Path) -> Path | None:
  215. """``source`` in ``target_dir`` under a free name (a link where it can)."""
  216. if source is None or not source.is_file():
  217. return None
  218. target_dir.mkdir(parents=True, exist_ok=True)
  219. target = target_dir / source.name
  220. n = 2
  221. while target.exists():
  222. if target.resolve() == source.resolve():
  223. return target
  224. target = target_dir / f"{source.stem}_{n}{source.suffix}"
  225. n += 1
  226. try:
  227. os.link(source, target)
  228. except OSError:
  229. try:
  230. shutil.copy2(source, target)
  231. except BaseException:
  232. target.unlink(missing_ok=True) # a partial copy
  233. raise
  234. self.created.append(target)
  235. return target
  236. def undo(self) -> None:
  237. for path in self.created:
  238. path.unlink(missing_ok=True)
  239. def _relpath(path: Path) -> str:
  240. return str(path.relative_to(settings.base_dir))
  241. def _under_base(relpath: str | None) -> Path | None:
  242. if not relpath:
  243. return None
  244. try:
  245. return safe_join_under(settings.base_dir, relpath, http=False)
  246. except PathTraversalError:
  247. return None
  248. def _missing(relpath: str | None) -> bool:
  249. path = _under_base(relpath)
  250. return path is None or not path.is_file()
  251. def _gets_a_run(archive: PrintArchive, runs: list[PrintLogEntry]) -> bool:
  252. """Whether the merge writes a run for the archive's print (``_run_from_archive``)."""
  253. return not runs and archive.status in _ENDED and bool(archive.started_at or archive.completed_at)
  254. def _orphan_video(archive: PrintArchive, runs: list[PrintLogEntry]) -> bool:
  255. """Whether the archive's video is its alone, with no print to take it."""
  256. from backend.app.services.print_log import owns_media
  257. if not archive.timelapse_path or _gets_a_run(archive, runs):
  258. return False
  259. if any(r.timelapse_path == archive.timelapse_path for r in runs):
  260. return False
  261. return not any(r.timelapse_path is None and not owns_media(r) for r in runs)
  262. def _videos_conflict(archives: list[PrintArchive], runs: dict[int, list[PrintLogEntry]]) -> bool:
  263. """Whether a video an archive alone holds (``_orphan_video``) is not the
  264. group's one video: the archive shows one, so the other would be lost."""
  265. orphans = {a.timelapse_path for a in archives if _orphan_video(a, runs.get(a.id, []))}
  266. if not orphans:
  267. return False
  268. videos = {r.timelapse_path for own in runs.values() for r in own if r.timelapse_path} | {
  269. a.timelapse_path for a in archives if a.timelapse_path
  270. }
  271. return len(orphans) > 1 or bool(videos - orphans)
  272. def _latest(runs, *, where) -> PrintLogEntry | None:
  273. """The latest by start of ``runs`` that ``where`` admits."""
  274. return max(
  275. (r for r in runs if where(r)),
  276. key=lambda r: (r.started_at is not None, r.started_at or datetime.min, r.id or 0),
  277. default=None,
  278. )
  279. def _run_from_archive(archive: PrintArchive, survivor_id: int, printer_name: str | None) -> PrintLogEntry:
  280. """The run of a print logged before runs were: what its archive says of it."""
  281. duration = None
  282. if archive.started_at and archive.completed_at and archive.completed_at > archive.started_at:
  283. duration = int((archive.completed_at - archive.started_at).total_seconds())
  284. return PrintLogEntry(
  285. archive_id=survivor_id,
  286. print_name=archive.print_name,
  287. plate_id=archive.plate_id,
  288. printer_id=archive.printer_id,
  289. printer_name=printer_name,
  290. status=archive.status,
  291. started_at=archive.started_at,
  292. completed_at=archive.completed_at,
  293. duration_seconds=duration,
  294. filament_type=archive.filament_type,
  295. filament_color=archive.filament_color,
  296. filament_used_grams=archive.filament_used_grams,
  297. cost=archive.cost,
  298. energy_kwh=archive.energy_kwh,
  299. energy_cost=archive.energy_cost,
  300. wear_cost=archive.wear_cost,
  301. failure_reason=archive.failure_reason,
  302. user_verdict=archive.user_verdict,
  303. user_verdict_source=archive.user_verdict_source,
  304. user_verdict_at=archive.user_verdict_at,
  305. thumbnail_path=archive.thumbnail_path,
  306. created_by_id=archive.created_by_id,
  307. quantity=archive.quantity,
  308. created_at=archive.completed_at or archive.created_at,
  309. )
  310. def _joined_tags(*tag_lists: str | None) -> str | None:
  311. seen: list[str] = []
  312. for tags in tag_lists:
  313. for tag in (tags or "").split(","):
  314. if (tag := tag.strip()) and tag.lower() not in {t.lower() for t in seen}:
  315. seen.append(tag)
  316. return ", ".join(seen) or None
  317. def _joined_notes(*notes: str | None) -> str | None:
  318. kept: list[str] = []
  319. for note in notes:
  320. if note and note.strip() and note.strip() not in kept:
  321. kept.append(note.strip())
  322. return "\n\n".join(kept) or None
  323. @dataclass
  324. class _Plan:
  325. """The files of one merged archive and where they go."""
  326. archive: PrintArchive
  327. photos: dict[str, Path] = field(default_factory=dict) # name -> source
  328. videos: dict[str, Path] = field(default_factory=dict) # relpath -> source
  329. extras: dict[str, Path] = field(default_factory=dict) # attribute -> source
  330. async def merge_group(db: AsyncSession, survivor_id: int, merged_ids: list[int]) -> str | None:
  331. """Merge ``merged_ids`` into ``survivor_id`` and commit. Returns why it
  332. was left out instead (a ``SKIPPED_*``), or None once merged."""
  333. with archive_work.merging([survivor_id, *merged_ids]) as marked:
  334. if not marked:
  335. return SKIPPED_BUSY
  336. return await _merge_marked(db, survivor_id, merged_ids)
  337. async def _merge_marked(db: AsyncSession, survivor_id: int, merged_ids: list[int]) -> str | None:
  338. """``_merge``, rolled back (with the files it placed) on any failure."""
  339. files = _Files()
  340. try:
  341. reason, leftover = await _merge(db, survivor_id, merged_ids, files)
  342. if reason is not None:
  343. # Left out before anything was written.
  344. return reason
  345. except BaseException:
  346. await db.rollback()
  347. await asyncio.to_thread(files.undo)
  348. raise
  349. for directory in leftover:
  350. await asyncio.to_thread(shutil.rmtree, directory, True)
  351. logger.info("Merged archives %s into archive %s", merged_ids, survivor_id)
  352. return None
  353. async def _merge(
  354. db: AsyncSession, survivor_id: int, merged_ids: list[int], files: _Files
  355. ) -> tuple[str | None, list[Path]]:
  356. """Merge and commit; returns why it was left out instead, or the folders
  357. to remove now that it has committed."""
  358. from backend.app.models.printer import Printer
  359. from backend.app.services.archive import ArchiveService
  360. from backend.app.services.print_log import mirror_latest_run, newest_run, owns_media
  361. ids = [survivor_id, *merged_ids]
  362. # As they are now, not as the preview or the group search found them.
  363. rows = {
  364. a.id: a
  365. for a in await db.scalars(
  366. select(PrintArchive).where(PrintArchive.id.in_(ids)).execution_options(populate_existing=True)
  367. )
  368. }
  369. if len(rows) != len(ids) or any(a.deleted_at is not None for a in rows.values()):
  370. return SKIPPED_BUSY, []
  371. survivor, merged = rows[survivor_id], [rows[i] for i in merged_ids]
  372. if len({_key(a) for a in rows.values()}) != 1 or await _busy(db, ids):
  373. return SKIPPED_BUSY, []
  374. if await asyncio.to_thread(_conflicting_extras, [survivor, *merged]):
  375. return SKIPPED_CONFLICTING_FILES, []
  376. runs: dict[int, list[PrintLogEntry]] = {a.id: [] for a in rows.values()}
  377. for run in await db.scalars(
  378. select(PrintLogEntry)
  379. .where(PrintLogEntry.archive_id.in_(ids))
  380. .execution_options(populate_existing=True, **{INCLUDE_OPEN_RUNS: True})
  381. ):
  382. if run.archive_id is not None:
  383. runs[run.archive_id].append(run)
  384. # A video on an archive alone that no print of it can take (attached by
  385. # hand to one never printed): the archive keeps it only if it is the one
  386. # video of the group, or it would be lost or hide another.
  387. orphans = {a.id: a.timelapse_path for a in rows.values() if _orphan_video(a, runs[a.id])}
  388. if _videos_conflict(list(rows.values()), runs):
  389. return SKIPPED_CONFLICTING_FILES, []
  390. # Plan, reading only: the files to place in the survivor's folder. The
  391. # survivor's own files where it lost them, from the first archive that
  392. # still has them.
  393. plans = []
  394. taken: set[str] = set()
  395. for archive in merged:
  396. plan = _Plan(archive)
  397. for name in archive.photos or []:
  398. if (source := find_archive_photo(archive, name)) is not None:
  399. plan.photos[name] = source
  400. for relpath in {r.timelapse_path for r in runs[archive.id] if r.timelapse_path} | (
  401. {archive.timelapse_path} if archive.timelapse_path else set()
  402. ):
  403. if (source := _under_base(relpath)) is not None:
  404. plan.videos[relpath] = source
  405. for attr in (*_FILE_EXTRAS, "file_path", "thumbnail_path"):
  406. source = _under_base(getattr(archive, attr))
  407. if attr not in taken and _missing(getattr(survivor, attr)) and source is not None and source.is_file():
  408. plan.extras[attr] = source
  409. taken.add(attr)
  410. plans.append(plan)
  411. # Place them off the event loop, before any write: SQLite would hold its
  412. # write lock through a copy that cannot link.
  413. survivor_dir = archive_dir(survivor)
  414. photos_dir = archive_photos_dir(survivor)
  415. def place(source: Path, target_dir: Path) -> Path | None:
  416. return files.place(source, target_dir)
  417. def place_all() -> list[tuple[dict[str, str], dict[str, str], dict[str, str]]]:
  418. placed = []
  419. for plan in plans:
  420. photos = {n: p.name for n, s in plan.photos.items() if (p := place(s, photos_dir)) is not None}
  421. videos = {r: _relpath(p) for r, s in plan.videos.items() if (p := place(s, survivor_dir)) is not None}
  422. extras = {a: _relpath(p) for a, s in plan.extras.items() if (p := place(s, survivor_dir)) is not None}
  423. placed.append((photos, videos, extras))
  424. return placed
  425. placed = await asyncio.to_thread(place_all)
  426. printer_ids = {a.printer_id for a in rows.values() if a.printer_id is not None}
  427. printer_names = dict(
  428. (await db.execute(select(Printer.id, Printer.name).where(Printer.id.in_(printer_ids)))).tuples().all()
  429. )
  430. for archive in [survivor, *merged]:
  431. own = runs[archive.id]
  432. # A print logged before runs keeps its only record on its archive.
  433. if _gets_a_run(archive, own):
  434. name = printer_names.get(archive.printer_id) if archive.printer_id is not None else None
  435. run = _run_from_archive(archive, archive.id, name)
  436. run.photos = list(archive.photos or [])
  437. run.timelapse_path = archive.timelapse_path
  438. db.add(run)
  439. own.append(run)
  440. # Its video, kept on the archive alone by prints logged before
  441. # runs kept their own: its latest such print's.
  442. elif (
  443. archive.timelapse_path
  444. and archive.id not in orphans
  445. and not any(r.timelapse_path == archive.timelapse_path for r in own)
  446. ):
  447. if (heir := _latest(own, where=lambda r: r.timelapse_path is None and not owns_media(r))) is not None:
  448. heir.timelapse_path = archive.timelapse_path
  449. # A one-tap link of the archive's question still answers it: on
  450. # its latest print, where questions are asked now.
  451. if archive.confirm_token and archive.confirm_token_used_at is None and archive.user_verdict is None:
  452. token, archive.confirm_token = archive.confirm_token, None
  453. await db.flush()
  454. if (heir := _latest(own, where=lambda r: r.confirm_token is None)) is not None:
  455. heir.confirm_token = token
  456. heir.confirm_requested = True
  457. survivor_photos = list(survivor.photos or [])
  458. for plan, (photos, videos, extras) in zip(plans, placed, strict=True):
  459. archive = plan.archive
  460. for name in archive.photos or []:
  461. if name in photos and photos[name] not in survivor_photos:
  462. survivor_photos.append(photos[name])
  463. for attr, relpath in extras.items():
  464. setattr(survivor, attr, relpath)
  465. if "file_path" in extras:
  466. # The same file (one hash), but the survivor's was missing: its
  467. # plates are read from the one it takes over (#3375).
  468. refresh_plate_estimates(survivor)
  469. for run in runs[archive.id]:
  470. if run.photos:
  471. run.photos = [photos[p] for p in run.photos if p in photos]
  472. if run.timelapse_path:
  473. run.timelapse_path = videos.get(run.timelapse_path)
  474. # Its thumbnail was its archive's copy of the file's.
  475. if run.thumbnail_path == archive.thumbnail_path:
  476. run.thumbnail_path = survivor.thumbnail_path
  477. run.archive_id = survivor.id
  478. # The file's description: one note, every tag, a favourite if any.
  479. survivor.notes = _joined_notes(survivor.notes, archive.notes)
  480. survivor.tags = _joined_tags(survivor.tags, archive.tags)
  481. survivor.is_favorite = bool(survivor.is_favorite or archive.is_favorite)
  482. for attr in ("library_file_id", "external_url", "makerworld_url", "designer"):
  483. if getattr(survivor, attr) is None and getattr(archive, attr) is not None:
  484. setattr(survivor, attr, getattr(archive, attr))
  485. await db.flush()
  486. for archive in merged:
  487. for model, column in _moved_references():
  488. await db.execute(
  489. update(model)
  490. .where(getattr(model, column) == archive.id)
  491. .values({column: survivor.id})
  492. .execution_options(**{INCLUDE_OPEN_RUNS: True})
  493. )
  494. await db.execute(
  495. update(ArchiveAlias).where(ArchiveAlias.archive_id == archive.id).values(archive_id=survivor.id)
  496. )
  497. db.add(ArchiveAlias(merged_archive_id=archive.id, archive_id=survivor.id))
  498. survivor.photos = survivor_photos or None
  499. await db.flush()
  500. # The archive shows its latest run, its video too.
  501. await mirror_latest_run(db, survivor.id)
  502. newest = await newest_run(db, survivor.id)
  503. if orphans:
  504. # The group's one video, which no print holds.
  505. [(holder_id, orphan)] = orphans.items()
  506. if holder_id != survivor.id and orphan is not None:
  507. orphan = placed[merged_ids.index(holder_id)][1].get(orphan)
  508. survivor.timelapse_path = orphan
  509. survivor.timelapse_plate_id = None
  510. elif newest is not None and owns_media(newest):
  511. survivor.timelapse_path = newest.timelapse_path
  512. survivor.timelapse_plate_id = newest.plate_id if newest.timelapse_path else None
  513. # Their folders go once committed, unless another archive still uses
  514. # one (archived in the same second under the same name).
  515. service = ArchiveService(db)
  516. leftover = []
  517. for archive in merged:
  518. for directory in service._resolve_archive_dirs_for_delete(archive):
  519. if not await _folder_in_use(db, directory, merged_ids=merged_ids):
  520. leftover.append(directory)
  521. for archive in merged:
  522. archive.deleted_at = datetime.now(timezone.utc)
  523. # Its files are the survivor's now: a later purge of the row must
  524. # find nothing of theirs to remove.
  525. archive.file_path = ""
  526. archive.thumbnail_path = archive.timelapse_path = None
  527. archive.source_3mf_path = archive.f3d_path = None
  528. archive.photos = None
  529. await db.commit()
  530. return None, leftover
  531. async def _folder_in_use(db: AsyncSession, directory: Path, *, merged_ids: list[int]) -> bool:
  532. """Whether ``directory`` holds a file of an archive other than the merged
  533. ones (the survivor's included). Compared as paths, not as text: stored
  534. paths use the platform's separator."""
  535. folder = directory.resolve()
  536. rows = await db.execute(
  537. select(PrintArchive.file_path, PrintArchive.source_3mf_path, PrintArchive.f3d_path).where(
  538. PrintArchive.id.not_in(merged_ids),
  539. PrintArchive.deleted_at.is_(None),
  540. or_(
  541. PrintArchive.file_path.contains(directory.name, autoescape=True),
  542. PrintArchive.source_3mf_path.contains(directory.name, autoescape=True),
  543. PrintArchive.f3d_path.contains(directory.name, autoescape=True),
  544. ),
  545. )
  546. )
  547. for row in rows.all():
  548. for relpath in row:
  549. if (path := _under_base(relpath)) is not None and path.resolve().is_relative_to(folder):
  550. return True
  551. return False
  552. async def merge(db: AsyncSession, survivor_ids: list[int], visible: list) -> tuple[list[dict], list[dict]]:
  553. """Merge the groups of ``survivor_ids`` as they stand now (each its
  554. group's oldest archive among those ``visible`` admits). Returns what was
  555. merged and what was left out, and why; one group failing leaves the
  556. others as they went."""
  557. # Read up front: a group that fails rolls back, which expires the rows.
  558. groups = {g.survivor.id: (g.skipped, [a.id for a in g.merged]) for g in await find_groups(db, visible)}
  559. done, skipped = [], []
  560. for survivor_id in survivor_ids:
  561. if survivor_id not in groups:
  562. skipped.append({"survivor_id": survivor_id, "reason": "not_a_group"})
  563. continue
  564. reason, merged_ids = groups[survivor_id]
  565. if reason is not None:
  566. skipped.append({"survivor_id": survivor_id, "reason": reason})
  567. continue
  568. try:
  569. reason = await merge_group(db, survivor_id, merged_ids)
  570. except Exception:
  571. logger.exception("Merging archives %s into archive %s failed", merged_ids, survivor_id)
  572. reason = SKIPPED_FAILED
  573. if reason is None:
  574. done.append({"survivor_id": survivor_id, "merged_ids": merged_ids})
  575. else:
  576. skipped.append({"survivor_id": survivor_id, "reason": reason})
  577. return done, skipped
  578. def merge_scope_clauses(printer_scope: PrinterScope) -> list:
  579. """Only archives every print of which the caller's printers made: a
  580. merge changes and removes whole archives (archive redesign, step 8)."""
  581. from backend.app.core.row_visibility import whole_archive_scope_clause
  582. clause = whole_archive_scope_clause(printer_scope)
  583. return [clause] if clause is not None else []