archive_merge.py 28 KB

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