[WIP] Timeout queue for posting updates

classes.py

@@ -1,5 +1,6 @@
-from typing import Dict, List, Tuple, TypedDict, NamedTuple
+from typing import Dict, List, Tuple, TypedDict, NamedTuple, override
 from dataclasses import dataclass
+from datetime import datetime
 
 Property = NamedTuple("Property", [("name", str), ("value", str)])
 
@@ -11,7 +12,6 @@ class Page(TypedDict):
     last_edited_time: str
     properties: Dict[str, Property]
     url: str
-
 
 class PropertyChange(TypedDict):
     """Represents a change in a property
@@ -28,9 +28,29 @@ class ChangeSet():
     """Pages that were added in the diff"""
     removed: List[Page]
     """Pages that were removed in the diff"""
-    changed: List[Tuple[Page, List[PropertyChange]]]
+    changed: Dict[str, Tuple[List[PropertyChange], datetime]]
     """Pages that had their properties changed in the diff. The key is the page id and the value is the list of 
     properties that were changed"""
 
     def is_empty(self) -> bool:
-        return len(self.added) == 0 and len(self.removed) == 0 and len(self.changed) == 0
+        return self.added == [] and self.removed == [] and self.changed == {}
+
+    def merge_set(self, other_changed_set):
+        """merges the changes from the other set into the current set
+        """
+
+        # For the added and removed pages, we simply add them to the new set if there is any
+        self.added += other_changed_set.added
+        self.removed += other_changed_set.removed
+
+        # For each change we check if its already in our dict
+        for page_id, (propertyList, timestamp) in other_changed_set.changed.items():
+            if page_id in self.changed:
+                property_changes, new_timestamp = self.changed[page_id]
+                if timestamp > new_timestamp:
+                    new_timestamp = timestamp
+
+                self.changed[page_id] = (propertyList + property_changes, new_timestamp)
+            else:
+                self.changed[page_id] = (propertyList, timestamp)
+

display.py

@@ -2,18 +2,30 @@
 from enum import Enum
 from typing import List, Dict, Tuple
 from webhook import WebhookEmbed
+from datetime import datetime, timedelta
 
 class MessageColor(Enum):
     ADD = 0x44CC00
     REMOVE = 0xFF2A00
     CHANGE = 0xFFD000
 
-def generate_changeset_messages(changeset: ChangeSet) -> List[WebhookEmbed]:
-    # changed: List[Tuple[Page, List[PropertyChange]]]
+def changeset_timeout(changeset: ChangeSet, timeout: timedelta, data: Dict[str, Page]) -> List[WebhookEmbed]:
+    """Processes the changeSet on timeout by parsing the set into embeds and purging expired elements in the set
+    """
     res = []
     res += generate_added_messages(changeset.added)
     res += generate_removed_messages(changeset.removed)
-    res += generate_changed_messages(changeset.changed)
+
+    # List[Tuple[Page, List[PropertyChange]]]
+    list_of_pages_to_purge = check_timeout_on_changed(changeset.changed, timeout)
+    res += generate_changed_messages(list_of_pages_to_purge, data)
+
+    # Cleanup the existing changeSet
+    changeset.added = []
+    changeset.removed = []
+    for page_id, _ in list_of_pages_to_purge:
+        del changeset.changed[page_id]
+
     return res
 
 def generate_properties_text(page: Page) -> str:
@@ -71,10 +83,28 @@ def generate_removed_messages(removals: List[Page]) -> List[WebhookEmbed]:
 
     return res
 
-def generate_changed_messages(pages: List[Tuple[Page, List[PropertyChange]]]) -> List[WebhookEmbed]:
+def check_timeout_on_changed(changes: Dict[str, Tuple[List[PropertyChange], datetime]], timeout: timedelta) -> List[Tuple[str, List[PropertyChange]]]:
+    res = []
+
+    current_time = datetime.now()
+    for page_id, (propertychange, timestamp) in changes.items():
+        time_delta = current_time - timestamp 
+
+        if time_delta > timeout:
+            res.append((page_id, propertychange))
+
+    return res
+
+
+def generate_changed_messages(pages: List[Tuple[str, List[PropertyChange]]], data: Dict[str, Page]) -> List[WebhookEmbed]:
+    """Generate the messages for changed by checking if any pages are expired
+        If expired then add to the output message and remove the page from the queue
+    """
+
     res = []
 
-    for page, changes in pages:
+    for page_id, changes in pages:
+        page = data[page_id]
         message = WebhookEmbed(
             title = f"Task Changed: {page['properties']['title'].value}",
             url = page["url"],

main.py

@@ -1,12 +1,13 @@
 from dotenv import load_dotenv
 from notion_client import Client
 from typing import Dict
-from classes import Page
+from classes import ChangeSet, Page
 from webhook import generate_message, send_message
-from display import generate_changeset_messages
+from display import changeset_timeout 
 import os
 from utils import parse_db_query_response, identify_changes
 from time import sleep
+from datetime import timedelta
 import logging
 import sys
 
@@ -23,26 +24,47 @@ def get_state() -> Dict[str, Page]:
     db_data = get_database_data()
     return parse_db_query_response(db_data)
 
+"""
+We identify the new changes in every polling interval.
+If there are new changes, we apply to the changeset.
+Then, if the changeset is non-empty, we try to generate a message:
+    Specifically, the message is only non-empty if there is any expired fields
+    Then send message if message is non-empty
+Set new state and poll again.
+"""
+
 def main():
     logging.basicConfig(stream=sys.stdout, level=logging.INFO)
     logger = logging.getLogger()
+    polling_interval = int(os.environ.get("NOTION_POLLING_INTERVAL_SECONDS", 60))
+    timeout = timedelta(minutes=2)
     prev_state = []
     prev_state = get_state()
+    changeset = ChangeSet([], [], {})
 
     while True:
         new_state = get_state()
-        changeset = identify_changes(prev_state, new_state)
+        new_changeset = identify_changes(prev_state, new_state)
 
-        if changeset.is_empty():
+        if new_changeset.is_empty():
             logger.info("No changes detected")
+
         else:
-            messages = generate_changeset_messages(changeset)
-            logger.info(f"Changes detected. {len(messages)} to be sent.")
+            changeset.merge_set(new_changeset)
+            logger.info("Changes detected.")
+
+        if not(changeset.is_empty()):
+            messages = changeset_timeout(changeset, timeout, new_state)
             message = generate_message(messages)
-            send_message(message)
-            prev_state = new_state
 
-        sleep(int(os.environ.get("NOTION_POLLING_INTERVAL_SECONDS", 60)))
+            # Expect message == "" if there is nothing returned from generate_message
+            if message != "":
+                logger.info("Message sent.")
+                send_message(message)
+
+        prev_state = new_state
+        sleep(polling_interval)
+
 
 if __name__ == "__main__":
     main()

utils.py

@@ -1,3 +1,4 @@
+from datetime import datetime
 from typing import Dict, List, Tuple, Optional, Any, Literal
 from classes import Page, ChangeSet, PropertyChange, Property
 from property_parsers import parse_properties
@@ -100,7 +101,7 @@ def parse_diff(data: Dict[str, Page], diff: List[Tuple]) -> ChangeSet:
     res = ChangeSet(
         added=[],
         removed=[],
-        changed=[]
+        changed={}
     ) 
 
     changes: Dict[str, List[PropertyChange]] = defaultdict(list) # A dictionary that maps the page_id to the list of changes
@@ -114,7 +115,9 @@ def parse_diff(data: Dict[str, Page], diff: List[Tuple]) -> ChangeSet:
             res.removed = parse_add_or_remove(change_value) 
         elif change_type == 'change':
             change_key_tokens = change_key.split('.')
-
+
+            timestamp = datetime.now()
+            # TODO: maybe use a timestamp here in this token?
             if len(change_key_tokens) <= 2: # We are only interested in changes to properties, not the page itself
                 continue
 
@@ -123,8 +126,9 @@ def parse_diff(data: Dict[str, Page], diff: List[Tuple]) -> ChangeSet:
 
             changes[page_id].append(change)
 
-    for page_id, change_list in changes.items():
-        res.changed.append((data[page_id], change_list))
+            # changed: Dict[page_id, Tuple[List[PropertyChange], datetime]]
+            for page_id, change_list in changes.items():
+                res.changed[page_id] = (change_list, timestamp)
 
     return res