implement backend for wire labels

server/cp/__init__.py

@@ -13,5 +13,6 @@
 
 HEADLINE2 = "headline_extended"
 CORRECTION = "correction"
+WIRE_LABELS_SCHEME = "wirelables"
 
 logging.basicConfig(level=logging.INFO)

server/cp/signals.py

@@ -3,6 +3,7 @@
 
 from typing import Literal, Optional
 from flask import current_app as app
+from flask_babel import gettext
 
 from datetime import datetime, timedelta
 from superdesk import get_resource_service
@@ -26,6 +27,7 @@ def on_publish_item(sender, item, **kwargs):
     copy_headline2_to_headline(item)
     copy_correction_to_body_html(item)
     handle_transcripts(item)
+    set_wire_labels(item)
 
 
 def copy_headline2_to_headline(item):
@@ -69,7 +71,9 @@ def on_push(sender, item, **kwargs):
         item["language"] = fix_language(item["language"])
 
     if get_media_type(item) and item.get("evolvedfrom"):
-        parent_item = get_resource_service("items").find_one(req=None, _id=item["evolvedfrom"])
+        parent_item = get_resource_service("items").find_one(
+            req=None, _id=item["evolvedfrom"]
+        )
         if parent_item is None:
             flask.abort(503)  # must be 50x error to trigger retry later
 
@@ -90,7 +94,8 @@ def user_auth_is_gip(user: User) -> bool:
 def get_media_type(item):
     media_type_scheme = get_media_type_scheme()
     return next(
-        (s for s in item.get("subject", []) if s.get("scheme") == media_type_scheme), None
+        (s for s in item.get("subject", []) if s.get("scheme") == media_type_scheme),
+        None,
     )
 
 
@@ -118,7 +123,9 @@ def handle_transcripts(item):
 
     if media_source:
         item["source"] = media_source.get("name") or media_source.get("code")
-        item["subject"] = [s for s in item["subject"] if s.get("scheme") != media_source_scheme]
+        item["subject"] = [
+            s for s in item["subject"] if s.get("scheme") != media_source_scheme
+        ]
 
     if media_type and media_type["code"] in ("tvstation", "radionstation"):
         # it might be already populated based on previous segment
@@ -142,6 +149,35 @@ def get_media_type_name(scheme: MediaType, language: Optional[str] = "en") -> st
     return MEDIA_TYPE_NAMES[scheme][1 if language and "fr" in language else 0]
 
 
+def item_has_code(item, field: Literal["genre", "service"], code: str) -> bool:
+    values = item.get(field) or []
+    return any(value.get("code") == code for value in values)
+
+
+def append_subject(item, name: str, code: str) -> None:
+    item.setdefault("subject", []).append(
+        dict(
+            name=name,
+            code=code,
+            scheme=cp.WIRE_LABELS_SCHEME,
+        )
+    )
+
+
+def set_wire_labels(item):
+    if item.get("slugline") and "-The-Latest" in item["slugline"]:
+        append_subject(item, gettext("THE LATEST"), "latest")
+
+    if item_has_code(item, "genre", "NewsAlert"):
+        append_subject(item, gettext("ALERT"), "alert")
+
+    if item_has_code(item, "service", "m"):
+        append_subject(item, gettext("ADVISORY"), "advisory")
+
+    if item_has_code(item, "service", "p"):
+        append_subject(item, gettext("PRESS RELEASE"), "press-release")
+
+
 def init_app(app):
     publish_item.connect(on_publish_item)
     user_created.connect(on_user_created)

server/newsroom-app-cp.pot

@@ -9,7 +9,7 @@ msgid ""
 msgstr ""
 "Project-Id-Version: Newsroom App CP 0.0.0\n"
 "Report-Msgid-Bugs-To: EMAIL@ADDRESS\n"
-"POT-Creation-Date: 2024-05-30 10:05+0200\n"
+"POT-Creation-Date: 2024-06-06 13:06+0200\n"
 "PO-Revision-Date: YEAR-MO-DA HO:MI+ZONE\n"
 "Last-Translator: FULL NAME <EMAIL@ADDRESS>\n"
 "Language-Team: LANGUAGE <[email protected]>\n"
@@ -55,63 +55,79 @@ msgstr ""
 msgid "Email"
 msgstr ""
 
-#: settings.py:143 settings.py:144
+#: cp/signals.py:169
+msgid "THE LATEST"
+msgstr ""
+
+#: cp/signals.py:172
+msgid "ALERT"
+msgstr ""
+
+#: cp/signals.py:175
+msgid "ADVISORY"
+msgstr ""
+
+#: cp/signals.py:178
+msgid "PRESS RELEASE"
+msgstr ""
+
+#: settings.py:144 settings.py:145
 msgid "No coverage"
 msgstr ""
 
-#: settings.py:149 settings.py:150
+#: settings.py:150 settings.py:151
 msgid "Cancelled / not planned"
 msgstr ""
 
-#: settings.py:155 settings.py:156
+#: settings.py:156 settings.py:157
 msgid "Not decided / on merit"
 msgstr ""
 
-#: settings.py:161 settings.py:162
+#: settings.py:162 settings.py:163
 msgid "Planned"
 msgstr ""
 
-#: settings.py:167 settings.py:168
+#: settings.py:168 settings.py:169
 msgid "Completed"
 msgstr ""
 
-#: settings.py:180 settings.py:225
+#: settings.py:182 settings.py:227
 msgid "Language"
 msgstr ""
 
-#: settings.py:184
+#: settings.py:186
 msgid "Media type"
 msgstr ""
 
-#: settings.py:193
+#: settings.py:195
 msgid "Source"
 msgstr ""
 
-#: settings.py:197
+#: settings.py:199
 msgid "Wire Category"
 msgstr ""
 
-#: settings.py:201 settings.py:243
+#: settings.py:203 settings.py:245
 msgid "Subject"
 msgstr ""
 
-#: settings.py:210
+#: settings.py:212
 msgid "Version"
 msgstr ""
 
-#: settings.py:229
+#: settings.py:231
 msgid "Category"
 msgstr ""
 
-#: settings.py:233
+#: settings.py:235
 msgid "Event Type"
 msgstr ""
 
-#: settings.py:337
+#: settings.py:339
 msgid "Calendar"
 msgstr ""
 
-#: settings.py:338
+#: settings.py:340
 msgid "Bookmarks"
 msgstr ""
 

server/settings.py

@@ -1,4 +1,5 @@
 import os
+import cp
 import pathlib
 from flask_babel import lazy_gettext
 from superdesk.default_settings import strtobool
@@ -169,6 +170,7 @@
             },
         },
         "agenda_top_story_scheme": "topstory",
+        "wire_labels_scheme": cp.WIRE_LABELS_SCHEME,
     },
 )
 

server/tests/test_signals.py

@@ -230,3 +230,42 @@ def test_handle_transcripts(app):
     signals.on_publish_item(None, transcript_item)
     assert 1 == len(transcript_item["subject"])
     assert "Station de télé" == transcript_item["subject"][0]["name"]
+
+
+def test_wire_labels(app):
+    def get_label(item):
+        return next(
+            (
+                s
+                for s in item.get("subject", [])
+                if s.get("scheme") == cp.WIRE_LABELS_SCHEME
+            ),
+            None,
+        )
+
+    def assert_label(item, code, name):
+        label = get_label(item)
+        assert label is not None
+        assert label["code"] == code
+        assert label["name"] == name
+
+    item = {}
+    signals.on_publish_item(None, item)
+    label = get_label(item)
+    assert label is None
+
+    item = {"slugline": "Something-The-Latest"}
+    signals.on_publish_item(None, item)
+    assert_label(item, "latest", "THE LATEST")
+
+    item = {"genre": [{"code": "NewsAlert", "name": "NewsAlert"}]}
+    signals.on_publish_item(None, item)
+    assert_label(item, "alert", "ALERT")
+
+    item = {"service": [{"code": "m", "name": "Advisory"}]}
+    signals.on_publish_item(None, item)
+    assert_label(item, "advisory", "ADVISORY")
+
+    item = {"service": [{"code": "p", "name": "Press Release"}]}
+    signals.on_publish_item(None, item)
+    assert_label(item, "press-release", "PRESS RELEASE")