m038
ae093881fa
- write-back: call upsert_tag inside _apply_tag's try so a tag-create failure is caught per-asset and apply_all no longer aborts mid-batch (was P1) - write-back: surface _pipeline/processed write failures in the result instead of discarding them (was reported as success) - ui: add title to the high confidence badge so approve-high-confidence's pre-action count is non-zero; confirm() before single-cluster apply - core: Store context manager; close DB connection even if a route raises; guard split_cluster against a first-member boundary (empty cluster); add writeback_log lookup index - ingest: split thumbnail download/write error handling and clean up the .tmp file on a write failure - tests: upsert/processed write-failure regression tests + split-guard test
63 lines
2.5 KiB
Python
63 lines
2.5 KiB
Python
from photoflow.immich import pipeline
|
|
|
|
APPLYABLE = ("approved", "non_trip", "skipped")
|
|
|
|
|
|
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:
|
|
store.log_writeback(a, action, tag, f"error:{e}")
|
|
return [], [(a, str(e)) for a in todo]
|
|
for a in todo:
|
|
store.log_writeback(a, action, tag, "ok")
|
|
return todo, []
|
|
|
|
|
|
def apply_cluster(client, store, cluster_id) -> dict:
|
|
c = store.get_cluster(cluster_id)
|
|
if c is None or c.status not in APPLYABLE:
|
|
return {"cluster_id": cluster_id, "status": c.status if c else None,
|
|
"succeeded": [], "failed": []}
|
|
|
|
included = [a.immich_id for a, m in store.cluster_members(cluster_id) if m.included]
|
|
succeeded, failed = [], []
|
|
|
|
if c.status == "approved":
|
|
tag = c.decided_name or c.suggested_name
|
|
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)
|
|
succeeded += ok
|
|
failed += fail
|
|
# 'skipped': no content/non-trip tag, only processed below.
|
|
|
|
failed_ids = {i for i, _ in failed}
|
|
proc_targets = [a for a in included if a not in failed_ids]
|
|
if proc_targets:
|
|
# 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)
|
|
|
|
return {"cluster_id": cluster_id, "status": c.status,
|
|
"succeeded": succeeded, "failed": failed}
|
|
|
|
|
|
def apply_all(client, store) -> list:
|
|
return [apply_cluster(client, store, c.id)
|
|
for c in store.all_clusters() if c.status in APPLYABLE]
|