From 77a0aa3784a46f61a9368f01ee2da7934b3e52e2 Mon Sep 17 00:00:00 2001 From: David Barnett Date: Thu, 26 Sep 2024 11:41:03 -0600 Subject: [PATCH] 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 | 34 +++++++++++++++++----------- gcalcli/ics.py | 60 +++++++++++++++++++++++++++++++++++++++++-------- 2 files changed, 72 insertions(+), 22 deletions(-) diff --git a/gcalcli/gcal.py b/gcalcli/gcal.py index e2303cf..df37a9e 100644 --- a/gcalcli/gcal.py +++ b/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,22 +1612,20 @@ 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( @@ -1634,7 +1633,7 @@ def ImportICS(self, verbose=False, dump=False, reminders=None, 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 diff --git a/gcalcli/ics.py b/gcalcli/ics.py index d7b51ef..3da9f23 100644 --- a/gcalcli/ics.py +++ b/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