ungana

commit 33422b8d62e35eb239fdfdc4ae17a94351547ba1
parent ddf991810b4e13478b2056236229aa5253d81269
Author: Carlosokumu <carlosokumu254@gmail.com>
Date:   Mon,  1 Sep 2025 12:54:05 +0300

add an IcalHelper

Diffstat:
A
ungana/ical/ical_helper.py
 | 
110
+++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++

1 file changed, 110 insertions(+), 0 deletions(-)
diff --git a/ungana/ical/ical_helper.py b/ungana/ical/ical_helper.py
@@ -0,0 +1,110 @@
+from datetime import datetime
+from typing import Any, Dict, Optional
+from zoneinfo import ZoneInfo
+from icalendar import Calendar, Event
+
+
+class ICalHelper:
+
+    @staticmethod
+    def update_event(cal: Calendar, updates: Dict[str, Any], attachments: list = None, filename: str = None) -> Calendar:
+        uid = cal.walk("VEVENT")[0].get("UID")
+        event_found = False
+
+        for component in cal.walk():
+            if component.name == "VEVENT" and str(component.get("UID")) == uid:
+                event_found = True
+                for key, value in updates.items():
+                    if key in component:
+                        component.pop(key)
+
+                    if isinstance(value, tuple) and len(value) == 2:
+                        v, params = value
+                        component.add(key, v, parameters=params)
+                    else:
+                        component[key] = value
+
+                # Deduplicate attachments
+                if attachments:
+                    existing_attachments = {
+                        str(a) for a in component.get("ATTACH", [])
+                    }
+
+                    for attachment in attachments:
+                        if isinstance(attachment, tuple) and len(attachment) == 3:
+                            prop, value, params = attachment
+                            if value not in existing_attachments:
+                                component.add(prop, value, parameters=params)
+                                existing_attachments.add(value)
+                        else:
+                            value = str(attachment)
+                            if value not in existing_attachments:
+                                component.add("ATTACH", value)
+                                existing_attachments.add(value)
+
+                break
+
+        if not event_found:
+            raise ValueError(f"No event with UID {uid} found.")
+        if filename:
+            ICalHelper.update_ical_file(cal, filename)
+
+        return cal
+
+    @staticmethod
+    def update_ical_file(cal: Calendar, filename: str) -> None:
+        ical_bytes = cal.to_ical()
+        unfolded = ical_bytes.replace(b"\r\n ", b"").replace(b"\n ", b"")
+        with open(filename, 'wb') as f:
+            f.write(unfolded)
+
+    @staticmethod
+    def get_first_event(cal):
+        """Return the first VEVENT component in the calendar (or None if no events exist)."""
+        for component in cal.walk():
+            if component.name == "VEVENT":
+                return component  
+        return None 
+    
+    @staticmethod
+    def get_all_events(cal):
+        """Return a list of all VEVENT components in the calendar."""
+        return [component for component in cal.walk() if component.name == "VEVENT"]
+    
+    @staticmethod
+    def normalize_ical_field(value, tzid: Optional[str] = None):
+        if isinstance(value, list):
+            value = value[0]
+        if hasattr(value, "dt"):
+            value = value.dt
+
+        if isinstance(value, datetime):
+            if value.tzinfo is None and tzid:
+                return value.replace(tzinfo=ZoneInfo(tzid))
+            return value
+
+        if hasattr(value, "to_ical"):
+            return value.to_ical().decode()
+
+        return str(value)  
+
+    @staticmethod
+    def check_existing_event(cal: Calendar, event_data: Event) -> bool:
+        tzid = event_data.get("tzid")
+
+        candidate_key = (
+            ICalHelper.normalize_ical_field(event_data.get("start"), tzid),
+            ICalHelper.normalize_ical_field(event_data.get("end"), tzid),
+            ICalHelper.normalize_ical_field(event_data.get("summary")),
+        )
+        for component in cal.walk("VEVENT"):
+            existing_key = (
+                ICalHelper.normalize_ical_field(component.get("DTSTART"), tzid),
+                ICalHelper.normalize_ical_field(component.get("DTEND"), tzid),
+                ICalHelper.normalize_ical_field(component.get("SUMMARY")),
+            )
+            if existing_key == candidate_key:
+                return True
+        return False
+
+