Merge branch 'main' into 599-feat-add-logic-to-download-results

Author: ERosendo

cl/api/templates/recap-api-docs-vlatest.html

@@ -229,7 +229,7 @@ <h4 id="dockets">Purchasing Dockets</h4>
   <h2 id="recap-upload">RECAP Upload API <small>— <code>{% url "processingqueue-list" version=version %}</code></small></h2>
   <p>This API is used by the RECAP extension and a handful of special partners to upload PACER content to the RECAP Archive. This API is not available to the public. If you have a collection of PACER data you wish to donate to the RECAP Archive so it is permanently available to the public, please <a href="{% url "contact" %}">get in touch</a>.
   </p>
-  <p>We describe the process for completing these uploads below, and you can see examples of them in <a href="https://github.com/freelawproject/courtlistener/blob/main/cl/recap/tests.py">CourtListener's automated test suite</a>. Uploads to these endpoints should be done using HTTP <code>POST</code> requests and multipart form data.
+  <p>We describe the process for completing these uploads below, and you can see examples of them in <a href="https://github.com/freelawproject/courtlistener/blob/main/cl/recap/tests/tests.py">CourtListener's automated test suite</a>. Uploads to these endpoints should be done using HTTP <code>POST</code> requests and multipart form data.
   </p>
   <p>When you make an upload, you create a <code>Processing Queue</code> object in the CourtListener system. This object will be returned in the HTTP response to your upload, so you will know its ID. This object will contain the fields you uploaded, and the following fields will be populated as the item is processed:
   </p>

cl/corpus_importer/signals.py

@@ -52,12 +52,19 @@ def update_latest_case_id_and_schedule_iquery_sweep(docket: Docket) -> None:
         tasks_to_schedule = (
             incoming_pacer_case_id - iquery_pacer_case_id_current
         )
+        logger.info(
+            "Found %s tasks to schedule for pacer case IDs ranging from %s to %s.",
+            tasks_to_schedule,
+            iquery_pacer_case_id_current,
+            incoming_pacer_case_id,
+        )
         if tasks_to_schedule > 10_800:
-            # Considering a Celery countdown of 1 second and a visibility_timeout
-            # of 6 hours, the maximum countdown time should be set to 21,600 to
-            # avoid a celery runaway. It's safer to abort if more than 10,800
-            # tasks are attempted to be scheduled. This could indicate an issue
-            # with retrieving the highest_known_pacer_case_id or a loss of the
+            # Considering a Celery countdown of 1 second applied via
+            # throttle_task and a visibility_timeout of 6 hours, the maximum
+            # countdown time should be set to 21,600 to avoid a celery runaway.
+            # It's safer to abort if more than 10,800 tasks are attempted to be
+            # scheduled. This could indicate an issue with retrieving the
+            # highest_known_pacer_case_id or a loss of the
             # iquery_pacer_case_id_current for the court in Redis.
             logger.error(
                 "Tried to schedule more than 10,800 iquery pages to scrape for "
@@ -66,20 +73,21 @@ def update_latest_case_id_and_schedule_iquery_sweep(docket: Docket) -> None:
             )
             release_redis_lock(r, update_lock_key, lock_value)
             return None
-        task_scheduled_countdown = 0
+        task_to_schedule_count = 0
         while iquery_pacer_case_id_current + 1 < incoming_pacer_case_id:
             iquery_pacer_case_id_current += 1
-            task_scheduled_countdown += 1
-            # Schedule the next task with a 1-second countdown increment
+            task_to_schedule_count += 1
+            # Schedule the next task.
             make_docket_by_iquery_sweep.apply_async(
                 args=(court_id, iquery_pacer_case_id_current),
                 kwargs={"skip_iquery_sweep": True},
-                countdown=task_scheduled_countdown,
                 queue=settings.CELERY_IQUERY_QUEUE,
             )
             logger.info(
-                f"Enqueued iquery docket case ID: {iquery_pacer_case_id_current} "
-                f"for court {court_id} with countdown {task_scheduled_countdown}"
+                "Enqueued %s iquery docket with case ID: %s for court %s",
+                task_to_schedule_count,
+                iquery_pacer_case_id_current,
+                court_id,
             )
 
         # Update the iquery_pacer_case_id_current in Redis

cl/corpus_importer/tasks.py

@@ -1964,47 +1964,6 @@ def create_attachment_pq(
     return pq
 
 
-# TODO: Remove after the new related methods have been rolled out.
-@app.task(bind=True, ignore_result=True)
-def make_attachment_pq_object(
-    self: Task,
-    attachment_report: AttachmentPage,
-    rd_pk: int,
-    user_pk: int,
-    att_report_text: str | None = None,
-) -> int:
-    """Create an item in the processing queue for an attachment page.
-
-    This is a helper shim to convert attachment page results into processing
-    queue objects that can be processed by our standard pipeline.
-
-    :param self: The celery task
-    :param attachment_report: An AttachmentPage object that's already queried
-    a page and populated its data attribute.
-    :param rd_pk: The RECAP document that the attachment page is associated
-    with
-    :param user_pk: The user to associate with the ProcessingQueue object when
-    it's created.
-    :param att_report_text: The attachment page report text if we got it from a
-    notification free look link.
-    :return: The pk of the ProcessingQueue object that's created.
-    """
-    rd = RECAPDocument.objects.get(pk=rd_pk)
-    user = User.objects.get(pk=user_pk)
-    pq = ProcessingQueue(
-        court_id=rd.docket_entry.docket.court_id,
-        uploader=user,
-        upload_type=UPLOAD_TYPE.ATTACHMENT_PAGE,
-        pacer_case_id=rd.docket_entry.docket.pacer_case_id,
-    )
-    if att_report_text is None:
-        att_report_text = attachment_report.response.text
-    pq.filepath_local.save(
-        "attachment_page.html", ContentFile(att_report_text.encode())
-    )
-    return pq.pk
-
-
 @app.task(bind=True, ignore_result=True)
 def save_attachment_pq_object(
     self: Task,

cl/lib/model_helpers.py

@@ -516,7 +516,7 @@ def linkify_orig_docket_number(agency: str, og_docket_number: str) -> str:
     """
     # Simple pattern for Federal Register citations
     fr_match = re.search(
-        r"(\d{1,3})\s*(?:FR|Fed\.?\s*Reg\.?)\s*(\d{1,5}(?:,\d{3})*)",
+        r"(\d{1,3})\s*(?:FR|Fed\.?\s*Reg\.?)\s*(\d{1,6}(?:,\d{3})*)",
         og_docket_number,
     )
 

cl/lib/tests.py

@@ -1394,6 +1394,16 @@ def test_linkify_orig_docket_number(self):
                 "89 Fed. Reg. 34,620",
                 "https://www.federalregister.gov/citation/89-FR-34,620",
             ),
+            (
+                "Animal and Plant Health Inspection Service",
+                "89 FR 106981",
+                "https://www.federalregister.gov/citation/89-FR-106981",
+            ),
+            (
+                "Animal and Plant Health Inspection Service",
+                "89 Fed. Reg. 106,981",
+                "https://www.federalregister.gov/citation/89-FR-106,981",
+            ),
             (
                 "Environmental Protection Agency",
                 "EPA-HQ-OW-2020-0005",

cl/search/management/commands/fix_rd_broken_links.py

@@ -132,6 +132,7 @@ def __init__(self, *args, **kwargs):
         self.chunk_size = None
         self.interval = None
         self.ids = None
+        self.docket_batch_size = None
 
     def add_arguments(self, parser):
         parser.add_argument(
@@ -146,6 +147,12 @@ def add_arguments(self, parser):
             default="100",
             help="The number of items to index in a single celery task.",
         )
+        parser.add_argument(
+            "--docket-batch-size",
+            type=int,
+            default="2000",
+            help="The number of docket_ids to process per batch.",
+        )
         parser.add_argument(
             "--auto-resume",
             action="store_true",
@@ -180,30 +187,24 @@ def add_arguments(self, parser):
             required=False,
         )
 
-    def get_and_fix_rds(self, cut_off_date: datetime) -> int:
-        """Get the dockets with broken RECAPDocument links and fix their RDs by
-        re-indexing.
-
-        :param cut_off_date: The cutoff date to filter docket events.
-        :return: The number of dockets affected.
+    def _process_docket_id_batch(
+        self, batch_number: int, batch_ids: list[int]
+    ) -> int:
+        """Process a batch of docket IDs with broken RECAPDocument links and
+        fix their RDs by re-indexing.
+        :param batch_ids: A batch list of docket IDs.
+        :return: The number of RECAP documents processed for this batch.
         """
 
-        chunk = []
         affected_rds = 0
-        docket_ids_to_fix_queryset = get_dockets_to_fix(
-            cut_off_date, self.pk_offset, self.ids
-        )
+        chunk = []
         rd_queryset = (
-            RECAPDocument.objects.filter(
-                docket_entry__docket_id__in=Subquery(
-                    docket_ids_to_fix_queryset
-                )
-            )
+            RECAPDocument.objects.filter(docket_entry__docket_id__in=batch_ids)
             .order_by("pk")
             .values_list("pk", flat=True)
         )
         logger.info(
-            "Getting the count of recap documents that need to be fixed."
+            "Getting the count of recap documents that need to be fixed in this batch."
         )
         count = rd_queryset.count()
         for rd_id_to_fix in rd_queryset.iterator():
@@ -226,21 +227,67 @@ def get_and_fix_rds(self, cut_off_date: datetime) -> int:
 
                 chunk = []
                 logger.info(
-                    "Processed %d/%d (%.0f%%), last PK fixed: %s",
+                    "Processed RDs in batch %s: %d/%d (%.0f%%), last PK fixed: %s",
+                    batch_number,
                     affected_rds,
                     count,
                     (affected_rds * 100.0) / count,
                     rd_id_to_fix,
                 )
 
-                if not affected_rds % 1000:
-                    # Log every 1000 documents processed.
-                    log_last_document_indexed(
-                        rd_id_to_fix, compose_redis_key()
-                    )
-
         return count
 
+    def get_and_fix_rds(self, cut_off_date: datetime) -> int:
+        """Get the dockets with broken RECAPDocument links and process them in
+        batches of docket IDs.
+
+        :param cut_off_date: The cutoff date to filter docket events.
+        :return: The total number of RECAP documents affected.
+        """
+
+        all_affected_rds = 0
+        docket_ids_queryset = get_dockets_to_fix(
+            cut_off_date, self.pk_offset, self.ids
+        )
+
+        batch_ids = []
+        batch_number = 0
+        affected_dockets = 0
+        for docket_id in docket_ids_queryset.iterator(
+            chunk_size=self.docket_batch_size
+        ):
+            batch_ids.append(docket_id)
+            affected_dockets += 1
+            if len(batch_ids) >= self.docket_batch_size:
+                batch_number += 1
+                logger.info(
+                    "Processing docket_ids batch %d with %d IDs",
+                    batch_number,
+                    len(batch_ids),
+                )
+                all_affected_rds += self._process_docket_id_batch(
+                    batch_number, batch_ids
+                )
+                batch_ids = []
+
+            if not affected_dockets % 1000:
+                # Log every 1000 documents processed.
+                log_last_document_indexed(docket_id, compose_redis_key())
+
+        # Process any remaining docket_ids in the final chunk.
+        if batch_ids:
+            batch_number += 1
+            logger.info(
+                "Processing final docket_ids batch %d with %d IDs",
+                batch_number,
+                len(batch_ids),
+            )
+            all_affected_rds += self._process_docket_id_batch(
+                batch_number, batch_ids
+            )
+
+        return all_affected_rds
+
     def handle(self, *args, **options):
         super().handle(*args, **options)
         self.options = options
@@ -250,6 +297,7 @@ def handle(self, *args, **options):
         self.throttle = CeleryThrottle(queue_name=self.queue)
         self.interval = self.options["interval"]
         self.ids = options.get("ids")
+        self.docket_batch_size = options.get("docket_batch_size")
         auto_resume = options["auto_resume"]
         if auto_resume:
             self.pk_offset = get_last_parent_document_id_processed(
@@ -259,9 +307,9 @@ def handle(self, *args, **options):
                 f"Auto-resume enabled starting indexing from ID: {self.pk_offset}."
             )
         start_date: datetime = options["start_date"]
-        affected_rds = self.get_and_fix_rds(start_date)
+        affected_dockets = self.get_and_fix_rds(start_date)
         logger.info(
             "Successfully fixed %d items from pk %s.",
-            affected_rds,
+            affected_dockets,
             self.pk_offset,
         )

cl/search/tasks.py

@@ -1188,106 +1188,6 @@ def index_parent_or_child_docs_in_es(
         parent_es_document._index.refresh()
 
 
-# TODO: Remove after the new task is rolled out and all scheduled tasks have been processed.
-@app.task(
-    bind=True,
-    autoretry_for=(ConnectionError,),
-    max_retries=3,
-    interval_start=5,
-    ignore_result=True,
-)
-def index_parent_or_child_docs(
-    self: Task,
-    instance_ids: list[int],
-    search_type: str,
-    document_type: str | None,
-    testing_mode: bool = False,
-) -> None:
-    """Index parent or child documents in Elasticsearch.
-
-    :param self: The Celery task instance
-    :param instance_ids: The parent instance IDs to index.
-    :param search_type: The Search Type to index parent and child docs.
-    :param document_type: The document type to index, 'parent' or 'child' documents
-    :param testing_mode: Set to True to enable streaming bulk, which is used in
-     TestCase-based tests because parallel_bulk is incompatible with them.
-    https://github.com/freelawproject/courtlistener/pull/3324#issue-1970675619
-    Default is False.
-    :return: None
-    """
-
-    parent_instances = QuerySet()
-    child_instances = QuerySet()
-    use_streaming_bulk = True if testing_mode else False
-    match search_type:
-        case SEARCH_TYPES.RECAP:
-            parent_es_document = DocketDocument
-            child_es_document = ESRECAPDocument
-            child_id_property = "RECAP"
-            if document_type == "parent":
-                parent_instances = Docket.objects.filter(pk__in=instance_ids)
-            elif document_type == "child":
-                child_instances = RECAPDocument.objects.filter(
-                    pk__in=instance_ids
-                )
-        case SEARCH_TYPES.OPINION:
-            parent_es_document = OpinionClusterDocument
-            child_es_document = OpinionDocument
-            child_id_property = "OPINION"
-            if document_type == "parent":
-                parent_instances = OpinionCluster.objects.filter(
-                    pk__in=instance_ids
-                )
-            elif document_type == "child":
-                child_instances = Opinion.objects.filter(pk__in=instance_ids)
-        case SEARCH_TYPES.ORAL_ARGUMENT:
-            parent_es_document = AudioDocument
-            if document_type == "parent":
-                parent_instances = Audio.objects.filter(pk__in=instance_ids)
-        case _:
-            return
-
-    base_doc = {
-        "_op_type": "index",
-        "_index": parent_es_document._index._name,
-    }
-    if document_type == "child":
-        # Then index only child documents in bulk.
-        failed_docs = index_documents_in_bulk_from_queryset(
-            child_instances,
-            child_es_document,
-            base_doc,
-            child_id_property=child_id_property,
-            use_streaming_bulk=use_streaming_bulk,
-        )
-
-        if failed_docs:
-            model_label = child_es_document.Django.model.__name__.capitalize()
-            logger.error(
-                f"Error indexing documents from {model_label}, "
-                f"Failed Doc IDs are: {failed_docs}"
-            )
-
-    if document_type == "parent":
-        # Index only parent documents.
-        failed_docs = index_documents_in_bulk_from_queryset(
-            parent_instances,
-            parent_es_document,
-            base_doc,
-            use_streaming_bulk=use_streaming_bulk,
-        )
-        if failed_docs:
-            model_label = parent_es_document.Django.model.__name__.capitalize()
-            logger.error(
-                f"Error indexing documents from {model_label}, "
-                f"Failed Doc IDs are: {failed_docs}"
-            )
-
-    if settings.ELASTICSEARCH_DSL_AUTO_REFRESH:
-        # Set auto-refresh, used for testing.
-        parent_es_document._index.refresh()
-
-
 @app.task(
     bind=True,
     autoretry_for=(ConnectionError, ConflictError, ConnectionTimeout),