Merge pull request #25 from akkaouim/labs-mbw-v3

Continued work on mbw v1 using updated pipeline work
Reviewed and tested

Author: akkaouim

commcare_connect/labs/analysis/backends/sql/backend.py

@@ -5,6 +5,7 @@
 All analysis is done via SQL queries, not Python/pandas.
 """
 
+import json
 import logging
 from collections.abc import Generator
 from datetime import date, datetime
@@ -54,6 +55,29 @@ def _model_to_visit_dict(row) -> dict:
     }
 
 
+def _build_visit_dict(row: dict) -> dict:
+    """Build a visit context dict from a raw SQL row for transform/extractor post-processing."""
+    form_json = row.get("form_json", {})
+    if isinstance(form_json, str):
+        try:
+            form_json = json.loads(form_json) if form_json else {}
+        except (ValueError, json.JSONDecodeError):
+            form_json = {}
+    images = row.get("images", [])
+    if isinstance(images, str):
+        try:
+            images = json.loads(images) if images else []
+        except (ValueError, json.JSONDecodeError):
+            images = []
+    return {
+        "form_json": form_json,
+        "images": images,
+        "username": row.get("username"),
+        "visit_date": row.get("visit_date"),
+        "entity_name": row.get("entity_name"),
+    }
+
+
 class SQLBackend:
     """
     SQL backend for analysis.
@@ -431,6 +455,8 @@ def _process_visit_level(
 
             # Apply post-processing transforms that need full visit context
             # (e.g., extract_images_with_question_ids needs both form_json and images)
+            # Build visit_dict once per row (lazy); transforms must not mutate it.
+            visit_dict = None
             for field in config.fields:
                 if field.name not in computed_field_names:
                     continue
@@ -445,30 +471,22 @@ def _process_visit_level(
                     # If transform takes 'visit_data' param, it needs full context
                     if "visit_data" in params or len(params) == 0:
                         try:
-                            import json
-
-                            # Build full visit dict for transform
-                            # Note: form_json and images come back as JSON strings from SQL
-                            form_json = row.get("form_json", {})
-                            if isinstance(form_json, str):
-                                form_json = json.loads(form_json) if form_json else {}
-
-                            images = row.get("images", [])
-                            if isinstance(images, str):
-                                images = json.loads(images) if images else []
-
-                            visit_dict = {
-                                "form_json": form_json,
-                                "images": images,
-                                "username": row.get("username"),
-                                "visit_date": row.get("visit_date"),
-                                "entity_name": row.get("entity_name"),
-                            }
+                            if visit_dict is None:
+                                visit_dict = _build_visit_dict(row)
                             computed[field.name] = field.transform(visit_dict)
                         except Exception as e:
                             logger.warning(f"Transform for {field.name} failed: {e}")
                             computed[field.name] = None
 
+                elif field.extractor and callable(field.extractor):
+                    try:
+                        if visit_dict is None:
+                            visit_dict = _build_visit_dict(row)
+                        computed[field.name] = field.extractor(visit_dict)
+                    except Exception as e:
+                        logger.warning(f"Extractor for {field.name} failed: {e}")
+                        computed[field.name] = None
+
             # Parse visit_date
             visit_date_val = row.get("visit_date")
             if visit_date_val and isinstance(visit_date_val, date):
@@ -495,7 +513,7 @@ def _process_visit_level(
         # Cache computed visits (store base fields as columns to avoid joins later)
         computed_cache_data = [
             {
-                "visit_id": int(row.id),
+                "visit_id": row.id,
                 "username": row.username,
                 # Handle both date and datetime objects
                 "visit_date": row.visit_date.date()

commcare_connect/labs/analysis/backends/sql/cchq_fetcher.py

@@ -46,7 +46,7 @@ def normalize_cchq_form_to_visit_dict(form: dict, index: int) -> dict:
         "entity_id": "",
         "entity_name": "",
         "deliver_unit": "",
-        "deliver_unit_id": "",
+        "deliver_unit_id": None,
         "location": "",
         "flagged": False,
         "flag_reason": "",
@@ -58,7 +58,7 @@ def normalize_cchq_form_to_visit_dict(form: dict, index: int) -> dict:
         "review_created_on": None,
         "justification": "",
         "date_created": received_on,
-        "completed_work_id": "",
+        "completed_work_id": None,
         "images": [],
     }
 

commcare_connect/labs/analysis/backends/sql/query_builder.py

@@ -360,6 +360,9 @@ def build_visit_extraction_query(
     # Check if any field needs full visit context (form_json, images)
     needs_full_context = False
     for field in config.fields:
+        if field.extractor and callable(field.extractor):
+            needs_full_context = True
+            break
         if field.transform and callable(field.transform):
             import inspect
 
@@ -378,6 +381,12 @@ def build_visit_extraction_query(
 
     # Add computed fields from config (no aggregation, just extraction + transform)
     for field in config.fields:
+        # Handle extractor fields — need post-processing with full visit context
+        if field.extractor and callable(field.extractor):
+            select_parts.append(f"NULL as {field.name}")
+            computed_field_names.append(field.name)
+            continue
+
         # Skip fields that will be computed from full visit context (special markers like __images__)
         if field.transform and callable(field.transform):
             import inspect

commcare_connect/labs/analysis/data_access.py

@@ -18,7 +18,12 @@
 logger = logging.getLogger(__name__)
 
 
-def fetch_flw_names(access_token: str, opportunity_id: int, use_cache: bool = True) -> dict[str, str]:
+def fetch_flw_names(
+    access_token: str,
+    opportunity_id: int,
+    use_cache: bool = True,
+    last_active_out: dict | None = None,
+) -> dict[str, str]:
     """
     Fetch FLW display names for an opportunity from Connect API.
 
@@ -29,6 +34,9 @@ def fetch_flw_names(access_token: str, opportunity_id: int, use_cache: bool = Tr
         access_token: OAuth Bearer token for Connect API
         opportunity_id: Opportunity ID to fetch FLW names for
         use_cache: Whether to use Django cache (default True)
+        last_active_out: Optional dict to populate with {username: last_active_str}.
+            When provided, last_active data is written directly into this dict,
+            avoiding reliance on Django cache for in-process data sharing.
 
     Returns:
         Dictionary mapping username to display name.
@@ -44,8 +52,16 @@ def fetch_flw_names(access_token: str, opportunity_id: int, use_cache: bool = Tr
         try:
             cached = cache.get(cache_key)
             if cached is not None:
-                logger.debug(f"FLW names loaded from cache for opp {opportunity_id}")
-                return cached
+                # If last_active was requested, only use cache if la data is also cached
+                if last_active_out is not None:
+                    la_cached = cache.get(f"flw_last_active_{opportunity_id}")
+                    if la_cached is not None:
+                        last_active_out.update(la_cached)
+                    else:
+                        cached = None  # Force fresh fetch to populate last_active
+                if cached is not None:
+                    logger.debug(f"FLW names loaded from cache for opp {opportunity_id}")
+                    return cached
         except Exception as e:
             logger.warning(f"Cache get failed for {cache_key}: {e}")
 
@@ -69,20 +85,30 @@ def fetch_flw_names(access_token: str, opportunity_id: int, use_cache: bool = Tr
 
     # Parse CSV response
     df = pd.read_csv(StringIO(response.text))
-    logger.info(f"Fetched {len(df)} FLWs from Connect")
+    logger.info(f"Fetched {len(df)} FLWs from Connect. CSV columns: {list(df.columns)}")
 
     # Build mapping: username -> name (fallback to username if name is empty)
+    # Also extract last_active for the "Last Active" dashboard column
     flw_names = {}
+    flw_last_active = {}
     for _, row in df.iterrows():
         username = row.get("username")
         name = row.get("name")
         if username:
             flw_names[username] = name if name else username
+            last_active = row.get("last_active")
+            if pd.notna(last_active):
+                flw_last_active[username] = str(last_active)
 
-    # Cache the result
+    # Populate caller's dict directly (avoids cache dependency)
+    if last_active_out is not None:
+        last_active_out.update(flw_last_active)
+
+    # Cache the results
     if use_cache:
         try:
             cache.set(cache_key, flw_names, DJANGO_CACHE_TTL)
+            cache.set(f"flw_last_active_{opportunity_id}", flw_last_active, DJANGO_CACHE_TTL)
             logger.debug(f"FLW names cached for opp {opportunity_id}")
         except Exception as e:
             logger.warning(f"Cache set failed for {cache_key}: {e}")

commcare_connect/labs/analysis/sse_streaming.py

@@ -7,6 +7,8 @@
 
 import json
 import logging
+import queue
+import threading
 import time
 from collections.abc import Callable, Generator
 
@@ -69,6 +71,9 @@ def stream_data(self, request) -> Generator[str, None, None]:
                 yield send_sse_event("Complete!", data={"result": 123})
     """
 
+    heartbeat_enabled = True
+    heartbeat_interval = 20  # seconds between heartbeat comments
+
     def get(self, request):
         """
         Handle GET request and return streaming response.
@@ -79,14 +84,82 @@ def get(self, request):
         if not request.user.is_authenticated:
             return JsonResponse({"error": "Not authenticated"}, status=401)
 
+        generator = self.stream_data(request)
+        if self.heartbeat_enabled:
+            generator = self._with_heartbeat(generator)
+
         response = StreamingHttpResponse(
-            self.stream_data(request),
+            generator,
             content_type="text/event-stream",
         )
         response["Cache-Control"] = "no-cache"
         response["X-Accel-Buffering"] = "no"  # Disable nginx buffering
         return response
 
+    def _with_heartbeat(self, generator, interval=None):
+        """Wrap a generator with periodic SSE heartbeat comments.
+
+        Prevents ALB/browser timeouts during long-running blocking operations
+        (CSV parsing, data processing) by sending SSE comment lines every
+        ``interval`` seconds when the generator isn't yielding real data.
+
+        SSE comment format ``: heartbeat\\n\\n`` keeps the TCP connection
+        alive but does not trigger EventSource.onmessage on the frontend.
+
+        Set ``heartbeat_enabled = False`` on a subclass to disable.
+        """
+        if interval is None:
+            interval = self.heartbeat_interval
+
+        data_queue: queue.Queue = queue.Queue(maxsize=100)
+        stop_event = threading.Event()
+
+        def _producer():
+            try:
+                for item in generator:
+                    if stop_event.is_set():
+                        break
+                    while not stop_event.is_set():
+                        try:
+                            data_queue.put(("data", item), timeout=1)
+                            break
+                        except queue.Full:
+                            continue
+            except Exception as e:  # noqa: BLE001
+                try:
+                    data_queue.put(("error", e), timeout=1)
+                except queue.Full:
+                    pass
+            finally:
+                try:
+                    data_queue.put(("done", None), timeout=1)
+                except queue.Full:
+                    pass
+                try:
+                    generator.close()
+                except (GeneratorExit, RuntimeError):
+                    pass
+
+        thread = threading.Thread(target=_producer, daemon=True)
+        thread.start()
+
+        try:
+            while True:
+                try:
+                    msg_type, value = data_queue.get(timeout=interval)
+                    if msg_type == "data":
+                        yield value
+                    elif msg_type == "done":
+                        break
+                    elif msg_type == "error":
+                        raise value
+                except queue.Empty:
+                    # No data for `interval` seconds — send SSE comment to keep alive
+                    yield ": heartbeat\n\n"
+        finally:
+            stop_event.set()
+            thread.join(timeout=2)
+
     def stream_data(self, request) -> Generator[str, None, None]:
         """
         Generator that yields SSE events.

commcare_connect/static/js/workflow-runner.tsx

@@ -545,6 +545,16 @@ function createActionHandlers(csrfToken: string): ActionHandlers {
       }
     },
 
+    getAISessions: async (taskId: number): Promise<Record<string, unknown>> => {
+      try {
+        const response = await fetch(`/tasks/${taskId}/ai/sessions/`);
+        const data = await response.json();
+        return data;
+      } catch (e) {
+        return { success: false, error: e instanceof Error ? e.message : 'Failed to fetch AI sessions' };
+      }
+    },
+
     updateTask: async (taskId: number, data: Record<string, unknown>): Promise<Record<string, unknown>> => {
       try {
         const response = await fetch(`/tasks/api/${taskId}/update/`, {