feat(trip-cluster): cluster orchestration (seed derivation + coverage) + CLI

apps/trip-cluster/app/cli.py

@@ -31,6 +31,16 @@ def cmd_ingest(deps, *, date_from, date_to, tag, subset, full) -> int:
     return 0
 
 
+def cmd_cluster(deps, *, gap_factor) -> int:
+    cfg = deps["config"]
+    store = _store(cfg)
+    from app.cluster_run import run_cluster
+    res = run_cluster(store, gap_factor=gap_factor)
+    store.close()
+    print(f"Built {res['clusters']} candidate cluster(s). Run 'categorize serve' to review.")
+    return 0
+
+
 def cmd_serve(deps) -> int:
     from app import create_app
     create_app(deps["config"]).run(host="0.0.0.0", port=8084)
@@ -67,7 +77,9 @@ def main(argv=None) -> int:
     if args.command == "ingest":
         return cmd_ingest(deps, date_from=args.date_from, date_to=args.date_to,
                           tag=args.tag, subset=args.subset, full=args.full)
-    # cluster / apply are wired in later tasks.
+    if args.command == "cluster":
+        return cmd_cluster(deps, gap_factor=args.gap_factor)
+    # apply is wired in a later task.
     print(f"Command '{args.command}' is not implemented yet.")
     return 1
 

apps/trip-cluster/app/cluster_run.py

@@ -0,0 +1,57 @@
+from collections import defaultdict
+
+from photoflow.core.models import Cluster
+from photoflow.immich import pipeline
+from app.clustering import cluster_assets, _epoch
+from app.coverage import coverage_members
+
+# A tag seeds only if its photos dominate the time window they span: among all
+# assets whose taken_at falls in [min, max] of the tag's photos, at least this
+# fraction must carry the tag. A name/event tag sprinkled among many untagged
+# photos in the same window fails this gate.
+SEED_DOMINANCE_MIN = 0.5
+
+
+def _seed_tags(assets, tags_by_asset, max_span_days: int,
+               dominance_min: float = SEED_DOMINANCE_MIN) -> set:
+    epochs = {a.immich_id: _epoch(a.taken_at) for a in assets}
+    tagged = defaultdict(list)  # tag -> epochs of its assets
+    for a in assets:
+        for t in tags_by_asset.get(a.immich_id, []):
+            if not pipeline.is_pipeline_tag(t):
+                tagged[t].append(epochs[a.immich_id])
+
+    seeds = set()
+    for tag, ts in tagged.items():
+        if len(ts) < 2:
+            continue
+        lo, hi = min(ts), max(ts)
+        # Loose safety cap only — reject absurdly long spans, but keep genuine
+        # long trips (this library has one > 60 days), so the default is high.
+        if hi - lo > max_span_days * 86400:
+            continue
+        # Dominance / temporal contiguity gate.
+        in_window = sum(1 for e in epochs.values() if lo <= e <= hi)
+        if in_window and len(ts) / in_window >= dominance_min:
+            seeds.add(tag)
+    return seeds
+
+
+def run_cluster(store, *, gap_factor: float = 6.0, seed_max_span_days: int = 400) -> dict:
+    assets = store.all_assets(include_processed=False)
+    tags_by_asset = {a.immich_id: store.asset_tags(a.immich_id) for a in assets}
+    seed_tags = _seed_tags(assets, tags_by_asset, seed_max_span_days)
+    candidates = cluster_assets(assets, tags_by_asset, seed_tags, gap_factor=gap_factor)
+    assets_by_id = {a.immich_id: a for a in assets}
+
+    store.clear_clusters()
+    n = 0
+    for cand in candidates:
+        members = coverage_members(cand, assets_by_id)
+        store.insert_cluster(Cluster(
+            start_at=cand.start_at, end_at=cand.end_at,
+            count=sum(1 for m in members if m.included),
+            suggested_name=cand.suggested_name, confidence=cand.confidence,
+            kind_guess=cand.kind_guess, status="pending"), members)
+        n += 1
+    return {"clusters": n}