API ledger client document url updates

legal-api/pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "legal-api"
-version = "3.1.7"
+version = "3.1.8"
 description = ""
 authors = [
     {name = "thor",email = "1042854+thorwolpert@users.noreply.github.com"}

legal-api/src/legal_api/reports/document_service.py

@@ -50,18 +50,32 @@
     }, {
         "documentType": "CORP_AFFIDAVIT",
         "documentClass": "CORP"
+    }],
+    "continuationIn": [{
+        "documentType": "CNTA",
+        "documentClass": "CORP"
+    }],
+    "continuationOut": [{
+        "documentType": "CNTO",
+        "documentClass": "CORP"
     }]
 }
+# Map to DRS document based on the default static document name set when the client does not submit a filename.
+# Included for migrated minio docs where submissions with no filename were allowed.
 STATIC_DOCUMENTS = {
     "Unlimited Liability Corporation Information": {
-        "documentType": "DIRECTOR_AFFIDAVIT",
-        "documentClass": "CORP"
+        "documentType": "DIRECTOR_AFFIDAVIT"
+    },
+    "Court Order": {
+        "documentType": "CRTO"
     }
 }
 APP_JSON = "application/json"
 APP_PDF = "application/pdf"
 DOC_PATH = "/documents"
 BEARER = "Bearer "
+DS_ID_PREFIX = "DS"
+DS_KEY_TOKENS = 2
 
 
 class ReportTypes(BaseEnum):
@@ -500,10 +514,10 @@ def replace_filing_report(
                 return response2
         return report_response
 
-
     def _update_static_document(self, doc: dict, doc_list: list) -> list:
         """
         Update static document urls in the document_list if a DRS match is found.
+        For static documents try obtaining download URL params from the file key first (DRS key).
 
         docs: The DRS document information to match on.
         doc_list: The business list of reports/documents for the filing.
@@ -516,12 +530,7 @@ def _update_static_document(self, doc: dict, doc_list: list) -> list:
         for key in doc_list:
             if key == "staticDocuments":
                 for static_doc in doc_list.get("staticDocuments"):
-                    name: str = static_doc.get("name")
-                    if (name and name == doc.get("name")) or (
-                            name and STATIC_DOCUMENTS.get(name) and
-                            doc.get("documentClass") == STATIC_DOCUMENTS[name].get("documentClass") and
-                            doc.get("documentType") == STATIC_DOCUMENTS[name].get("documentType")
-                        ):
+                    if self.is_static_doc_match(doc, static_doc):
                         static_doc["url"] = static_doc["url"] + query_params
                         break
             elif any(
@@ -533,6 +542,61 @@ def _update_static_document(self, doc: dict, doc_list: list) -> list:
                 break
         return doc_list
 
+    def get_url_drs_id(self, url: str) -> str:
+        """
+        Try to extract a DRS ID from the static document url (either minio or DRS).
+
+        url: URL to use.
+        return: DRS ID if url ends with DRS file key.
+        """
+        file_key: str = str(url).split("/")[-1]
+        if file_key:
+            tokens = file_key.split("-")
+            if (
+                len(tokens) == DS_KEY_TOKENS and
+                tokens[0] in ("COOP", "FIRM", "CORP") and
+                str(tokens[1]).startswith(DS_ID_PREFIX)
+            ):
+                return tokens[1]
+        return ""
+
+
+    def is_static_doc_match(self, doc: dict, static_doc: dict) -> bool:
+        """
+        Match the current DRS document with a ledger filing static document entry.
+        Try obtaining download URL params from the file key first (DRS key).
+
+        doc: Current DRS document to match with static document.
+        doc_list: The business list of reports/documents for the filing.
+        return: True if the DRS doc matches the static document information.
+        """
+        # Name is the ledger link name and should always exist, otherwise there is a defect somewhere else.
+        name: str = static_doc.get("name", "")
+        if not name or str(static_doc.get("url")).find("documentClass=") > 0:
+            return False
+        drs_id = self.get_url_drs_id(static_doc.get("url", ""))
+        if drs_id:
+            return drs_id == doc.get("identifier")
+        doc_name: str = doc.get("name", "")
+        if (doc_name):
+            if name.removesuffix(".pdf") == doc_name.removesuffix(".pdf"):
+                return True
+            if (
+                STATIC_DOCUMENTS.get(name) and
+                doc.get("documentType") == STATIC_DOCUMENTS[name].get("documentType")
+            ):
+                return True
+        # Try partial matching static doc name with default name.
+        for key, value in STATIC_DOCUMENTS.items():
+            if (
+                name.startswith(str(key)) and
+                (doc_name.startswith(str(key)) or doc_name == "") and
+                value.get("documentType") == doc.get("documentType")
+            ):
+                return True
+        return False
+
+
     def _get_request_headers(self, account_id: str, accept_mime_type = None) -> dict:
         """
         Get request headers for the DRS api call.

legal-api/tests/unit/reports/test_document_service.py

@@ -78,11 +78,11 @@
     "staticDocuments": [
       {
         "name": "Unlimited Liability Corporation Information",
-        "url": "https://test.com/C9900863/filings/155753/documents/static/DS0000100741"
+        "url": "https://test.com/C9900863/filings/155753/documents/static/CORP-DS0000100741"
       },
       {
         "name": "20250107-Authorization1.pdf",
-        "url": "https://test.com/C9900863/filings/155753/documents/static/DS0000100740"
+        "url": "https://test.com/C9900863/filings/155753/documents/static/CORP-DS0000100740"
       }
     ]
   }
@@ -250,6 +250,24 @@
     "url": ""
   }
 ]
+DRS_TEST_DOC = {
+    "consumerDocumentId": "0100000527",
+    "dateCreated": "2026-06-29T19:19:32+00:00",
+    "datePublished": "2026-06-29T00:00:00+00:00",
+    "documentClass": "COOP",
+    "documentType": "COSD",
+    "documentTypeDescription": "Statement of Dissolution",
+    "entityIdentifier": "CP1044808",
+    "eventIdentifier": 233801,
+    "identifier": "DS0000101951",
+    "name": "",
+    "url": ""
+}
+STATIC_URL1 = "https://test/business/api/v2/businesses/BC0888572/filings/3671021/documents/static/798da472-4f2d-4299-a3ee-84388d89f9ce.pdf"
+STATIC_URL2 = "https://test/business/api/v2/businesses/BC0888572/filings/3671021/documents/static/CORP-DS0000101951"
+STATIC_URL3 = "https://test/business/api/v2/businesses/BC0888572/filings/3671021/documents/static/CORP-DS0000101952"
+STATIC_URL4 = "https://test/business/api/v2/businesses/BC0888572/filings/3671021/documents/static/COOP-DS0000101953"
+STATIC_URL5 = "https://test/business/api/v2/businesses/BC0888572/filings/3671021/documents/static/FIRM-DS0000101954"
 
 # testdata pattern is ({description}, {doc_data}, {drs_data}, {receipt}, {filing}, {noa}, {cert}, {static})
 TEST_FILING_UPDATE_DATA = [
@@ -309,7 +327,32 @@
     (True, "ceaseReceiver", "FILING"),
     (True, "default", "FILING"),
 ]
-
+# testdata pattern is ({match}, {url}, {name}, {drs_doc}, {drs_class}, {drs_type}, {drs_id}, {drs_filename})
+TEST_STATIC_DOC_MATCH_DATA = [
+   (True, STATIC_URL2, "Test court order", DRS_TEST_DOC, "CORP", "CRTO", "DS0000101951", "Test court order.pdf"),
+   (True, STATIC_URL2, "Test court order", DRS_TEST_DOC, "CORP", "CRTO", "DS0000101951", ""),
+   (True, STATIC_URL1, "Test court order", DRS_TEST_DOC, "CORP", "CRTO", "DS0000101951", "Test court order.pdf"),
+   (True, STATIC_URL1, "Unlimited Liability Corporation Information", DRS_TEST_DOC, "CORP", "DIRECTOR_AFFIDAVIT", "DS0000101951", ""),
+   (True, STATIC_URL1, "Test court order", DRS_TEST_DOC, "FIRM", "CRTO", "DS0000101951", "Test court order"),
+   (True, STATIC_URL1, "Test court order", DRS_TEST_DOC, "COOP", "CRTO", "DS0000101951", "Test court order.pdf"),
+   (True, STATIC_URL1, "Court Order 1234", DRS_TEST_DOC, "CORP", "CRTO", "DS0000101951", "Court Order 1234"),
+   (True, STATIC_URL1, "Court Order 1234", DRS_TEST_DOC, "CORP", "CRTO", "DS0000101951", ""),
+   (False, STATIC_URL3, "Test court order", DRS_TEST_DOC, "CORP", "CRTO", "DS0000101951", "Test court order.pdf"),
+   (False, STATIC_URL2, "Test court order", DRS_TEST_DOC, "CORP", "CRTO", "DS0000101952", ""),
+   (False, STATIC_URL2, "Test court order", DRS_TEST_DOC, "CORP", "CRTO", "DS0000101952", None),
+   (False, STATIC_URL1, "Court Order 1234", DRS_TEST_DOC, "CORP", "COSD", "DS0000101951", ""),
+   (False, STATIC_URL1, "Court Order 1234", DRS_TEST_DOC, "CORP", "COSD", "DS0000101951", "Court Order.pdf"),
+   (False, STATIC_URL1, None, DRS_TEST_DOC, "CORP", "COSD", "DS0000101951", "Court Order.pdf"),
+   (False, STATIC_URL1, "", DRS_TEST_DOC, "CORP", "COSD", "DS0000101951", "Court Order.pdf"),
+   (False, STATIC_URL1, "Unlimited Liability Corporation Information", DRS_TEST_DOC, "CORP", "CRTO", "DS0000101951", ""),
+ ]
+# testdata pattern is ({url}, {drs_id})
+TEST_STATIC_URL_DRS_ID_DATA = [
+   (STATIC_URL1, ""),
+   (STATIC_URL2, "DS0000101951"),
+   (STATIC_URL4, "DS0000101953"),
+   (STATIC_URL5, "DS0000101954"),
+]
 
 @pytest.mark.parametrize("has_data,report_key,report_type", TEST_REPORT_META_DATA)
 def test_report_meta_type(session, has_data, report_key, report_type):
@@ -390,6 +433,30 @@ def test_update_ledger_docs(session, desc, doc_data, drs_data, receipt, filing,
         assert text_results.find(static) > 0
 
 
+@pytest.mark.parametrize("match,url,name,drs_doc,drs_class,drs_type,drs_id,drs_filename", TEST_STATIC_DOC_MATCH_DATA)
+def test_match_static_doc(session, match, url, name, drs_doc, drs_class, drs_type, drs_id, drs_filename):
+    """Assert that DRS matching on static documents when building download URLs works as expected."""
+    doc = copy.deepcopy(drs_doc)
+    doc["documentClass"] = drs_class
+    doc["documentType"] = drs_type
+    doc["name"] = drs_filename
+    if drs_id:
+        doc["identifier"] = drs_id
+    static_doc = {
+        "name": name,
+        "url": url
+    }
+    result = DocumentService().is_static_doc_match(doc, static_doc)
+    assert result == match
+
+
+@pytest.mark.parametrize("url,drs_id", TEST_STATIC_URL_DRS_ID_DATA)
+def test_url_drs_id_doc(session, url, drs_id):
+    """Assert that extracting a DRS ID from a url file key when building download URLs works as expected."""
+    result = DocumentService().get_url_drs_id(url)
+    assert result == drs_id
+
+
 def test_create_document(app, session, mock_bearer_token, mock_doc_service):
     founding_date = datetime.now(UTC)
     business = factory_business('CP1234567', founding_date=founding_date)