[SDESK-7462] - Events, Planning & History resource, service and REST API

server/planning/events/__init__.py

@@ -47,11 +47,14 @@
 from planning.autosave import AutosaveService
 
 from .events_service import EventsAsyncService
-from .module import events_resource_config
+from .events_history_async_service import EventsHistoryAsyncService
+from .module import events_resource_config, events_history_resource_config
 
 __all__ = [
     "EventsAsyncService",
     "events_resource_config",
+    "EventsHistoryAsyncService",
+    "events_history_resource_config",
 ]
 
 

server/planning/events/events_history.py

@@ -32,6 +32,7 @@ class EventsHistoryResource(Resource):
         "operation": {"type": "string"},
         "update": {"type": "dict", "nullable": True},
     }
+    internal_resource = True
 
 
 class EventsHistoryService(HistoryService):

server/planning/events/events_history_async_service.py

@@ -0,0 +1,83 @@
+# This file is part of Superdesk.
+#
+# Copyright 2013, 2014 Sourcefabric z.u. and contributors.
+#
+# For the full copyright and license information, please see the
+# AUTHORS and LICENSE files distributed with this source code, or
+# at https://www.sourcefabric.org/superdesk/license
+
+"""Superdesk Files"""
+
+from copy import deepcopy
+import logging
+from typing import Any
+
+from planning.types import EventResourceModel
+
+from planning.types import EventsHistoryResourceModel
+from superdesk.resource_fields import ID_FIELD
+from planning.utils import get_related_planning_for_events
+from planning.history_async_service import HistoryAsyncService
+from planning.item_lock import LOCK_ACTION
+
+logger = logging.getLogger(__name__)
+
+
+class EventsHistoryAsyncService(HistoryAsyncService[EventsHistoryResourceModel]):
+    async def on_item_created(self, items: list[dict[str, Any]], operation: str | None = None):
+        created_from_planning = []
+        regular_events = []
+        for item in items:
+            if isinstance(item, EventResourceModel):
+                item = item.to_dict()
+
+            planning_items = get_related_planning_for_events([item[ID_FIELD]], "primary")
+            if len(planning_items) > 0:
+                item["created_from_planning"] = planning_items[0].get("_id")
+                created_from_planning.append(item)
+            else:
+                regular_events.append((item))
+
+        await super().on_item_created(created_from_planning, "created_from_planning")
+        await super().on_item_created(regular_events)
+
+    async def on_item_deleted(self, doc: dict[str, Any]):
+        lookup = {"event_id": doc[ID_FIELD]}
+        await self.delete_many(lookup=lookup)
+
+    async def on_item_updated(self, updates: dict[str, Any], original: dict[str, Any], operation: str | None = None):
+        item = deepcopy(original)
+        if list(item.keys()) == ["_id"]:
+            diff = self._remove_unwanted_fields(updates)
+        else:
+            diff = await self._changes(original, updates)
+            if updates:
+                item.update(updates)
+
+        if not operation:
+            operation = "convert_recurring" if original.get(LOCK_ACTION) == "convert_recurring" else "edited"
+
+        await self._save_history(item, diff, operation)
+
+    async def _save_history(self, item: dict[str, Any], update: dict[str, Any], operation: str | None = None):
+        history = {
+            "event_id": item[ID_FIELD],
+            "user_id": self.get_user_id(),
+            "operation": operation,
+            "update": update,
+        }
+        # a post action is recorded as a special case
+        if operation == "update":
+            if "scheduled" == update.get("state", ""):
+                history["operation"] = "post"
+            elif "canceled" == update.get("state", ""):
+                history["operation"] = "unpost"
+        elif operation == "create" and "ingested" == update.get("state", ""):
+            history["operation"] = "ingested"
+        await self.create([history])
+
+    async def on_update_repetitions(self, updates: dict[str, Any], event_id: str, operation: str | None = None):
+        await self.on_item_updated(updates, {"_id": event_id}, operation or "update_repetitions")
+
+    async def on_update_time(self, updates: dict[str, Any], original: dict[str, Any]):
+        await self.on_item_updated(updates, original, "update_time")

server/planning/events/events_service.py

@@ -1,3 +1,4 @@
+from planning.events.events_history_async_service import EventsHistoryAsyncService
 import pytz
 import itertools
 
@@ -214,8 +215,8 @@ async def prepare_events_data(self, docs: list[EventResourceModel]) -> None:
                 event.planning_item = original_planning_item
 
             if event.state == WorkflowStates.INGESTED:
-                events_history = get_resource_service("events_history")
-                events_history.on_item_created([event.to_dict()])
+                events_history = EventsHistoryAsyncService()
+                await events_history.on_item_created([event.to_dict()])
 
             if original_planning_item:
                 await self._link_to_planning(event)
@@ -231,7 +232,7 @@ async def on_created(self, docs: list[EventResourceModel]):
         then send this list off to the clients so they can fetch these events
         """
         notifications_sent = []
-        history_service = get_resource_service("events_history")
+        history_service = EventsHistoryAsyncService()
 
         for doc in docs:
             event_id = doc.id
@@ -243,8 +244,8 @@ async def on_created(self, docs: list[EventResourceModel]):
                 if not parent_event:
                     raise SuperdeskApiError.badRequestError("Parent event not found")
 
-                history_service.on_item_updated({"duplicate_id": event_id}, parent_event.to_dict(), "duplicate")
-                history_service.on_item_updated({"duplicate_id": parent_id}, doc.to_dict(), "duplicate_from")
+                await history_service.on_item_updated({"duplicate_id": event_id}, parent_event.to_dict(), "duplicate")
+                await history_service.on_item_updated({"duplicate_id": parent_id}, doc.to_dict(), "duplicate_from")
 
                 duplicate_ids = parent_event.duplicate_to or []
                 duplicate_ids.append(event_id)
@@ -671,8 +672,8 @@ async def _convert_to_recurring_events(self, updates: dict[str, Any], original:
             event_reschedule_service.update_single_event(updates, original)
 
             if updates.get("state") == WorkflowState.RESCHEDULED:
-                history_service = get_resource_service("events_history")
-                history_service.on_reschedule(updates, original.to_dict())
+                history_service = EventsHistoryAsyncService()
+                await history_service.on_reschedule(updates, original.to_dict())
         else:
             # Original event falls as a part of the series
             # Remove the first element in the list (the current event being updated)

server/planning/events/module.py

@@ -3,10 +3,12 @@
     MongoIndexOptions,
     MongoResourceConfig,
     ElasticResourceConfig,
+    RestEndpointConfig,
 )
 
-from planning.types import EventResourceModel
+from planning.types import EventResourceModel, EventsHistoryResourceModel
 from .events_service import EventsAsyncService
+from .events_history_async_service import EventsHistoryAsyncService
 
 events_resource_config = ResourceConfig(
     name="events",
@@ -28,3 +30,19 @@
     ),
     elastic=ElasticResourceConfig(),
 )
+
+events_history_resource_config = ResourceConfig(
+    name="events_history",
+    data_class=EventsHistoryResourceModel,
+    service=EventsHistoryAsyncService,
+    mongo=MongoResourceConfig(
+        indexes=[
+            MongoIndexOptions(
+                name="event_id",
+                keys=[("event_id", 1)],
+                unique=False,
+            ),
+        ],
+    ),
+    rest_endpoints=RestEndpointConfig(resource_methods=["GET"], item_methods=["GET"]),
+)

server/planning/history_async_service.py

@@ -0,0 +1,122 @@
+# This file is part of Superdesk.
+#
+# Copyright 2013, 2014 Sourcefabric z.u. and contributors.
+#
+# For the full copyright and license information, please see the
+# AUTHORS and LICENSE files distributed with this source code, or
+# at https://www.sourcefabric.org/superdesk/license
+
+"""Superdesk Files"""
+
+from copy import deepcopy
+from typing import Any, Generic, TypeVar
+from bson import ObjectId
+
+from planning.types import HistoryResourceModel
+from superdesk.core import get_current_app
+from superdesk.core.resources import AsyncResourceService
+from superdesk.resource_fields import ID_FIELD
+from .item_lock import LOCK_ACTION, LOCK_USER, LOCK_TIME, LOCK_SESSION
+from superdesk.metadata.item import ITEM_TYPE
+
+
+HistoryResourceModelType = TypeVar("HistoryResourceModelType", bound=HistoryResourceModel)
+
+fields_to_remove = [
+    "_id",
+    "_etag",
+    "_current_version",
+    "_updated",
+    "_created",
+    "_links",
+    "version_creator",
+    "guid",
+    LOCK_ACTION,
+    LOCK_USER,
+    LOCK_TIME,
+    LOCK_SESSION,
+    "planning_ids",
+    "_updates_schedule",
+    "_planning_schedule",
+    "_planning_date",
+    "_reschedule_from_schedule",
+    "versioncreated",
+]
+
+
+class HistoryAsyncService(AsyncResourceService[Generic[HistoryResourceModelType]]):
+    """Provide common async methods for tracking history of Creation, Updates and Spiking to collections"""
+
+    async def on_item_created(self, items: list[dict[str, Any]], operation: str | None = None):
+        for item in items:
+            if not item.get("duplicate_from"):
+                await self._save_history(
+                    {ID_FIELD: ObjectId(item[ID_FIELD]) if ObjectId.is_valid(item[ID_FIELD]) else str(item[ID_FIELD])},
+                    deepcopy(item),
+                    operation or "create",
+                )
+
+    async def on_item_updated(self, updates: dict[str, Any], original: dict[str, Any], operation: str | None = None):
+        item = deepcopy(original)
+        if list(item.keys()) == ["_id"]:
+            diff = updates
+        else:
+            diff = await self._changes(original, updates)
+            if updates:
+                item.update(updates)
+
+        await self._save_history(item, diff, operation or "edited")
+
+    async def on_spike(self, updates: dict[str, Any], original: dict[str, Any]):
+        await self.on_item_updated(updates, original, "spiked")
+
+    async def on_unspike(self, updates: dict[str, Any], original: dict[str, Any]):
+        await self.on_item_updated(updates, original, "unspiked")
+
+    async def on_cancel(self, updates: dict[str, Any], original: dict[str, Any]):
+        operation = "events_cancel" if original.get(ITEM_TYPE) == "event" else "planning_cancel"
+        await self.on_item_updated(updates, original, operation)
+
+    async def on_reschedule(self, updates: dict[str, Any], original: dict[str, Any]):
+        await self.on_item_updated(updates, original, "reschedule")
+
+    async def on_reschedule_from(self, item: dict[str, Any]):
+        new_item = deepcopy(item)
+        await self._save_history({ID_FIELD: str(item[ID_FIELD])}, new_item, "reschedule_from")
+
+    async def on_postpone(self, updates: dict[str, Any], original: dict[str, Any]):
+        await self.on_item_updated(updates, original, "postpone")
+
+    async def get_user_id(self):
+        user = get_current_app().get_current_user_dict()
+        if user:
+            return user.get("_id")
+
+    async def _changes(self, original: dict[str, Any], updates: dict[str, Any]):
+        """
+        Given the original record and the updates calculate what has changed and what is new
+
+        :param original:
+        :param updates:
+        :return: dictionary of what was changed and what was added
+        """
+        original_keys = set(original.keys())
+        updates_keys = set(updates.keys())
+        intersect_keys = original_keys.intersection(updates_keys)
+        modified = {o: updates[o] for o in intersect_keys if original[o] != updates[o]}
+        added_keys = updates_keys - original_keys
+        added = {a: updates[a] for a in added_keys}
+        modified.update(added)
+        return self._remove_unwanted_fields(modified)
+
+    def _remove_unwanted_fields(self, update: dict[str, Any]):
+        if update:
+            update_copy = deepcopy(update)
+            for field in fields_to_remove:
+                update_copy.pop(field, None)
+
+            return update_copy
+        return update
+
+    async def _save_history(self, item: Any, update: dict[str, Any], operation: str | None = None):
+        raise NotImplementedError()

server/planning/module.py

@@ -1,6 +1,6 @@
 from superdesk.core.module import Module
-from planning.events import events_resource_config
-from planning.planning import planning_resource_config
+from planning.events import events_resource_config, events_history_resource_config
+from planning.planning import planning_resource_config, planning_history_resource_config
 from planning.assignments import assignments_resource_config, delivery_resource_config
 from planning.published import published_resource_config
 
@@ -13,5 +13,7 @@
         assignments_resource_config,
         published_resource_config,
         delivery_resource_config,
+        events_history_resource_config,
+        planning_history_resource_config,
     ],
 )

server/planning/planning/__init__.py

@@ -43,13 +43,16 @@
 from .planning_featured import PlanningFeaturedResource, PlanningFeaturedService
 from .planning_files import PlanningFilesResource, PlanningFilesService
 
-from .module import planning_resource_config
+from .module import planning_resource_config, planning_history_resource_config
 from .service import PlanningAsyncService
+from .planning_history_async_service import PlanningHistoryAsyncService
 
 
 __all__ = [
     "planning_resource_config",
     "PlanningAsyncService",
+    "PlanningHistoryAsyncService",
+    "planning_history_resource_config",
 ]
 
 

server/planning/planning/module.py

@@ -3,11 +3,13 @@
     MongoIndexOptions,
     MongoResourceConfig,
     ElasticResourceConfig,
+    RestEndpointConfig,
 )
 
-from planning.types import PlanningResourceModel
+from planning.types import PlanningResourceModel, PlanningHistoryResourceModel
 
 from .service import PlanningAsyncService
+from .planning_history_async_service import PlanningHistoryAsyncService
 
 planning_resource_config = ResourceConfig(
     name="planning",
@@ -24,3 +26,19 @@
     ),
     elastic=ElasticResourceConfig(),
 )
+
+planning_history_resource_config = ResourceConfig(
+    name="planning_history",
+    data_class=PlanningHistoryResourceModel,
+    service=PlanningHistoryAsyncService,
+    mongo=MongoResourceConfig(
+        indexes=[
+            MongoIndexOptions(
+                name="planning_id",
+                keys=[("planning_id", 1)],
+                unique=False,
+            ),
+        ],
+    ),
+    rest_endpoints=RestEndpointConfig(resource_methods=["GET"], item_methods=["GET"]),
+)

server/planning/planning/planning_history.py

@@ -42,6 +42,7 @@ class PlanningHistoryResource(Resource):
         "operation": {"type": "string"},
         "update": {"type": "dict", "nullable": True},
     }
+    internal_resource = True
 
 
 class PlanningHistoryService(HistoryService):