diff --git a/.git_archival.txt b/.git_archival.txt
new file mode 100644
index 0000000..082d6c2
--- /dev/null
+++ b/.git_archival.txt
@@ -0,0 +1 @@
+ref-names: $Format:%D$
\ No newline at end of file
diff --git a/.gitattributes b/.gitattributes
new file mode 100644
index 0000000..82bf71c
--- /dev/null
+++ b/.gitattributes
@@ -0,0 +1 @@
+.git_archival.txt export-subst
\ No newline at end of file
diff --git a/.github/dependabot.yml b/.github/dependabot.yml
new file mode 100644
index 0000000..700d735
--- /dev/null
+++ b/.github/dependabot.yml
@@ -0,0 +1,9 @@
+version: 2
+updates:
+- package-ecosystem: pip
+ directory: "/"
+ schedule:
+ interval: monthly
+ time: '02:00'
+ open-pull-requests-limit: 10
+ target-branch: develop
diff --git a/.github/workflows/codeql-analysis.yml b/.github/workflows/codeql-analysis.yml
new file mode 100644
index 0000000..c72037a
--- /dev/null
+++ b/.github/workflows/codeql-analysis.yml
@@ -0,0 +1,38 @@
+name: "CodeQL"
+
+on:
+ push:
+ branches: [develop, ]
+ pull_request:
+ # The branches below must be a subset of the branches above
+ branches: [develop]
+ schedule:
+ - cron: '0 12 10 * *'
+
+jobs:
+ analyse:
+ name: Analyse
+ runs-on: ubuntu-latest
+
+ steps:
+ - name: Checkout repository
+ uses: actions/checkout@v2
+ with:
+ # We must fetch at least the immediate parents so that if this is
+ # a pull request then we can checkout the head.
+ fetch-depth: 2
+
+ # If this run was triggered by a pull request event, then checkout
+ # the head of the pull request instead of the merge commit.
+ - run: git checkout HEAD^2
+ if: ${{ github.event_name == 'pull_request' }}
+
+ # Initializes the CodeQL tools for scanning.
+ - name: Initialize CodeQL
+ uses: github/codeql-action/init@v1
+ # Override language selection by uncommenting this and choosing your languages
+ with:
+ languages: python
+
+ - name: Perform CodeQL Analysis
+ uses: github/codeql-action/analyze@v1
diff --git a/.github/workflows/pythonpackage.yml b/.github/workflows/pythonpackage.yml
index 55c17b1..30d2477 100644
--- a/.github/workflows/pythonpackage.yml
+++ b/.github/workflows/pythonpackage.yml
@@ -17,7 +17,7 @@ jobs:
strategy:
max-parallel: 4
matrix:
- python-version: [3.5, 3.6, 3.7, 3.8]
+ python-version: [3.6, 3.7, 3.8, 3.9]
steps:
- uses: actions/checkout@v1
diff --git a/.github/workflows/pythonpublish.yml b/.github/workflows/pythonpublish.yml
new file mode 100644
index 0000000..9e383e8
--- /dev/null
+++ b/.github/workflows/pythonpublish.yml
@@ -0,0 +1,26 @@
+name: Upload Python Package
+
+on:
+ release:
+ types: [created]
+
+jobs:
+ deploy:
+ runs-on: ubuntu-latest
+ steps:
+ - uses: actions/checkout@v2
+ - name: Set up Python
+ uses: actions/setup-python@v1
+ with:
+ python-version: '3.x'
+ - name: Install dependencies
+ run: |
+ python -m pip install --upgrade pip
+ pip install setuptools setuptools_scm setuptools_scm_git_archive wheel twine
+ - name: Build and publish
+ env:
+ TWINE_USERNAME: __token__
+ TWINE_PASSWORD: ${{ secrets.pypi_token }}
+ run: |
+ python setup.py sdist bdist_wheel
+ twine upload dist/*
diff --git a/.gitignore b/.gitignore
index 3fd4fef..b5c5b91 100644
--- a/.gitignore
+++ b/.gitignore
@@ -2,4 +2,10 @@ config.yml
service-account.json
*.pyc
my-test*.ics
-.vscode/*
+.vscode/
+.idea/
+/dist/
+/*.egg-info/
+/build/
+/.eggs/
+venv/
diff --git a/.travis.yml b/.travis.yml
index f913379..09d1177 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,10 +1,10 @@
language: python
python:
- - "3.5"
- "3.6"
- "3.7"
- - "3.8"
+ - "3.8"
+ - "3.9"
script:
- pytest -v
diff --git a/MANIFEST.in b/MANIFEST.in
new file mode 100644
index 0000000..346766b
--- /dev/null
+++ b/MANIFEST.in
@@ -0,0 +1,7 @@
+include pyproject.toml
+
+# Include the README
+include *.md
+
+# Include the license file
+include LICENSE
diff --git a/README.md b/README.md
index 79d7c92..90dd467 100644
--- a/README.md
+++ b/README.md
@@ -1,15 +1,96 @@
# sync_ics2gcal
+[](https://badge.fury.io/py/sync-ics2gcal)
[](https://travis-ci.org/b4tman/sync_ics2gcal)
-[](https://dependabot.com)
[](https://app.fossa.io/projects/git%2Bgithub.com%2Fb4tman%2Fsync_ics2gcal?ref=badge_shield)
Python scripts for sync .ics file with Google calendar
+## Installation
+
+To install from [PyPI](https://pypi.org/project/sync-ics2gcal/) with [pip](https://pypi.python.org/pypi/pip), run:
+
+```sh
+pip install sync-ics2gcal
+```
+
+Or download source code and install:
+
+```sh
+python setup.py install
+```
+
+## Configuration
+
+### Create application in Google API Console
+
+1. Create a new project: [console.developers.google.com/project](https://console.developers.google.com/project)
+2. Choose the new project from the top right project dropdown (only if another project is selected)
+3. In the project Dashboard, choose "Library"
+4. Find and Enable "Google Calendar API"
+5. In the project Dashboard, choose "Credentials"
+6. In the "Service Accounts" group, click to "Manage service accounts"
+7. Click "Create service account"
+8. Choose service account name and ID
+9. Go back to "Service Accounts" group in "Credentials"
+10. Edit service account and click "Create key", choose JSON and download key file.
+
+### Create working directory
+
+For example: `/home/user/myfolder`.
+
+1. Save service account key in file `service-account.json`.
+2. Download [sample config](https://github.com/b4tman/sync_ics2gcal/blob/develop/sample-config.yml) and save to file `config.yml`. For example:
+
+```sh
+wget https://raw.githubusercontent.com/b4tman/sync_ics2gcal/develop/sample-config.yml -O config.yml
+```
+
+3. *(Optional)* Place source `.ics` file, `my-calendar.ics` for example.
+
+### Configuration parameters
+
+* `start_from` - start date:
+ * full format datetime, `2018-04-03T13:23:25.000001Z` for example
+ * or just `now`
+* *(Optional)* `service_account` - service account filename, remove it from config to use [default credentials](https://developers.google.com/identity/protocols/application-default-credentials)
+* *(Optional)* `logging` - [config](https://docs.python.org/3.8/library/logging.config.html#dictionary-schema-details) to setup logging
+* `google_id` - target google calendar id, `my-calendar@group.calendar.google.com` for example
+* `source` - source `.ics` filename, `my-calendar.ics` for example
+
+## Usage
+
+### Manage calendars
+
+```sh
+manage-ics2gcal GROUP | COMMAND
+```
+
+**GROUPS**:
+
+* **property** - get/set properties (see [CalendarList resource](https://developers.google.com/calendar/v3/reference/calendarList#resource)), subcommands:
+ - **get** - get calendar property
+ - **set** - set calendar property
+
+**COMMANDS**:
+
+* **list** - list calendars
+* **create** - create calendar
+* **add_owner** - add owner to calendar
+* **remove** - remove calendar
+* **rename** - rename calendar
+
+
+Use **-h** for more info.
+
+### Sync calendar
+
+just type:
+
+```sh
+sync-ics2gcal
+```
+
## How it works
-
-
-## License
-[](https://app.fossa.io/projects/git%2Bgithub.com%2Fb4tman%2Fsync_ics2gcal?ref=badge_large)
\ No newline at end of file
diff --git a/manage-calendars.py b/manage-calendars.py
deleted file mode 100644
index e528c76..0000000
--- a/manage-calendars.py
+++ /dev/null
@@ -1,104 +0,0 @@
-import argparse
-import datetime
-import logging.config
-
-import yaml
-from pytz import utc
-
-from gcal_sync import GoogleCalendar, GoogleCalendarService
-
-
-def parse_args():
- parser = argparse.ArgumentParser(
- description="manage google calendars in service account")
- command_subparsers = parser.add_subparsers(help='command', dest='command')
- command_subparsers.add_parser('list', help='list calendars')
- parser_create = command_subparsers.add_parser(
- 'create', help='create calendar')
- parser_create.add_argument(
- 'summary', action='store', help='new calendar summary')
- parser_create.add_argument('--timezone', action='store',
- default=None, required=False, help='new calendar timezone')
- parser_create.add_argument(
- '--public', default=False, action='store_true', help='make calendar public')
- parser_add_owner = command_subparsers.add_parser(
- 'add_owner', help='add owner to calendar')
- parser_add_owner.add_argument('id', action='store', help='calendar id')
- parser_add_owner.add_argument(
- 'owner_email', action='store', help='new owner email')
- parser_remove = command_subparsers.add_parser(
- 'remove', help='remove calendar')
- parser_remove.add_argument(
- 'id', action='store', help='calendar id to remove')
- parser_rename = command_subparsers.add_parser(
- 'rename', help='rename calendar')
- parser_rename.add_argument(
- 'id', action='store', help='calendar id')
- parser_rename.add_argument(
- 'summary', action='store', help='new summary')
-
- args = parser.parse_args()
- if args.command is None:
- parser.print_usage()
- return args
-
-
-def load_config():
- with open('config.yml', 'r', encoding='utf-8') as f:
- result = yaml.safe_load(f)
- return result
-
-
-def list_calendars(service):
- response = service.calendarList().list(fields='items(id,summary)').execute()
- for calendar in response.get('items'):
- print('{summary}: {id}'.format_map(calendar))
-
-
-def create_calendar(service, summary, timezone, public):
- calendar = GoogleCalendar(service, None)
- calendar.create(summary, timezone)
- if public:
- calendar.make_public()
- print('{}: {}'.format(summary, calendar.calendarId))
-
-
-def add_owner(service, id, owner_email):
- calendar = GoogleCalendar(service, id)
- calendar.add_owner(owner_email)
- print('to {} added owner: {}'.format(id, owner_email))
-
-
-def remove_calendar(service, id):
- calendar = GoogleCalendar(service, id)
- calendar.delete()
- print('removed: {}'.format(id))
-
-def rename_calendar(service, id, summary):
- calendar = {'summary': summary}
- service.calendars().patch(body=calendar, calendarId=id).execute()
- print('{}: {}'.format(summary, id))
-
-def main():
- args = parse_args()
- config = load_config()
-
- if 'logging' in config:
- logging.config.dictConfig(config['logging'])
-
- srv_acc_file = config['service_account']
- service = GoogleCalendarService.from_srv_acc_file(srv_acc_file)
-
- if 'list' == args.command:
- list_calendars(service)
- elif 'create' == args.command:
- create_calendar(service, args.summary, args.timezone, args.public)
- elif 'add_owner' == args.command:
- add_owner(service, args.id, args.owner_email)
- elif 'remove' == args.command:
- remove_calendar(service, args.id)
- elif 'rename' == args.command:
- rename_calendar(service, args.id, args.summary)
-
-if __name__ == '__main__':
- main()
diff --git a/pyproject.toml b/pyproject.toml
new file mode 100644
index 0000000..47626ae
--- /dev/null
+++ b/pyproject.toml
@@ -0,0 +1,3 @@
+[build-system]
+requires = ["setuptools>=40.8.0", "wheel", "setuptools_scm"]
+build-backend = "setuptools.build_meta"
diff --git a/requirements.txt b/requirements.txt
index 5369fe9..08659f2 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,5 +1,6 @@
-google-auth==1.11.0
-google-api-python-client==1.7.11
-icalendar==4.0.4
-pytz==2019.3
-PyYAML==5.3
+google-auth==1.30.0
+google-api-python-client==2.3.0
+icalendar==4.0.7
+pytz==2021.1
+PyYAML==5.4.1
+fire==0.4.0
\ No newline at end of file
diff --git a/setup.cfg b/setup.cfg
new file mode 100644
index 0000000..5ed97a9
--- /dev/null
+++ b/setup.cfg
@@ -0,0 +1,7 @@
+[metadata]
+license_files = LICENSE
+
+[options]
+setup_requires =
+ setuptools_scm
+ setuptools_scm_git_archive
\ No newline at end of file
diff --git a/setup.py b/setup.py
new file mode 100644
index 0000000..bd6dc4c
--- /dev/null
+++ b/setup.py
@@ -0,0 +1,44 @@
+import setuptools
+
+with open('README.md', 'r') as fh:
+ long_description = fh.read()
+
+setuptools.setup(
+ name='sync-ics2gcal',
+ author='Dmitry Belyaev',
+ author_email='b4tm4n@mail.ru',
+ license='MIT',
+ description='Sync ics file with Google calendar',
+ long_description=long_description,
+ long_description_content_type='text/markdown',
+ url='https://github.com/b4tman/sync_ics2gcal',
+ use_scm_version={
+ 'fallback_version': '0.1',
+ 'local_scheme': 'no-local-version'
+ },
+ setup_requires=['setuptools_scm', 'setuptools_scm_git_archive'],
+ packages=setuptools.find_packages(exclude=['tests']),
+ classifiers=[
+ 'License :: OSI Approved :: MIT License',
+ 'Operating System :: OS Independent',
+
+ 'Programming Language :: Python :: 3.6',
+ 'Programming Language :: Python :: 3.7',
+ 'Programming Language :: Python :: 3.8',
+ 'Programming Language :: Python :: 3.9',
+ ],
+ python_requires='>=3.6',
+ install_requires = [
+ 'google-auth>=1.5.0',
+ 'google-api-python-client>=1.7.0',
+ 'icalendar>=4.0.1',
+ 'pytz',
+ 'PyYAML>=3.13'
+ ],
+ entry_points={
+ "console_scripts": [
+ "sync-ics2gcal = sync_ics2gcal.sync_calendar:main",
+ "manage-ics2gcal = sync_ics2gcal.manage_calendars:main",
+ ]
+ }
+)
\ No newline at end of file
diff --git a/gcal_sync/__init__.py b/sync_ics2gcal/__init__.py
similarity index 92%
rename from gcal_sync/__init__.py
rename to sync_ics2gcal/__init__.py
index 5c5fcb1..24b73f5 100644
--- a/gcal_sync/__init__.py
+++ b/sync_ics2gcal/__init__.py
@@ -1,14 +1,14 @@
-
-from .ical import (
- CalendarConverter,
- EventConverter
-)
-
-from .gcal import (
- GoogleCalendarService,
- GoogleCalendar
-)
-
-from .sync import (
- CalendarSync
-)
+
+from .ical import (
+ CalendarConverter,
+ EventConverter
+)
+
+from .gcal import (
+ GoogleCalendarService,
+ GoogleCalendar
+)
+
+from .sync import (
+ CalendarSync
+)
diff --git a/gcal_sync/gcal.py b/sync_ics2gcal/gcal.py
similarity index 62%
rename from gcal_sync/gcal.py
rename to sync_ics2gcal/gcal.py
index 4d02e14..6c96cbd 100644
--- a/gcal_sync/gcal.py
+++ b/sync_ics2gcal/gcal.py
@@ -1,283 +1,341 @@
-import logging
-import sys
-
-import google.auth
-from google.oauth2 import service_account
-from googleapiclient import discovery
-from pytz import utc
-
-
-class GoogleCalendarService():
- """class for make google calendar service Resource
-
- Returns:
- service Resource
- """
-
- @staticmethod
- def from_srv_acc_file(service_account_file):
- """make service Resource from service account filename (authorize)
-
- Returns:
- service Resource
- """
-
- scopes = ['https://www.googleapis.com/auth/calendar']
- credentials = service_account.Credentials.from_service_account_file(service_account_file)
- scoped_credentials = credentials.with_scopes(scopes)
- service = discovery.build('calendar', 'v3', credentials=scoped_credentials)
- return service
-
-def select_event_key(event):
- """select event key for logging
-
- Arguments:
- event -- event resource
-
- Returns:
- key name or None if no key found
- """
-
- key = None
- if 'iCalUID' in event:
- key = 'iCalUID'
- elif 'id' in event:
- key = 'id'
- return key
-
-
-class GoogleCalendar():
- """class to interact with calendar on google
- """
-
- logger = logging.getLogger('GoogleCalendar')
-
- def __init__(self, service, calendarId):
- self.service = service
- self.calendarId = calendarId
-
- def _make_request_callback(self, action, events_by_req):
- """make callback for log result of batch request
-
- Arguments:
- action -- action name
- events_by_req -- list of events ordered by request id
-
- Returns:
- callback function
- """
-
- def callback(request_id, response, exception):
- event = events_by_req[int(request_id)]
- key = select_event_key(event)
-
- if exception is not None:
- self.logger.error('failed to %s event with %s: %s, exception: %s',
- action, key, event.get(key), str(exception))
- else:
- resp_key = select_event_key(response)
- if resp_key is not None:
- event = response
- key = resp_key
- self.logger.info('event %s ok, %s: %s',
- action, key, event.get(key))
- return callback
-
- def list_events_from(self, start):
- """ list events from calendar, where start date >= start
- """
- fields = 'nextPageToken,items(id,iCalUID,updated)'
- events = []
- page_token = None
- timeMin = utc.normalize(start.astimezone(utc)).replace(
- tzinfo=None).isoformat() + 'Z'
- while True:
- response = self.service.events().list(calendarId=self.calendarId, pageToken=page_token,
- singleEvents=True, timeMin=timeMin, fields=fields).execute()
- if 'items' in response:
- events.extend(response['items'])
- page_token = response.get('nextPageToken')
- if not page_token:
- break
- self.logger.info('%d events listed', len(events))
- return events
-
- def find_exists(self, events):
- """ find existing events from list, by 'iCalUID' field
-
- Arguments:
- events {list} -- list of events
-
- Returns:
- tuple -- (events_exist, events_not_found)
- events_exist - list of tuples: (new_event, exists_event)
- """
-
- fields = 'items(id,iCalUID,updated)'
- events_by_req = []
- exists = []
- not_found = []
-
- def list_callback(request_id, response, exception):
- found = False
- event = events_by_req[int(request_id)]
- if exception is None:
- found = ([] != response['items'])
- else:
- self.logger.error('exception %s, while listing event with UID: %s', str(
- exception), event['iCalUID'])
- if found:
- exists.append(
- (event, response['items'][0]))
- else:
- not_found.append(events_by_req[int(request_id)])
-
- batch = self.service.new_batch_http_request(callback=list_callback)
- i = 0
- for event in events:
- events_by_req.append(event)
- batch.add(self.service.events().list(calendarId=self.calendarId,
- iCalUID=event['iCalUID'], showDeleted=True, fields=fields), request_id=str(i))
- i += 1
- batch.execute()
- self.logger.info('%d events exists, %d not found',
- len(exists), len(not_found))
- return exists, not_found
-
- def insert_events(self, events):
- """ insert list of events
-
- Arguments:
- events - events list
- """
-
- fields = 'id'
- events_by_req = []
-
- insert_callback = self._make_request_callback('insert', events_by_req)
- batch = self.service.new_batch_http_request(callback=insert_callback)
- i = 0
- for event in events:
- events_by_req.append(event)
- batch.add(self.service.events().insert(
- calendarId=self.calendarId, body=event, fields=fields), request_id=str(i))
- i += 1
- batch.execute()
-
- def patch_events(self, event_tuples):
- """ patch (update) events
-
- Arguments:
- event_tuples -- list of tuples: (new_event, exists_event)
- """
-
- fields = 'id'
- events_by_req = []
-
- patch_callback = self._make_request_callback('patch', events_by_req)
- batch = self.service.new_batch_http_request(callback=patch_callback)
- i = 0
- for event_new, event_old in event_tuples:
- if 'id' not in event_old:
- continue
- events_by_req.append(event_new)
- batch.add(self.service.events().patch(
- calendarId=self.calendarId, eventId=event_old['id'], body=event_new), fields=fields, request_id=str(i))
- i += 1
- batch.execute()
-
- def update_events(self, event_tuples):
- """ update events
-
- Arguments:
- event_tuples -- list of tuples: (new_event, exists_event)
- """
-
- fields = 'id'
- events_by_req = []
-
- update_callback = self._make_request_callback('update', events_by_req)
- batch = self.service.new_batch_http_request(callback=update_callback)
- i = 0
- for event_new, event_old in event_tuples:
- if 'id' not in event_old:
- continue
- events_by_req.append(event_new)
- batch.add(self.service.events().update(
- calendarId=self.calendarId, eventId=event_old['id'], body=event_new, fields=fields), request_id=str(i))
- i += 1
- batch.execute()
-
- def delete_events(self, events):
- """ delete events
-
- Arguments:
- events -- list of events
- """
-
- events_by_req = []
-
- delete_callback = self._make_request_callback('delete', events_by_req)
- batch = self.service.new_batch_http_request(callback=delete_callback)
- i = 0
- for event in events:
- events_by_req.append(event)
- batch.add(self.service.events().delete(
- calendarId=self.calendarId, eventId=event['id']), request_id=str(i))
- i += 1
- batch.execute()
-
- def create(self, summary, timeZone=None):
- """create calendar
-
- Arguments:
- summary -- new calendar summary
-
- Keyword Arguments:
- timeZone -- new calendar timezone as string (optional)
-
- Returns:
- calendar Resource
- """
-
- calendar = {'summary': summary}
- if timeZone is not None:
- calendar['timeZone'] = timeZone
-
- created_calendar = self.service.calendars().insert(body=calendar).execute()
- self.calendarId = created_calendar['id']
- return created_calendar
-
- def delete(self):
- """delete calendar
- """
-
- self.service.calendars().delete(calendarId=self.calendarId).execute()
-
- def make_public(self):
- """make calendar puplic
- """
-
- rule_public = {
- 'scope': {
- 'type': 'default',
- },
- 'role': 'reader'
- }
- return self.service.acl().insert(calendarId=self.calendarId, body=rule_public).execute()
-
- def add_owner(self, email):
- """add calendar owner by email
-
- Arguments:
- email -- email to add
- """
-
- rule_owner = {
- 'scope': {
- 'type': 'user',
- 'value': email,
- },
- 'role': 'owner'
- }
- return self.service.acl().insert(calendarId=self.calendarId, body=rule_owner).execute()
+import logging
+
+import google.auth
+from google.oauth2 import service_account
+from googleapiclient import discovery
+from pytz import utc
+from datetime import datetime
+from typing import List, Dict, Any, Callable, Tuple, Optional
+
+
+class GoogleCalendarService:
+ """class for make google calendar service Resource
+
+ Returns:
+ service Resource
+ """
+
+ @staticmethod
+ def default():
+ """make service Resource from default credentials (authorize)
+ ( https://developers.google.com/identity/protocols/application-default-credentials )
+ ( https://googleapis.dev/python/google-auth/latest/reference/google.auth.html#google.auth.default )
+ """
+
+ scopes = ['https://www.googleapis.com/auth/calendar']
+ credentials, _ = google.auth.default(scopes=scopes)
+ service = discovery.build(
+ 'calendar', 'v3', credentials=credentials, cache_discovery=False)
+ return service
+
+ @staticmethod
+ def from_srv_acc_file(service_account_file: str):
+ """make service Resource from service account filename (authorize)
+ """
+
+ scopes = ['https://www.googleapis.com/auth/calendar']
+ credentials = service_account.Credentials.from_service_account_file(
+ service_account_file)
+ scoped_credentials = credentials.with_scopes(scopes)
+ service = discovery.build(
+ 'calendar', 'v3', credentials=scoped_credentials,
+ cache_discovery=False)
+ return service
+
+ @staticmethod
+ def from_config(config: Optional[Dict[str, Optional[str]]] = None):
+ """make service Resource from config dict
+
+ Arguments:
+ config -- config with keys:
+ (optional) service_account: - service account filename
+ if key not in dict then default credentials will be used
+ ( https://developers.google.com/identity/protocols/application-default-credentials )
+ -- None: default credentials will be used
+ """
+
+ if config is not None and 'service_account' in config:
+ service = GoogleCalendarService.from_srv_acc_file(
+ config['service_account'])
+ else:
+ service = GoogleCalendarService.default()
+ return service
+
+
+def select_event_key(event: Dict[str, Any]) -> Optional[str]:
+ """select event key for logging
+
+ Arguments:
+ event -- event resource
+
+ Returns:
+ key name or None if no key found
+ """
+
+ key = None
+ if 'iCalUID' in event:
+ key = 'iCalUID'
+ elif 'id' in event:
+ key = 'id'
+ return key
+
+
+class GoogleCalendar:
+ """class to interact with calendar on google
+ """
+
+ logger = logging.getLogger('GoogleCalendar')
+
+ def __init__(self, service: discovery.Resource, calendarId: Optional[str]):
+ self.service: discovery.Resource = service
+ self.calendarId: str = calendarId
+
+ def _make_request_callback(self, action: str, events_by_req: List[Dict[str, Any]]) -> Callable:
+ """make callback for log result of batch request
+
+ Arguments:
+ action -- action name
+ events_by_req -- list of events ordered by request id
+
+ Returns:
+ callback function
+ """
+
+ def callback(request_id, response, exception):
+ event = events_by_req[int(request_id)]
+ key = select_event_key(event)
+
+ if exception is not None:
+ self.logger.error(
+ 'failed to %s event with %s: %s, exception: %s',
+ action, key, event.get(key), str(exception)
+ )
+ else:
+ resp_key = select_event_key(response)
+ if resp_key is not None:
+ event = response
+ key = resp_key
+ self.logger.info('event %s ok, %s: %s',
+ action, key, event.get(key))
+ return callback
+
+ def list_events_from(self, start: datetime) -> List[Dict[str, Any]]:
+ """ list events from calendar, where start date >= start
+ """
+ fields = 'nextPageToken,items(id,iCalUID,updated)'
+ events = []
+ page_token = None
+ timeMin = utc.normalize(start.astimezone(utc)).replace(
+ tzinfo=None).isoformat() + 'Z'
+ while True:
+ response = self.service.events().list(calendarId=self.calendarId,
+ pageToken=page_token,
+ singleEvents=True,
+ timeMin=timeMin,
+ fields=fields).execute()
+ if 'items' in response:
+ events.extend(response['items'])
+ page_token = response.get('nextPageToken')
+ if not page_token:
+ break
+ self.logger.info('%d events listed', len(events))
+ return events
+
+ def find_exists(self, events: List) -> Tuple[List[Tuple[Dict[str, Any], Dict[str, Any]]], List[Dict[str, Any]]]:
+ """ find existing events from list, by 'iCalUID' field
+
+ Arguments:
+ events {list} -- list of events
+
+ Returns:
+ tuple -- (events_exist, events_not_found)
+ events_exist - list of tuples: (new_event, exists_event)
+ """
+
+ fields = 'items(id,iCalUID,updated)'
+ events_by_req = []
+ exists = []
+ not_found = []
+
+ def list_callback(request_id, response, exception):
+ found = False
+ cur_event = events_by_req[int(request_id)]
+ if exception is None:
+ found = ([] != response['items'])
+ else:
+ self.logger.error(
+ 'exception %s, while listing event with UID: %s',
+ str(exception), cur_event['iCalUID'])
+ if found:
+ exists.append(
+ (cur_event, response['items'][0]))
+ else:
+ not_found.append(events_by_req[int(request_id)])
+
+ batch = self.service.new_batch_http_request(callback=list_callback)
+ i = 0
+ for event in events:
+ events_by_req.append(event)
+ batch.add(self.service.events().list(calendarId=self.calendarId,
+ iCalUID=event['iCalUID'],
+ showDeleted=True,
+ fields=fields
+ ),
+ request_id=str(i)
+ )
+ i += 1
+ batch.execute()
+ self.logger.info('%d events exists, %d not found',
+ len(exists), len(not_found))
+ return exists, not_found
+
+ def insert_events(self, events: List[Dict[str, Any]]):
+ """ insert list of events
+
+ Arguments:
+ events - events list
+ """
+
+ fields = 'id'
+ events_by_req = []
+
+ insert_callback = self._make_request_callback('insert', events_by_req)
+ batch = self.service.new_batch_http_request(callback=insert_callback)
+ i = 0
+ for event in events:
+ events_by_req.append(event)
+ batch.add(self.service.events().insert(
+ calendarId=self.calendarId, body=event, fields=fields),
+ request_id=str(i)
+ )
+ i += 1
+ batch.execute()
+
+ def patch_events(self, event_tuples: List[Tuple[Dict[str, Any], Dict[str, Any]]]):
+ """ patch (update) events
+
+ Arguments:
+ event_tuples -- list of tuples: (new_event, exists_event)
+ """
+
+ fields = 'id'
+ events_by_req = []
+
+ patch_callback = self._make_request_callback('patch', events_by_req)
+ batch = self.service.new_batch_http_request(callback=patch_callback)
+ i = 0
+ for event_new, event_old in event_tuples:
+ if 'id' not in event_old:
+ continue
+ events_by_req.append(event_new)
+ batch.add(self.service.events().patch(
+ calendarId=self.calendarId, eventId=event_old['id'],
+ body=event_new), fields=fields, request_id=str(i))
+ i += 1
+ batch.execute()
+
+ def update_events(self, event_tuples: List[Tuple[Dict[str, Any], Dict[str, Any]]]):
+ """ update events
+
+ Arguments:
+ event_tuples -- list of tuples: (new_event, exists_event)
+ """
+
+ fields = 'id'
+ events_by_req = []
+
+ update_callback = self._make_request_callback('update', events_by_req)
+ batch = self.service.new_batch_http_request(callback=update_callback)
+ i = 0
+ for event_new, event_old in event_tuples:
+ if 'id' not in event_old:
+ continue
+ events_by_req.append(event_new)
+ batch.add(self.service.events().update(
+ calendarId=self.calendarId, eventId=event_old['id'],
+ body=event_new, fields=fields), request_id=str(i))
+ i += 1
+ batch.execute()
+
+ def delete_events(self, events: List[Dict[str, Any]]):
+ """ delete events
+
+ Arguments:
+ events -- list of events
+ """
+
+ events_by_req = []
+
+ delete_callback = self._make_request_callback('delete', events_by_req)
+ batch = self.service.new_batch_http_request(callback=delete_callback)
+ i = 0
+ for event in events:
+ events_by_req.append(event)
+ batch.add(self.service.events().delete(
+ calendarId=self.calendarId,
+ eventId=event['id']), request_id=str(i))
+ i += 1
+ batch.execute()
+
+ def create(self, summary: str, timeZone: Optional[str] = None) -> Any:
+ """create calendar
+
+ Arguments:
+ summary -- new calendar summary
+
+ Keyword Arguments:
+ timeZone -- new calendar timezone as string (optional)
+
+ Returns:
+ calendar Resource
+ """
+
+ calendar = {'summary': summary}
+ if timeZone is not None:
+ calendar['timeZone'] = timeZone
+
+ created_calendar = self.service.calendars().insert(
+ body=calendar
+ ).execute()
+ self.calendarId = created_calendar['id']
+ return created_calendar
+
+ def delete(self):
+ """delete calendar
+ """
+
+ self.service.calendars().delete(calendarId=self.calendarId).execute()
+
+ def make_public(self):
+ """make calendar puplic
+ """
+
+ rule_public = {
+ 'scope': {
+ 'type': 'default',
+ },
+ 'role': 'reader'
+ }
+ return self.service.acl().insert(
+ calendarId=self.calendarId,
+ body=rule_public
+ ).execute()
+
+ def add_owner(self, email: str):
+ """add calendar owner by email
+
+ Arguments:
+ email -- email to add
+ """
+
+ rule_owner = {
+ 'scope': {
+ 'type': 'user',
+ 'value': email,
+ },
+ 'role': 'owner'
+ }
+ return self.service.acl().insert(
+ calendarId=self.calendarId,
+ body=rule_owner
+ ).execute()
diff --git a/gcal_sync/ical.py b/sync_ics2gcal/ical.py
similarity index 73%
rename from gcal_sync/ical.py
rename to sync_ics2gcal/ical.py
index ec21b8d..678c471 100644
--- a/gcal_sync/ical.py
+++ b/sync_ics2gcal/ical.py
@@ -1,184 +1,195 @@
-import datetime
-import logging
-
-from icalendar import Calendar, Event
-from pytz import utc
-
-
-def format_datetime_utc(value):
- """utc datetime as string from date or datetime value
- Arguments:
- value -- date or datetime value
-
- Returns:
- utc datetime value as string in iso format
- """
- if not isinstance(value, datetime.datetime):
- value = datetime.datetime(
- value.year, value.month, value.day, tzinfo=utc)
- value = value.replace(microsecond=1)
- return utc.normalize(value.astimezone(utc)).replace(tzinfo=None).isoformat() + 'Z'
-
-
-def gcal_date_or_dateTime(value, check_value=None):
- """date or dateTime to gcal (start or end dict)
- Arguments:
- value -- date or datetime value
- check_value - date or datetime to choise result type (if not None)
-
- Returns:
- dict { 'date': ... } or { 'dateTime': ... }
- """
-
- if check_value is None:
- check_value = value
-
- result = {}
- if isinstance(check_value, datetime.datetime):
- result['dateTime'] = format_datetime_utc(value)
- else:
- if isinstance(check_value, datetime.date):
- if isinstance(value, datetime.datetime):
- value = datetime.date(value.year, value.month, value.day)
- result['date'] = value.isoformat()
- return result
-
-
-class EventConverter(Event):
- """Convert icalendar event to google calendar resource
- ( https://developers.google.com/calendar/v3/reference/events#resource-representations )
- """
-
- def _str_prop(self, prop):
- """decoded string property
-
- Arguments:
- prop - propperty name
-
- Returns:
- string value
- """
-
- return self.decoded(prop).decode(encoding='utf-8')
-
- def _datetime_str_prop(self, prop):
- """utc datetime as string from property
-
- Arguments:
- prop -- property name
-
- Returns:
- utc datetime value as string in iso format
- """
-
- return format_datetime_utc(self.decoded(prop))
-
- def _gcal_start(self):
- """ event start dict from icalendar event
-
- Raises:
- ValueError -- if DTSTART not date or datetime
-
- Returns:
- dict
- """
-
- value = self.decoded('DTSTART')
- return gcal_date_or_dateTime(value)
-
- def _gcal_end(self):
- """event end dict from icalendar event
-
- Raises:
- ValueError -- if no DTEND or DURATION
- Returns:
- dict
- """
-
- result = None
- if 'DTEND' in self:
- value = self.decoded('DTEND')
- result = gcal_date_or_dateTime(value)
- elif 'DURATION' in self:
- start_val = self.decoded('DTSTART')
- duration = self.decoded('DURATION')
- end_val = start_val + duration
-
- result = gcal_date_or_dateTime(end_val, check_value=start_val)
- else:
- raise ValueError('no DTEND or DURATION')
- return result
-
- def _put_to_gcal(self, gcal_event, prop, func, ics_prop=None):
- """get property from ical event if exist, and put to gcal event
-
- Arguments:
- gcal_event -- dest event
- prop -- property name
- func -- function to convert
- ics_prop -- ical property name (default: {None})
- """
-
- if not ics_prop:
- ics_prop = prop
- if ics_prop in self:
- gcal_event[prop] = func(ics_prop)
-
- def to_gcal(self):
- """Convert
-
- Returns:
- dict - google calendar#event resource
- """
-
- event = {
- 'iCalUID': self._str_prop('UID')
- }
-
- event['start'] = self._gcal_start()
- event['end'] = self._gcal_end()
-
- self._put_to_gcal(event, 'summary', self._str_prop)
- self._put_to_gcal(event, 'description', self._str_prop)
- self._put_to_gcal(event, 'location', self._str_prop)
- self._put_to_gcal(event, 'created', self._datetime_str_prop)
- self._put_to_gcal(
- event, 'updated', self._datetime_str_prop, 'LAST-MODIFIED')
- self._put_to_gcal(
- event, 'transparency', lambda prop: self._str_prop(prop).lower(), 'TRANSP')
-
- return event
-
-
-class CalendarConverter():
- """Convert icalendar events to google calendar resources
- """
-
- logger = logging.getLogger('CalendarConverter')
-
- def __init__(self, calendar=None):
- self.calendar = calendar
-
- def load(self, filename):
- """ load calendar from ics file
- """
- with open(filename, 'r', encoding='utf-8') as f:
- self.calendar = Calendar.from_ical(f.read())
- self.logger.info('%s loaded', filename)
-
- def loads(self, string):
- """ load calendar from ics string
- """
- self.calendar = Calendar.from_ical(string)
-
- def events_to_gcal(self):
- """Convert events to google calendar resources
- """
-
- ics_events = self.calendar.walk(name='VEVENT')
- self.logger.info('%d events readed', len(ics_events))
-
- result = list(
- map(lambda event: EventConverter(event).to_gcal(), ics_events))
- self.logger.info('%d events converted', len(result))
- return result
+import datetime
+import logging
+from typing import Union, Dict, Any, Callable, Optional, List
+
+from icalendar import Calendar, Event
+from pytz import utc
+
+
+def format_datetime_utc(value: Union[datetime.date, datetime.datetime]) -> str:
+ """utc datetime as string from date or datetime value
+
+ Arguments:
+ value -- date or datetime value
+
+ Returns:
+ utc datetime value as string in iso format
+ """
+ if not isinstance(value, datetime.datetime):
+ value = datetime.datetime(
+ value.year, value.month, value.day, tzinfo=utc)
+ value = value.replace(microsecond=1)
+
+ return utc.normalize(
+ value.astimezone(utc)
+ ).replace(tzinfo=None).isoformat() + 'Z'
+
+
+def gcal_date_or_dateTime(value: Union[datetime.date, datetime.datetime],
+ check_value: Union[datetime.date, datetime.datetime, None] = None)\
+ -> Dict[str, str]:
+ """date or dateTime to gcal (start or end dict)
+
+ Arguments:
+ value: date or datetime
+ check_value: optional for choose result type
+
+ Returns:
+ { 'date': ... } or { 'dateTime': ... }
+ """
+
+ if check_value is None:
+ check_value = value
+
+ result: Dict[str, str] = {}
+ if isinstance(check_value, datetime.datetime):
+ result['dateTime'] = format_datetime_utc(value)
+ else:
+ if isinstance(check_value, datetime.date):
+ if isinstance(value, datetime.datetime):
+ value = datetime.date(value.year, value.month, value.day)
+ result['date'] = value.isoformat()
+ return result
+
+
+class EventConverter(Event):
+ """Convert icalendar event to google calendar resource
+ ( https://developers.google.com/calendar/v3/reference/events#resource-representations )
+ """
+
+ def _str_prop(self, prop: str) -> str:
+ """decoded string property
+
+ Arguments:
+ prop - propperty name
+
+ Returns:
+ string value
+ """
+
+ return self.decoded(prop).decode(encoding='utf-8')
+
+ def _datetime_str_prop(self, prop: str) -> str:
+ """utc datetime as string from property
+
+ Arguments:
+ prop -- property name
+
+ Returns:
+ utc datetime value as string in iso format
+ """
+
+ return format_datetime_utc(self.decoded(prop))
+
+ def _gcal_start(self) -> Dict[str, str]:
+ """ event start dict from icalendar event
+
+ Raises:
+ ValueError -- if DTSTART not date or datetime
+
+ Returns:
+ dict
+ """
+
+ value = self.decoded('DTSTART')
+ return gcal_date_or_dateTime(value)
+
+ def _gcal_end(self) -> Dict[str, str]:
+ """event end dict from icalendar event
+
+ Raises:
+ ValueError -- if no DTEND or DURATION
+ Returns:
+ dict
+ """
+
+ result = None
+ if 'DTEND' in self:
+ value = self.decoded('DTEND')
+ result = gcal_date_or_dateTime(value)
+ elif 'DURATION' in self:
+ start_val = self.decoded('DTSTART')
+ duration = self.decoded('DURATION')
+ end_val = start_val + duration
+
+ result = gcal_date_or_dateTime(end_val, check_value=start_val)
+ else:
+ raise ValueError('no DTEND or DURATION')
+ return result
+
+ def _put_to_gcal(self, gcal_event: Dict[str, Any],
+ prop: str, func: Callable[[str], str],
+ ics_prop: Optional[str] = None):
+ """get property from ical event if exist, and put to gcal event
+
+ Arguments:
+ gcal_event -- dest event
+ prop -- property name
+ func -- function to convert
+ ics_prop -- ical property name (default: {None})
+ """
+
+ if not ics_prop:
+ ics_prop = prop
+ if ics_prop in self:
+ gcal_event[prop] = func(ics_prop)
+
+ def to_gcal(self) -> Dict[str, Any]:
+ """Convert
+
+ Returns:
+ dict - google calendar#event resource
+ """
+
+ event = {
+ 'iCalUID': self._str_prop('UID'),
+ 'start': self._gcal_start(),
+ 'end': self._gcal_end()
+ }
+
+ self._put_to_gcal(event, 'summary', self._str_prop)
+ self._put_to_gcal(event, 'description', self._str_prop)
+ self._put_to_gcal(event, 'location', self._str_prop)
+ self._put_to_gcal(event, 'created', self._datetime_str_prop)
+ self._put_to_gcal(
+ event, 'updated', self._datetime_str_prop, 'LAST-MODIFIED')
+ self._put_to_gcal(
+ event,
+ 'transparency',
+ lambda prop: self._str_prop(prop).lower(), 'TRANSP')
+
+ return event
+
+
+class CalendarConverter:
+ """Convert icalendar events to google calendar resources
+ """
+
+ logger = logging.getLogger('CalendarConverter')
+
+ def __init__(self, calendar: Optional[Calendar] = None):
+ self.calendar: Optional[Calendar] = calendar
+
+ def load(self, filename: str):
+ """ load calendar from ics file
+ """
+ with open(filename, 'r', encoding='utf-8') as f:
+ self.calendar = Calendar.from_ical(f.read())
+ self.logger.info('%s loaded', filename)
+
+ def loads(self, string: str):
+ """ load calendar from ics string
+ """
+ self.calendar = Calendar.from_ical(string)
+
+ def events_to_gcal(self) -> List[Dict[str, Any]]:
+ """Convert events to google calendar resources
+ """
+
+ ics_events = self.calendar.walk(name='VEVENT')
+ self.logger.info('%d events readed', len(ics_events))
+
+ result = list(
+ map(lambda event: EventConverter(event).to_gcal(), ics_events))
+ self.logger.info('%d events converted', len(result))
+ return result
diff --git a/sync_ics2gcal/manage_calendars.py b/sync_ics2gcal/manage_calendars.py
new file mode 100644
index 0000000..f3e9cdc
--- /dev/null
+++ b/sync_ics2gcal/manage_calendars.py
@@ -0,0 +1,144 @@
+import logging.config
+from typing import Optional, Dict, Any, List
+
+import fire
+import yaml
+
+from . import GoogleCalendar, GoogleCalendarService
+
+
+def load_config(filename: str) -> Optional[Dict[str, Any]]:
+ result = None
+ try:
+ with open(filename, 'r', encoding='utf-8') as f:
+ result = yaml.safe_load(f)
+ except FileNotFoundError:
+ pass
+
+ return result
+
+
+class PropertyCommands:
+ """ get/set google calendar properties """
+
+ def __init__(self, _service):
+ self._service = _service
+
+ def get(self, calendar_id: str, property_name: str) -> None:
+ """ get calendar property
+
+ Args:
+ calendar_id: calendar id
+ property_name: property key
+ """
+ response = self._service.calendarList().get(calendarId=calendar_id,
+ fields=property_name).execute()
+ print(response.get(property_name))
+
+ def set(self, calendar_id: str, property_name: str, property_value: str) -> None:
+ """ set calendar property
+
+ Args:
+ calendar_id: calendar id
+ property_name: property key
+ property_value: property value
+ """
+ body = {property_name: property_value}
+ response = self._service.calendarList().patch(body=body, calendarId=calendar_id).execute()
+ print(response)
+
+
+class Commands:
+ """ manage google calendars in service account """
+
+ def __init__(self, config: str = 'config.yml'):
+ """
+
+ Args:
+ config(str): config filename
+ """
+ self._config: Optional[Dict[str, Any]] = load_config(config)
+ if self._config is not None and 'logging' in self._config:
+ logging.config.dictConfig(self._config['logging'])
+ self._service = GoogleCalendarService.from_config(self._config)
+ self.property = PropertyCommands(self._service)
+
+ def list(self, show_hidden: bool = False, show_deleted: bool = False) -> None:
+ """ list calendars
+
+ Args:
+ show_hidden: show hidden calendars
+ show_deleted: show deleted calendars
+ """
+
+ fields: str = 'nextPageToken,items(id,summary)'
+ calendars: List[Dict[str, Any]] = []
+ page_token: Optional[str] = None
+ while True:
+ calendars_api = self._service.calendarList()
+ response = calendars_api.list(fields=fields,
+ pageToken=page_token,
+ showHidden=show_hidden,
+ showDeleted=show_deleted
+ ).execute()
+ if 'items' in response:
+ calendars.extend(response['items'])
+ page_token = response.get('nextPageToken')
+ if page_token is None:
+ break
+ for calendar in calendars:
+ print('{summary}: {id}'.format_map(calendar))
+
+ def create(self, summary: str, timezone: Optional[str] = None, public: bool = False) -> None:
+ """ create calendar
+
+ Args:
+ summary: new calendar summary
+ timezone: new calendar timezone
+ public: make calendar public
+ """
+ calendar = GoogleCalendar(self._service, None)
+ calendar.create(summary, timezone)
+ if public:
+ calendar.make_public()
+ print('{}: {}'.format(summary, calendar.calendarId))
+
+ def add_owner(self, calendar_id: str, email: str) -> None:
+ """ add owner to calendar
+
+ Args:
+ calendar_id: calendar id
+ email: new owner email
+ """
+ calendar = GoogleCalendar(self._service, calendar_id)
+ calendar.add_owner(email)
+ print('to {} added owner: {}'.format(calendar_id, email))
+
+ def remove(self, calendar_id: str) -> None:
+ """ remove calendar
+
+ Args:
+ calendar_id: calendar id
+ """
+ calendar = GoogleCalendar(self._service, calendar_id)
+ calendar.delete()
+ print('removed: {}'.format(calendar_id))
+
+ def rename(self, calendar_id: str, summary: str) -> None:
+ """ rename calendar
+
+ Args:
+ calendar_id: calendar id
+ summary:
+ """
+ calendar = {'summary': summary}
+ self._service.calendars().patch(body=calendar, calendarId=calendar_id).execute()
+ print('{}: {}'.format(summary, calendar_id))
+
+
+def main():
+ fire.Fire(Commands, name='manage-ics2gcal')
+
+
+if __name__ == '__main__':
+ main()
diff --git a/gcal_sync/sync.py b/sync_ics2gcal/sync.py
similarity index 60%
rename from gcal_sync/sync.py
rename to sync_ics2gcal/sync.py
index 6ca4fc8..475ad95 100644
--- a/gcal_sync/sync.py
+++ b/sync_ics2gcal/sync.py
@@ -1,173 +1,195 @@
-import datetime
-import dateutil.parser
-import logging
-import operator
-from pytz import utc
-
-
-class CalendarSync():
- """class for syncronize calendar with google
- """
-
- logger = logging.getLogger('CalendarSync')
-
- def __init__(self, gcalendar, converter):
- self.gcalendar = gcalendar
- self.converter = converter
-
- @staticmethod
- def _events_list_compare(items_src, items_dst, key='iCalUID'):
- """ compare list of events by key
-
- Arguments:
- items_src {list of dict} -- source events
- items_dst {list of dict} -- dest events
- key {str} -- name of key to compare (default: {'iCalUID'})
-
- Returns:
- tuple -- (items_to_insert,
- items_to_update,
- items_to_delete)
- """
-
- def get_key(item): return item[key]
-
- keys_src = set(map(get_key, items_src))
- keys_dst = set(map(get_key, items_dst))
-
- keys_to_insert = keys_src - keys_dst
- keys_to_update = keys_src & keys_dst
- keys_to_delete = keys_dst - keys_src
-
- def items_by_keys(items, key_name, keys):
- return list(filter(lambda item: item[key_name] in keys, items))
-
- items_to_insert = items_by_keys(items_src, key, keys_to_insert)
- items_to_delete = items_by_keys(items_dst, key, keys_to_delete)
-
- to_upd_src = items_by_keys(items_src, key, keys_to_update)
- to_upd_dst = items_by_keys(items_dst, key, keys_to_update)
- to_upd_src.sort(key=get_key)
- to_upd_dst.sort(key=get_key)
- items_to_update = list(zip(to_upd_src, to_upd_dst))
-
- return items_to_insert, items_to_update, items_to_delete
-
- def _filter_events_to_update(self):
- """ filter 'to_update' events by 'updated' datetime
- """
-
- def filter_updated(event_tuple):
- new, old = event_tuple
- return dateutil.parser.parse(new['updated']) > dateutil.parser.parse(old['updated'])
-
- self.to_update = list(filter(filter_updated, self.to_update))
-
- @staticmethod
- def _filter_events_by_date(events, date, op):
- """ filter events by start datetime
-
- Arguments:
- events -- events list
- date {datetime} -- datetime to compare
- op {operator} -- comparsion operator
-
- Returns:
- list of filtred events
- """
-
- def filter_by_date(event):
- date_cmp = date
- event_start = event['start']
- event_date = None
- compare_dates = False
-
- if 'date' in event_start:
- event_date = event_start['date']
- compare_dates = True
- elif 'dateTime' in event_start:
- event_date = event_start['dateTime']
-
- event_date = dateutil.parser.parse(event_date)
- if compare_dates:
- date_cmp = datetime.date(date.year, date.month, date.day)
- event_date = datetime.date(event_date.year, event_date.month, event_date.day)
-
- return op(event_date, date_cmp)
-
- return list(filter(filter_by_date, events))
-
- @staticmethod
- def _tz_aware_datetime(date):
- """make tz aware datetime from datetime/date (utc if no tzinfo)
-
- Arguments:
- date - date or datetime / with or without tzinfo
-
- Returns:
- datetime with tzinfo
- """
-
- if not isinstance(date, datetime.datetime):
- date = datetime.datetime(date.year, date.month, date.day)
- if date.tzinfo is None:
- date = date.replace(tzinfo=utc)
- return date
-
- def prepare_sync(self, start_date):
- """prepare sync lists by comparsion of events
-
- Arguments:
- start_date -- date/datetime to start sync
- """
-
- start_date = CalendarSync._tz_aware_datetime(start_date)
-
- events_src = self.converter.events_to_gcal()
- events_dst = self.gcalendar.list_events_from(start_date)
-
- # divide source events by start datetime
- events_src_pending = CalendarSync._filter_events_by_date(
- events_src, start_date, operator.ge)
- events_src_past = CalendarSync._filter_events_by_date(
- events_src, start_date, operator.lt)
-
- events_src = None
-
- # first events comparsion
- self.to_insert, self.to_update, self.to_delete = CalendarSync._events_list_compare(
- events_src_pending, events_dst)
-
- events_src_pending, events_dst = None, None
-
- # find in events 'to_delete' past events from source, for update (move to past)
- _, add_to_update, self.to_delete = CalendarSync._events_list_compare(
- events_src_past, self.to_delete)
- self.to_update.extend(add_to_update)
-
- events_src_past = None
-
- # find if events 'to_insert' exists in gcalendar, for update them
- add_to_update, self.to_insert = self.gcalendar.find_exists(
- self.to_insert)
- self.to_update.extend(add_to_update)
-
- add_to_update = None
-
- # exclude outdated events from 'to_update' list, by 'updated' field
- self._filter_events_to_update()
-
- self.logger.info('prepared to sync: ( insert: %d, update: %d, delete: %d )',
- len(self.to_insert), len(self.to_update), len(self.to_delete))
-
- def apply(self):
- """apply sync (insert, update, delete), using prepared lists of events
- """
-
- self.gcalendar.insert_events(self.to_insert)
- self.gcalendar.update_events(self.to_update)
- self.gcalendar.delete_events(self.to_delete)
-
- self.logger.info('sync done')
-
- self.to_insert, self.to_update, self.to_delete = [], [], []
+import datetime
+import logging
+import operator
+from typing import List, Any, Dict, Set, Tuple, Union, Callable
+
+import dateutil.parser
+from pytz import utc
+
+from .gcal import GoogleCalendar
+from .ical import CalendarConverter
+
+
+class CalendarSync:
+ """class for syncronize calendar with google
+ """
+
+ logger = logging.getLogger('CalendarSync')
+
+ def __init__(self, gcalendar: GoogleCalendar, converter: CalendarConverter):
+ self.gcalendar: GoogleCalendar = gcalendar
+ self.converter: CalendarConverter = converter
+ self.to_insert: List[Dict[str, Any]] = []
+ self.to_update: List[Tuple[Dict[str, Any], Dict[str, Any]]] = []
+ self.to_delete: List[Dict[str, Any]] = []
+
+ @staticmethod
+ def _events_list_compare(items_src: List[Dict[str, Any]],
+ items_dst: List[Dict[str, Any]],
+ key: str = 'iCalUID') \
+ -> Tuple[List[Dict[str, Any]], List[Tuple[Dict[str, Any], Dict[str, Any]]], List[Dict[str, Any]]]:
+ """ compare list of events by key
+
+ Arguments:
+ items_src {list of dict} -- source events
+ items_dst {list of dict} -- dest events
+ key {str} -- name of key to compare (default: {'iCalUID'})
+
+ Returns:
+ tuple -- (items_to_insert,
+ items_to_update,
+ items_to_delete)
+ """
+
+ def get_key(item: Dict[str, Any]) -> str: return item[key]
+
+ keys_src: Set[str] = set(map(get_key, items_src))
+ keys_dst: Set[str] = set(map(get_key, items_dst))
+
+ keys_to_insert = keys_src - keys_dst
+ keys_to_update = keys_src & keys_dst
+ keys_to_delete = keys_dst - keys_src
+
+ def items_by_keys(items: List[Dict[str, Any]],
+ key_name: str,
+ keys: Set[str]) -> List[Dict[str, Any]]:
+ return list(filter(lambda item: item[key_name] in keys, items))
+
+ items_to_insert = items_by_keys(items_src, key, keys_to_insert)
+ items_to_delete = items_by_keys(items_dst, key, keys_to_delete)
+
+ to_upd_src = items_by_keys(items_src, key, keys_to_update)
+ to_upd_dst = items_by_keys(items_dst, key, keys_to_update)
+ to_upd_src.sort(key=get_key)
+ to_upd_dst.sort(key=get_key)
+ items_to_update = list(zip(to_upd_src, to_upd_dst))
+
+ return items_to_insert, items_to_update, items_to_delete
+
+ def _filter_events_to_update(self):
+ """ filter 'to_update' events by 'updated' datetime
+ """
+
+ def filter_updated(event_tuple: Tuple[Dict[str, Any], Dict[str, Any]]) -> bool:
+ new, old = event_tuple
+ new_date = dateutil.parser.parse(new['updated'])
+ old_date = dateutil.parser.parse(old['updated'])
+ return new_date > old_date
+
+ self.to_update = list(filter(filter_updated, self.to_update))
+
+ @staticmethod
+ def _filter_events_by_date(events: List[Dict[str, Any]],
+ date: Union[datetime.date, datetime.datetime],
+ op: Callable[[Union[datetime.date, datetime.datetime],
+ Union[datetime.date, datetime.datetime]], bool]) -> List[Dict[str, Any]]:
+ """ filter events by start datetime
+
+ Arguments:
+ events -- events list
+ date {datetime} -- datetime to compare
+ op {operator} -- comparsion operator
+
+ Returns:
+ list of filtred events
+ """
+
+ def filter_by_date(event: Dict[str, Any]) -> bool:
+ date_cmp = date
+ event_start: Dict[str, str] = event['start']
+ event_date: Union[datetime.date, datetime.datetime, str, None] = None
+ compare_dates = False
+
+ if 'date' in event_start:
+ event_date = event_start['date']
+ compare_dates = True
+ elif 'dateTime' in event_start:
+ event_date = event_start['dateTime']
+
+ event_date = dateutil.parser.parse(event_date)
+ if compare_dates:
+ date_cmp = datetime.date(date.year, date.month, date.day)
+ event_date = datetime.date(
+ event_date.year, event_date.month, event_date.day)
+
+ return op(event_date, date_cmp)
+
+ return list(filter(filter_by_date, events))
+
+ @staticmethod
+ def _tz_aware_datetime(date: Union[datetime.date, datetime.datetime]) -> datetime.datetime:
+ """make tz aware datetime from datetime/date (utc if no tzinfo)
+
+ Arguments:
+ date - date or datetime / with or without tzinfo
+
+ Returns:
+ datetime with tzinfo
+ """
+
+ if not isinstance(date, datetime.datetime):
+ date = datetime.datetime(date.year, date.month, date.day)
+ if date.tzinfo is None:
+ date = date.replace(tzinfo=utc)
+ return date
+
+ def prepare_sync(self, start_date: Union[datetime.date, datetime.datetime]) -> None:
+ """prepare sync lists by comparsion of events
+
+ Arguments:
+ start_date -- date/datetime to start sync
+ """
+
+ start_date = CalendarSync._tz_aware_datetime(start_date)
+
+ events_src = self.converter.events_to_gcal()
+ events_dst = self.gcalendar.list_events_from(start_date)
+
+ # divide source events by start datetime
+ events_src_pending = CalendarSync._filter_events_by_date(
+ events_src, start_date, operator.ge)
+ events_src_past = CalendarSync._filter_events_by_date(
+ events_src, start_date, operator.lt)
+
+ # first events comparsion
+ self.to_insert, self.to_update, self.to_delete = CalendarSync._events_list_compare(
+ events_src_pending, events_dst)
+
+ # find in events 'to_delete' past events from source, for update (move to past)
+ _, add_to_update, self.to_delete = CalendarSync._events_list_compare(
+ events_src_past, self.to_delete)
+ self.to_update.extend(add_to_update)
+
+ # find if events 'to_insert' exists in gcalendar, for update them
+ add_to_update, self.to_insert = self.gcalendar.find_exists(
+ self.to_insert)
+ self.to_update.extend(add_to_update)
+
+ # exclude outdated events from 'to_update' list, by 'updated' field
+ self._filter_events_to_update()
+
+ self.logger.info(
+ 'prepared to sync: ( insert: %d, update: %d, delete: %d )',
+ len(self.to_insert),
+ len(self.to_update),
+ len(self.to_delete)
+ )
+
+ def clear(self) -> None:
+ """ clear prepared sync lists (insert, update, delete)
+ """
+ self.to_insert.clear()
+ self.to_update.clear()
+ self.to_delete.clear()
+
+ def apply(self) -> None:
+ """ apply sync (insert, update, delete), using prepared lists of events
+ """
+
+ self.gcalendar.insert_events(self.to_insert)
+ self.gcalendar.update_events(self.to_update)
+ self.gcalendar.delete_events(self.to_delete)
+
+ self.clear()
+
+ self.logger.info('sync done')
diff --git a/sync-calendar.py b/sync_ics2gcal/sync_calendar.py
similarity index 69%
rename from sync-calendar.py
rename to sync_ics2gcal/sync_calendar.py
index 6df9a79..2e90b6a 100644
--- a/sync-calendar.py
+++ b/sync_ics2gcal/sync_calendar.py
@@ -1,53 +1,54 @@
-import yaml
-
-import dateutil.parser
-import datetime
-import logging
-import logging.config
-from gcal_sync import (
- CalendarConverter,
- GoogleCalendarService,
- GoogleCalendar,
- CalendarSync
-)
-
-
-def load_config():
- with open('config.yml', 'r', encoding='utf-8') as f:
- result = yaml.safe_load(f)
- return result
-
-
-def get_start_date(date_str):
- result = datetime.datetime(1,1,1)
- if 'now' == date_str:
- result = datetime.datetime.utcnow()
- else:
- result = dateutil.parser.parse(date_str)
- return result
-
-
-def main():
- config = load_config()
-
- if 'logging' in config:
- logging.config.dictConfig(config['logging'])
-
- calendarId = config['calendar']['google_id']
- ics_filepath = config['calendar']['source']
- srv_acc_file = config['service_account']
-
- start = get_start_date(config['start_from'])
-
- converter = CalendarConverter()
- converter.load(ics_filepath)
-
- service = GoogleCalendarService.from_srv_acc_file(srv_acc_file)
- gcalendar = GoogleCalendar(service, calendarId)
-
- sync = CalendarSync(gcalendar, converter)
- sync.prepare_sync(start)
- sync.apply()
-
-if __name__ == '__main__':
- main()
+from typing import Dict, Any
+
+import yaml
+
+import dateutil.parser
+import datetime
+import logging
+import logging.config
+from . import (
+ CalendarConverter,
+ GoogleCalendarService,
+ GoogleCalendar,
+ CalendarSync
+)
+
+
+def load_config() -> Dict[str, Any]:
+ with open('config.yml', 'r', encoding='utf-8') as f:
+ result = yaml.safe_load(f)
+ return result
+
+
+def get_start_date(date_str: str) -> datetime.datetime:
+ if 'now' == date_str:
+ result = datetime.datetime.utcnow()
+ else:
+ result = dateutil.parser.parse(date_str)
+ return result
+
+
+def main():
+ config = load_config()
+
+ if 'logging' in config:
+ logging.config.dictConfig(config['logging'])
+
+ calendarId: str = config['calendar']['google_id']
+ ics_filepath: str = config['calendar']['source']
+
+ start = get_start_date(config['start_from'])
+
+ converter = CalendarConverter()
+ converter.load(ics_filepath)
+
+ service = GoogleCalendarService.from_config(config)
+ gcalendar = GoogleCalendar(service, calendarId)
+
+ sync = CalendarSync(gcalendar, converter)
+ sync.prepare_sync(start)
+ sync.apply()
+
+
+if __name__ == '__main__':
+ main()
diff --git a/tests/test_converter.py b/tests/test_converter.py
index c5bb563..37a9f97 100644
--- a/tests/test_converter.py
+++ b/tests/test_converter.py
@@ -1,6 +1,8 @@
+from typing import Tuple
+
import pytest
-from gcal_sync import CalendarConverter
+from sync_ics2gcal import CalendarConverter
uid = "UID:uisgtr8tre93wewe0yr8wqy@test.com"
only_start_date = uid + """
@@ -26,11 +28,11 @@ LAST-MODIFIED:20180326T120235Z
"""
-def ics_test_cal(content):
+def ics_test_cal(content: str) -> str:
return "BEGIN:VCALENDAR\r\n{}END:VCALENDAR\r\n".format(content)
-def ics_test_event(content):
+def ics_test_event(content: str) -> str:
return ics_test_cal("BEGIN:VEVENT\r\n{}END:VEVENT\r\n".format(content))
@@ -68,7 +70,7 @@ def param_events_start_end(request):
return request.param
-def test_event_start_end(param_events_start_end):
+def test_event_start_end(param_events_start_end: Tuple[str, str, str, str]):
(date_type, ics_str, start, end) = param_events_start_end
converter = CalendarConverter()
converter.loads(ics_str)
diff --git a/tests/test_sync.py b/tests/test_sync.py
index 52c2099..f480260 100644
--- a/tests/test_sync.py
+++ b/tests/test_sync.py
@@ -3,15 +3,16 @@ import hashlib
import operator
from copy import deepcopy
from random import shuffle
+from typing import Union, List, Dict, Optional
import dateutil.parser
import pytest
from pytz import timezone, utc
-from gcal_sync import CalendarSync
+from sync_ics2gcal import CalendarSync
-def sha1(string):
+def sha1(string: Union[str, bytes]) -> str:
if isinstance(string, str):
string = string.encode('utf8')
h = hashlib.sha1()
@@ -19,55 +20,57 @@ def sha1(string):
return h.hexdigest()
-def gen_events(start, stop, start_time, no_time=False):
+def gen_events(start: int,
+ stop: int,
+ start_time: Union[datetime.datetime, datetime.date],
+ no_time: bool = False) -> List[Dict[str, Union[str, Dict[str, str]]]]:
if no_time:
start_time = datetime.date(
start_time.year, start_time.month, start_time.day)
- duration = datetime.date(1, 1, 2) - datetime.date(1, 1, 1)
- date_key = "date"
- suff = ''
+ duration: datetime.timedelta = datetime.date(1, 1, 2) - datetime.date(1, 1, 1)
+ date_key: str = "date"
+ date_end: str = ''
else:
start_time = utc.normalize(
start_time.astimezone(utc)).replace(tzinfo=None)
- duration = datetime.datetime(
- 1, 1, 1, 2) - datetime.datetime(1, 1, 1, 1)
- date_key = "dateTime"
- suff = 'Z'
+ duration: datetime.timedelta = datetime.datetime(1, 1, 1, 2) - datetime.datetime(1, 1, 1, 1)
+ date_key: str = "dateTime"
+ date_end: str = 'Z'
- result = []
+ result: List[Dict[str, Union[str, Dict[str, str]]]] = []
for i in range(start, stop):
event_start = start_time + (duration * i)
event_end = event_start + duration
- updated = event_start
+ updated: Union[datetime.datetime, datetime.date] = event_start
if no_time:
updated = datetime.datetime(
updated.year, updated.month, updated.day, 0, 0, 0, 1, tzinfo=utc)
- event = {
+ event: Dict[str, Union[str, Dict[str, str]]] = {
'summary': 'test event __ {}'.format(i),
'location': 'la la la {}'.format(i),
'description': 'test TEST -- test event {}'.format(i),
"iCalUID": "{}@test.com".format(sha1("test - event {}".format(i))),
"updated": updated.isoformat() + 'Z',
- "created": updated.isoformat() + 'Z'
+ "created": updated.isoformat() + 'Z',
+ 'start': {date_key: event_start.isoformat() + date_end},
+ 'end': {date_key: event_end.isoformat() + date_end}
}
- event['start'] = {date_key: event_start.isoformat() + suff}
- event['end'] = {date_key: event_end.isoformat() + suff}
result.append(event)
return result
-def gen_list_to_compare(start, stop):
- result = []
+def gen_list_to_compare(start: int, stop: int) -> List[Dict[str, str]]:
+ result: List[Dict[str, str]] = []
for i in range(start, stop):
result.append({'iCalUID': 'test{:06d}'.format(i)})
return result
-def get_start_date(event):
- event_start = event['start']
- start_date = None
+def get_start_date(event: Dict[str, Union[str, Dict[str, str]]]) -> Union[datetime.datetime, datetime.date]:
+ event_start: Dict[str, str] = event['start']
+ start_date: Optional[str] = None
is_date = False
if 'date' in event_start:
start_date = event_start['date']
@@ -113,7 +116,7 @@ def test_compare():
@pytest.mark.parametrize("no_time", [True, False], ids=['date', 'dateTime'])
-def test_filter_events_by_date(no_time):
+def test_filter_events_by_date(no_time: bool):
msk = timezone('Europe/Moscow')
now = utc.localize(datetime.datetime.utcnow())
msk_now = msk.normalize(now.astimezone(msk))