Add Apprise notifications to quality-scan (self-hosted apprise server); dashboard ready

quality_scan.py

@@ -87,13 +87,15 @@ def _move(p: Path, dest_root: Path, src_root: Path):
 
 
 @flow(name="photo-quality-scan")
-def quality_scan(base_dir: str, move: bool = False):
+def quality_scan(base_dir: str, move: bool = False, notify: bool = True):
     """Audit image quality with CleanVision; classify into keep/review/delete."""
     audit = audit_folder(base_dir)
     print(f"Issue summary: {audit['summary']}")
     result = classify_and_sort(base_dir, audit["per_image"], move=move)
     print(f"Verdicts: {result['counts']}")
     return {"audit": audit["summary"], **result}
+    if notify:
+        notify_result(result["counts"], audit["summary"], base_dir)
 
 
 if __name__ == "__main__":
@@ -101,3 +103,23 @@ if __name__ == "__main__":
     d = sys.argv[1] if len(sys.argv) > 1 else "/tmp/cvtest"
     move = "--move" in sys.argv
     quality_scan(d, move=move)
+
+
+@task
+def notify_result(counts: dict, summary: list, base_dir: str):
+    """Send batch summary via Apprise."""
+    import apprise_helper
+
+    hard = counts.get("delete_candidate", 0)
+    soft = counts.get("review", 0)
+    keep = counts.get("keep", 0)
+    flagged = [s for s in summary if s["num_images"] > 0]
+    lines = "; ".join(f"{s[issue_type]}: {s[num_images]}" for s in flagged) or "none"
+    body = (
+        f"Scanned: {base_dir}\n"
+        f"Keep: {keep} | Review: {soft} | Delete-candidates: {hard}\n"
+        f"Issues: {lines}\n"
+        f"Review: http://192.168.20.13:8092/review"
+    )
+    apprise_helper.notify("📸 photo-pipeline batch complete", body)
+    return True