As a user I want to export my selected events in csv format [CPCN-15]

newsroom/agenda/__init__.py

@@ -37,6 +37,7 @@ def init_app(app):
     )
 
     app.download_formatter("ical", formatters.iCalFormatter(), "iCalendar", ["agenda"])
+    app.download_formatter("Csv", formatters.CSVFormatter(), "CSV", ["agenda"])
     app.add_template_global(url_for_agenda)
     app.add_template_global(get_coverage_email_text)
     app.add_template_global(get_coverage_content_type_name, "get_coverage_content_type")

newsroom/agenda/formatters/__init__.py

@@ -1 +1,2 @@
 from .ical_formatter import iCalFormatter  # noqa
+from .csv_formatter import CSVFormatter  # noqa

newsroom/agenda/formatters/csv_formatter.py

@@ -0,0 +1,102 @@
+from newsroom.formatter import BaseFormatter
+import csv
+import io
+
+
+class CSVFormatter(BaseFormatter):
+    VERSION = "1.0"
+    PRODID = "Newshub"
+    FILE_EXTENSION = "csv"
+    MIMETYPE = "text/csv"
+
+    def format_item(self, item, item_type=None):
+        event_item = self.format_event(item)
+        return self.serialize_to_csv(event_item)
+
+    def serialize_to_csv(self, data):
+        csv_string = io.StringIO()
+        csv_writer = csv.DictWriter(csv_string, delimiter=",", fieldnames=data.keys())
+        csv_writer.writeheader()
+        csv_writer.writerow(data)  # Write a single row for the provided data
+        csv_string.seek(0)  # Reset the buffer position
+        return csv_string.getvalue().encode("utf-8")
+
+    def format_event(self, item):
+        event = item.get("event", {})
+        return {
+            "Event name": item.get("name", ""),
+            "Description": item.get("definition_long", item.get("definition_short", "")),
+            "Language": item.get("language", ""),
+            "Event start date": self.format_date(item, "start"),
+            "Event end date": self.format_date(item, "end"),
+            "Event time": self.format_time(item),
+            "Event timezone": item.get("dates", {}).get("tz", ""),
+            "Location": self.parse_location(item, "name"),
+            "Country": self.parse_location(item, "country"),
+            "Subject": self.parse_list(event, "subject"),
+            "Website": event.get("links")[0] if event.get("links") else "",
+            "Category": self.parse_list(event, "anpa_category"),
+            "Event type": item.get("item_type", ""),
+            "Organization name": event.get("event_contact_info")[0].get("organisation", " ")
+            if event.get("event_contact_info")
+            else "",
+            "Contact": self.parse_contact_info(item),
+            "Coverage type": self.parse_coverage(item, "coverage_type"),
+            "Coverage status": self.parse_coverage(item, "coverage_status"),
+        }
+
+    def format_date(self, item, date_type):
+        date_obj = item.get("dates", {}).get(date_type)
+        if date_obj:
+            return date_obj.strftime("%Y-%m-%d")
+        return ""
+
+    def format_time(self, item):
+        date_obj = item.get("dates", {})
+        if date_obj.get("all_day"):
+            return ""
+        elif date_obj.get("no_end_time"):
+            return f"{date_obj.get('start').strftime('%H:%M:%S')}"
+        else:
+            return f"{date_obj.get('start').strftime('%H:%M:%S')}-{date_obj.get('end').strftime('%H:%M:%S')}"
+
+    def parse_location(self, item, field):
+        """
+        parse location info
+        """
+        if item.get("location"):
+            for loc in item.get("location"):
+                return loc.get(field, "") if not field == "country" else loc.get("address", {}).get(field)
+        return ""
+
+    def parse_list(self, item, key):
+        values = [v.get("name", "") for v in item.get(key, [])]
+        return ",".join(values)
+
+    def parse_contact_info(self, item):
+        """
+        parse contact information
+        """
+        event_contact_info = item.get("event", {}).get("event_contact_info", [])
+
+        for contact in event_contact_info:
+            contact_values = [
+                contact.get("honorific", ""),
+                contact.get("first_name", ""),
+                contact.get("last_name", ""),
+                contact.get("organisation", ""),
+                ",".join(contact.get("contact_email", [])),
+                ",".join(contact.get("mobile", [])),
+            ]
+            return ",".join(contact_values)
+
+    def parse_coverage(self, item, field):
+        """
+        parse coverage information
+        """
+        coverages = item.get("event", {}).get("coverages", {})
+        value = []
+        if coverages:
+            for coverage in coverages:
+                value.append(coverage.get(field, ""))
+        return ",".join(value)

tests/core/test_csv_formatter.py

@@ -0,0 +1,90 @@
+from .test_push_events import test_event
+from flask import json
+import copy
+from newsroom.utils import get_entity_or_404
+from newsroom.agenda.formatters import CSVFormatter
+import csv
+
+
+event = copy.deepcopy(test_event)
+event["coverages"] = [
+    {
+        "planning_id": "urn:newsml:stt.fi:20280911:631023",
+        "coverage_id": "ID_EVENT_286323",
+        "scheduled": "2028-09-11T00:00:00+0000",
+        "coverage_type": "text",
+        "workflow_status": "draft",
+        "coverage_status": "coverage intended",
+        "slugline": "Testi/Luotsi",
+    },
+    {
+        "planning_id": "urn:newsml:stt.fi:20280911:631025",
+        "coverage_id": "ID_WORKREQUEST_187845",
+        "scheduled": "2023-10-03T22:00:00+0000",
+        "coverage_type": "picture",
+        "workflow_status": "draft",
+        "coverage_status": "coverage intended",
+        "slugline": "Testi/Luotsi",
+        "genre": [{"qcode": "sttimage:20", "name": "Kuvaaja paikalla"}],
+    },
+]
+event["subject"] = [
+    {"name": "Statistics & Economic Indicators", "qcode": "150", "scheme": "event_types", "code": "150"},
+    {"name": "Economic Indicators", "qcode": "3", "scheme": "categories", "code": "3"},
+]
+event["anpa_category"].append({"name": "Economic News", "qcode": "b", "code": "b"})
+
+
+def test_csv_formatter_item(client, app):
+    client.post("/push", data=json.dumps(event), content_type="application/json")
+    parsed = get_entity_or_404(event["guid"], "agenda")
+    formatter = CSVFormatter()
+    assert formatter.format_filename(parsed).endswith("new-press-conference.csv")
+    csv_data = formatter.format_item(parsed, item_type="agenda")
+    csv_string = csv_data.decode("utf-8")
+    csv_lines = csv_string.split("\n")
+    csv_reader = csv.reader(csv_lines)
+
+    header = next(csv_reader)
+    expected_header_fields = [
+        "Event name",
+        "Description",
+        "Language",
+        "Event start date",
+        "Event end date",
+        "Event time",
+        "Event timezone",
+        "Location",
+        "Country",
+        "Subject",
+        "Website",
+        "Category",
+        "Event type",
+        "Organization name",
+        "Contact",
+        "Coverage type",
+        "Coverage status",
+    ]
+    assert header == expected_header_fields
+
+    data_fields = next(csv_reader)
+    expected_data_values = [
+        "Prime minister press conference",
+        "Ministers will check the environmental issues",
+        "en-CA",
+        "2018-05-28",
+        "2018-05-28",
+        "04:00:00-05:00:00",
+        "Australia/Sydney",
+        "Sydney",
+        "Australia",
+        "Statistics & Economic Indicators,Economic Indicators",
+        "www.earthhour.com",
+        "Australian General News,Economic News",
+        "event",
+        "AAP",
+        "Professor,Tom,Jones,AAP,jones@foo.com,",
+        "text,picture",
+        "coverage intended,coverage intended",
+    ]
+    assert data_fields == expected_data_values