Support Gift transaction event, map timeline events using customerSupportChat payload when available

pyproject.toml

@@ -32,6 +32,8 @@ dependencies = [
     "shtab",
     "websockets>=14",
     "babel",
+    "python-jsonpath>=1.3.0",
+    "typing-extensions>=4.12.2",
 ]
 
 [project.scripts]

pytr/dl.py

@@ -6,7 +6,7 @@
 from requests_futures.sessions import FuturesSession  # type: ignore[import-untyped]
 
 from pytr.api import TradeRepublicError
-from pytr.timeline import Timeline, UnsupportedEventError
+from pytr.timeline import Timeline
 from pytr.utils import get_logger, preview
 
 
@@ -77,12 +77,7 @@ async def dl_loop(self):
             elif subscription.get("type", "") == "timelineActivityLog":
                 await self.tl.get_next_timeline_activity_log(response)
             elif subscription.get("type", "") == "timelineDetailV2":
-                try:
-                    self.tl.process_timelineDetail(response, self)
-                except UnsupportedEventError:
-                    self.log.warning("Ignoring unsupported event %s", response)
-                    self.tl.skipped_detail += 1
-                    self.tl.check_if_done(self)
+                self.tl.process_timelineDetail(response, self)
             else:
                 self.log.warning(f"unmatched subscription of type '{subscription['type']}':\n{preview(response)}")
 

pytr/event.py

@@ -34,6 +34,7 @@ class PPEventType(EventType):
     TAX_REFUND = "TAX_REFUND"
     TRANSFER_IN = "TRANSFER_IN"  # Currently not mapped to
     TRANSFER_OUT = "TRANSFER_OUT"  # Currently not mapped to
+    OTHER = "OTHER"  # Events that have no impact on the account balance.
 
 
 tr_event_type_mapping = {
@@ -74,6 +75,10 @@ class PPEventType(EventType):
     "SAVINGS_PLAN_INVOICE_CREATED": ConditionalEventType.TRADE_INVOICE,
     "benefits_spare_change_execution": ConditionalEventType.TRADE_INVOICE,
     "TRADE_INVOICE": ConditionalEventType.TRADE_INVOICE,
+    # Gifting
+    "GIFTER_TRANSACTION": PPEventType.TRANSFER_OUT,
+    # Other
+    "card_successful_verification": PPEventType.OTHER,
 }
 
 

pytr/timeline.py

@@ -1,12 +1,13 @@
 import json
 from datetime import datetime
+from typing import Optional, cast
 
-from .transactions import export_transactions
-from .utils import get_logger
+import jsonpath
 
+from pytr.types import TimelineDetailV2, TimelineDetailV2_CustomerSupportChatAction
 
-class UnsupportedEventError(Exception):
-    pass
+from .transactions import export_transactions
+from .utils import get_logger
 
 
 class Timeline:
@@ -39,7 +40,7 @@ async def get_next_timeline_transactions(self, response=None):
             for event in response["items"]:
                 if (
                     self.max_age_timestamp == 0
-                    or datetime.fromisoformat(event["timestamp"][:19]).timestamp() >= self.max_age_timestamp
+                    or datetime.fromisoformat(event["timestamp"]).timestamp() >= self.max_age_timestamp
                 ):
                     event["source"] = "timelineTransaction"
                     self.timeline_events[event["id"]] = event
@@ -125,9 +126,19 @@ def process_timelineDetail(self, response, dl):
         create other_events.json, events_with_documents.json and account_transactions.csv
         """
 
-        event = self.timeline_events.get(response["id"], None)
+        # Find the ID of the corresponding timeline event. This is burried deep in the last section of the
+        # response that contains the customer support information.
+        support_action = get_customer_support_chat_action(response)
+        if support_action and (timeline_event_id := support_action["payload"]["contextParams"].get("timelineEventId")):
+            pass
+        else:
+            timeline_event_id = response["id"]
+
+        event = self.timeline_events.get(timeline_event_id, None)
         if event is None:
-            raise UnsupportedEventError(response["id"])
+            self.log.warning("Missing timeline event %r for detail: %s", timeline_event_id, json.dumps(response))
+            self.skipped_detail += 1
+            return
 
         self.received_detail += 1
         event["details"] = response
@@ -205,3 +216,18 @@ def finish_timeline_details(self, dl):
         )
 
         dl.work_responses()
+
+
+def get_customer_support_chat_action(
+    timeline_detail: TimelineDetailV2,
+) -> Optional[TimelineDetailV2_CustomerSupportChatAction]:
+    """
+    From a `timelineDetailV2` object, find the `customerSupportChat` object.
+    """
+
+    JSONPATH = '$.sections[*].data[?(@.detail.action.type == "customerSupportChat")].detail.action'
+
+    for action in jsonpath.finditer(JSONPATH, timeline_detail):
+        return cast(TimelineDetailV2_CustomerSupportChatAction, action.obj)
+
+    return None

pytr/types.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+from typing import Any, Literal, TypedDict
+
+from typing_extensions import NotRequired
+
+
+class TimelineDetailV2(TypedDict):
+    """
+    Incomplete typed representation of the TR `timelineDetailV2` object.
+    """
+
+    id: str
+    sections: list[TimelineDetailV2_Section]
+
+
+class TimelineDetailV2_Section(TypedDict):
+    title: str
+    type: Literal["header", "table", "steps"]
+    data: dict[str, Any] | list[dict[str, Any]]
+
+
+class TimelineDetailV2_CustomerSupportChatAction(TypedDict):
+    type: Literal["customerSupportChat"]
+    payload: TimelineDetailV2_CustomerSupportChatAction_Payload
+    style: str
+
+
+class TimelineDetailV2_CustomerSupportChatAction_Payload(TypedDict):
+    contextParams: TimelineDetailV2_CustomerSupportChatAction_ContextParamms
+    contextCategory: str
+
+
+class TimelineDetailV2_CustomerSupportChatAction_ContextParamms(TypedDict):
+    chat_flow_key: str
+    timelineEventId: NotRequired[str]
+    savingsPlanId: NotRequired[str]
+    primId: NotRequired[str]
+    groupId: NotRequired[str]
+    createdAt: NotRequired[str]
+    amount: NotRequired[str]
+    iban: NotRequired[str]
+    interestPayoutId: NotRequired[str]

tests/test_timeline.py

@@ -0,0 +1,21 @@
+import json
+
+from pytr.timeline import get_customer_support_chat_action
+
+
+def test__get_customer_support_chat_action() -> None:
+    with open("tests/sample_event.json", "r") as file:
+        sample_data = json.load(file)
+
+    data = get_customer_support_chat_action(sample_data["details"])
+    assert data is not None
+    assert data == {
+        "payload": {
+            "contextParams": {
+                "timelineEventId": "d8a5aa3d-12a4-465a-90ad-3fca36eff19a",
+                "chat_flow_key": "NHC_0024_deposit_report_an_issue",
+            },
+            "contextCategory": "NHC",
+        },
+        "type": "customerSupportChat",
+    }