Merge pull request #203 from OpenSPP/fix/spp-dci-dr-spec-envelope

fix(spp_dci_client_dr): parse DCI v1.0.0 spec envelope and record fields

Author: gonzalesedwin1123

spp_dci_client_dr/routers/callback.py

@@ -6,6 +6,7 @@
 from datetime import UTC, datetime
 from typing import Annotated
 
+from odoo import fields
 from odoo.api import Environment
 
 from odoo.addons.fastapi.dependencies import odoo_env
@@ -14,6 +15,7 @@
 from fastapi import APIRouter, Depends, HTTPException, Request, status
 
 from ..middleware.signature import verify_dr_signature
+from ..services.dr_parsing import extract_disability_data, unwrap_search_data
 
 _logger = logging.getLogger(__name__)
 
@@ -149,8 +151,7 @@ def _process_dr_search_result(env: Environment, result: dict, source_registry: s
             )
             return
 
-        data = result.get("data", {})
-        reg_records = data.get("reg_records", [])
+        reg_records = unwrap_search_data(result.get("data"))
 
         for record in reg_records:
             # Extract identifiers to find matching partner
@@ -245,11 +246,8 @@ def _update_disability_status(
     # Use sudo() for API access - authentication is handled by signature verification
     DisabilityStatus = env["spp.dci.disability.status"].sudo()  # nosemgrep: odoo-sudo-without-context
 
-    # Extract disability data from record
-    has_disability = record.get("has_disability", False) or record.get("is_pwd", False)
-    disability_types = record.get("disability_types", [])
-    functional_scores = record.get("functional_scores", {})
-    assessment_date = record.get("assessment_date")
+    # Extract disability data using spec-aware parsing
+    extracted = extract_disability_data(record)
 
     # Find existing status
     existing = DisabilityStatus.search(
@@ -259,16 +257,14 @@ def _update_disability_status(
 
     vals = {
         "partner_id": partner.id,
-        "has_disability": has_disability,
-        "disability_types": json.dumps(disability_types) if isinstance(disability_types, list) else disability_types,
-        "functional_scores": json.dumps(functional_scores)
-        if isinstance(functional_scores, dict)
-        else functional_scores,
-        "assessment_date": assessment_date,
+        "has_disability": extracted["has_disability"],
+        "disability_types": json.dumps(extracted["disability_types"]),
+        "functional_scores": json.dumps(extracted["functional_scores"]),
+        "assessment_date": extracted["assessment_date"],
         "source_registry": source_registry,
         "raw_data": json.dumps(record),
         "state": "synced",
-        "last_sync_date": datetime.now(UTC),
+        "last_sync_date": fields.Datetime.now(),
         "synced_by": env.user.id,
     }
 

spp_dci_client_dr/services/__init__.py

@@ -1 +1,2 @@
+from . import dr_parsing
 from . import dr_service

spp_dci_client_dr/services/dr_parsing.py

@@ -0,0 +1,151 @@
+# Part of OpenSPP. See LICENSE file for full copyright and licensing details.
+"""Pure parsing helpers for DCI v1.0.0 Disability Registry records.
+
+All functions are stateless and require no Odoo env, making them
+independently testable and easy to reuse.
+"""
+
+import logging
+from datetime import date, datetime
+
+_logger = logging.getLogger(__name__)
+
+# Status values that indicate the person is registered as disabled.
+# The SP DCI v1.0.0 spec does not declare an enum; these are the
+# spec-aligned workflow tokens. Empty / missing / other values trigger
+# the impairment-list fallback in extract_disability_data.
+_APPROVED_STATUSES = {"approved", "registered"}
+
+# Status values that explicitly reject disability registration.
+_REJECTED_STATUSES = {"rejected", "denied"}
+
+
+def _coerce_date(value) -> date | None:
+    """Coerce a DCI date/datetime value into a ``date`` object.
+
+    Accepts ISO date strings (``YYYY-MM-DD``), ISO datetime strings with an
+    optional trailing ``Z`` (``YYYY-MM-DDTHH:MM:SSZ``), naive/aware datetimes,
+    and date objects. Returns ``None`` for empty input or unparseable values
+    (with a WARNING logged).
+
+    For tz-aware inputs, the local wall-clock date is returned;
+    no UTC normalization is applied.
+    """
+    if not value:
+        return None
+    if isinstance(value, datetime):
+        return value.date()
+    if isinstance(value, date):
+        return value
+    try:
+        return datetime.fromisoformat(str(value).removesuffix("Z")).date()
+    except ValueError:
+        _logger.warning("Could not parse date from value: %r", value)
+        return None
+
+
+def unwrap_search_data(data) -> list:
+    """Extract the reg_records list from a DCI v1.0.0 search response data envelope.
+
+    Args:
+        data: The value of ``search_response[*].data`` from the API response.
+              Expected to be a dict with a ``reg_records`` key per the spec.
+
+    Returns:
+        list: The contents of ``data["reg_records"]``, or an empty list when
+              the envelope is absent, empty, or malformed.
+    """
+    if data is None:
+        return []
+
+    if not isinstance(data, dict):
+        _logger.warning(
+            "Unexpected type for search response data envelope: %s; expected dict",
+            type(data).__name__,
+        )
+        return []
+
+    if not data:
+        return []
+
+    records = data.get("reg_records")
+    if records is None:
+        return []
+    if not isinstance(records, list):
+        _logger.warning(
+            "Unexpected type for reg_records: %s; expected list",
+            type(records).__name__,
+        )
+        return []
+    return records
+
+
+def extract_disability_data(record: dict) -> dict:
+    """Extract structured disability information from a DCI v1.0.0 record.
+
+    Args:
+        record: A single record dict from ``reg_records``.
+
+    Returns:
+        dict with keys:
+            - ``has_disability`` (bool)
+            - ``disability_types`` (list[str])
+            - ``functional_scores`` (dict, always ``{}``: spec has no numeric scores)
+            - ``assessment_date`` (``date`` | None)
+            - ``source_registry`` (str | None)
+            - ``raw_data`` (the input record, unchanged)
+    """
+    # Extract impairment types from disability_details.
+    # Use `or []` so an explicit null on the wire does not crash the loop.
+    details = record.get("disability_details") or []
+    disability_types = [d["impairment_type"] for d in details if isinstance(d, dict) and d.get("impairment_type")]
+
+    # Resolve has_disability from the disability_status string
+    status_str = str(record.get("disability_status", "")).strip().lower()
+
+    if status_str in _APPROVED_STATUSES:
+        has_disability = True
+    elif status_str in _REJECTED_STATUSES:
+        has_disability = False
+    elif status_str == "":
+        # No explicit status: fall back to impairment list presence
+        has_disability = bool(disability_types)
+    else:
+        _logger.warning(
+            "Unknown disability_status value: %s; falling back to impairment list",
+            record.get("disability_status"),
+        )
+        has_disability = bool(disability_types)
+
+    # Assessment date: prefer last_updated, fall back to registration_date.
+    # The spec uses ISO datetime strings; coerce to a date for the ORM.
+    assessment_date = _coerce_date(record.get("last_updated") or record.get("registration_date"))
+
+    # Source registry: prefer source_registry, fall back to registry_name
+    source_registry = record.get("source_registry") or record.get("registry_name")
+
+    return {
+        "has_disability": has_disability,
+        "disability_types": disability_types,
+        "functional_scores": {},
+        "assessment_date": assessment_date,
+        "source_registry": source_registry,
+        "raw_data": record,
+    }
+
+
+def extract_functional_scores(record: dict) -> dict:
+    """Return functional assessment scores from a DCI v1.0.0 record.
+
+    The DCI v1.0.0 spec does not define numeric functional scores.
+    ``impairment_level`` is a free-text string, not a number.
+    This function always returns ``{}`` and exists as a hook for future
+    spec versions that may introduce numeric scoring.
+
+    Args:
+        record: A single record dict from ``reg_records``.
+
+    Returns:
+        dict: Always ``{}``.
+    """
+    return {}

spp_dci_client_dr/services/dr_service.py

@@ -9,6 +9,8 @@
 
 from odoo.addons.spp_dci_client.services import DCIClient
 
+from .dr_parsing import extract_disability_data, extract_functional_scores, unwrap_search_data
+
 _logger = logging.getLogger(__name__)
 
 
@@ -122,14 +124,11 @@ def get_disability_status(self, partner) -> dict | None:
 
             # Extract first result
             search_response = message["search_response"][0]
-            if "data" not in search_response or not search_response["data"]:
+            records = unwrap_search_data(search_response.get("data"))
+            if not records:
                 return None
 
-            record_data = (
-                search_response["data"][0] if isinstance(search_response["data"], list) else search_response["data"]
-            )
-
-            # Extract disability information
+            record_data = records[0]
             disability_data = self._extract_disability_data(record_data)
 
             _logger.info(
@@ -207,14 +206,11 @@ def get_functional_assessment(self, identifier_type: str, identifier_value: str)
 
             # Extract first result
             search_response = message["search_response"][0]
-            if "data" not in search_response or not search_response["data"]:
+            records = unwrap_search_data(search_response.get("data"))
+            if not records:
                 return None
 
-            record_data = (
-                search_response["data"][0] if isinstance(search_response["data"], list) else search_response["data"]
-            )
-
-            # Extract functional scores
+            record_data = records[0]
             scores = self._extract_functional_scores(record_data)
 
             _logger.info(
@@ -389,103 +385,29 @@ def _get_partner_identifier(self, partner):
         return None
 
     def _extract_disability_data(self, record_data: dict) -> dict:
-        """Extract disability information from DCI record data.
+        """Extract disability information from a DCI v1.0.0 record.
+
+        Delegates to the stateless module-level helper in dr_parsing.
 
         Args:
-            record_data: DCI record data from search response
+            record_data: A single record dict from reg_records
 
         Returns:
             dict: Extracted disability data
         """
-        disability_data = {
-            "has_disability": False,
-            "disability_types": [],
-            "functional_scores": {},
-            "raw_data": record_data,
-        }
-
-        # Check for disability flag
-        if "has_disability" in record_data:
-            disability_data["has_disability"] = bool(record_data["has_disability"])
-        elif "is_pwd" in record_data:
-            disability_data["has_disability"] = bool(record_data["is_pwd"])
-
-        # Extract disability types
-        if "disability_types" in record_data:
-            types_data = record_data["disability_types"]
-            if isinstance(types_data, list):
-                disability_data["disability_types"] = types_data
-            elif isinstance(types_data, str):
-                # Handle comma-separated string
-                disability_data["disability_types"] = [t.strip() for t in types_data.split(",") if t.strip()]
-
-        # Extract functional scores
-        disability_data["functional_scores"] = self._extract_functional_scores(record_data)
-
-        # Extract assessment date
-        if "assessment_date" in record_data:
-            disability_data["assessment_date"] = record_data["assessment_date"]
-        elif "disability_assessment_date" in record_data:
-            disability_data["assessment_date"] = record_data["disability_assessment_date"]
-
-        # Extract source registry
-        if "source_registry" in record_data:
-            disability_data["source_registry"] = record_data["source_registry"]
-        elif "registry_name" in record_data:
-            disability_data["source_registry"] = record_data["registry_name"]
-
-        return disability_data
+        return extract_disability_data(record_data)
 
     def _extract_functional_scores(self, record_data: dict) -> dict:
-        """Extract functional assessment scores from record data.
+        """Return functional assessment scores from a DCI v1.0.0 record.
+
+        Delegates to the stateless module-level helper in dr_parsing.
+        The DCI v1.0.0 spec has no numeric functional scores, so this always
+        returns ``{}``.
 
         Args:
-            record_data: DCI record data
+            record_data: A single record dict from reg_records
 
         Returns:
-            dict: Functional scores by domain
-            Example: {'Vision': 3, 'Hearing': 1, 'Mobility': 4, ...}
+            dict: Always ``{}``
         """
-        scores = {}
-
-        # Try to extract from functional_scores field
-        if "functional_scores" in record_data:
-            scores_data = record_data["functional_scores"]
-            if isinstance(scores_data, dict):
-                scores = scores_data
-            elif isinstance(scores_data, str):
-                # Try to parse JSON string
-                try:
-                    scores = json.loads(scores_data)
-                except json.JSONDecodeError:
-                    _logger.warning("Failed to parse functional_scores JSON")
-
-        # Try to extract individual domain scores
-        functional_domains = [
-            "Vision",
-            "Hearing",
-            "Mobility",
-            "Cognition",
-            "SelfCare",
-            "Communication",
-        ]
-
-        for domain in functional_domains:
-            # Try various field name formats
-            for field_name in [
-                f"functional_{domain.lower()}",
-                f"{domain.lower()}_score",
-                domain.lower(),
-                domain,
-            ]:
-                if field_name in record_data and record_data[field_name]:
-                    try:
-                        scores[domain] = int(record_data[field_name])
-                    except (ValueError, TypeError):
-                        _logger.warning(
-                            "Invalid functional score for %s: %s",
-                            domain,
-                            record_data[field_name],
-                        )
-
-        return scores
+        return extract_functional_scores(record_data)

spp_dci_client_dr/tests/__init__.py

@@ -1,4 +1,6 @@
 # Part of OpenSPP. See LICENSE file for full copyright and licensing details.
 
+from . import test_callback
 from . import test_disability_status
+from . import test_dr_parsing
 from . import test_dr_service