diff --git a/apps/trip-cluster/app/ingest.py b/apps/trip-cluster/app/ingest.py index d207ad0..87e2ec6 100644 --- a/apps/trip-cluster/app/ingest.py +++ b/apps/trip-cluster/app/ingest.py @@ -45,14 +45,23 @@ def run_ingest(client, store, thumbs_dir, *, date_from=None, date_to=None, if need_thumb: try: data = client.download_thumbnail(a["id"]) - # Write to a temp path and os.replace into place so a failure - # never leaves a 0-byte .jpg behind. - tmp_path = f"{thumb_path}.tmp" + except Exception: + log.exception("Thumbnail download failed for asset %s", a["id"]) + continue + # Write to a temp path and os.replace into place so a failure + # never leaves a 0-byte .jpg behind; clean up the temp file + # if the write itself fails (e.g. disk full). + tmp_path = f"{thumb_path}.tmp" + try: with open(tmp_path, "wb") as f: f.write(data) os.replace(tmp_path, thumb_path) except Exception: - log.exception("Thumbnail download failed for asset %s", a["id"]) + log.exception("Thumbnail write failed for asset %s", a["id"]) + try: + os.unlink(tmp_path) + except OSError: + pass continue store.upsert_asset(Asset( diff --git a/apps/trip-cluster/app/routes/review.py b/apps/trip-cluster/app/routes/review.py index 0d26f18..6ee5805 100644 --- a/apps/trip-cluster/app/routes/review.py +++ b/apps/trip-cluster/app/routes/review.py @@ -113,9 +113,8 @@ def apply_one(cid): # Lazy import: app.writeback is owned by a later task and may be absent # at app-startup; importing here keeps the blueprint importable regardless. from app.writeback import apply_cluster - s = _store() - res = apply_cluster(_client(), s, cid) - s.close() + with _store() as s: # close even if the Immich write-back raises + res = apply_cluster(_client(), s, cid) return jsonify(res) @@ -123,7 +122,6 @@ def apply_one(cid): def apply_everything(): # Lazy import: see apply_one above. from app.writeback import apply_all - s = _store() - results = apply_all(_client(), s) - s.close() + with _store() as s: # close even if the Immich write-back raises + results = apply_all(_client(), s) return jsonify({"results": results}) diff --git a/apps/trip-cluster/app/static/app.js b/apps/trip-cluster/app/static/app.js index 7745dc0..9cd3235 100644 --- a/apps/trip-cluster/app/static/app.js +++ b/apps/trip-cluster/app/static/app.js @@ -136,6 +136,8 @@ function clusterReview() { location.reload(); }, async apply() { + // Write-back needs explicit confirmation (mirrors applyAll/merge). + if (!confirm('Apply this cluster’s decision to Immich?')) return; const r = await this.post(`/cluster/${this.selected}/apply`, {}); alert(`Applied ${r.succeeded.length}, failed ${r.failed.length}.`); }, diff --git a/apps/trip-cluster/app/writeback.py b/apps/trip-cluster/app/writeback.py index 7416583..663e94e 100644 --- a/apps/trip-cluster/app/writeback.py +++ b/apps/trip-cluster/app/writeback.py @@ -3,11 +3,15 @@ from photoflow.immich import pipeline APPLYABLE = ("approved", "non_trip", "skipped") -def _apply_tag(client, store, asset_ids, action, tag, tag_id): +def _apply_tag(client, store, asset_ids, action, tag): todo = [a for a in asset_ids if not store.already_applied(a, action, tag)] if not todo: return [], [] try: + # upsert_tag is part of the write: a failure here must be caught and + # recorded like a tag_assets failure, not propagate out of apply_cluster + # (which would abort the whole apply_all batch and skip later clusters). + tag_id = client.upsert_tag(tag) client.tag_assets(tag_id, todo) except Exception as e: # noqa: BLE001 — recorded, surfaced for a in todo: @@ -29,12 +33,11 @@ def apply_cluster(client, store, cluster_id) -> dict: if c.status == "approved": tag = c.decided_name or c.suggested_name - ok, fail = _apply_tag(client, store, included, "trip", tag, client.upsert_tag(tag)) + ok, fail = _apply_tag(client, store, included, "trip", tag) succeeded += ok failed += fail elif c.status == "non_trip": - ok, fail = _apply_tag(client, store, included, "non-trip", pipeline.NON_TRIP, - client.upsert_tag(pipeline.NON_TRIP)) + ok, fail = _apply_tag(client, store, included, "non-trip", pipeline.NON_TRIP) succeeded += ok failed += fail # 'skipped': no content/non-trip tag, only processed below. @@ -42,8 +45,10 @@ def apply_cluster(client, store, cluster_id) -> dict: failed_ids = {i for i, _ in failed} proc_targets = [a for a in included if a not in failed_ids] if proc_targets: - _apply_tag(client, store, proc_targets, "processed", pipeline.PROCESSED, - client.upsert_tag(pipeline.PROCESSED)) + # Surface processed-marker failures too — a discarded return here makes a + # failed _pipeline/processed write look like success in the CLI/UI summary. + _, proc_fail = _apply_tag(client, store, proc_targets, "processed", pipeline.PROCESSED) + failed += proc_fail for a in proc_targets: if store.already_applied(a, "processed", pipeline.PROCESSED): store.mark_processed(a) diff --git a/apps/trip-cluster/tests/test_writeback.py b/apps/trip-cluster/tests/test_writeback.py index 9deb8d6..7233c8e 100644 --- a/apps/trip-cluster/tests/test_writeback.py +++ b/apps/trip-cluster/tests/test_writeback.py @@ -8,9 +8,12 @@ class FakeImmich: def __init__(self): self.tagged = [] self.fail_tag_id = None + self.fail_upsert = None # tag name whose upsert_tag should raise self._ids = {} def upsert_tag(self, name): + if self.fail_upsert is not None and name == self.fail_upsert: + raise RuntimeError("upsert boom") self._ids.setdefault(name, f"id:{name}") return self._ids[name] @@ -111,3 +114,35 @@ def test_apply_skipped(tmp_path): assert s.get_asset("a").processed is True assert s.get_asset("b").processed is True s.close() + + +def test_upsert_tag_failure_is_caught_and_does_not_abort_batch(tmp_path): + # A failing upsert_tag must be recorded as a per-asset failure (retryable), + # not propagate out of apply_cluster and abort apply_all (review finding R1). + s = _store(tmp_path) + _approved(s, name="Venice") + _approved(s, name="Rome") # second cluster must still be applied + client = FakeImmich() + client.fail_upsert = "Venice" # first cluster's trip-tag upsert fails + results = apply_all(client, s) + assert len(results) == 2 # batch was not aborted by the first failure + venice = next(r for r in results if r["cluster_id"] == 1) + assert venice["succeeded"] == [] and sorted(i for i, _ in venice["failed"]) == ["a", "b"] + assert s.already_applied("a", "trip", "Venice") is False # retryable + rome = next(r for r in results if r["cluster_id"] == 2) + assert sorted(rome["succeeded"]) == ["a", "b"] + s.close() + + +def test_processed_write_failure_is_surfaced_in_result(tmp_path): + # Trip tag succeeds but the _pipeline/processed write fails: the failure must + # appear in result["failed"] and the asset must NOT be marked processed. + s = _store(tmp_path) + cid = _approved(s, name="Venice") + client = FakeImmich() + client.fail_upsert = pipeline.PROCESSED + res = apply_cluster(client, s, cid) + assert sorted(res["succeeded"]) == ["a", "b"] # trip tag still applied + assert sorted(i for i, _ in res["failed"]) == ["a", "b"] # processed surfaced + assert s.get_asset("a").processed is False + s.close() diff --git a/shared/photoflow/core/store.py b/shared/photoflow/core/store.py index 32c67ab..9fafa52 100644 --- a/shared/photoflow/core/store.py +++ b/shared/photoflow/core/store.py @@ -67,6 +67,8 @@ CREATE TABLE IF NOT EXISTS writeback_log ( CREATE INDEX IF NOT EXISTS idx_assets_taken_at ON assets(taken_at); CREATE INDEX IF NOT EXISTS idx_members_cluster ON cluster_members(cluster_id); CREATE INDEX IF NOT EXISTS idx_members_asset ON cluster_members(immich_id); +CREATE INDEX IF NOT EXISTS idx_writeback_lookup + ON writeback_log(immich_id, action, tag, result); """ @@ -109,6 +111,12 @@ class Store: self._conn.close() self._conn = None + def __enter__(self) -> "Store": + return self + + def __exit__(self, *exc) -> None: + self.close() + def upsert_asset(self, a: Asset) -> None: has_gps = 1 if (a.gps_lat is not None and a.gps_lon is not None) else 0 self.conn.execute( @@ -284,6 +292,10 @@ class Store: if boundary_immich_id not in ids: raise ValueError(f"boundary {boundary_immich_id!r} not in cluster") idx = ids.index(boundary_immich_id) + if idx == 0: + raise ValueError( + f"boundary {boundary_immich_id!r} is the first member; " + "split would leave an empty cluster") base = self.get_cluster(cluster_id) left, right = pairs[:idx], pairs[idx:] diff --git a/shared/photoflow/ui/templates/macros.html b/shared/photoflow/ui/templates/macros.html index 76430b0..872e6be 100644 --- a/shared/photoflow/ui/templates/macros.html +++ b/shared/photoflow/ui/templates/macros.html @@ -10,7 +10,7 @@ {% elif confidence < 0.75 %} med {% else %} - high + high {% endif %} {% endmacro %} diff --git a/shared/tests/test_store_clusters.py b/shared/tests/test_store_clusters.py index d8bddd8..bee53ad 100644 --- a/shared/tests/test_store_clusters.py +++ b/shared/tests/test_store_clusters.py @@ -97,3 +97,14 @@ def test_writeback_log_idempotency(tmp_path): s.log_writeback("b", "trip", "Venice", "error:boom") assert s.already_applied("b", "trip", "Venice") is False # only ok counts s.close() + + +def test_split_cluster_rejects_first_member_boundary(tmp_path): + import pytest + s = _seed(tmp_path) + cid = s.insert_cluster(Cluster(start_at="2019-06-01", end_at="2019-06-03", + suggested_name="Trip"), _members(["a", "b", "c"])) + with pytest.raises(ValueError): + s.split_cluster(cid, "a") # boundary == first member -> empty left + assert s.get_cluster(cid).status != "split" # original untouched on rejection + s.close()