Merge develop back in

Author: fritzbrand

CHANGELOG.md

@@ -1,4 +1,30 @@
 # Changelog
+
+# 0.10.28
+1. Add intent label tasks
+   ([#646](https://github.com/praekeltfoundation/ndoh-hub/pull/646))
+
+# 0.10.27
+1. Bump django from 4.2.15 to 4.2.16
+([#628](https://github.com/praekeltfoundation/ndoh-hub/pull/628))
+2. Update CHANGELOG
+([#636](https://github.com/praekeltfoundation/ndoh-hub/pull/636))
+3. add test coverage to PR comment using py-cov-action
+([#638](https://github.com/praekeltfoundation/ndoh-hub/pull/638))
+4. update gibberish detection function
+([#640](https://github.com/praekeltfoundation/ndoh-hub/pull/640))
+5. update gibberish response
+([#641](https://github.com/praekeltfoundation/ndoh-hub/pull/641))
+6. update gibberish response
+([#642](https://github.com/praekeltfoundation/ndoh-hub/pull/642))
+7. update gibberish response dict
+([#643](https://github.com/praekeltfoundation/ndoh-hub/pull/643))
+8. return a dict in utils when gibberish detected
+([#644](https://github.com/praekeltfoundation/ndoh-hub/pull/644))
+# 0.10.26
+1. Remove seaworthy from hub
+([#635](https://github.com/praekeltfoundation/ndoh-hub/pull/635))
+([#634](https://github.com/praekeltfoundation/ndoh-hub/pull/634))
 # 0.10.25
 1. Remove mqr from hub
 ([#633](https://github.com/praekeltfoundation/ndoh-hub/pull/633))

aaq/tests/test_utils.py

@@ -133,6 +133,6 @@ def test_search_gibberish(self):
         search_request = responses.calls[0]
 
         assert search_request.response.status_code == 400
-        assert response.data == {
+        assert response == {
             "message": "Gibberish Detected",
         }

aaq/utils.py

@@ -3,7 +3,6 @@
 import requests
 from django.conf import settings
 from rest_framework import status
-from rest_framework.response import Response
 
 
 def check_urgency_v2(message_text):
@@ -41,10 +40,7 @@ def search(query_text, generate_llm_response, query_metadata):
     ):
         error_detail = response.json().get("detail", "")
         if "Gibberish text detected" in error_detail:
-            json_msg = {
-                "message": "Gibberish Detected",
-            }
-            return Response(json_msg, status=status.HTTP_200_OK)
+            return {"message": "Gibberish Detected"}
 
     response.raise_for_status()
 

eventstore/tasks.py

@@ -12,6 +12,7 @@
 from celery.exceptions import SoftTimeLimitExceeded
 from django.conf import settings
 from django.utils import dateparse, timezone
+from requests.auth import HTTPBasicAuth
 from requests.exceptions import RequestException
 from temba_client.exceptions import TembaHttpError
 
@@ -778,3 +779,46 @@ def process_whatsapp_template_send_status():
         status.status = WhatsAppTemplateSendStatus.Status.ACTION_COMPLETED
         status.action_completed_at = timezone.now()
         status.save()
+
+
+@app.task(
+    autoretry_for=(RequestException, SoftTimeLimitExceeded),
+    retry_backoff=True,
+    max_retries=15,
+    acks_late=True,
+    soft_time_limit=10,
+    time_limit=15,
+)
+def get_inbound_intent(text):
+    params = {"question": text}
+    response = requests.get(
+        urljoin(settings.INTENT_CLASSIFIER_URL, "/nlu/"),
+        params=params,
+        auth=HTTPBasicAuth(
+            settings.INTENT_CLASSIFIER_USER, settings.INTENT_CLASSIFIER_PASS
+        ),
+    )
+    response.raise_for_status()
+    return response.json()["intent"]
+
+
+@app.task(
+    autoretry_for=(RequestException, SoftTimeLimitExceeded),
+    retry_backoff=True,
+    max_retries=15,
+    acks_late=True,
+    soft_time_limit=10,
+    time_limit=15,
+)
+def label_whatsapp_message(label, message_id):
+    headers = {
+        "Authorization": "Bearer {}".format(settings.TURN_TOKEN),
+        "content-type": "application/json",
+        "Accept": "application/vnd.v1+json",
+    }
+    response = requests.post(
+        urljoin(settings.TURN_URL, f"/v1/messages/{message_id}/labels"),
+        json={"labels": [label]},
+        headers=headers,
+    )
+    response.raise_for_status()

eventstore/tests/test_whatsapp_actions.py

@@ -1,3 +1,4 @@
+import json
 from datetime import timedelta
 from unittest.mock import Mock, patch
 
@@ -203,6 +204,76 @@ def test_handle_edd_label_disabled(self):
             handle_inbound(message)
             handle.assert_not_called()
 
+    @responses.activate
+    def test_intent_classification(self):
+        inbound_text = "The test inbound message body"
+        responses.add(
+            responses.GET,
+            "http://intent-classifier/nlu/",
+            json={
+                "question": "The test inbound message body",
+                "intent": "Test Label",
+                "confidence": 90,
+            },
+        )
+
+        responses.add(
+            responses.POST, "http://turn/v1/messages/msg-id-1/labels", json={}
+        )
+
+        message = Mock()
+        message.has_label.return_value = False
+        message.id = "msg-id-1"
+        message.type = "text"
+        message.data = {"text": {"body": inbound_text}}
+
+        handle_inbound(message)
+
+        [intent_call, label_call] = responses.calls
+
+        self.assertEqual(intent_call.request.params, {"question": inbound_text})
+        self.assertEqual(
+            intent_call.request.headers["Authorization"],
+            "Basic bmx1X3VzZXI6bmx1X3Bhc3M=",
+        )
+
+        self.assertEqual(
+            json.loads(label_call.request.body), {"labels": ["Test Label"]}
+        )
+        self.assertEqual(
+            label_call.request.headers["Authorization"],
+            "Bearer turn-token",
+        )
+
+    @responses.activate
+    @override_settings(INTENT_CLASSIFIER_URL=None)
+    def test_intent_classification_disabled(self):
+        inbound_text = "The test inbound message body"
+
+        message = Mock()
+        message.has_label.return_value = False
+        message.id = "msg-id-1"
+        message.type = "text"
+        message.data = {"text": {"body": inbound_text}}
+
+        handle_inbound(message)
+
+        self.assertEqual(len(responses.calls), 0)
+
+    @responses.activate
+    def test_intent_classification_yes(self):
+        inbound_text = "YES"
+
+        message = Mock()
+        message.has_label.return_value = False
+        message.id = "msg-id-1"
+        message.type = "text"
+        message.data = {"text": {"body": inbound_text}}
+
+        handle_inbound(message)
+
+        self.assertEqual(len(responses.calls), 0)
+
 
 class UpdateRapidproAlertOptoutTests(DjangoTestCase):
     def test_contact_update_is_called(self):

eventstore/whatsapp_actions.py

@@ -4,7 +4,9 @@
 from eventstore.models import SMS_CHANNELTYPE, DeliveryFailure, Event, OptOut
 from eventstore.tasks import (
     async_create_flow_start,
+    get_inbound_intent,
     get_rapidpro_contact_by_msisdn,
+    label_whatsapp_message,
     send_helpdesk_response_to_dhis2,
     update_rapidpro_contact,
 )
@@ -57,6 +59,14 @@ def handle_inbound(message):
     if not settings.DISABLE_EDD_LABEL_FLOW and message.has_label("EDD ISSUE"):
         handle_edd_message(message)
 
+    if settings.INTENT_CLASSIFIER_URL and message.type == "text":
+        text = message.data["text"]["body"]
+        if text.lower() != "yes":
+            chain(
+                get_inbound_intent.s(),
+                label_whatsapp_message.s(message.id),
+            ).delay(text)
+
 
 def update_rapidpro_alert_optout(message):
     update_rapidpro_contact.delay(

ndoh_hub/settings.py

@@ -408,3 +408,7 @@
 WHATSAPP_TEMPLATE_SEND_TIMEOUT_HOURS = env.int(
     "WHATSAPP_TEMPLATE_SEND_TIMEOUT_HOURS", 3
 )
+
+INTENT_CLASSIFIER_URL = env.str("INTENT_CLASSIFIER_URL", None)
+INTENT_CLASSIFIER_USER = env.str("INTENT_CLASSIFIER_USER", None)
+INTENT_CLASSIFIER_PASS = env.str("INTENT_CLASSIFIER_PASS", None)

ndoh_hub/testsettings.py

@@ -26,3 +26,7 @@
 AAQ_CORE_API_URL = "http://aaqcore"
 AAQ_UD_API_URL = "http://aaqud"
 AAQ_V2_API_URL = "http://aaq_v2"
+
+INTENT_CLASSIFIER_URL = "http://intent-classifier"
+INTENT_CLASSIFIER_USER = "nlu_user"
+INTENT_CLASSIFIER_PASS = "nlu_pass"

scripts/migrate_to_turn/README.md

@@ -0,0 +1,90 @@
+# Migrate contacts from Rapidpro to Turn
+
+The scripts in this folder will be used to get all the contact information from Rapidpro and update Turn contacts.
+
+The fetch part is based on the modified date we have on Rapidpro contacts and the the idea is to start backfilling before we go live.
+We can do this in batches and repeat until we do the actual switch over from Rapidpro to Turn.
+
+## SCRIPTS
+
+There is one fetch script and 2 update script options. We can test out the different update option with larger batches and see which one works the best.
+
+###  fetch_rapidpro_contacts.py
+
+This fetches all the contacts from Rapidpro based on the start and end date provided, you can also configure a limit.
+
+The `FIELD_MAPPING` variable should be updated with all the fields we want to move over.
+
+The script will write all the contacts to a file with the start and end date in the name.
+
+It will also output the latest modified on date in the batch, this can then be used as a start date to get the next batch.
+
+### update_turn_contacts.py
+
+Update Turn contacts using the Turn contacts api asynchronously. 
+
+This script takes a filename of a file generated by the `fetch_rapidpro_contacts.py` script as a parameter and updates all the contact in the file on Turn.
+
+It is an async script and `CONCURRENCY` can be updated to control the speed, to avoid hitting the Turn API rate limits.
+
+Command to run:
+`python scripts/migrate_to_turn/update_turn_contacts.py contacts-2025-01-01-2025-01-07.csv > update_turn_contacts.json`
+
+The output is sent to a json file, which can be used to retry failed requests.
+
+### update_turn_contacts_queue.py
+
+Update Turn contacts using the Turn contacts api asynchronously but using a queue and workers. It will sleep if it gets rate limited by Turn.
+
+This script takes a filename of a file generated by the `fetch_rapidpro_contacts.py` script as a parameter and updates all the contact in the file on Turn.
+
+It is an async script and `WORKER_COUNT` can be configured, to change the amount being processed at a time.
+
+Command to run:
+`python scripts/migrate_to_turn/update_turn_contacts_queue.py contacts-2025-01-01-2025-01-07.csv > update_turn_contacts.json`
+
+The output is sent to a json file, which can be used to retry failed requests.
+
+### update_turn_contacts_bulk.py
+
+Update Turn contacts using the Turn bulk update contacts API. The API is currently limited to allow files of a maximum of 1 Megabyte in size.
+
+This script takes the csv provided and sends it directly yto the Turn API. It will output the results to a new csv file.
+
+Command to run:
+`python scripts/migrate_to_turn/update_turn_contacts_bulk.py contacts-2024-01-01-2025-01-07.csv`
+
+### compare_contacts.py
+
+This script can be used to compare specific contacts.
+
+Add the WhatsApp IDs of the contacts you want to compare to the WA_IDS list in the script.
+
+The script will get their Rapidpro and Turn contact details and output everything to a `compare.csv` file.
+
+Command to run:
+`python scripts/migrate_to_turn/compare_contacts.py`
+
+## FIELD_MAPPING
+
+This is a dictionary the script uses to figure out where to get the data, how to process it and where it should go.
+
+The key is the field name in Rapidpro, the value determinues the rest:
+
+`turn_name` - The destination field name in Turn
+
+`process` - The function to call to process the value, like change datetime to format Turn understands. The functions live in the `process_fields.py` file and that's where any new process field functions should be added.
+
+`type` - Where the value comes from, `default` Rapidpro contacts field or a `custom` one added by Reach.
+
+## STEPS
+
+1. Make sure the `FIELD_MAPPING` is configured
+1. Set the required environment variables: `TURN_TOKEN` and `RAPIDPRO_TOKEN`.
+1. Update the start date, end date, limit and field mapping in `fetch_rapidpro_contacts.py` script
+1. Run `python scripts/migrate_to_turn/fetch_rapidpro_contacts.py` and take note of the last modified date and the filename.
+1. Run `python scripts/migrate_to_turn/update_turn_contacts.py contacts-2025-01-01-2025-01-07.csv > update_turn_contacts.json`
+1. Use jq to check if there were any errors `jq .response.status update_turn_contacts.json | sort | uniq -c`
+1. To retry errors, run `cat update_turn_contacts.json | python scripts/migrate_to_rapidpro/retry_requests.py > update_turn_contacts2.json`
+1. Repeat previous two steps until all contacts successfully completed.
+1. Update the start and end date in `fetch_rapidpro_contacts.py` script. Repeat from step 3.

scripts/migrate_to_turn/compare_contacts.py

@@ -0,0 +1,70 @@
+import csv
+import os
+from urllib.parse import urljoin
+
+import requests
+from fetch_rapidpro_contacts import FIELD_MAPPING
+from temba_client.v2 import TembaClient
+
+RAPIDPRO_URL = "https://rapidpro.qa.momconnect.co.za"
+TURN_URL = "https://whatsapp-praekelt-cloud.turn.io"
+
+WA_IDS = ["27836378531"]
+
+rapidpro_client = TembaClient(RAPIDPRO_URL, os.environ["RAPIDPRO_TOKEN"])
+
+
+def get_rapidpro_contact(wa_id):
+    urn = f"whatsapp:{wa_id}"
+    contact = rapidpro_client.get_contacts(urn=urn).first(retry_on_rate_exceed=True)
+    data = {}
+
+    for rapidpro_field, turn_details in FIELD_MAPPING.items():
+        if turn_details["type"] == "default":
+            data[rapidpro_field] = getattr(contact, rapidpro_field)
+        else:
+            data[rapidpro_field] = contact.fields[rapidpro_field]
+
+    return data
+
+
+def get_turn_contact(wa_id):
+    headers = {
+        "Authorization": "Bearer {}".format(os.environ["TURN_TOKEN"]),
+        "content-type": "application/json",
+        "Accept": "application/vnd.v1+json",
+    }
+    response = requests.get(
+        urljoin(TURN_URL, "/v1/contacts/{}/profile".format(wa_id)),
+        headers=headers,
+    )
+    contact = response.json()["fields"]
+
+    data = {}
+    for rapidpro_field, turn_details in FIELD_MAPPING.items():
+        data[rapidpro_field] = contact[turn_details["turn_name"]]
+    return data
+
+
+def compare_contacts():
+    rows = []
+    for wa_id in WA_IDS:
+        rapidpro_data = get_rapidpro_contact(wa_id)
+        turn_data = get_turn_contact(wa_id)
+
+        row = {"wa_id": wa_id}
+        for rapidpro_field in FIELD_MAPPING.keys():
+            row[f"RP {rapidpro_field}"] = rapidpro_data[rapidpro_field]
+            row[f"TURN {rapidpro_field}"] = turn_data[rapidpro_field]
+
+        rows.append(row)
+
+    with open("compare.csv", "w", encoding="utf-8") as f:
+        fieldnames = rows[0].keys()
+        writer = csv.DictWriter(f, fieldnames=fieldnames)
+        writer.writeheader()
+        writer.writerows(rows)
+
+
+if __name__ == "__main__":
+    compare_contacts()