Merge pull request #712 from insanum/respect_uid

Avoid creating duplicate events by respecting the UID field

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

@@ -1396,6 +1396,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():
@@ -1435,6 +1450,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
@@ -1443,32 +1470,56 @@ def ImportICS(self, verbose=False, dump=False, reminders=None,
                 continue
 
             self._add_reminders(event, reminders)
+
             if not verbose:
+                # Don't prompt, just assume user wants to import.
+                pass
+            else:
+                # Prompt for whether to import.
+                self.printer.msg('\n[S]kip [i]mport [q]uit: ', 'magenta')
+                val = input()
+                if not val or val.lower() == 's':
+                    continue
+                elif val.lower() == 'q':
+                    sys.exit(0)
+                elif val.lower() == 'i':
+                    # Import requested, continue
+                    pass
+                else:
+                    self.printer.err_msg('Error: invalid input\n')
+                    sys.exit(1)
+
+            # Import event
+            import_method = (
+                self.get_events().import_ if (
+                    self._event_should_use_new_import_api(event, self.cals[0]))
+                else self.get_events().insert)
+            try:
                 new_event = self._retry_with_backoff(
-                    self.get_events().insert(
-                        calendarId=self.cals[0]['id'], body=event
-                    )
+                    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']}"
                 )
+                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('New event added: %s\n' % hlink, 'green')
-                continue
+                self.printer.msg(f'New event added: {hlink}\n', 'green')
 
-            self.printer.msg('\n[S]kip [i]mport [q]uit: ', 'magenta')
-            val = input()
-            if not val or val.lower() == 's':
-                continue
-            if val.lower() == 'i':
-                new_event = self._retry_with_backoff(
-                    self.get_events().insert(
-                        calendarId=self.cals[0]['id'], body=event
-                    )
-                )
-                hlink = new_event.get('htmlLink')
-                self.printer.msg('New event added: %s\n' % hlink, 'green')
-            elif val.lower() == 'q':
-                sys.exit(0)
-            else:
-                self.printer.err_msg('Error: invalid input\n')
-                sys.exit(1)
         # 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(
@@ -134,4 +138,15 @@ def CreateEventFromVOBJ(
                 {'displayName': attendee.name, 'email': email}
             )
 
+    if hasattr(ve, 'uid'):
+        uid = ve.uid.value.strip()
+        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