Merge pull request #5952 from nyaruka/broadcast_quickreplies_update

Update broadcast views to support quick replies as structs

Author: rowanseymour

temba/msgs/migrations/0286_update_quick_replies.py

@@ -0,0 +1,35 @@
+# Generated by Django 5.1.4 on 2025-03-21 15:34
+
+from django.db import migrations
+
+
+def update_quick_replies(apps, schema_editor):
+    Broadcast = apps.get_model("msgs", "Broadcast")
+
+    num_updated = 0
+
+    for broadcast in Broadcast.objects.filter(is_active=True).only("translations"):
+        updated = False
+        for trans in broadcast.translations.values():
+            if "quick_replies" in trans:
+                for i, qr in enumerate(trans["quick_replies"]):
+                    if isinstance(qr, str):
+                        trans["quick_replies"][i] = {"text": qr}
+                        updated = True
+        if updated:
+            broadcast.save(update_fields=("translations",))
+            num_updated += 1
+
+        if num_updated % 1000 == 0:  # pragma: no cover
+            print(f"Updated {num_updated} broadcasts with quick replies")
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ("msgs", "0285_delete_systemlabelcount"),
+    ]
+
+    operations = [
+        migrations.RunPython(update_quick_replies, migrations.RunPython.noop),
+    ]

temba/msgs/models.py

@@ -412,6 +412,9 @@ def bulk_delete(cls, attachments):
     def as_json(self) -> dict:
         return {"content_type": self.content_type, "url": self.url}
 
+    def __str__(self) -> str:
+        return f"{self.content_type}:{self.url}"
+
 
 @dataclass
 class QuickReply:

temba/msgs/tests/test_migrations.py

@@ -0,0 +1,46 @@
+from temba.msgs.models import Broadcast
+from temba.tests import MigrationTest
+
+
+class UpdateQuickRepliesTest(MigrationTest):
+    app = "msgs"
+    migrate_from = "0285_delete_systemlabelcount"
+    migrate_to = "0286_update_quick_replies"
+
+    def setUpBeforeMigration(self, apps):
+        def create_broadcast(translations, is_active=True):
+            return Broadcast.objects.create(org=self.org, translations=translations, is_active=is_active)
+
+        self.bcast1 = create_broadcast(  # active broadcast with quick replies as strings
+            {
+                "eng": {"text": "Hi there", "quick_replies": ["yes", "no"]},
+                "spa": {"text": "Hola", "quick_replies": ["si", "no"]},
+            }
+        )
+        self.bcast2 = create_broadcast(  # active broadcast with quick replies already as objects
+            {"eng": {"text": "Hi there", "quick_replies": [{"text": "yes"}, {"text": "no"}]}}
+        )
+        self.bcast3 = create_broadcast({"eng": {"text": "Hi there"}})  # active broadcast with no quick replies
+        self.bcast4 = create_broadcast(  # inactive broadcast with quick replies as strings
+            {"eng": {"text": "Hi there", "quick_replies": ["yes", "no"]}},
+            is_active=False,
+        )
+
+    def test_migration(self):
+        def assert_translations(bcast, expected: dict):
+            bcast.refresh_from_db()
+            self.assertEqual(bcast.translations, expected)
+
+        assert_translations(
+            self.bcast1,
+            {
+                "eng": {"text": "Hi there", "quick_replies": [{"text": "yes"}, {"text": "no"}]},
+                "spa": {"text": "Hola", "quick_replies": [{"text": "si"}, {"text": "no"}]},
+            },
+        )
+        assert_translations(
+            self.bcast2,
+            {"eng": {"text": "Hi there", "quick_replies": [{"text": "yes"}, {"text": "no"}]}},  # unchanged
+        )
+        assert_translations(self.bcast3, {"eng": {"text": "Hi there"}})  # unchanged
+        assert_translations(self.bcast4, {"eng": {"text": "Hi there", "quick_replies": ["yes", "no"]}})  # unchanged

temba/utils/compose.py

@@ -4,59 +4,62 @@
 from temba.msgs.models import Attachment, Media, Q
 
 
-def compose_serialize(translation=None, json_encode=False, *, base_language=None, optin=None):
+def compose_serialize(translations, *, json_encode=False, base_language=None, optin=None) -> dict:
     """
-    Serializes attachments from db to compose widget for populating initial widget values
+    Serializes translations to format for compose widget
     """
 
-    if not translation:
+    if not translations:
         return {}
 
-    translation = copy.deepcopy(translation)
+    translations = copy.deepcopy(translations)
 
     if base_language and optin:
-        translation[base_language]["optin"] = {"uuid": str(optin.uuid), "name": optin.name}
+        translations[base_language]["optin"] = {"uuid": str(optin.uuid), "name": optin.name}
 
-    for details in translation.values():
-        if "attachments" in details:
-            details["attachments"] = compose_serialize_attachments(details["attachments"])
+    for translation in translations.values():
+        if "attachments" in translation:
+            translation["attachments"] = compose_serialize_attachments(translation["attachments"])
 
-    if json_encode:
-        return json.dumps(translation)
+        # for now compose widget only supports simple text quick replies
+        if "quick_replies" in translation:
+            translation["quick_replies"] = [qr["text"] for qr in translation["quick_replies"]]
 
-    return translation
+    return json.dumps(translations) if json_encode else translations
 
 
-def compose_serialize_attachments(attachments):
-    if not attachments:
-        return []
-    parsed_attachments = Attachment.parse_all(attachments)
-    serialized_attachments = []
-    for parsed_attachment in parsed_attachments:
-        media = Media.objects.filter(
-            Q(content_type=parsed_attachment.content_type) and Q(url=parsed_attachment.url)
-        ).first()
-        serialized_attachment = {
-            "uuid": str(media.uuid),
-            "content_type": media.content_type,
-            "url": media.url,
-            "filename": media.filename,
-            "size": str(media.size),
-        }
-        serialized_attachments.append(serialized_attachment)
-    return serialized_attachments
-
-
-def compose_deserialize(compose):
+def compose_serialize_attachments(attachments: list) -> list:
+    serialized = []
+
+    for parsed in Attachment.parse_all(attachments):
+        media = Media.objects.filter(Q(content_type=parsed.content_type) and Q(url=parsed.url)).first()
+        serialized.append(
+            {
+                "uuid": str(media.uuid),
+                "content_type": media.content_type,
+                "url": media.url,
+                "filename": media.filename,
+                "size": str(media.size),
+            }
+        )
+    return serialized
+
+
+def compose_deserialize(compose: dict) -> dict:
     """
     Deserializes attachments from compose widget to db for saving final db values
     """
-    for details in compose.values():
-        details["attachments"] = compose_deserialize_attachments(details.get("attachments", []))
+    for translation in compose.values():
+        translation["attachments"] = compose_deserialize_attachments(translation.get("attachments", []))
+
+        # for now compose widget only supports simple text quick replies
+        if "quick_replies" in translation:
+            translation["quick_replies"] = [{"text": qr} for qr in translation["quick_replies"]]
     return compose
 
 
-def compose_deserialize_attachments(attachments):
+def compose_deserialize_attachments(attachments: list) -> list:
     if not attachments:
         return []
-    return [f"{a['content_type']}:{a['url']}" for a in attachments]
+
+    return [str(Attachment(a["content_type"], a["url"])) for a in attachments]

temba/utils/tests/test_compose.py

@@ -1,7 +1,41 @@
 from temba.tests import TembaTest
-from temba.utils.compose import compose_serialize
+from temba.utils.compose import compose_deserialize, compose_serialize
 
 
 class ComposeTest(TembaTest):
-    def test_empty_compose(self):
-        self.assertEqual(compose_serialize(), {})
+    def test_serialize(self):
+        self.assertEqual(compose_serialize(None), {})
+        self.assertEqual(compose_serialize({"eng": {"text": "Hello"}}), {"eng": {"text": "Hello"}})
+        self.assertEqual(
+            compose_serialize({"eng": {"text": "Hello", "quick_replies": [{"text": "Yes"}, {"text": "No"}]}}),
+            {"eng": {"text": "Hello", "quick_replies": ["Yes", "No"]}},
+        )
+
+    def test_deserialize(self):
+        self.assertEqual(compose_deserialize({"eng": {"text": "Hello"}}), {"eng": {"text": "Hello", "attachments": []}})
+        self.assertEqual(
+            compose_deserialize(
+                {
+                    "eng": {
+                        "text": "Hello",
+                        "attachments": [
+                            {
+                                "uuid": "8a798c81-c890-4fe5-b9c7-617c06096b94",
+                                "content_type": "image/jpeg",
+                                "url": "https://example.com/image.jpg",
+                                "filename": "image.jpg",
+                                "size": "12345",
+                            }
+                        ],
+                        "quick_replies": ["Yes", "No"],
+                    }
+                }
+            ),
+            {
+                "eng": {
+                    "text": "Hello",
+                    "attachments": ["image/jpeg:https://example.com/image.jpg"],
+                    "quick_replies": [{"text": "Yes"}, {"text": "No"}],
+                }
+            },
+        )

templates/msgs/includes/broadcast.html

@@ -24,9 +24,9 @@
         <div class="text">
           {{ translation.text }}
           <br />
-          {% for reply in translation.quick_replies %}
+          {% for qr in translation.quick_replies %}
             <temba-label class="mr-2 mt-2" backgroundcolor="#fafafa">
-              {{ reply }}
+              {{ qr.text }}
             </temba-label>
           {% endfor %}
           {% if broadcast.optin or broadcast.template %}