From 9ec65124284b0b88018b8d8a12185eab3da80dce Mon Sep 17 00:00:00 2001 From: mischa Date: Sat, 27 Jun 2026 17:19:06 +0200 Subject: [PATCH] =?UTF-8?q?feat(core):=20cluster=20data-access=20=E2=80=94?= =?UTF-8?q?=20attention=20sort,=20neighbors,=20split/merge,=20writeback=20?= =?UTF-8?q?log?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- shared/photoflow/core/store.py | 158 +++++++++++++++++++++++++++- shared/tests/test_store_clusters.py | 99 +++++++++++++++++ 2 files changed, 256 insertions(+), 1 deletion(-) create mode 100644 shared/tests/test_store_clusters.py diff --git a/shared/photoflow/core/store.py b/shared/photoflow/core/store.py index 3cf889c..32c67ab 100644 --- a/shared/photoflow/core/store.py +++ b/shared/photoflow/core/store.py @@ -1,6 +1,6 @@ import sqlite3 -from photoflow.core.models import Asset, Tag +from photoflow.core.models import Asset, Cluster, ClusterMember, Tag SCHEMA_VERSION = 1 @@ -179,3 +179,159 @@ class Store: def all_tags(self) -> list: return [Tag(name=r["name"], immich_tag_id=r["immich_tag_id"], count=r["count"]) for r in self.conn.execute("SELECT * FROM tags ORDER BY name")] + + def insert_cluster(self, c: Cluster, members: list) -> int: + cur = self.conn.execute( + """INSERT INTO clusters + (start_at, end_at, count, suggested_name, confidence, kind_guess, + status, decided_name, reviewed_at, notes) + VALUES (?,?,?,?,?,?,?,?,?,?)""", + (c.start_at, c.end_at, c.count or len(members), c.suggested_name, + c.confidence, c.kind_guess, c.status, c.decided_name, c.reviewed_at, c.notes)) + cid = cur.lastrowid + self.conn.executemany( + """INSERT OR REPLACE INTO cluster_members + (cluster_id, immich_id, member_confidence, is_outlier, included, flagged_coverage) + VALUES (?,?,?,?,?,?)""", + [(cid, m.immich_id, m.member_confidence, 1 if m.is_outlier else 0, + 1 if m.included else 0, 1 if m.flagged_coverage else 0) for m in members]) + self.conn.commit() + return cid + + def clear_clusters(self) -> None: + self.conn.execute("DELETE FROM cluster_members") + self.conn.execute("DELETE FROM clusters") + self.conn.commit() + + def _cluster_from_row(self, r) -> Cluster: + return Cluster( + id=r["id"], start_at=r["start_at"], end_at=r["end_at"], count=r["count"], + suggested_name=r["suggested_name"], confidence=r["confidence"], + kind_guess=r["kind_guess"], status=r["status"], decided_name=r["decided_name"], + reviewed_at=r["reviewed_at"], notes=r["notes"]) + + def get_cluster(self, cluster_id: int): + r = self.conn.execute("SELECT * FROM clusters WHERE id=?", (cluster_id,)).fetchone() + return self._cluster_from_row(r) if r else None + + def all_clusters(self) -> list: + return [self._cluster_from_row(r) + for r in self.conn.execute("SELECT * FROM clusters ORDER BY start_at, id")] + + def clusters_by_attention(self) -> list: + return [self._cluster_from_row(r) for r in self.conn.execute( + """SELECT * FROM clusters + ORDER BY CASE status WHEN 'pending' THEN 0 ELSE 1 END, + confidence ASC, + (julianday(end_at) - julianday(start_at)) DESC, + id""")] + + def cluster_members(self, cluster_id: int) -> list: + rows = self.conn.execute( + """SELECT a.*, m.member_confidence AS m_conf, m.is_outlier AS m_out, + m.included AS m_inc, m.flagged_coverage AS m_cov + FROM cluster_members m JOIN assets a ON a.immich_id = m.immich_id + WHERE m.cluster_id=? ORDER BY a.taken_at, a.immich_id""", (cluster_id,)) + out = [] + for r in rows: + asset = self._asset_from_row(r) + member = ClusterMember( + cluster_id=cluster_id, immich_id=r["immich_id"], + member_confidence=r["m_conf"], is_outlier=bool(r["m_out"]), + included=bool(r["m_inc"]), flagged_coverage=bool(r["m_cov"])) + out.append((asset, member)) + return out + + def chronological_neighbors(self, cluster_id: int): + ids = [r["id"] for r in self.conn.execute( + "SELECT id FROM clusters ORDER BY start_at, id")] + if cluster_id not in ids: + return (None, None) + i = ids.index(cluster_id) + prev_id = ids[i - 1] if i > 0 else None + next_id = ids[i + 1] if i < len(ids) - 1 else None + return (prev_id, next_id) + + def update_cluster(self, cluster_id: int, *, status=None, decided_name=None, + suggested_name=None, notes=None, reviewed_at=None) -> None: + sets, vals = [], [] + for col, val in [("status", status), ("decided_name", decided_name), + ("suggested_name", suggested_name), ("notes", notes), + ("reviewed_at", reviewed_at)]: + if val is not None: + sets.append(f"{col}=?") + vals.append(val) + if not sets: + return + vals.append(cluster_id) + self.conn.execute(f"UPDATE clusters SET {', '.join(sets)} WHERE id=?", vals) + self.conn.commit() + + def set_member_inclusion(self, cluster_id: int, immich_id: str, included: bool) -> None: + self.conn.execute( + "UPDATE cluster_members SET included=? WHERE cluster_id=? AND immich_id=?", + (1 if included else 0, cluster_id, immich_id)) + self.conn.commit() + + def _recompute_span(self, members: list) -> tuple: + pairs = members + starts = [a.taken_at for a, _ in pairs] + return (min(starts), max(starts)) if starts else ("", "") + + def split_cluster(self, cluster_id: int, boundary_immich_id: str): + pairs = self.cluster_members(cluster_id) + ids = [a.immich_id for a, _ in pairs] + if boundary_immich_id not in ids: + raise ValueError(f"boundary {boundary_immich_id!r} not in cluster") + idx = ids.index(boundary_immich_id) + base = self.get_cluster(cluster_id) + left, right = pairs[:idx], pairs[idx:] + + def _new(part, suffix): + ms = [ClusterMember(cluster_id=0, immich_id=a.immich_id, + member_confidence=m.member_confidence, + is_outlier=m.is_outlier, included=m.included, + flagged_coverage=m.flagged_coverage) for a, m in part] + start, end = self._recompute_span(part) + return self.insert_cluster(Cluster( + start_at=start, end_at=end, count=len(ms), + suggested_name=f"{base.suggested_name} ({suffix})", + confidence=base.confidence, kind_guess=base.kind_guess, + status="pending"), ms) + + id1, id2 = _new(left, 1), _new(right, 2) + self.update_cluster(cluster_id, status="split") + return (id1, id2) + + def merge_clusters(self, cluster_id_a: int, cluster_id_b: int): + pairs = self.cluster_members(cluster_id_a) + self.cluster_members(cluster_id_b) + pairs.sort(key=lambda p: (p[0].taken_at, p[0].immich_id)) + base = self.get_cluster(cluster_id_a) + ms = [ClusterMember(cluster_id=0, immich_id=a.immich_id, + member_confidence=m.member_confidence, is_outlier=m.is_outlier, + included=m.included, flagged_coverage=m.flagged_coverage) + for a, m in pairs] + start, end = self._recompute_span(pairs) + new = self.insert_cluster(Cluster( + start_at=start, end_at=end, count=len(ms), + suggested_name=base.suggested_name, confidence=base.confidence, + kind_guess=base.kind_guess, status="pending"), ms) + self.update_cluster(cluster_id_a, status="merged") + self.update_cluster(cluster_id_b, status="merged") + return new + + def log_writeback(self, immich_id: str, action: str, tag, result: str) -> None: + import datetime + self.conn.execute( + "INSERT INTO writeback_log(immich_id, action, tag, result, applied_at) " + "VALUES (?,?,?,?,?)", + (immich_id, action, tag, result, + datetime.datetime.now(datetime.timezone.utc).isoformat())) + self.conn.commit() + + def already_applied(self, immich_id: str, action: str, tag) -> bool: + row = self.conn.execute( + "SELECT 1 FROM writeback_log WHERE immich_id=? AND action=? " + "AND IFNULL(tag,'')=IFNULL(?, '') AND result='ok' LIMIT 1", + (immich_id, action, tag)).fetchone() + return row is not None diff --git a/shared/tests/test_store_clusters.py b/shared/tests/test_store_clusters.py new file mode 100644 index 0000000..d8bddd8 --- /dev/null +++ b/shared/tests/test_store_clusters.py @@ -0,0 +1,99 @@ +from photoflow.core import Store +from photoflow.core.models import Asset, Cluster, ClusterMember + + +def _seed(tmp_path): + s = Store(str(tmp_path / "t.db")).connect() + for i, t in [("a", "2019-06-01"), ("b", "2019-06-02"), ("c", "2019-06-03"), + ("d", "2019-07-01"), ("e", "2019-07-02")]: + s.upsert_asset(Asset(immich_id=i, taken_at=t)) + return s + + +def _members(ids): + return [ClusterMember(cluster_id=0, immich_id=i) for i in ids] + + +def test_insert_and_members_ordered(tmp_path): + s = _seed(tmp_path) + cid = s.insert_cluster(Cluster(start_at="2019-06-01", end_at="2019-06-03", + count=3, suggested_name="Trip 1", confidence=0.9), + _members(["c", "a", "b"])) + pairs = s.cluster_members(cid) + assert [a.immich_id for a, m in pairs] == ["a", "b", "c"] + assert s.get_cluster(cid).suggested_name == "Trip 1" + s.close() + + +def test_attention_sort(tmp_path): + s = _seed(tmp_path) + s.insert_cluster(Cluster(start_at="2019-06-01", end_at="2019-06-10", + confidence=0.9, status="pending"), _members(["a"])) + low = s.insert_cluster(Cluster(start_at="2019-07-01", end_at="2019-07-02", + confidence=0.2, status="pending"), _members(["d"])) + s.insert_cluster(Cluster(start_at="2019-06-02", end_at="2019-06-03", + confidence=0.1, status="approved"), _members(["b"])) + order = [c.id for c in s.clusters_by_attention()] + assert order[0] == low # lowest-confidence pending first + assert order[-1] != low # approved sinks to the bottom + s.close() + + +def test_chronological_neighbors(tmp_path): + s = _seed(tmp_path) + c1 = s.insert_cluster(Cluster(start_at="2019-06-01", end_at="2019-06-03"), _members(["a"])) + c2 = s.insert_cluster(Cluster(start_at="2019-07-01", end_at="2019-07-02"), _members(["d"])) + assert s.chronological_neighbors(c1) == (None, c2) + assert s.chronological_neighbors(c2) == (c1, None) + s.close() + + +def test_update_and_member_inclusion(tmp_path): + s = _seed(tmp_path) + cid = s.insert_cluster(Cluster(start_at="2019-06-01", end_at="2019-06-03"), + _members(["a", "b"])) + s.update_cluster(cid, status="approved", decided_name="Venice", reviewed_at="now") + c = s.get_cluster(cid) + assert c.status == "approved" and c.decided_name == "Venice" + s.set_member_inclusion(cid, "b", False) + inc = {m.immich_id: m.included for _, m in s.cluster_members(cid)} + assert inc == {"a": True, "b": False} + s.close() + + +def test_split_cluster(tmp_path): + 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"])) + id1, id2 = s.split_cluster(cid, "c") # boundary "c" begins the second cluster + assert s.get_cluster(cid).status == "split" + left = [a.immich_id for a, _ in s.cluster_members(id1)] + right = [a.immich_id for a, _ in s.cluster_members(id2)] + assert left == ["a", "b"] and right == ["c"] + assert s.get_cluster(id1).status == "pending" + s.close() + + +def test_merge_clusters(tmp_path): + s = _seed(tmp_path) + a = s.insert_cluster(Cluster(start_at="2019-06-01", end_at="2019-06-03", + suggested_name="A"), _members(["a", "b"])) + b = s.insert_cluster(Cluster(start_at="2019-07-01", end_at="2019-07-02", + suggested_name="B"), _members(["d", "e"])) + new = s.merge_clusters(a, b) + assert s.get_cluster(a).status == "merged" and s.get_cluster(b).status == "merged" + ids = [x.immich_id for x, _ in s.cluster_members(new)] + assert ids == ["a", "b", "d", "e"] + c = s.get_cluster(new) + assert c.start_at == "2019-06-01" and c.end_at == "2019-07-02" and c.status == "pending" + s.close() + + +def test_writeback_log_idempotency(tmp_path): + s = _seed(tmp_path) + assert s.already_applied("a", "trip", "Venice") is False + s.log_writeback("a", "trip", "Venice", "ok") + assert s.already_applied("a", "trip", "Venice") is True + s.log_writeback("b", "trip", "Venice", "error:boom") + assert s.already_applied("b", "trip", "Venice") is False # only ok counts + s.close()