Change import to dump ics file for any failed events (#778)

Quality-of-life improvement for scenarios like #730 so that it's easier to retry failed events with different settings (--use-legacy-import etc).

gcalcli/gcal.py

@@ -1562,14 +1562,14 @@ def ImportICS(self, verbose=False, dump=False, reminders=None,
                 self.printer.err_msg('Error: ' + str(e) + '!\n')
                 sys.exit(1)
 
-        events_to_import = ics.get_events(
+        ical_data = ics.get_ics_data(
             f,
             verbose=verbose,
             default_tz=self.cals[0]['timeZone'],
             printer=self.printer)
         if not dump and any(
-                self._event_should_use_new_import_api(event, self.cals[0])
-                for event in events_to_import):
+                self._event_should_use_new_import_api(event.body, self.cals[0])
+                for event in ical_data.events):
             self.printer.msg(
                 '\n'
                 'NOTE: This import will use a new graceful import feature in '
@@ -1581,14 +1581,15 @@ def ImportICS(self, verbose=False, dump=False, reminders=None,
 
         cal = self.cals[0]
         imported_cnt = 0
-        for event in events_to_import:
-            if not event:
+        failed_events = []
+        for event in ical_data.events:
+            if not event.body:
                 continue
 
             if dump:
                 continue
 
-            self._add_reminders(event, reminders)
+            self._add_reminders(event.body, reminders)
 
             if not verbose:
                 # Don't prompt, just assume user wants to import.
@@ -1611,30 +1612,28 @@ def ImportICS(self, verbose=False, dump=False, reminders=None,
             # Import event
             import_method = (
                 self.get_events().import_ if (
-                    self._event_should_use_new_import_api(event, cal))
+                    self._event_should_use_new_import_api(event.body, cal))
                 else self.get_events().insert)
             try:
                 new_event = self._retry_with_backoff(
-                    import_method(calendarId=cal['id'], body=event))
+                    import_method(calendarId=cal['id'], body=event.body))
             except HttpError as e:
+                failed_events.append(event)
                 try:
                     is_skipped_dupe = any(detail.get('reason') == 'duplicate'
                                           for detail in e.error_details)
                 except Exception:
                     # Fail gracefully so weird error responses don't blow up.
                     is_skipped_dupe = False
-                event_label = (
-                    f'"{event["summary"]}"' if event.get('summary')
-                    else f"with start {event['start']}"
-                )
+                event_label = event.label_str()
                 if is_skipped_dupe:
                     # TODO: #492 - Offer to force import dupe anyway?
                     self.printer.msg(
                         f'Skipped duplicate event {event_label}.\n')
                 else:
                     self.printer.err_msg(
                         f'Failed to import event {event_label}.\n')
-                    self.printer.msg(f'Event details: {event}\n')
+                    self.printer.msg(f'Event details: {event.body}\n')
                     self.printer.debug_msg(f'Error details: {e}\n')
             else:
                 imported_cnt += 1
@@ -1645,4 +1644,13 @@ def ImportICS(self, verbose=False, dump=False, reminders=None,
             f"Added {imported_cnt} events to calendar {cal['id']}\n"
         )
 
+        if failed_events:
+            ics_dump_path = ics.dump_partial_ical(
+                failed_events, ical_data.raw_components
+            )
+            self.printer.msg(
+                f"Dumped {len(failed_events)} failed events to "
+                f"{ics_dump_path!s}.\n"
+            )
+
         return True

gcalcli/ics.py

@@ -1,27 +1,52 @@
 """Helpers for working with iCal/ics format."""
 
+from dataclasses import dataclass
 import importlib.util
 import io
 from datetime import datetime
-from typing import Any, Optional
+import pathlib
+import tempfile
+from typing import Any, NamedTuple, Optional
 
 from gcalcli.printer import Printer
 from gcalcli.utils import localize_datetime
 
-EventBody = dict[str, Any]
+
+@dataclass
+class EventData:
+    body: Optional[dict[str, Any]]
+    source: Any
+
+    def label_str(self):
+        if self.source.get('summary'):
+            return f'"{self.source.summary}"'
+        elif self.source.get('dtstart') and self.source.dtstart.get('value'):
+            return f"with start {self.source.dtstart.value}"
+        else:
+            return None
+
+
+class IcalData(NamedTuple):
+    events: list[EventData]
+    raw_components: list[Any]
 
 
 def has_vobject_support() -> bool:
     return importlib.util.find_spec('vobject') is not None
 
 
-def get_events(
+def get_ics_data(
     ics: io.TextIOBase, verbose: bool, default_tz: str, printer: Printer
-) -> list[Optional[EventBody]]:
+) -> IcalData:
     import vobject
 
-    events: list[Optional[EventBody]] = []
+    events: list[EventData] = []
+    raw_components: list[Any] = []
     for v in vobject.readComponents(ics):
+        if v.name == 'VCALENDAR' and hasattr(v, 'components'):
+            raw_components.extend(
+                c for c in v.components() if c.name != 'VEVENT'
+            )
         # Strangely, in empty calendar cases vobject sometimes returns
         # Components with no vevent_list attribute at all.
         vevents = getattr(v, 'vevent_list', [])
@@ -31,12 +56,12 @@ def get_events(
             )
             for ve in vevents
         )
-    return events
+    return IcalData(events, raw_components)
 
 
 def CreateEventFromVOBJ(
     ve, verbose: bool, default_tz: str, printer: Printer
-) -> Optional[EventBody]:
+) -> EventData:
     event = {}
 
     if verbose:
@@ -56,7 +81,7 @@ def CreateEventFromVOBJ(
 
     if not hasattr(ve, 'dtstart') or not hasattr(ve, 'dtend'):
         printer.err_msg('Error: event does not have a dtstart and dtend!\n')
-        return None
+        return EventData(body=None, source=ve)
 
     if verbose:
         if ve.dtstart.value:
@@ -152,4 +177,21 @@ def CreateEventFromVOBJ(
             print(f'Sequence.....{sequence}')
         event['sequence'] = sequence
 
-    return event
+    return EventData(body=event, source=ve)
+
+
+def dump_partial_ical(
+    events: list[EventData], raw_components: list[Any]
+) -> pathlib.Path:
+    import vobject
+
+    tmp_dir = pathlib.Path(tempfile.mkdtemp(prefix="gcalcli."))
+    f_path = tmp_dir.joinpath("rej.ics")
+    cal = vobject.iCalendar()
+    for c in raw_components:
+        cal.add(c)
+    for event in events:
+        cal.add(event.source)
+    with open(f_path, 'w') as f:
+        f.write(cal.serialize())
+    return f_path