Extend promotional classification toggle to OCR review

.env.sample

@@ -40,6 +40,17 @@ FEATURE_ADMIN_OCR=false
 FEATURE_OCR_QUEUE_V2=false
 FEATURE_OCR_SHADOW_LAUNCH=false
 OCR_SHADOW_PROVIDER=
+# When true, uploaded queue images are OCR'd automatically in the background; when false they
+# land as `pending` for manual review on the admin OCR queue page. Requires FEATURE_ADMIN_OCR.
+FEATURE_OCR_AUTO_EXTRACT=false
+# Admin UI toggle to tag mail as promotional. When true, the admin record-entry form and
+# the OCR queue review screen expose a "Promotional" checkbox and the flag is persisted on
+# the resulting MAIL document (`isPromotional: true`, omitted otherwise).
+FEATURE_PROMO_CLASSIFICATION=false
+
+# Local photo staging: uploaded queue images are written here (bind-mounted into backend),
+# and anything older than the retention window is pruned nightly by the scheduler.
+IMAGE_RETENTION_HOURS=24
 
 OCR_PROVIDER=paddleocr
 # OCR_PROVIDER: paddleocr (self-hosted, default), easyocr, tesseract, or ocrspace (external API)
@@ -48,6 +59,7 @@ OCR_SPACE_API_KEY=
 
 BACKEND_API_URL=http://backend:8000
 SCHEDULER_CRON=0 8 * * 1
+IMAGE_PRUNE_CRON=0 3 * * *
 SCHEDULER_TIMEZONE=America/New_York
 SCHEDULER_TICK_SECONDS=20
 
@@ -59,3 +71,6 @@ GRAFANA_ROOT_URL=https://hub.avenuworkspaces.com/mail/grafana
 PROMETHEUS_DATA_DIR=/mnt/Main/other/PrometheusMail/
 PROMETHEUS_CONFIG_DIR=/mnt/Main/other/PrometheusMailEtc/
 GRAFANA_DATA_DIR=/mnt/Main/other/GrafanaMail/
+# Host directory bind-mounted into backend at /var/lib/avenu/images. Must exist and be
+# writable by the backend container uid. Pruned nightly per IMAGE_PRUNE_CRON + IMAGE_RETENTION_HOURS.
+MAIL_IMAGES_DIR=/mnt/Main/other/AvenuMailImages/

backend/config.py

@@ -79,12 +79,25 @@ def _env_positive_int(name: str, default: int) -> int:
 FEATURE_OCR_SHADOW_LAUNCH = _env_bool("FEATURE_OCR_SHADOW_LAUNCH", False)
 OCR_SHADOW_PROVIDER = os.getenv("OCR_SHADOW_PROVIDER", "").strip().lower() or None
 
+# Admin toggle to mark incoming mail as promotional. Independent of OCR; default off.
+FEATURE_PROMO_CLASSIFICATION = _env_bool("FEATURE_PROMO_CLASSIFICATION", False)
+# When true, uploaded images are OCR'd in a background thread; when false, items land as
+# `pending` for manual review. OCR auto-sort is descoped; default off.
+FEATURE_OCR_AUTO_EXTRACT = _env_bool("FEATURE_OCR_AUTO_EXTRACT", False)
+
+# Host-mounted directory where uploaded mail photos live (see docker-compose volumes).
+IMAGE_STORE_DIR = os.getenv("IMAGE_STORE_DIR", "/var/lib/avenu/images").strip() or "/var/lib/avenu/images"
+# Nightly prune deletes anything older than this many hours (both files and queue rows).
+IMAGE_RETENTION_HOURS = _env_positive_int("IMAGE_RETENTION_HOURS", 24)
+
 
 def get_feature_flags() -> dict[str, bool]:
     return {
         "adminOcr": FEATURE_ADMIN_OCR,
         "ocrQueueV2": FEATURE_OCR_QUEUE_V2 and FEATURE_ADMIN_OCR,
         "ocrShadowLaunch": FEATURE_OCR_SHADOW_LAUNCH and FEATURE_ADMIN_OCR,
+        "ocrAutoExtract": FEATURE_OCR_AUTO_EXTRACT and FEATURE_ADMIN_OCR,
+        "promoClassification": FEATURE_PROMO_CLASSIFICATION,
     }
 
 

backend/controllers/internal_jobs_controller.py

@@ -8,6 +8,7 @@
 from controllers.common import parse_iso_date
 from errors import APIError
 from services.idempotency_service import begin_request, commit_response, rollback_reservation
+from services.image_pruner import prune_expired_images
 from services.notifications.channels.factory import build_notification_channels
 from services.notifications.weekly_summary_cron_job import run_weekly_summary_cron_job
 from services.notifications.weekly_summary_notifier import WeeklySummaryNotifier
@@ -84,3 +85,36 @@ def internal_weekly_summary_job_route():
     except Exception:
         rollback_reservation(key=idempotency_key, route=route, method="POST")
         raise
+
+
+@internal_jobs_bp.route("/api/internal/jobs/image-prune", methods=["POST"])
+def internal_image_prune_job_route():
+    _require_scheduler_token()
+
+    body_raw = request.get_json(silent=True)
+    payload = {} if body_raw is None else require_dict(body_raw)
+    retention_override = payload.get("retentionHours")
+    retention_hours: int | None = None
+    if retention_override is not None:
+        if not isinstance(retention_override, int) or isinstance(retention_override, bool) or retention_override <= 0:
+            raise APIError(422, "retentionHours must be a positive integer")
+        retention_hours = retention_override
+
+    route = "/api/internal/jobs/image-prune"
+    idempotency_key, replay = begin_request(
+        headers=request.headers,
+        payload=payload,
+        route=route,
+        method="POST",
+    )
+    if replay is not None:
+        return jsonify(replay["body"]), replay["status"]
+
+    try:
+        result = prune_expired_images(retention_hours=retention_hours)
+        body = result.to_dict()
+        commit_response(key=idempotency_key, route=route, method="POST", status=200, body=body)
+        return jsonify(body), 200
+    except Exception:
+        rollback_reservation(key=idempotency_key, route=route, method="POST")
+        raise

backend/controllers/ocr_queue_controller.py

@@ -10,7 +10,15 @@
 from bson import ObjectId
 from flask import Blueprint, jsonify, request, send_file
 
-from config import FEATURE_ADMIN_OCR, FEATURE_OCR_QUEUE_V2, OCR_MAX_FILE_BYTES, ocr_queue_items_collection, fs
+from config import (
+    FEATURE_ADMIN_OCR,
+    FEATURE_OCR_AUTO_EXTRACT,
+    FEATURE_OCR_QUEUE_V2,
+    FEATURE_PROMO_CLASSIFICATION,
+    OCR_MAX_FILE_BYTES,
+    fs,
+    ocr_queue_items_collection,
+)
 from controllers.auth_guard import require_admin_session
 from errors import APIError
 from idempotency import payload_hash
@@ -27,6 +35,7 @@
     update_ocr_queue_item,
 )
 from repositories import to_api_doc
+from services import image_store
 from services.mail_service import create_mail
 from services.ocr.ocr_parser import parse_ocr_text
 
@@ -127,6 +136,8 @@ def _serialize_item(doc: dict[str, Any]) -> dict:
         "error": doc.get("error"),
         "mailboxId": str(doc["mailboxId"]) if doc.get("mailboxId") else None,
         "fileId": str(doc["fileId"]) if doc.get("fileId") else None,
+        "imagePath": doc.get("imagePath"),
+        "isPromotional": bool(doc.get("isPromotional")),
         "confirmedAt": doc["confirmedAt"].isoformat() if doc.get("confirmedAt") else None,
     }
 
@@ -163,27 +174,34 @@ def create_job():
     if not image_payloads:
         raise APIError(422, "no valid images to process")
 
-    # Store images in GridFS
-    file_ids: list[ObjectId] = []
-    for img_data, content_type in image_payloads:
-        fid = fs.put(img_data, content_type=content_type, filename="ocr_upload")
-        file_ids.append(fid)
+    image_paths: list[str] = [
+        image_store.save_bytes(img_data, content_type)
+        for img_data, content_type in image_payloads
+    ]
 
     date = request.form.get("date") or datetime.utcnow().strftime("%Y-%m-%d")
 
     job_id = create_ocr_job(created_by=admin_id, date=date, item_count=len(image_payloads))
-    create_ocr_queue_items(job_id, len(image_payloads), file_ids=file_ids)
-
-    from flask import current_app
-    app = current_app._get_current_object()
-    thread = threading.Thread(target=_process_ocr_job, args=(app, job_id, image_payloads), daemon=True)
-    thread.start()
+    create_ocr_queue_items(job_id, len(image_payloads), image_paths=image_paths)
+
+    if FEATURE_OCR_AUTO_EXTRACT:
+        from flask import current_app
+        app = current_app._get_current_object()
+        thread = threading.Thread(target=_process_ocr_job, args=(app, job_id, image_payloads), daemon=True)
+        thread.start()
+        status = "processing"
+    else:
+        # Auto-extract is off: items stay `pending`; admin fills in receiver/sender
+        # manually during review. Mark the job itself as processed so the UI stops
+        # showing a spinner — there's no background work to wait for.
+        update_ocr_job_status(job_id, "processed", completed_count=len(image_payloads))
+        status = "processed"
 
     return jsonify({
         "id": str(job_id),
-        "status": "processing",
+        "status": status,
         "totalCount": len(image_payloads),
-        "completedCount": 0,
+        "completedCount": len(image_payloads) if not FEATURE_OCR_AUTO_EXTRACT else 0,
         "date": date,
     }), 201
 
@@ -245,6 +263,8 @@ def update_queue_item(item_id: str):
     if "mailboxId" in data:
         mb = data["mailboxId"]
         repo_kwargs["mailbox_id"] = ObjectId(mb) if mb and ObjectId.is_valid(mb) else None
+    if "isPromotional" in data and FEATURE_PROMO_CLASSIFICATION:
+        repo_kwargs["is_promotional"] = bool(data["isPromotional"])
 
     if repo_kwargs:
         update_ocr_queue_item(oid, **repo_kwargs)
@@ -303,6 +323,8 @@ def confirm_queue_item(item_id: str):
             "receiverName": item.get("receiverName"),
             "senderInfo": item.get("senderInfo"),
         }
+        if FEATURE_PROMO_CLASSIFICATION and item.get("isPromotional"):
+            payload["isPromotional"] = True
         created_mail = create_mail(payload)
 
         now = datetime.utcnow()
@@ -354,19 +376,35 @@ def update_job_stage(job_id: str):
 @ocr_queue_bp.route("/api/ocr/queue/<item_id>/image", methods=["GET"])
 @require_admin_session
 def get_item_image(item_id: str):
-    """Get the image for a queue item."""
+    """Stream the image for a queue item.
+
+    Prefers the filesystem path (new uploads). Falls back to GridFS for any
+    legacy items written before the volume-backed store existed.
+    """
     if not ObjectId.is_valid(item_id):
         raise APIError(404, "item not found")
-    
+
     item = get_ocr_queue_item(ObjectId(item_id))
-    if not item or not item.get("fileId"):
-        raise APIError(404, "image not found")
+    if not item:
+        raise APIError(404, "item not found")
 
-    try:
-        grid_out = fs.get(item["fileId"])
-        return send_file(grid_out, mimetype=grid_out.content_type)
-    except Exception:
-        raise APIError(404, "image file missing")
+    rel_path = item.get("imagePath")
+    if rel_path:
+        try:
+            stream, content_type = image_store.open_path(rel_path)
+        except FileNotFoundError:
+            raise APIError(404, "image file missing")
+        return send_file(stream, mimetype=content_type)
+
+    file_id = item.get("fileId")
+    if file_id:
+        try:
+            grid_out = fs.get(file_id)
+            return send_file(grid_out, mimetype=grid_out.content_type)
+        except Exception:
+            raise APIError(404, "image file missing")
+
+    raise APIError(404, "image not found")
 
 
 @ocr_queue_bp.route("/api/ocr/jobs", methods=["OPTIONS"], endpoint="ocr_jobs_options")

backend/models/builders.py

@@ -162,6 +162,9 @@ def build_mail_create(payload: dict[str, Any]) -> dict[str, Any]:
     sender = _optional_text(payload, "senderInfo", 500)
     if sender is not None:
         doc["senderInfo"] = sender
+    if "isPromotional" in payload:
+        if optional_bool(payload, "isPromotional"):
+            doc["isPromotional"] = True
     return doc
 
 
@@ -187,6 +190,8 @@ def build_mail_patch(payload: dict[str, Any]) -> dict[str, Any]:
         patch["receiverName"] = _optional_text(payload, "receiverName", 2000)
     if "senderInfo" in payload:
         patch["senderInfo"] = _optional_text(payload, "senderInfo", 500)
+    if "isPromotional" in payload:
+        patch["isPromotional"] = optional_bool(payload, "isPromotional")
 
     if "count" in payload:
         n = _optional_mail_piece_count(payload)

backend/repositories/ocr_queue_repository.py

@@ -24,7 +24,13 @@ def create_ocr_job(*, created_by: ObjectId, date: str, item_count: int) -> Objec
     return r.inserted_id
 
 
-def create_ocr_queue_items(job_id: ObjectId, count: int, file_ids: list[ObjectId] | None = None) -> list[ObjectId]:
+def create_ocr_queue_items(
+    job_id: ObjectId,
+    count: int,
+    *,
+    file_ids: list[ObjectId] | None = None,
+    image_paths: list[str] | None = None,
+) -> list[ObjectId]:
     now = datetime.utcnow()
     docs = [
         {
@@ -38,6 +44,8 @@ def create_ocr_queue_items(job_id: ObjectId, count: int, file_ids: list[ObjectId
             "error": None,
             "mailboxId": None,
             "fileId": file_ids[i] if file_ids and i < len(file_ids) else None,
+            "imagePath": image_paths[i] if image_paths and i < len(image_paths) else None,
+            "isPromotional": False,
             "confirmedAt": None,
             "createdAt": now,
             "updatedAt": now,
@@ -92,6 +100,7 @@ def update_ocr_queue_item(
     raw_text: str | None = None,
     error: str | None = None,
     mailbox_id: ObjectId | None | object = _SENTINEL,
+    is_promotional: bool | object = _SENTINEL,
     confirmed_at: datetime | None = None,
 ) -> bool:
     updates: dict[str, Any] = {"updatedAt": datetime.utcnow()}
@@ -109,6 +118,8 @@ def update_ocr_queue_item(
         updates["error"] = error
     if mailbox_id is not _SENTINEL:
         updates["mailboxId"] = mailbox_id
+    if is_promotional is not _SENTINEL:
+        updates["isPromotional"] = bool(is_promotional)
     if confirmed_at is not None:
         updates["confirmedAt"] = confirmed_at
     r = ocr_queue_items_collection.update_one({"_id": item_id}, {"$set": updates})