4081 Get latest cases from iquery pages

cl/corpus_importer/apps.py

@@ -0,0 +1,9 @@
+from django.apps import AppConfig
+
+
+class CorpusImporterConfig(AppConfig):
+    name = "cl.corpus_importer"
+
+    def ready(self):
+        # Implicitly connect a signal handlers decorated with @receiver.
+        from cl.corpus_importer import signals

cl/corpus_importer/factories.py

@@ -71,3 +71,14 @@ class FreeOpinionRowDataFactory(factory.DictFactory):
     pacer_case_id = Faker("random_id_string")
     pacer_doc_id = Faker("random_id_string")
     pacer_seq_no = Faker("pyint", min_value=1, max_value=10000)
+
+
+class CaseQueryDataFactory(factory.DictFactory):
+    assigned_to_str = Faker("name_female")
+    case_name = Faker("case_name")
+    case_name_raw = Faker("case_name")
+    court_id = FuzzyText(length=4, chars=string.ascii_lowercase, suffix="d")
+    date_filed = Faker("date_object")
+    date_last_filing = Faker("date_object")
+    date_terminated = Faker("date_object")
+    docket_number = Faker("federal_district_docket_number")

cl/corpus_importer/management/commands/iquery_pages_probing_daemon.py

@@ -0,0 +1,97 @@
+import time
+
+from django.conf import settings
+from redis import ConnectionError
+
+from cl.corpus_importer.tasks import iquery_pages_probing
+from cl.corpus_importer.utils import make_iquery_probing_key
+from cl.lib.celery_utils import CeleryThrottle
+from cl.lib.command_utils import VerboseCommand, logger
+from cl.lib.redis_utils import create_redis_semaphore, get_redis_interface
+from cl.search.models import Court
+
+
+def enqueue_iquery_probing(court_id: str) -> bool:
+    """Get iquery forward probing semaphore.
+
+    :param court_id: The identifier for the court.
+    :return: A boolean indicating if the semaphore was successfully created.
+    """
+    key = make_iquery_probing_key(court_id)
+    return create_redis_semaphore("CACHE", key, ttl=60 * 10)
+
+
+def get_all_pacer_courts() -> list[str]:
+    """Retrieve all district and bankruptcy  PACER courts from the database.
+
+    :return: A list of Court IDs.
+    """
+    courts = (
+        Court.federal_courts.district_or_bankruptcy_pacer_courts().exclude(
+            pk__in=["uscfc", "arb", "cit"]
+        )
+    )
+    return list(courts.values_list("pk", flat=True))
+
+
+class Command(VerboseCommand):
+    help = "Run the iquery pages probing daemon."
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.options = {}
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--testing-iterations",
+            type=int,
+            default="0",
+            required=False,
+            help="The number of iterations to run on testing mode.",
+        )
+
+    def handle(self, *args, **options):
+        super().handle(*args, **options)
+
+        testing_iterations = options["testing_iterations"]
+        # If a new court is added to the DB. We should restart the daemon.
+        court_ids = get_all_pacer_courts()
+        iterations_completed = 0
+        q = settings.CELERY_IQUERY_QUEUE
+        # Create a queue equal than the number of courts we're doing.
+        throttle = CeleryThrottle(queue_name=q, min_items=len(court_ids))
+        r = get_redis_interface("CACHE")
+        testing = True if testing_iterations else False
+        while True:
+            for court_id in court_ids:
+                if r.exists(f"court_wait:{court_id}"):
+                    continue
+                throttle.maybe_wait()
+                try:
+                    newly_enqueued = enqueue_iquery_probing(court_id)
+                    if newly_enqueued:
+                        # No other probing being conducted for the court.
+                        # Enqueue it.
+                        iquery_pages_probing.delay(court_id, testing)
+                except ConnectionError:
+                    logger.info(
+                        "Failed to connect to redis. Waiting a bit and making "
+                        "a new connection."
+                    )
+                    time.sleep(10)
+                    # Continuing here will skip this court for this iteration; not
+                    # a huge deal.
+                    continue
+
+                if not testing_iterations:
+                    # Avoid waiting in testing mode.
+                    time.sleep(settings.IQUERY_PROBE_WAIT / len(court_ids))
+
+            if testing_iterations:
+                iterations_completed += 1
+            if (
+                testing_iterations
+                and iterations_completed >= testing_iterations
+            ):
+                # Perform only the indicated iterations for testing purposes.
+                break

cl/corpus_importer/signals.py

@@ -0,0 +1,122 @@
+from functools import partial
+
+from django.conf import settings
+from django.db import transaction
+from django.db.models.signals import post_save
+from django.dispatch import receiver
+
+from cl.corpus_importer.tasks import make_docket_by_iquery
+from cl.lib.redis_utils import (
+    acquire_atomic_redis_lock,
+    get_redis_interface,
+    make_update_pacer_case_id_key,
+    release_atomic_redis_lock,
+)
+from cl.search.models import Court, Docket
+
+
+def update_latest_case_id_and_schedule_iquery_sweep(docket: Docket) -> None:
+    """Updates the latest PACER case ID and schedules iquery retrieval tasks.
+
+    :param docket: The incoming Docket instance.
+    :return: None
+    """
+
+    r = get_redis_interface("CACHE")
+    court_id = docket.court.pk
+    # Get the latest pacer_case_id from Redis using a lock to avoid race conditions
+    # when getting and updating it.
+    update_lock_key = make_update_pacer_case_id_key(court_id)
+    # ttl one hour.
+    lock_value = acquire_atomic_redis_lock(r, update_lock_key, 60 * 60 * 1000)
+
+    current_iquery_pacer_case_id_final = int(
+        r.hget("iquery_pacer_case_id_final", court_id) or 0
+    )
+    iquery_pacer_case_id_status = int(
+        r.hget("iquery_pacer_case_id_status", court_id) or 0
+    )
+    incoming_pacer_case_id = int(docket.pacer_case_id)
+    updated_pacer_case_id = False
+    if incoming_pacer_case_id > current_iquery_pacer_case_id_final:
+        r.hset("iquery_pacer_case_id_final", court_id, incoming_pacer_case_id)
+        updated_pacer_case_id = True
+
+    if updated_pacer_case_id:
+        task_scheduled_countdown = 0
+
+        while iquery_pacer_case_id_status + 1 < incoming_pacer_case_id:
+            tasks_processed_in_this_batch = 0
+
+            # Schedule tasks in batches of IQUERY_SWEEP_BATCH_SIZE to avoid
+            # a celery runaway scheduling tasks with countdowns larger than
+            # the celery visibility_timeout.
+            while (
+                tasks_processed_in_this_batch
+                < settings.IQUERY_SWEEP_BATCH_SIZE
+                and iquery_pacer_case_id_status + 1 < incoming_pacer_case_id
+            ):
+                iquery_pacer_case_id_status += 1
+                tasks_processed_in_this_batch += 1
+                task_scheduled_countdown += 1
+                # Schedule the next task with a 1-second countdown increment
+                make_docket_by_iquery.apply_async(
+                    args=(court_id, iquery_pacer_case_id_status),
+                    kwargs={"from_iquery_scrape": True},
+                    countdown=task_scheduled_countdown,
+                )
+
+            # Update the status in Redis after each batch
+            r.hset(
+                "iquery_pacer_case_id_status",
+                court_id,
+                iquery_pacer_case_id_status,
+            )
+
+    # Release the lock once the whole process is complete.
+    release_atomic_redis_lock(r, update_lock_key, lock_value)
+
+
+@receiver(
+    post_save,
+    sender=Docket,
+    dispatch_uid="handle_update_latest_case_id_and_schedule_iquery_sweep",
+)
+def handle_update_latest_case_id_and_schedule_iquery_sweep(
+    sender, instance: Docket, created=False, update_fields=None, **kwargs
+) -> None:
+    """post_save Docket signal receiver to handle
+    update_latest_case_id_and_schedule_iquery_sweep
+    """
+
+    if hasattr(instance, "from_iquery_scrape") and instance.from_iquery_scrape:
+        # Early abort if this is an instance added by the iquery probing task
+        # or the iquery sweep scraper.
+        return None
+
+    # Only call update_latest_case_id_and_schedule_iquery_sweep if this is a
+    # new RECAP district or bankruptcy docket with pacer_case_id not added by
+    # iquery sweep tasks.
+    if (
+        created
+        and instance.pacer_case_id
+        and getattr(instance, "court", None)
+        and instance.court_id
+        in list(
+            Court.federal_courts.district_or_bankruptcy_pacer_courts()
+            .exclude(pk__in=["uscfc", "arb", "cit"])
+            .values_list("pk", flat=True)
+        )
+    ):
+        transaction.on_commit(
+            partial(update_latest_case_id_and_schedule_iquery_sweep, instance)
+        )
+
+
+if settings.TESTING:
+    # Disconnect handle_update_latest_case_id_and_schedule_iquery_sweep
+    # for all tests. It will be enabled only for tests where it is required.
+    post_save.disconnect(
+        sender=Docket,
+        dispatch_uid="handle_update_latest_case_id_and_schedule_iquery_sweep",
+    )