add /claim-linker endpoint - log VBMS and VA.gov claim id's

domain-cc/cc-app/src/python_src/api.py

@@ -1,19 +1,21 @@
+import json
 import logging
 import sys
+import time
 from typing import Optional
 
 from fastapi import FastAPI, HTTPException
 
-from .pydantic_models import Claim, PredictedClassification
+from .pydantic_models import Claim, ClaimLinkInfo, PredictedClassification
 from .util.brd_classification_codes import get_classification_name
 from .util.logging_dropdown_selections import build_logging_table
 from .util.lookup_table import ConditionDropdownLookupTable, DiagnosticCodeLookupTable
+from .util.sanitizer import sanitize
 
 dc_lookup_table = DiagnosticCodeLookupTable()
 dropdown_lookup_table = ConditionDropdownLookupTable()
 dropdown_values = build_logging_table()
 
-
 app = FastAPI(
     title="Contention Classification",
     description="Mapping VA.gov disability form contentions to actual classifications defined in the [Benefits Reference Data API](https://developer.va.gov/explore/benefits/docs/benefits_reference_data) for use in downstream VA systems.",
@@ -32,7 +34,7 @@
 )
 
 logging.basicConfig(
-    format="[%(asctime)s] %(levelname)-8s %(message)s",
+    format="%(message)s",
     level=logging.INFO,
     datefmt="%Y-%m-%d %H:%M:%S",
     stream=sys.stdout,
@@ -47,35 +49,56 @@
     return {"status": "ok"}
 
 
+def log_lookup_table_match(
+    classification_code: int, is_in_dropdown: bool, contention_text: str
+):
+    log_as_json({"is_in_dropdown": sanitize(is_in_dropdown)})
+    log_contention_text = contention_text if is_in_dropdown else "Not in dropdown"
+
+    if classification_code:
+        already_mapped_text = contention_text.strip().lower()  # do not leak PII
+        log_as_json({"lookup_table_match": sanitize(already_mapped_text)})
+    elif is_in_dropdown:
+        log_as_json(
+            {
+                "lookup_table_match": sanitize(
+                    f"No table match for {log_contention_text}"
+                )
+            }
+        )
+    else:
+        log_as_json(
+            {"lookup_table_match": sanitize("No table match for free text entry")}
+        )
+
+
+def log_as_json(log: dict):
+    if "date" not in log.keys():
+        log.update({"date": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())})
+    if "level" not in log.keys():
+        log.update({"level": "info"})
+    logging.info(json.dumps(log))
+
+
 @app.post("/classifier")
 def get_classification(claim: Claim) -> Optional[PredictedClassification]:
-    logging.info(
-        f"claim_id: {claim.claim_id}, form526_submission_id: {claim.form526_submission_id}"
+    log_as_json(
+        {
+            "claim_id": sanitize(claim.claim_id),
+            "form526_submission_id": sanitize(claim.form526_submission_id),
+        }
     )
     classification_code = None
     if claim.claim_type == "claim_for_increase":
-        logging.info(f"diagnostic code: {claim.diagnostic_code}")
+        log_as_json({"diagnostic code": sanitize(claim.diagnostic_code)})
         classification_code = dc_lookup_table.get(claim.diagnostic_code, None)
 
     if claim.contention_text and not classification_code:
         classification_code = dropdown_lookup_table.get(claim.contention_text, None)
         is_in_dropdown = claim.contention_text.strip().lower() in dropdown_values
-        log_contention_text = (
-            claim.contention_text if is_in_dropdown else "Not in dropdown"
+        log_lookup_table_match(
+            classification_code, is_in_dropdown, claim.contention_text
         )
-        if classification_code:
-            already_mapped_text = (  # being explicit, do not leak PII
-                claim.contention_text.strip().lower()
-            )
-            logging.info(
-                f"In Dropdown: {is_in_dropdown}, "
-                f" Contention Text: {log_contention_text}, Lookup table match: {already_mapped_text}"
-            )
-        else:
-            logging.info(
-                f"In Dropdown: {is_in_dropdown}, "
-                f"Contention Text: {log_contention_text}, No Lookup table match"
-            )
 
     if classification_code:
         classification_name = get_classification_name(classification_code)
@@ -86,5 +109,19 @@
     else:
         classification = None
 
-    logging.info(f"classification: {classification}")
+    log_as_json({"classification": classification})
     return classification
+
+
+@app.post("/claim-linker")
+def link_vbms_claim_id(claim_link_info: ClaimLinkInfo):
+    log_as_json(
+        {
+            "message": "linking claims",
+            "va_gov_claim_id": claim_link_info.va_gov_claim_id,
+            "vbms_claim_id": claim_link_info.vbms_claim_id,
+        }
+    )
+    return {
+        "success": True,
+    }

domain-cc/cc-app/src/python_src/pydantic_models.py

@@ -27,3 +27,10 @@ class PredictedClassification(BaseModel):
 
     classification_code: int
     classification_name: str
+
+
+class ClaimLinkInfo(BaseModel):
+    """used for connecting VA.gov and VBMS claims to each other in order to track contention changes downstream"""
+
+    va_gov_claim_id: int
+    vbms_claim_id: int

domain-cc/cc-app/src/python_src/util/sanitizer.py

@@ -0,0 +1,6 @@
+def sanitize(obj):
+    if isinstance(obj, bool):
+        return bool(str(obj).replace("\r\n", "").replace("\n", ""))
+    if isinstance(obj, int):
+        return int(str(obj).replace("\r\n", "").replace("\n", ""))
+    return str(obj).replace("\r\n", "").replace("\n", "")

domain-cc/cc-app/tests/test_claim_linker.py

@@ -0,0 +1,15 @@
+from fastapi.testclient import TestClient
+
+
+def test_claim_ids_logged(client: TestClient, caplog):
+    json_post_dict = {
+        "va_gov_claim_id": 100,
+        "vbms_claim_id": 200,
+    }
+    client.post("/claim-linker", json=json_post_dict)
+    expected_claim_link_json = {
+        "message": "linking claims",
+        "va_gov_claim_id": 100,
+        "vbms_claim_id": 200,
+    }
+    assert eval(caplog.records[0].message) == expected_claim_link_json