612 Introduced RECAP Search Alerts sweep index

cl/alerts/management/commands/cl_send_recap_alerts.py

@@ -0,0 +1,112 @@
+import traceback
+import datetime
+
+from asgiref.sync import async_to_sync
+from django.contrib.auth.models import User
+from django.http import QueryDict
+from django.utils.timezone import now
+from elasticsearch.exceptions import RequestError, TransportError
+
+from cl.lib.command_utils import VerboseCommand, logger
+from cl.lib.elasticsearch_utils import do_es_sweep_alert_query
+from cl.search.documents import DocketSweepDocument
+from cl.search.models import SEARCH_TYPES
+from cl.stats.utils import tally_stat
+from cl.alerts.tasks import send_search_alert_emails
+from cl.alerts.models import Alert
+from cl.search.exception import (
+    BadProximityQuery,
+    UnbalancedParenthesesQuery,
+    UnbalancedQuotesQuery,
+)
+from cl.alerts.utils import recap_document_hl_matched, query_includes_rd_field
+
+
+def index_daily_recap_documents():
+    # TODO implement
+    pass
+
+def has_rd_hit_been_triggered():
+    # TODO implement
+    return False
+
+def has_docket_hit_been_triggered():
+    # TODO implement
+    return True
+
+def query_and_send_alerts(rate):
+    alert_users = User.objects.filter(alerts__rate=rate).distinct()
+    alerts_sent_count = 0
+    now_time = datetime.datetime.now()
+    for user in alert_users:
+        alerts = user.alerts.filter(rate=rate)
+        logger.info(f"Running alerts for user '{user}': {alerts}")
+
+        hits = []
+        alerts_to_update = []
+        for alert in alerts:
+            search_params = QueryDict(alert.query.encode(), mutable=True)
+            includes_rd_fields = query_includes_rd_field(search_params)
+
+            try:
+                search_query = DocketSweepDocument.search()
+                results, total_hits = do_es_sweep_alert_query(
+                    search_query,
+                    search_params,
+                )
+            except (UnbalancedParenthesesQuery,
+        UnbalancedQuotesQuery,
+        BadProximityQuery,TransportError, ConnectionError, RequestError):
+                traceback.print_exc()
+                logger.info(
+                    f"Search for this alert failed: {alert.query}\n"
+                )
+                continue
+
+            alerts_to_update.append(alert.pk)
+            if len(results) > 0:
+                search_type = search_params.get("type", SEARCH_TYPES.OPINION)
+                results_to_send = []
+                for hit in results:
+                    if not includes_rd_fields:
+                        rds_to_send = [rd_hit for rd_hit in hit["child_docs"]
+                                       if not recap_document_hl_matched(
+                                rd_hit) and not has_rd_hit_been_triggered()]
+                        if rds_to_send:
+                            hit["child_docs"] = rds_to_send
+                            results_to_send.append(hit)
+
+                hits.append(
+                    [alert, search_type, results_to_send, len(results_to_send)]
+                )
+                alert.query_run = search_params.urlencode()
+                alert.date_last_hit = now()
+                alert.save()
+
+        if hits:
+            send_search_alert_emails.delay([(user.pk, hits)])
+            alerts_sent_count += 1
+
+        # Update Alert's date_last_hit in bulk.
+        Alert.objects.filter(id__in=alerts_to_update).update(
+            date_last_hit=now_time
+        )
+        async_to_sync(tally_stat)(f"alerts.sent.{rate}", inc=alerts_sent_count)
+        logger.info(f"Sent {alerts_sent_count} {rate} email alerts.")
+
+
+def query_and_schedule_wly_and_mly_alerts():
+    # TODO implement
+    pass
+
+
+class Command(VerboseCommand):
+    help = "Send RECAP Search Alerts."
+
+    def handle(self, *args, **options):
+        super().handle(*args, **options)
+
+        index_daily_recap_documents()
+        query_and_send_alerts(Alert.REAL_TIME)
+        query_and_send_alerts(Alert.DAILY)
+        query_and_schedule_wly_and_mly_alerts()

cl/alerts/tests/tests_recap_alerts.py

@@ -0,0 +1,190 @@
+from unittest import mock
+
+import time_machine
+from asgiref.sync import sync_to_async
+from django.core import mail
+from django.core.management import call_command
+from django.utils.timezone import now
+
+from cl.alerts.factories import AlertFactory
+from cl.alerts.models import SEARCH_TYPES, Alert
+from cl.alerts.utils import query_includes_rd_field, recap_document_hl_matched
+from cl.api.factories import WebhookFactory
+from cl.api.models import WebhookEventType
+from cl.donate.models import NeonMembership
+from cl.lib.elasticsearch_utils import do_es_sweep_alert_query
+from cl.lib.test_helpers import RECAPSearchTestCase
+from cl.search.documents import DocketSweepDocument
+from cl.tests.cases import ESIndexTestCase, TestCase
+from cl.tests.utils import MockResponse
+from cl.users.factories import UserProfileWithParentsFactory
+
+
+class RECAPAlertsSweepIndexTest(
+    RECAPSearchTestCase, ESIndexTestCase, TestCase
+):
+    """
+    RECAP Alerts Sweep Index Tests
+    """
+
+    @classmethod
+    def setUpTestData(cls):
+        cls.rebuild_index("people_db.Person")
+        cls.rebuild_index("search.Docket")
+        cls.mock_date = now().replace(day=15, hour=0)
+        with time_machine.travel(cls.mock_date, tick=False):
+            super().setUpTestData()
+            call_command(
+                "cl_index_parent_and_child_docs",
+                search_type=SEARCH_TYPES.RECAP,
+                queue="celery",
+                pk_offset=0,
+                testing_mode=True,
+                sweep_index=True,
+            )
+
+            cls.user_profile = UserProfileWithParentsFactory()
+            NeonMembership.objects.create(
+                level=NeonMembership.LEGACY, user=cls.user_profile.user
+            )
+            cls.webhook_enabled = WebhookFactory(
+                user=cls.user_profile.user,
+                event_type=WebhookEventType.SEARCH_ALERT,
+                url="https://example.com/",
+                enabled=True,
+            )
+            cls.search_alert = AlertFactory(
+                user=cls.user_profile.user,
+                rate=Alert.REAL_TIME,
+                name="Test Alert Docket Only",
+                query='q="401 Civil"&type=r',
+            )
+            cls.search_alert_2 = AlertFactory(
+                user=cls.user_profile.user,
+                rate=Alert.REAL_TIME,
+                name="Test Alert RECAP Only",
+                query='q="Mauris iaculis, leo sit amet hendrerit vehicula"&type=r',
+            )
+            cls.search_alert_3 = AlertFactory(
+                user=cls.user_profile.user,
+                rate=Alert.DAILY,
+                name="Test Cross object",
+                query="q=SUBPOENAS SERVED OFF Mauris iaculis&type=r",
+            )
+
+    async def test_recap_document_hl_matched(self) -> None:
+        """Test recap_document_hl_matched method that determines weather a hit
+        contains RECAPDocument HL fields."""
+        # Docket-only query
+        search_params = {
+            "type": SEARCH_TYPES.RECAP,
+            "q": '"401 Civil"',
+        }
+        search_query = DocketSweepDocument.search()
+        results, total_hits = await sync_to_async(do_es_sweep_alert_query)(
+            search_query,
+            search_params,
+        )
+        docket_result = results[0]
+        for rd in docket_result["child_docs"]:
+            rd_field_matched = recap_document_hl_matched(rd)
+            self.assertEqual(rd_field_matched, False)
+
+        # RECAPDocument-only query
+        search_params = {
+            "type": SEARCH_TYPES.RECAP,
+            "q": '"Mauris iaculis, leo sit amet hendrerit vehicula"',
+        }
+        search_query = DocketSweepDocument.search()
+        results, total_hits = await sync_to_async(do_es_sweep_alert_query)(
+            search_query,
+            search_params,
+        )
+        docket_result = results[0]
+        for rd in docket_result["child_docs"]:
+            rd_field_matched = recap_document_hl_matched(rd)
+            self.assertEqual(rd_field_matched, True)
+
+        # Cross-object query
+        search_params = {
+            "type": SEARCH_TYPES.RECAP,
+            "q": "SUBPOENAS SERVED OFF Mauris iaculis",
+        }
+        search_query = DocketSweepDocument.search()
+        results, total_hits = await sync_to_async(do_es_sweep_alert_query)(
+            search_query,
+            search_params,
+        )
+        docket_result = results[0]
+        for rd in docket_result["child_docs"]:
+            rd_field_matched = recap_document_hl_matched(rd)
+            self.assertEqual(rd_field_matched, True)
+
+    async def test_query_includes_rd_field(self) -> None:
+        """Test query_includes_rd_field method that checks if a query
+        includes any indexed fields in the query string or filters specific to
+        RECAP Documents.
+        """
+
+        # Docket-only query
+        search_params = {
+            "type": SEARCH_TYPES.RECAP,
+            "q": '"401 Civil"',
+        }
+        self.assertEqual(query_includes_rd_field(search_params), False)
+
+        # RECAPDocument-only query
+        search_params = {
+            "type": SEARCH_TYPES.RECAP,
+            "q": 'description:"lorem ipsum"',
+        }
+        self.assertEqual(query_includes_rd_field(search_params), True)
+
+        # Cross-object query
+        search_params = {
+            "type": SEARCH_TYPES.RECAP,
+            "q": 'case_name:"American v." description:"lorem ipsum"',
+        }
+        self.assertEqual(query_includes_rd_field(search_params), True)
+
+        # Docket-only query
+        search_params = {
+            "type": SEARCH_TYPES.RECAP,
+            "q": "",
+            "case_name": "SUBPOENAS",
+        }
+        self.assertEqual(query_includes_rd_field(search_params), False)
+
+        # RECAPDocument-only query
+        search_params = {
+            "type": SEARCH_TYPES.RECAP,
+            "q": "",
+            "description": "Lorem",
+        }
+        self.assertEqual(query_includes_rd_field(search_params), True)
+
+        # Cross-object query
+        search_params = {
+            "type": SEARCH_TYPES.RECAP,
+            "q": "",
+            "case_name": "SUBPOENAS",
+            "document_number": 1,
+        }
+        self.assertEqual(query_includes_rd_field(search_params), True)
+
+    def test_filter_out_alerts_to_send(self) -> None:
+        """Test RECAP alerts hit can be properly filtered out according to
+        their query and hits matched conditions.
+        """
+
+        with mock.patch(
+            "cl.api.webhooks.requests.post",
+            side_effect=lambda *args, **kwargs: MockResponse(
+                200, mock_raw=True
+            ),
+        ), time_machine.travel(self.mock_date, tick=False):
+            call_command("cl_send_recap_alerts")
+
+        self.assertEqual(
+            len(mail.outbox), 2, msg="Outgoing emails don't match."
+        )

cl/alerts/utils.py

@@ -4,7 +4,7 @@
 from django.conf import settings
 from django.http import QueryDict
 from elasticsearch_dsl import Q, Search
-from elasticsearch_dsl.response import Response
+from elasticsearch_dsl.response import Hit, Response
 
 from cl.alerts.models import (
     SCHEDULED_ALERT_HIT_STATUS,
@@ -14,9 +14,15 @@
 )
 from cl.lib.command_utils import logger
 from cl.lib.elasticsearch_utils import add_es_highlighting
+from cl.lib.types import CleanData
+from cl.search.constants import (
+    ALERTS_HL_TAG,
+    SEARCH_RECAP_CHILD_HL_FIELDS,
+    recap_document_filters,
+    recap_document_indexed_fields,
+)
 from cl.search.documents import AudioPercolator
 from cl.search.models import SEARCH_TYPES, Docket
-from cl.users.models import UserProfile
 
 
 @dataclass
@@ -138,3 +144,46 @@ def alert_hits_limit_reached(alert_pk: int, user_pk: int) -> bool:
         )
         return True
     return False
+
+
+def recap_document_hl_matched(rd_hit: Hit) -> bool:
+    """Determine whether HL matched a RECAPDocument text field.
+
+    :param rd_hit: The ES hit.
+    :return: True if the hit matched a RECAPDocument field. Otherwise, False.
+    """
+
+    matched_rd_hl = set()
+    rd_hl_fields = set(SEARCH_RECAP_CHILD_HL_FIELDS.keys())
+    if hasattr(rd_hit, "highlight"):
+        highlights = rd_hit.highlight.to_dict()
+        matched_rd_hl.update(
+            hl_key
+            for hl_key, hl_value in highlights.items()
+            for hl in hl_value
+            if f"<{ALERTS_HL_TAG}>" in hl
+        )
+    if matched_rd_hl and matched_rd_hl.issubset(rd_hl_fields):
+        return True
+    return False
+
+
+def query_includes_rd_field(query_params: CleanData) -> bool:
+    """Determine whether the query includes any indexed fields in the query
+    string or filters specific to RECAP Documents.
+
+    :param query_params: The query parameters.
+    :return: True if any recap document fields or filters are included in the
+    query, otherwise False.
+    """
+
+    query_string = query_params.get("q", "")
+    for rd_field in recap_document_indexed_fields:
+        if f"{rd_field}:" in query_string:
+            return True
+
+    for rd_filter in recap_document_filters:
+        if query_params.get(rd_filter, ""):
+            return True
+
+    return False