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

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()