Fix import corner case bugs, add test, update ChangeLog

ChangeLog

@@ -2,6 +2,8 @@ v4.4.0
   * Fix lots of bugs by switching from deprecated oauth2client to
     google_auth_oauthlib
   * Friendlier help output when `import` command is missing vobject extra
+  * `import` command more gracefully handles existing events to avoid duplicates
+    and unnecessary edits (tsheinen, cryhot)
   * Handle encoding/decoding errors more gracefully by replacing with
     placeholder chars instead of blowing up
   * Fix `--lineart` option failing with unicode errors

gcalcli/argparsers.py

@@ -415,6 +415,13 @@ def get_argument_parser():
     _import.add_argument(
         '--dump', '-d', action='store_true',
         help='Print events and don\'t import')
+    _import.add_argument(
+        '--use-legacy-import',
+        action='store_true',
+        help='Use legacy "insert" operation instead of new graceful "import" '
+        'operation when importing calendar events. Note this option will be '
+        'removed in future releases.',
+    )
 
     default_cmd = 'notify-send -u critical -i appointment-soon -a gcalcli %s'
     remind = sub.add_parser(

gcalcli/gcal.py

@@ -22,7 +22,7 @@
 from google.auth.transport.requests import Request  # type: ignore
 
 from . import actions, ics, utils
-from ._types import Cache, CalendarListEntry
+from ._types import Cache, CalendarListEntry, Event
 from .actions import ACTIONS
 from .conflicts import ShowConflicts
 from .details import _valid_title, ACTION_DEFAULT, DETAILS_DEFAULT, HANDLERS
@@ -1417,6 +1417,21 @@ def Remind(self, minutes, command, use_reminders=False):
         if not pid:
             os.execvp(cmd[0], cmd)
 
+    def _event_should_use_new_import_api(
+            self, event: Event, cal: CalendarListEntry
+    ) -> bool:
+        """Evaluates whether a given event should use the newer "import" API.
+
+        Returns true if the user hasn't opted out and the event is cleanly
+        importable.
+        """
+        if not self.options.get('use_legacy_import'):
+            return False
+        event_includes_self = any(
+                'self' in a or a['email'] == cal['id']
+                for a in event.get('attendees', []))
+        return 'iCalUID' in event and event_includes_self
+
     def ImportICS(self, verbose=False, dump=False, reminders=None,
                   icsFile=None):
         if not ics.has_vobject_support():
@@ -1456,6 +1471,18 @@ def ImportICS(self, verbose=False, dump=False, reminders=None,
             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.printer.msg(
+                '\n'
+                'NOTE: This import will use a new graceful import feature in '
+                'gcalcli to avoid creating duplicate events (see '
+                'https://github.com/insanum/gcalcli/issues/492).\n'
+                'If you see any issues, you can cancel and retry with '
+                '--use-legacy-import to restore the old behavior.\n\n')
+            time.sleep(1)
+
         for event in events_to_import:
             if not event:
                 continue
@@ -1485,15 +1512,35 @@ def ImportICS(self, verbose=False, dump=False, reminders=None,
 
             # Import event
             import_method = (
-                self.get_events().import_ if 'iCalUID' in event
+                self.get_events().import_ if (
+                    self._event_should_use_new_import_api(event, self.cals[0]))
                 else self.get_events().insert)
-            new_event = self._retry_with_backoff(
-                import_method(
-                    calendarId=self.cals[0]['id'], body=event
+            try:
+                new_event = self._retry_with_backoff(
+                    import_method(calendarId=self.cals[0]['id'], body=event))
+            except HttpError as e:
+                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']}"
                 )
-            )
-            hlink = new_event.get('htmlLink')
-            self.printer.msg(f'New event added: {hlink}\n', 'green')
+                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.debug_msg(f'Error details: {e}\n')
+            else:
+                hlink = new_event.get('htmlLink')
+                self.printer.msg(f'New event added: {hlink}\n', 'green')
 
         # TODO: return the number of events added
         return True

gcalcli/ics.py

@@ -3,7 +3,7 @@
 import importlib.util
 import io
 from datetime import datetime
-from typing import Any, Dict, Iterable, Optional
+from typing import Any, Dict, List, Optional
 
 from gcalcli.printer import Printer
 from gcalcli.utils import localize_datetime
@@ -17,14 +17,18 @@ def has_vobject_support() -> bool:
 
 def get_events(
     ics: io.TextIOBase, verbose: bool, default_tz: str, printer: Printer
-) -> Iterable[Optional[EventBody]]:
+) -> List[Optional[EventBody]]:
     import vobject
 
+    events: List[Optional[EventBody]] = []
     for v in vobject.readComponents(ics):
-        for ve in v.vevent_list:
-            yield CreateEventFromVOBJ(
+        events.extend(
+            CreateEventFromVOBJ(
                 ve, verbose=verbose, default_tz=default_tz, printer=printer
             )
+            for ve in v.vevent_list
+        )
+    return events
 
 
 def CreateEventFromVOBJ(
@@ -139,5 +143,10 @@ def CreateEventFromVOBJ(
         if verbose:
             print(f'UID..........{uid}')
         event['iCalUID'] = uid
+    if hasattr(ve, 'sequence'):
+        sequence = ve.sequence.value.strip()
+        if verbose:
+            print(f'Sequence.....{sequence}')
+        event['sequence'] = sequence
 
     return event

tests/test_gcalcli.py

@@ -277,6 +277,14 @@ def test_import(PatchedGCalI):
     assert gcal.ImportICS(icsFile=open(vcal_path, errors='replace'))
 
 
+def test_legacy_import(PatchedGCalI):
+    cal_names = parse_cal_names(['[email protected]'])
+    gcal = PatchedGCalI(
+        cal_names=cal_names, default_reminders=True, use_legacy_import=True)
+    vcal_path = TEST_DATA_DIR + '/vv.txt'
+    assert gcal.ImportICS(icsFile=open(vcal_path, errors='replace'))
+
+
 def test_parse_reminder():
     MINS_PER_DAY = 60 * 24
     MINS_PER_WEEK = MINS_PER_DAY * 7