Improve DigestPlugin

7 files changed, 206 insertions, 109 deletions

diff --git a/Aggrofile b/Aggrofile
index 3c28000..470d2c0 100644
--- a/Aggrofile
+++ b/Aggrofile
@@ -19,11 +19,12 @@
             "plugin": "MapPlugin",
             "map_expr": "set_field(item, 'author', 'Marius Schulz')"
         },
-        "mariusschulz_digest": {
+        "typescript_digest": {
             "plugin": "DigestPlugin",
-            "schedule_expr": "schedule.every(30).seconds",
-            "digest_title": "Marius Schulz digest",
-            "digest_description": "Marius Schulz digest"
+            "from_datetime": "2010-01-01T00:00:00Z",
+            "interval": "2m",
+            "max_length": 1000,
+            "digest_title_prefix": "Only Typescript Bimonthly"
         },
         "concat": {
             "plugin": "ConcatPlugin"
@@ -32,17 +33,11 @@
             "plugin": "FilterPlugin",
             "filter_expr": "'TypeScript' in item.title"
         },
-        "output_feed": {
+        "typescript_digest_feed": {
             "plugin": "FeedSinkPlugin",
-            "feed_id": "test_feed",
-            "feed_title": "Only Typescript posts",
-            "feed_description": "Feed where all posts have Typescript in the title"
-        },
-        "mariusschulz_digest_feed": {
-            "plugin": "FeedSinkPlugin",
-            "feed_id": "marius_digest_feed",
-            "feed_title": "Marius Schulz digest",
-            "feed_description": "Marius Schulz digest"
+            "feed_id": "typescript_digest",
+            "feed_title": "Only Typescript Digest",
+            "feed_description": "Bimonthly digest where all posts have Typescript in the title"
         }
     },
     "graph": {
@@ -56,17 +51,16 @@
             "mariusschulz_add_author"
         ],
         "mariusschulz_add_author": [
-            "concat",
-            "mariusschulz_digest"
+            "concat"
         ],
         "concat": [
             "title_filter"
         ],
         "title_filter": [
-            "output_feed"
+            "typescript_digest"
         ],
-        "mariusschulz_digest": [
-            "mariusschulz_digest_feed"
+        "typescript_digest": [
+            "typescript_digest_feed"
         ]
     }
 }
\ No newline at end of file
diff --git a/app/Item.py b/app/Item.py
index 23a841e..7376301 100644
--- a/app/Item.py
+++ b/app/Item.py
@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-import time
+from datetime import datetime
 
 
 @dataclass
@@ -19,7 +19,8 @@ class Item:
     link: str | None
     description: str | None
     author: str | None
-    pub_date: time.struct_time | None
+    pub_date: datetime | None
     category: str | None
     comments: str | None
     enclosures: list[ItemEnclosure]
+    guid: str
diff --git a/app/utils.py b/app/utils.py
index 77ea775..626ee24 100644
--- a/app/utils.py
+++ b/app/utils.py
@@ -1,4 +1,4 @@
-import time
+from datetime import datetime
 from typing import TypeAlias, Union
 from app.Item import Item
 from dataclasses import asdict
@@ -17,9 +17,11 @@ def get_param(key: str, params: Params) -> str:
 
 
 def item_to_dict(item: Item) -> ItemDict:
-    return asdict(item)
+    d = asdict(item)
+    d["pub_date"] = datetime.isoformat(d["pub_date"]) if d["pub_date"] else None
+    return d
 
 
 def dict_to_item(d: ItemDict) -> Item:
     pub_date, rest = (lambda pub_date, **rest: (pub_date, rest))(**d)  # type: ignore
-    return Item(pub_date=time.struct_time(pub_date), **rest)  # type: ignore
+    return Item(pub_date=datetime.fromisoformat(pub_date), **rest)  # type: ignore
diff --git a/plugins/ConcatPlugin.py b/plugins/ConcatPlugin.py
index 43d7db2..a0317cf 100644
--- a/plugins/ConcatPlugin.py
+++ b/plugins/ConcatPlugin.py
@@ -1,5 +1,5 @@
 from typing import Any
-import time
+from datetime import datetime
 
 from tinydb import Query
 from app.Item import Item
@@ -13,9 +13,9 @@ class Plugin(PluginInterface):
         super().__init__(id, params)
         print(f"[ConcatPlugin#{self.id}] initialized")
 
-    def item_sort_key(self, item: Item) -> time.struct_time:
+    def item_sort_key(self, item: Item) -> datetime:
         if item.pub_date is None:
-            return time.localtime()
+            return datetime.now()
 
         return item.pub_date
 
diff --git a/plugins/DigestPlugin.py b/plugins/DigestPlugin.py
index 9a097ba..d6e8eb4 100644
--- a/plugins/DigestPlugin.py
+++ b/plugins/DigestPlugin.py
@@ -1,113 +1,193 @@
+from dataclasses import dataclass
 from typing import Any
-import time
+from datetime import datetime, timedelta
+import re
 
 from tinydb import Query
 from app.Item import Item
 from app.PluginInterface import Params, PluginInterface
 from app.DatabaseManager import database_manager
-from app.utils import ItemDict, dict_to_item, get_param, item_to_dict
+from app.utils import dict_to_item, get_param, item_to_dict
 
 
-def remove_duplicates_by_key(lst: list[ItemDict], key: str) -> list[ItemDict]:
-    return list({x[key]: x for x in lst}.values())
+def add_days(start: datetime, days: int):
+    return start + timedelta(days=days)
 
 
-class Plugin(PluginInterface):
-    def __init__(self, id: str, params: Params) -> None:
-        super().__init__(id, params)
-        self.digest_title = get_param("digest_title", params)
-        self.digest_description = get_param("digest_description", params)
-        print(f"[DigestPlugin#{self.id}] initialized")
+def add_weeks(start: datetime, weeks: int):
+    return start + timedelta(weeks=weeks)
 
-    def item_sort_key(self, item: Item) -> time.struct_time:
-        if item.pub_date is None:
-            return time.localtime()
 
-        return item.pub_date
+def add_months(start: datetime, months: int):
+    year, month = divmod(start.month - 1 + months, 12)
+    return datetime(
+        year=start.year + year, month=month + 1, day=start.day, tzinfo=start.tzinfo
+    )
 
-    def make_digest(self) -> list[Item]:
-        plugin_state_q = Query().plugin_id == self.id
-        _d: Any = database_manager.plugin_states.get(plugin_state_q)  # type: ignore
-        data: dict[str, Any] = (
-            _d
-            if _d is not None
-            else {
-                "plugin_id": self.id,
-                "state": {"cutoff_timestamp": time.localtime(), "channels": {}},
-            }
-        )
-        state = data["state"]
-        channels: dict[str, list[ItemDict]] = state["channels"]
-        cutoff_timestamp = time.struct_time(state["cutoff_timestamp"])
 
-        aggregated_items: list[Item] = []
-        for data_source_id in channels:
-            item_dicts = channels[data_source_id]
-            items = list(map(dict_to_item, item_dicts))
-            for item in items:
-                if item.pub_date is not None and cutoff_timestamp < item.pub_date:
-                    aggregated_items.append(item)
+def add_years(start: datetime, years: int):
+    return datetime(
+        year=start.year + years, month=start.month, day=start.day, tzinfo=start.tzinfo
+    )
 
-        state["cutoff_timestamp"] = time.localtime()
 
-        digest_desc = (
-            f"{self.digest_description}<br><br>" if self.digest_description else ""
-        )
+add_interval_map = {"d": add_days, "w": add_weeks, "m": add_months, "y": add_years}
 
-        for item in aggregated_items:
-            pub_date_stamp = (
-                time.strftime("%a, %d %b %Y %H:%M:%S +0000", item.pub_date)
-                if item.pub_date
-                else ""
-            )
-            author = item.author if item.author else ""
-            digest_desc += f"<strong>{item.title}</strong><br>"
-            digest_desc += f"<small>{pub_date_stamp} <i>{author}</i></small><br><br>"
-            digest_desc += item.description if item.description else ""
-            digest_desc += "<br><br>"
 
-        digest_item: Item = Item(
-            title=self.digest_title,
-            description=digest_desc,
-            link=None,
-            author=None,
-            pub_date=time.localtime(),
-            category=None,
-            comments=None,
-            enclosures=[],
+@dataclass
+class Span:
+    span_title: str
+    items: list[Item]
+
+
+def group_by_time_span(
+    lst: list[Item], interval: tuple[int, str], start_datetime: datetime
+) -> list[Span]:
+    interval_n, interval_suffix = interval
+    datetime_1970: datetime = datetime.fromtimestamp(0)
+    sorted_lst = sorted(
+        lst,
+        key=lambda item: item.pub_date if item.pub_date else datetime_1970,
+    )
+
+    groups: list[Span] = []
+    start = start_datetime
+    end = add_interval_map[interval_suffix](start, interval_n)
+    group: list[Item] = []
+    for item in sorted_lst:
+        item_datetime = item.pub_date if item.pub_date else datetime_1970
+        while item_datetime >= end:
+            if group:
+                formatted_start = start.strftime("%Y-%m-%d")
+                formatted_end = end.strftime("%Y-%m-%d")
+                groups.append(
+                    Span(
+                        span_title=f"{formatted_start} to {formatted_end}", items=group
+                    )
+                )
+                group = []
+            start = end
+            end = add_interval_map[interval_suffix](start, interval_n)
+
+        if start <= item_datetime < end:
+            group.append(item)
+
+    if group:
+        formatted_start = start.strftime("%Y-%m-%d")
+        formatted_end = end.strftime("%Y-%m-%d")
+        groups.append(
+            Span(span_title=f"{formatted_start} to {formatted_end}", items=group)
         )
 
-        database_manager.plugin_states.upsert(  # type: ignore
-            {"plugin_id": self.id, "state": state}, plugin_state_q
+    return groups
+
+
+def parse_interval(interval: str) -> tuple[int, str]:
+    # Parse the interval string into a number and a suffix
+    match = re.match(r"(\d+)([dwmy])", interval)
+    if not match:
+        raise ValueError(
+            'Invalid interval format. Must be a number followed by one of "d", "w", "m", or "y".'
         )
 
-        return [digest_item]
+    num, suffix = match.groups()
+    num = int(num)
+
+    return (num, suffix)
+
+
+class Plugin(PluginInterface):
+    def __init__(self, id: str, params: Params) -> None:
+        super().__init__(id, params)
+        self.digest_title_prefix = get_param("digest_title_prefix", params)
+        self.digest_description = params.get("digest_description", None)
+        self.digest_link = params.get("digest_link", None)
+        from_datatime_str = get_param("from_datetime", params)
+        self.from_datetime = datetime.fromisoformat(from_datatime_str)
+        interval = get_param("interval", params)
+        self.interval_pair = parse_interval(interval)
+        self.max_length = int(params.get("max_length", "1000"))
+        print(f"[DigestPlugin#{self.id}] initialized")
+
+    def process(self, source_id: str | None, items: list[Item]) -> list[Item]:
+        print(f"[DigestPlugin#{self.id}] process called, n={len(items)}")
+        if source_id is None:
+            raise Exception(f"[DigestPlugin#{self.id}] can not be scheduled")
 
-    def add_to_state(self, source_id: str, items: list[Item]) -> list[Item]:
         plugin_state_q = Query().plugin_id == self.id
         _d: Any = database_manager.plugin_states.get(plugin_state_q)  # type: ignore
         data: dict[str, Any] = (
-            _d
-            if _d is not None
-            else {
-                "plugin_id": self.id,
-                "state": {"cutoff_timestamp": time.localtime(), "channels": {}},
-            }
+            _d if _d is not None else {"plugin_id": self.id, "state": {}}
         )
         state = data["state"]
 
         items_as_dicts = list(map(item_to_dict, items))
-        state["channels"][source_id] = items_as_dicts
+        state[source_id] = items_as_dicts
 
         database_manager.plugin_states.upsert(  # type: ignore
             {"plugin_id": self.id, "state": state}, plugin_state_q
         )
 
-        return []
+        aggregated_items: list[Item] = []
+        for data_source_id in state:
+            source_item_dicts = state[data_source_id]
+            source_items = map(dict_to_item, source_item_dicts)
+            aggregated_items += source_items
 
-    def process(self, source_id: str | None, items: list[Item]) -> list[Item]:
-        print(f"[DigestPlugin#{self.id}] process called, n={len(items)}")
-        if source_id is None:
-            return self.make_digest()
-        else:
-            return self.add_to_state(source_id, items)
+        spans: list[Span] = group_by_time_span(
+            aggregated_items, self.interval_pair, self.from_datetime
+        )
+
+        digest_items: list[Item] = []
+        for span in spans:
+            if len(span.items) == 0:
+                continue
+
+            digest_title = (
+                f"{self.digest_title_prefix} {span.span_title}"
+                if self.digest_title_prefix
+                else span.span_title
+            )
+            digest_desc = (
+                f"{self.digest_description}<br><br>" if self.digest_description else ""
+            )
+            for span_item in span.items:
+                pub_date_stamp = (
+                    span_item.pub_date.strftime("%a, %d %b %Y %H:%M:%S +0000")
+                    if span_item.pub_date
+                    else ""
+                )
+                author = span_item.author if span_item.author else ""
+                digest_desc += f'<strong><a href="{span_item.link}">{span_item.title}</a></strong><br>'
+                digest_desc += (
+                    f"<small>{pub_date_stamp} <i>{author}</i></small><br><br>"
+                )
+                digest_desc += span_item.description if span_item.description else ""
+                if len(digest_desc) > self.max_length:
+                    digest_desc = f"{digest_desc[0:self.max_length]}..."
+
+                digest_desc += "<br><br>"
+
+            datetime_1970: datetime = datetime.fromtimestamp(0)
+            digest_pub_date = (
+                span.items[-1].pub_date if span.items[-1].pub_date else datetime_1970
+            )
+            digest_item: Item = Item(
+                title=digest_title,
+                description=digest_desc,
+                link=self.digest_link,
+                author=None,
+                pub_date=digest_pub_date,
+                category=None,
+                comments=None,
+                enclosures=[],
+                guid=f"aggro__{self.id}__{digest_pub_date.isoformat()}",
+            )
+
+            digest_items.append(digest_item)
+
+        print(
+            f"[DigestPlugin#{self.id}] process returning items, n={len(digest_items)}"
+        )
+
+        return digest_items
diff --git a/plugins/FeedSinkPlugin.py b/plugins/FeedSinkPlugin.py
index c0fc4c9..02e20e9 100644
--- a/plugins/FeedSinkPlugin.py
+++ b/plugins/FeedSinkPlugin.py
@@ -1,4 +1,3 @@
-import time
 from typing import Any
 import xml.etree.ElementTree as ET
 
@@ -50,13 +49,16 @@ class Plugin(PluginInterface):
                 item_comments.text = item.comments
             if item.pub_date is not None:
                 item_pub_date = ET.SubElement(item_elem, "pubDate")
-                item_pub_date.text = time.strftime(
-                    "%a, %d %b %Y %H:%M:%S +0000", item.pub_date
+                item_pub_date.text = item.pub_date.strftime(
+                    "%a, %d %b %Y %H:%M:%S +0000"
                 )
             if item.author is not None:
                 item_author = ET.SubElement(item_elem, "author")
                 item_author.text = item.author
 
+            item_guid = ET.SubElement(item_elem, "guid")
+            item_guid.text = item.guid
+
             for enc in item.enclosures:
                 ET.SubElement(
                     item_elem,
diff --git a/plugins/FeedSourcePlugin.py b/plugins/FeedSourcePlugin.py
index 575e5d6..e271092 100644
--- a/plugins/FeedSourcePlugin.py
+++ b/plugins/FeedSourcePlugin.py
@@ -1,10 +1,19 @@
 import feedparser  # type: ignore
+import time
+from datetime import datetime, timezone
 from typing import Any
 from app.Item import Item, ItemEnclosure
 from app.PluginInterface import Params, PluginInterface
 from app.utils import ItemDict, get_param
 
 
+def struct_time_to_utc_datetime(struct_time: time.struct_time) -> datetime:
+    timestamp = time.mktime(struct_time)
+    naive_datetime = datetime.fromtimestamp(timestamp)
+    aware_datetime = naive_datetime.replace(tzinfo=timezone.utc)
+    return aware_datetime
+
+
 class Plugin(PluginInterface):
     def __init__(self, id: str, params: Params) -> None:
         super().__init__(id, params)
@@ -39,16 +48,25 @@ class Plugin(PluginInterface):
                     )
                 )
 
+            datetime_1970: datetime = datetime.fromtimestamp(0)
+            published_time_struct: Any = d["published_parsed"]
+            published_datetime = (
+                struct_time_to_utc_datetime(published_time_struct)
+                if published_time_struct
+                else datetime_1970
+            )
+
             result_items.append(
                 Item(
                     title=d.get("title", None),  # type: ignore
-                    link=d.get("link", None),  # type: ignore
+                    link=d["link"],  # type: ignore
                     description=d.get("description", None),  # type: ignore
                     author=d.get("author", None),  # type: ignore
-                    pub_date=d.get("published_parsed", None),  # type: ignore
+                    pub_date=published_datetime,
                     category=d.get("category", None),  # type: ignore
                     comments=d.get("comments"),  # type: ignore
                     enclosures=item_enclosures,
+                    guid=d["link"],  # type: ignore
                 )
             )