def get_doc_list_data()

in microservices/hitl_service/src/routes/hitl.py [0:0]

• 79 lines of code
• 27 McCabe index (conditional complexity)

def get_doc_list_data(docs_list: list):
  for doc in docs_list:
    name = "N/A"
    if doc["entities"]:
      for entity in doc["entities"]:
        if entity["entity"] == "name":
          if entity["corrected_value"]:
            name = entity["corrected_value"]
          elif entity["value"]:
            name = entity["value"]
    doc["applicant_name"] = name

    process_stage = "-"
    current_status = "-"
    status_last_updated_by = "-"
    last_status = None
    last_update_timestamp = None
    last_system_status = ""

    all_status_list = (doc.get("hitl_status", []) or []) + (
        doc.get("system_status", []) or [])
    audit_trail = sorted(all_status_list, key=lambda d: d["timestamp"])
    if audit_trail and len(audit_trail) > 0:
      last_status = audit_trail[-1]
      last_update_timestamp = last_status["timestamp"]
      status_last_updated_by = last_status.get("last_status", "System")

    system_status = doc.get("system_status", None)
    if system_status:
      last_system_status = system_status[-1]
      process_stage = (last_system_status["stage"]).lower()

    hitl_status = doc.get("hitl_status", None)
    last_hitl_status = hitl_status[-1] if hitl_status else None

    if doc["system_status"]:
      system_status = doc["system_status"]
      last_system_status = system_status[-1]
      status_last_updated_by = "System"

      # If there's HITL status, use the latest HITL status.
      if doc["hitl_status"]:
        last_hitl_status = doc["hitl_status"][-1]

        if last_system_status["timestamp"] > last_hitl_status["timestamp"]:
          if last_system_status["stage"] == "auto_approval":
            if last_system_status["status"] == STATUS_SUCCESS:
              current_status = doc["auto_approval"].title()
            else:
              current_status = STATUS_IN_PROGRESS
          elif last_system_status["status"] == STATUS_SUCCESS:
            current_status = STATUS_IN_PROGRESS
          else:
            current_status = STATUS_ERROR
        else:
          if last_hitl_status["status"] == "reassigned":
            current_status = STATUS_IN_PROGRESS
          else:
            current_status = last_hitl_status["status"].title()
            status_last_updated_by = last_hitl_status["user"]
            last_update_timestamp = last_hitl_status["timestamp"]

      # Otherwise, check the last system status.
      else:
        if last_system_status["stage"] == "auto_approval":
          if last_system_status["status"] == STATUS_SUCCESS:
            current_status = doc["auto_approval"].title()
          else:
            current_status = STATUS_REVIEW

        elif last_system_status["status"] == STATUS_SUCCESS:
          current_status = STATUS_IN_PROGRESS
        else:
          current_status = STATUS_ERROR

    # Show next stage process status.
    if current_status == STATUS_IN_PROGRESS and process_stage in PROCESS_NEXT_STAGE:
      process_stage = PROCESS_NEXT_STAGE[process_stage.lower()] + "..."

    # Update process detail status
    if current_status == STATUS_IN_PROGRESS:
      time_difference = datetime.datetime.now() - last_update_timestamp.replace(
          tzinfo=None)
      if time_difference.seconds > PROCESS_TIMEOUT_SECONDS:
        process_stage = last_system_status["stage"] + " " + STATUS_TIMEOUT
        current_status = STATUS_ERROR

    else:
      process_stage = process_stage + " " + last_system_status["status"]

    doc["process_status"] = process_stage.title()
    doc["current_status"] = current_status
    doc["status_last_updated_by"] = status_last_updated_by
    doc["last_update_timestamp"] = last_update_timestamp
    doc["audit_trail"] = audit_trail

  return docs_list