Add rulemaking document sorting logic and tests

tests/test_legal/test_rm_load.py

@@ -4,10 +4,10 @@
 
 import pytest
 from tests.common import TEST_CONN, BaseTestCase
-
+import datetime
 from webservices.common.models import db
 from sqlalchemy import text
-# from webservices.rulemaking_docs.rulemaking import get_rulemaking
+from webservices.legal.rulemaking_docs.rulemaking import sort_documents_tier_one
 
 EMPTY_SET = set()
 
@@ -55,6 +55,46 @@ def create_rm(self, rm_id, rm):
             }
         )
 
+    def test_rm_sort(self):
+        documents = [
+            {
+                "doc_date": datetime.date(2026, 3, 17),
+                "level_1": 10,
+                "doc_id": 7
+            },
+            {
+                "doc_date": None,
+                "level_1": 3,
+                "doc_id": 8,
+                "level_2_labels": [
+                    {
+                        "level_2_docs": [
+                            {"doc_date": datetime.date(2024, 12, 19)},
+                            {"doc_date": datetime.date(2026, 3, 17)},
+                            {"doc_date": None}
+                        ]
+                    }
+                ]
+            },
+            {
+                "doc_date": None,
+                "level_1": 15,
+                "doc_id": 9,
+                "level_2_labels": [
+                    {
+                        "level_2_docs": [
+                            {"doc_date": None},
+                            {"doc_date": None}
+                        ]
+                    }
+                ]
+            }
+        ]
+        sort_documents_tier_one(documents)
+        self.assertEqual(documents[0]["doc_id"], 9)
+        self.assertEqual(documents[1]["doc_id"], 8)
+        self.assertEqual(documents[2]["doc_id"], 7)
+
     # @patch("webservices.legal.legal_docs.advisory_opinions.get_bucket")
     # @patch("webservices.legal.legal_docs.advisory_opinions.create_opensearch_client")
     # @patch("webservices.legal.legal_docs.opensearch_management.create_index")

webservices/legal/constants.py

@@ -108,6 +108,20 @@
     21: "Termination of Rulemaking",
 }
 
+TIER_PRECEDENCE = {
+    15: 0,  # Notice of Disposition (highest priority)
+    5: 1,   # Advance NPRM
+    13: 2,  # Interim Final Rules (swapped with tier 14)
+    14: 3,  # Notice of Availability
+    2: 4,   # NPRM
+    17: 5,  # Supplemental NPRM
+    3: 6,   # Notice of Hearing
+    6: 7,   # Notice of Change of Hearing Date
+    8: 8,   # Correction to Final Rules
+    9: 9,   # Correction to Final Rules and E&J
+    1: 10,  # Final Rules
+    10: 11,  # Explanation and Justification
+}
 
 LEVEL_1_2_MAP = {
     1: {

webservices/legal/rulemaking_docs/rulemaking.py

@@ -1,6 +1,7 @@
 import logging
 import json
 import copy
+import datetime
 import webservices.legal.constants as constants
 from webservices.common.models import db
 from webservices.legal.utils_opensearch import (
@@ -589,9 +590,35 @@ def get_documents(rm_no, rm_id, bucket):
                     "level_2_labels": get_level_2_labels(rm_no, rm_id, row_list["level_1"], bucket),
                 }
                 documents.append(document)
+        if len(documents) > 1:
+            sort_documents_tier_one(documents)
+
         return documents
 
 
+def sort_documents_tier_one(documents):
+    def sort_key(doc):
+        sort_date = get_doc_date(doc)
+
+        tier = constants.TIER_PRECEDENCE.get(doc.get('level_1'), 999)
+
+        return (sort_date, -tier)
+
+    documents.sort(key=sort_key, reverse=True)
+
+
+def get_doc_date(doc):
+    if doc.get("doc_date"):
+        return doc["doc_date"]
+
+    for l2 in reversed(doc.get("level_2_labels", [])):
+        for l2_doc in reversed(l2.get("level_2_docs", [])):
+            if l2_doc.get("doc_date"):
+                return l2_doc["doc_date"]
+
+    return datetime.date.today()
+
+
 def get_level_2_labels(rm_no, rm_id, level_1, bucket):
     level_2_labels = []
     with db.engine.begin() as conn: