b4tman/sync_ics2gcal
1
0
Fork 0
mirror of https://github.com/b4tman/sync_ics2gcal synced 2025-05-15 21:34:31 +00:00
Code Issues Releases Activity

Merge branch 'develop'

Browse Source
This commit is contained in:
Dmitry Belyaev 2021-05-01 18:11:56 +03:00
commit 06f3776b1a
Signed by: b4tman
GPG Key ID: 41A00BF15EA7E5F3
23 changed files with 1212 additions and 851 deletions

1
.git_archival.txt Normal file

@ -0,0 +1 @@
ref-names: $Format:%D$

1
.gitattributes vendored Normal file

@ -0,0 +1 @@
.git_archival.txt export-subst

9
.github/dependabot.yml vendored Normal file

@ -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

38
.github/workflows/codeql-analysis.yml vendored Normal file

@ -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

2
.github/workflows/pythonpackage.yml vendored

@ -17,7 +17,7 @@ jobs:
strategy: strategy:
max-parallel: 4 max-parallel: 4
matrix: matrix:
python-version: [3.5, 3.6, 3.7, 3.8] python-version: [3.6, 3.7, 3.8, 3.9]
steps: steps:
- uses: actions/checkout@v1 - uses: actions/checkout@v1

26
.github/workflows/pythonpublish.yml vendored Normal file

@ -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/*

8
.gitignore vendored

@ -2,4 +2,10 @@ config.yml
service-account.json service-account.json
*.pyc *.pyc
my-test*.ics my-test*.ics
.vscode/* .vscode/
.idea/
/dist/
/*.egg-info/
/build/
/.eggs/
venv/

4
.travis.yml

@ -1,10 +1,10 @@
language: python language: python
python: python:
- "3.5"
- "3.6" - "3.6"
- "3.7" - "3.7"
- "3.8" - "3.8"
- "3.9"
script: script:
- pytest -v - pytest -v

7
MANIFEST.in Normal file

@ -0,0 +1,7 @@
include pyproject.toml
# Include the README
include *.md
# Include the license file
include LICENSE

91
README.md

@ -1,15 +1,96 @@
# sync_ics2gcal # sync_ics2gcal
[![PyPI version](https://badge.fury.io/py/sync-ics2gcal.svg)](https://badge.fury.io/py/sync-ics2gcal)
[![Build Status](https://travis-ci.org/b4tman/sync_ics2gcal.svg?branch=master)](https://travis-ci.org/b4tman/sync_ics2gcal) [![Build Status](https://travis-ci.org/b4tman/sync_ics2gcal.svg?branch=master)](https://travis-ci.org/b4tman/sync_ics2gcal)
[![Dependabot Status](https://api.dependabot.com/badges/status?host=github&repo=b4tman/sync_ics2gcal)](https://dependabot.com) [![FOSSA Status](https://app.fossa.io/api/projects/git%2Bgithub.com%2Fb4tman%2Fsync_ics2gcal.svg?type=shield)](https://app.fossa.io/projects/git%2Bgithub.com%2Fb4tman%2Fsync_ics2gcal?ref=badge_shield)
![Python package status](https://github.com/b4tman/sync_ics2gcal/workflows/Python%20package/badge.svg) ![Python package status](https://github.com/b4tman/sync_ics2gcal/workflows/Python%20package/badge.svg)
Python scripts for sync .ics file with Google calendar 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 ## How it works
![How it works](how-it-works.png) ![How it works](how-it-works.png)
## License
[![FOSSA Status](https://app.fossa.io/api/projects/git%2Bgithub.com%2Fb4tman%2Fsync_ics2gcal.svg?type=large)](https://app.fossa.io/projects/git%2Bgithub.com%2Fb4tman%2Fsync_ics2gcal?ref=badge_large)

104
manage-calendars.py

@ -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()

3
pyproject.toml Normal file

@ -0,0 +1,3 @@
[build-system]
requires = ["setuptools>=40.8.0", "wheel", "setuptools_scm"]
build-backend = "setuptools.build_meta"

11
requirements.txt

@ -1,5 +1,6 @@
google-auth==1.11.0 google-auth==1.30.0
google-api-python-client==1.7.11 google-api-python-client==2.3.0
icalendar==4.0.4 icalendar==4.0.7
pytz==2019.3 pytz==2021.1
PyYAML==5.3 PyYAML==5.4.1
fire==0.4.0

7
setup.cfg Normal file

@ -0,0 +1,7 @@
[metadata]
license_files = LICENSE
[options]
setup_requires =
setuptools_scm
setuptools_scm_git_archive

44
setup.py Normal file

@ -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",
]
}
)

28
gcal_sync/__init__.py → sync_ics2gcal/__init__.py

@ -1,14 +1,14 @@
from .ical import ( from .ical import (
CalendarConverter, CalendarConverter,
EventConverter EventConverter
) )
from .gcal import ( from .gcal import (
GoogleCalendarService, GoogleCalendarService,
GoogleCalendar GoogleCalendar
) )
from .sync import ( from .sync import (
CalendarSync CalendarSync
) )

624
gcal_sync/gcal.py → sync_ics2gcal/gcal.py

@ -1,283 +1,341 @@
import logging import logging
import sys
import google.auth
import google.auth from google.oauth2 import service_account
from google.oauth2 import service_account from googleapiclient import discovery
from googleapiclient import discovery from pytz import utc
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 class GoogleCalendarService:
"""class for make google calendar service Resource
Returns:
service Resource Returns:
""" service Resource
"""
@staticmethod
def from_srv_acc_file(service_account_file): @staticmethod
"""make service Resource from service account filename (authorize) def default():
"""make service Resource from default credentials (authorize)
Returns: ( https://developers.google.com/identity/protocols/application-default-credentials )
service Resource ( https://googleapis.dev/python/google-auth/latest/reference/google.auth.html#google.auth.default )
""" """
scopes = ['https://www.googleapis.com/auth/calendar'] scopes = ['https://www.googleapis.com/auth/calendar']
credentials = service_account.Credentials.from_service_account_file(service_account_file) credentials, _ = google.auth.default(scopes=scopes)
scoped_credentials = credentials.with_scopes(scopes) service = discovery.build(
service = discovery.build('calendar', 'v3', credentials=scoped_credentials) 'calendar', 'v3', credentials=credentials, cache_discovery=False)
return service return service
def select_event_key(event): @staticmethod
"""select event key for logging def from_srv_acc_file(service_account_file: str):
"""make service Resource from service account filename (authorize)
Arguments: """
event -- event resource
scopes = ['https://www.googleapis.com/auth/calendar']
Returns: credentials = service_account.Credentials.from_service_account_file(
key name or None if no key found service_account_file)
""" scoped_credentials = credentials.with_scopes(scopes)
service = discovery.build(
key = None 'calendar', 'v3', credentials=scoped_credentials,
if 'iCalUID' in event: cache_discovery=False)
key = 'iCalUID' return service
elif 'id' in event:
key = 'id' @staticmethod
return key def from_config(config: Optional[Dict[str, Optional[str]]] = None):
"""make service Resource from config dict
class GoogleCalendar(): Arguments:
"""class to interact with calendar on google config -- config with keys:
""" (optional) service_account: - service account filename
if key not in dict then default credentials will be used
logger = logging.getLogger('GoogleCalendar') ( https://developers.google.com/identity/protocols/application-default-credentials )
-- None: default credentials will be used
def __init__(self, service, calendarId): """
self.service = service
self.calendarId = calendarId if config is not None and 'service_account' in config:
service = GoogleCalendarService.from_srv_acc_file(
def _make_request_callback(self, action, events_by_req): config['service_account'])
"""make callback for log result of batch request else:
service = GoogleCalendarService.default()
Arguments: return service
action -- action name
events_by_req -- list of events ordered by request id
def select_event_key(event: Dict[str, Any]) -> Optional[str]:
Returns: """select event key for logging
callback function
""" Arguments:
event -- event resource
def callback(request_id, response, exception):
event = events_by_req[int(request_id)] Returns:
key = select_event_key(event) key name or None if no key found
"""
if exception is not None:
self.logger.error('failed to %s event with %s: %s, exception: %s', key = None
action, key, event.get(key), str(exception)) if 'iCalUID' in event:
else: key = 'iCalUID'
resp_key = select_event_key(response) elif 'id' in event:
if resp_key is not None: key = 'id'
event = response return key
key = resp_key
self.logger.info('event %s ok, %s: %s',
action, key, event.get(key)) class GoogleCalendar:
return callback """class to interact with calendar on google
"""
def list_events_from(self, start):
""" list events from calendar, where start date >= start logger = logging.getLogger('GoogleCalendar')
"""
fields = 'nextPageToken,items(id,iCalUID,updated)' def __init__(self, service: discovery.Resource, calendarId: Optional[str]):
events = [] self.service: discovery.Resource = service
page_token = None self.calendarId: str = calendarId
timeMin = utc.normalize(start.astimezone(utc)).replace(
tzinfo=None).isoformat() + 'Z' def _make_request_callback(self, action: str, events_by_req: List[Dict[str, Any]]) -> Callable:
while True: """make callback for log result of batch request
response = self.service.events().list(calendarId=self.calendarId, pageToken=page_token,
singleEvents=True, timeMin=timeMin, fields=fields).execute() Arguments:
if 'items' in response: action -- action name
events.extend(response['items']) events_by_req -- list of events ordered by request id
page_token = response.get('nextPageToken')
if not page_token: Returns:
break callback function
self.logger.info('%d events listed', len(events)) """
return events
def callback(request_id, response, exception):
def find_exists(self, events): event = events_by_req[int(request_id)]
""" find existing events from list, by 'iCalUID' field key = select_event_key(event)
Arguments: if exception is not None:
events {list} -- list of events self.logger.error(
'failed to %s event with %s: %s, exception: %s',
Returns: action, key, event.get(key), str(exception)
tuple -- (events_exist, events_not_found) )
events_exist - list of tuples: (new_event, exists_event) else:
""" resp_key = select_event_key(response)
if resp_key is not None:
fields = 'items(id,iCalUID,updated)' event = response
events_by_req = [] key = resp_key
exists = [] self.logger.info('event %s ok, %s: %s',
not_found = [] action, key, event.get(key))
return callback
def list_callback(request_id, response, exception):
found = False def list_events_from(self, start: datetime) -> List[Dict[str, Any]]:
event = events_by_req[int(request_id)] """ list events from calendar, where start date >= start
if exception is None: """
found = ([] != response['items']) fields = 'nextPageToken,items(id,iCalUID,updated)'
else: events = []
self.logger.error('exception %s, while listing event with UID: %s', str( page_token = None
exception), event['iCalUID']) timeMin = utc.normalize(start.astimezone(utc)).replace(
if found: tzinfo=None).isoformat() + 'Z'
exists.append( while True:
(event, response['items'][0])) response = self.service.events().list(calendarId=self.calendarId,
else: pageToken=page_token,
not_found.append(events_by_req[int(request_id)]) singleEvents=True,
timeMin=timeMin,
batch = self.service.new_batch_http_request(callback=list_callback) fields=fields).execute()
i = 0 if 'items' in response:
for event in events: events.extend(response['items'])
events_by_req.append(event) page_token = response.get('nextPageToken')
batch.add(self.service.events().list(calendarId=self.calendarId, if not page_token:
iCalUID=event['iCalUID'], showDeleted=True, fields=fields), request_id=str(i)) break
i += 1 self.logger.info('%d events listed', len(events))
batch.execute() return events
self.logger.info('%d events exists, %d not found',
len(exists), len(not_found)) def find_exists(self, events: List) -> Tuple[List[Tuple[Dict[str, Any], Dict[str, Any]]], List[Dict[str, Any]]]:
return exists, not_found """ find existing events from list, by 'iCalUID' field
def insert_events(self, events): Arguments:
""" insert list of events events {list} -- list of events
Arguments: Returns:
events - events list tuple -- (events_exist, events_not_found)
""" events_exist - list of tuples: (new_event, exists_event)
"""
fields = 'id'
events_by_req = [] fields = 'items(id,iCalUID,updated)'
events_by_req = []
insert_callback = self._make_request_callback('insert', events_by_req) exists = []
batch = self.service.new_batch_http_request(callback=insert_callback) not_found = []
i = 0
for event in events: def list_callback(request_id, response, exception):
events_by_req.append(event) found = False
batch.add(self.service.events().insert( cur_event = events_by_req[int(request_id)]
calendarId=self.calendarId, body=event, fields=fields), request_id=str(i)) if exception is None:
i += 1 found = ([] != response['items'])
batch.execute() else:
self.logger.error(
def patch_events(self, event_tuples): 'exception %s, while listing event with UID: %s',
""" patch (update) events str(exception), cur_event['iCalUID'])
if found:
Arguments: exists.append(
event_tuples -- list of tuples: (new_event, exists_event) (cur_event, response['items'][0]))
""" else:
not_found.append(events_by_req[int(request_id)])
fields = 'id'
events_by_req = [] batch = self.service.new_batch_http_request(callback=list_callback)
i = 0
patch_callback = self._make_request_callback('patch', events_by_req) for event in events:
batch = self.service.new_batch_http_request(callback=patch_callback) events_by_req.append(event)
i = 0 batch.add(self.service.events().list(calendarId=self.calendarId,
for event_new, event_old in event_tuples: iCalUID=event['iCalUID'],
if 'id' not in event_old: showDeleted=True,
continue fields=fields
events_by_req.append(event_new) ),
batch.add(self.service.events().patch( request_id=str(i)
calendarId=self.calendarId, eventId=event_old['id'], body=event_new), fields=fields, request_id=str(i)) )
i += 1 i += 1
batch.execute() batch.execute()
self.logger.info('%d events exists, %d not found',
def update_events(self, event_tuples): len(exists), len(not_found))
""" update events return exists, not_found
Arguments: def insert_events(self, events: List[Dict[str, Any]]):
event_tuples -- list of tuples: (new_event, exists_event) """ insert list of events
"""
Arguments:
fields = 'id' events - events list
events_by_req = [] """
update_callback = self._make_request_callback('update', events_by_req) fields = 'id'
batch = self.service.new_batch_http_request(callback=update_callback) events_by_req = []
i = 0
for event_new, event_old in event_tuples: insert_callback = self._make_request_callback('insert', events_by_req)
if 'id' not in event_old: batch = self.service.new_batch_http_request(callback=insert_callback)
continue i = 0
events_by_req.append(event_new) for event in events:
batch.add(self.service.events().update( events_by_req.append(event)
calendarId=self.calendarId, eventId=event_old['id'], body=event_new, fields=fields), request_id=str(i)) batch.add(self.service.events().insert(
i += 1 calendarId=self.calendarId, body=event, fields=fields),
batch.execute() request_id=str(i)
)
def delete_events(self, events): i += 1
""" delete events batch.execute()
Arguments: def patch_events(self, event_tuples: List[Tuple[Dict[str, Any], Dict[str, Any]]]):
events -- list of events """ patch (update) events
"""
Arguments:
events_by_req = [] event_tuples -- list of tuples: (new_event, exists_event)
"""
delete_callback = self._make_request_callback('delete', events_by_req)
batch = self.service.new_batch_http_request(callback=delete_callback) fields = 'id'
i = 0 events_by_req = []
for event in events:
events_by_req.append(event) patch_callback = self._make_request_callback('patch', events_by_req)
batch.add(self.service.events().delete( batch = self.service.new_batch_http_request(callback=patch_callback)
calendarId=self.calendarId, eventId=event['id']), request_id=str(i)) i = 0
i += 1 for event_new, event_old in event_tuples:
batch.execute() if 'id' not in event_old:
continue
def create(self, summary, timeZone=None): events_by_req.append(event_new)
"""create calendar batch.add(self.service.events().patch(
calendarId=self.calendarId, eventId=event_old['id'],
Arguments: body=event_new), fields=fields, request_id=str(i))
summary -- new calendar summary i += 1
batch.execute()
Keyword Arguments:
timeZone -- new calendar timezone as string (optional) def update_events(self, event_tuples: List[Tuple[Dict[str, Any], Dict[str, Any]]]):
""" update events
Returns:
calendar Resource Arguments:
""" event_tuples -- list of tuples: (new_event, exists_event)
"""
calendar = {'summary': summary}
if timeZone is not None: fields = 'id'
calendar['timeZone'] = timeZone events_by_req = []
created_calendar = self.service.calendars().insert(body=calendar).execute() update_callback = self._make_request_callback('update', events_by_req)
self.calendarId = created_calendar['id'] batch = self.service.new_batch_http_request(callback=update_callback)
return created_calendar i = 0
for event_new, event_old in event_tuples:
def delete(self): if 'id' not in event_old:
"""delete calendar continue
""" events_by_req.append(event_new)
batch.add(self.service.events().update(
self.service.calendars().delete(calendarId=self.calendarId).execute() calendarId=self.calendarId, eventId=event_old['id'],
body=event_new, fields=fields), request_id=str(i))
def make_public(self): i += 1
"""make calendar puplic batch.execute()
"""
def delete_events(self, events: List[Dict[str, Any]]):
rule_public = { """ delete events
'scope': {
'type': 'default', Arguments:
}, events -- list of events
'role': 'reader' """
}
return self.service.acl().insert(calendarId=self.calendarId, body=rule_public).execute() events_by_req = []
def add_owner(self, email): delete_callback = self._make_request_callback('delete', events_by_req)
"""add calendar owner by email batch = self.service.new_batch_http_request(callback=delete_callback)
i = 0
Arguments: for event in events:
email -- email to add events_by_req.append(event)
""" batch.add(self.service.events().delete(
calendarId=self.calendarId,
rule_owner = { eventId=event['id']), request_id=str(i))
'scope': { i += 1
'type': 'user', batch.execute()
'value': email,
}, def create(self, summary: str, timeZone: Optional[str] = None) -> Any:
'role': 'owner' """create calendar
}
return self.service.acl().insert(calendarId=self.calendarId, body=rule_owner).execute() 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()

379
gcal_sync/ical.py → sync_ics2gcal/ical.py

@ -1,184 +1,195 @@
import datetime import datetime
import logging import logging
from typing import Union, Dict, Any, Callable, Optional, List
from icalendar import Calendar, Event
from pytz import utc from icalendar import Calendar, Event
from pytz import utc
def format_datetime_utc(value):
"""utc datetime as string from date or datetime value def format_datetime_utc(value: Union[datetime.date, datetime.datetime]) -> str:
Arguments: """utc datetime as string from date or datetime value
value -- date or datetime value
Arguments:
Returns: value -- date or datetime value
utc datetime value as string in iso format
""" Returns:
if not isinstance(value, datetime.datetime): utc datetime value as string in iso format
value = datetime.datetime( """
value.year, value.month, value.day, tzinfo=utc) if not isinstance(value, datetime.datetime):
value = value.replace(microsecond=1) value = datetime.datetime(
return utc.normalize(value.astimezone(utc)).replace(tzinfo=None).isoformat() + 'Z' value.year, value.month, value.day, tzinfo=utc)
value = value.replace(microsecond=1)
def gcal_date_or_dateTime(value, check_value=None): return utc.normalize(
"""date or dateTime to gcal (start or end dict) value.astimezone(utc)
Arguments: ).replace(tzinfo=None).isoformat() + 'Z'
value -- date or datetime value
check_value - date or datetime to choise result type (if not None)
def gcal_date_or_dateTime(value: Union[datetime.date, datetime.datetime],
Returns: check_value: Union[datetime.date, datetime.datetime, None] = None)\
dict { 'date': ... } or { 'dateTime': ... } -> Dict[str, str]:
""" """date or dateTime to gcal (start or end dict)
if check_value is None: Arguments:
check_value = value value: date or datetime
check_value: optional for choose result type
result = {}
if isinstance(check_value, datetime.datetime): Returns:
result['dateTime'] = format_datetime_utc(value) { 'date': ... } or { 'dateTime': ... }
else: """
if isinstance(check_value, datetime.date):
if isinstance(value, datetime.datetime): if check_value is None:
value = datetime.date(value.year, value.month, value.day) check_value = value
result['date'] = value.isoformat()
return result result: Dict[str, str] = {}
if isinstance(check_value, datetime.datetime):
result['dateTime'] = format_datetime_utc(value)
class EventConverter(Event): else:
"""Convert icalendar event to google calendar resource if isinstance(check_value, datetime.date):
( https://developers.google.com/calendar/v3/reference/events#resource-representations ) if isinstance(value, datetime.datetime):
""" value = datetime.date(value.year, value.month, value.day)
result['date'] = value.isoformat()
def _str_prop(self, prop): return result
"""decoded string property
Arguments: class EventConverter(Event):
prop - propperty name """Convert icalendar event to google calendar resource
( https://developers.google.com/calendar/v3/reference/events#resource-representations )
Returns: """
string value
""" def _str_prop(self, prop: str) -> str:
"""decoded string property
return self.decoded(prop).decode(encoding='utf-8')
Arguments:
def _datetime_str_prop(self, prop): prop - propperty name
"""utc datetime as string from property
Returns:
Arguments: string value
prop -- property name """
Returns: return self.decoded(prop).decode(encoding='utf-8')
utc datetime value as string in iso format
""" def _datetime_str_prop(self, prop: str) -> str:
"""utc datetime as string from property
return format_datetime_utc(self.decoded(prop))
Arguments:
def _gcal_start(self): prop -- property name
""" event start dict from icalendar event
Returns:
Raises: utc datetime value as string in iso format
ValueError -- if DTSTART not date or datetime """
Returns: return format_datetime_utc(self.decoded(prop))
dict
""" def _gcal_start(self) -> Dict[str, str]:
""" event start dict from icalendar event
value = self.decoded('DTSTART')
return gcal_date_or_dateTime(value) Raises:
ValueError -- if DTSTART not date or datetime
def _gcal_end(self):
"""event end dict from icalendar event Returns:
dict
Raises: """
ValueError -- if no DTEND or DURATION
Returns: value = self.decoded('DTSTART')
dict return gcal_date_or_dateTime(value)
"""
def _gcal_end(self) -> Dict[str, str]:
result = None """event end dict from icalendar event
if 'DTEND' in self:
value = self.decoded('DTEND') Raises:
result = gcal_date_or_dateTime(value) ValueError -- if no DTEND or DURATION
elif 'DURATION' in self: Returns:
start_val = self.decoded('DTSTART') dict
duration = self.decoded('DURATION') """
end_val = start_val + duration
result = None
result = gcal_date_or_dateTime(end_val, check_value=start_val) if 'DTEND' in self:
else: value = self.decoded('DTEND')
raise ValueError('no DTEND or DURATION') result = gcal_date_or_dateTime(value)
return result elif 'DURATION' in self:
start_val = self.decoded('DTSTART')
def _put_to_gcal(self, gcal_event, prop, func, ics_prop=None): duration = self.decoded('DURATION')
"""get property from ical event if exist, and put to gcal event end_val = start_val + duration
Arguments: result = gcal_date_or_dateTime(end_val, check_value=start_val)
gcal_event -- dest event else:
prop -- property name raise ValueError('no DTEND or DURATION')
func -- function to convert return result
ics_prop -- ical property name (default: {None})
""" def _put_to_gcal(self, gcal_event: Dict[str, Any],
prop: str, func: Callable[[str], str],
if not ics_prop: ics_prop: Optional[str] = None):
ics_prop = prop """get property from ical event if exist, and put to gcal event
if ics_prop in self:
gcal_event[prop] = func(ics_prop) Arguments:
gcal_event -- dest event
def to_gcal(self): prop -- property name
"""Convert func -- function to convert
ics_prop -- ical property name (default: {None})
Returns: """
dict - google calendar#event resource
""" if not ics_prop:
ics_prop = prop
event = { if ics_prop in self:
'iCalUID': self._str_prop('UID') gcal_event[prop] = func(ics_prop)
}
def to_gcal(self) -> Dict[str, Any]:
event['start'] = self._gcal_start() """Convert
event['end'] = self._gcal_end()
Returns:
self._put_to_gcal(event, 'summary', self._str_prop) dict - google calendar#event resource
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) event = {
self._put_to_gcal( 'iCalUID': self._str_prop('UID'),
event, 'updated', self._datetime_str_prop, 'LAST-MODIFIED') 'start': self._gcal_start(),
self._put_to_gcal( 'end': self._gcal_end()
event, 'transparency', lambda prop: self._str_prop(prop).lower(), 'TRANSP') }
return event 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)
class CalendarConverter(): self._put_to_gcal(event, 'created', self._datetime_str_prop)
"""Convert icalendar events to google calendar resources self._put_to_gcal(
""" event, 'updated', self._datetime_str_prop, 'LAST-MODIFIED')
self._put_to_gcal(
logger = logging.getLogger('CalendarConverter') event,
'transparency',
def __init__(self, calendar=None): lambda prop: self._str_prop(prop).lower(), 'TRANSP')
self.calendar = calendar
return event
def load(self, filename):
""" load calendar from ics file
""" class CalendarConverter:
with open(filename, 'r', encoding='utf-8') as f: """Convert icalendar events to google calendar resources
self.calendar = Calendar.from_ical(f.read()) """
self.logger.info('%s loaded', filename)
logger = logging.getLogger('CalendarConverter')
def loads(self, string):
""" load calendar from ics string def __init__(self, calendar: Optional[Calendar] = None):
""" self.calendar: Optional[Calendar] = calendar
self.calendar = Calendar.from_ical(string)
def load(self, filename: str):
def events_to_gcal(self): """ load calendar from ics file
"""Convert events to google calendar resources """
""" with open(filename, 'r', encoding='utf-8') as f:
self.calendar = Calendar.from_ical(f.read())
ics_events = self.calendar.walk(name='VEVENT') self.logger.info('%s loaded', filename)
self.logger.info('%d events readed', len(ics_events))
def loads(self, string: str):
result = list( """ load calendar from ics string
map(lambda event: EventConverter(event).to_gcal(), ics_events)) """
self.logger.info('%d events converted', len(result)) self.calendar = Calendar.from_ical(string)
return result
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

144
sync_ics2gcal/manage_calendars.py Normal file

@ -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()

368
gcal_sync/sync.py → sync_ics2gcal/sync.py

@ -1,173 +1,195 @@
import datetime import datetime
import dateutil.parser import logging
import logging import operator
import operator from typing import List, Any, Dict, Set, Tuple, Union, Callable
from pytz import utc
import dateutil.parser
from pytz import utc
class CalendarSync():
"""class for syncronize calendar with google from .gcal import GoogleCalendar
""" from .ical import CalendarConverter
logger = logging.getLogger('CalendarSync')
class CalendarSync:
def __init__(self, gcalendar, converter): """class for syncronize calendar with google
self.gcalendar = gcalendar """
self.converter = converter
logger = logging.getLogger('CalendarSync')
@staticmethod
def _events_list_compare(items_src, items_dst, key='iCalUID'): def __init__(self, gcalendar: GoogleCalendar, converter: CalendarConverter):
""" compare list of events by key self.gcalendar: GoogleCalendar = gcalendar
self.converter: CalendarConverter = converter
Arguments: self.to_insert: List[Dict[str, Any]] = []
items_src {list of dict} -- source events self.to_update: List[Tuple[Dict[str, Any], Dict[str, Any]]] = []
items_dst {list of dict} -- dest events self.to_delete: List[Dict[str, Any]] = []
key {str} -- name of key to compare (default: {'iCalUID'})
@staticmethod
Returns: def _events_list_compare(items_src: List[Dict[str, Any]],
tuple -- (items_to_insert, items_dst: List[Dict[str, Any]],
items_to_update, key: str = 'iCalUID') \
items_to_delete) -> Tuple[List[Dict[str, Any]], List[Tuple[Dict[str, Any], Dict[str, Any]]], List[Dict[str, Any]]]:
""" """ compare list of events by key
def get_key(item): return item[key] Arguments:
items_src {list of dict} -- source events
keys_src = set(map(get_key, items_src)) items_dst {list of dict} -- dest events
keys_dst = set(map(get_key, items_dst)) key {str} -- name of key to compare (default: {'iCalUID'})
keys_to_insert = keys_src - keys_dst Returns:
keys_to_update = keys_src & keys_dst tuple -- (items_to_insert,
keys_to_delete = keys_dst - keys_src items_to_update,
items_to_delete)
def items_by_keys(items, key_name, keys): """
return list(filter(lambda item: item[key_name] in keys, items))
def get_key(item: Dict[str, Any]) -> str: return item[key]
items_to_insert = items_by_keys(items_src, key, keys_to_insert)
items_to_delete = items_by_keys(items_dst, key, keys_to_delete) keys_src: Set[str] = set(map(get_key, items_src))
keys_dst: Set[str] = set(map(get_key, items_dst))
to_upd_src = items_by_keys(items_src, key, keys_to_update)
to_upd_dst = items_by_keys(items_dst, key, keys_to_update) keys_to_insert = keys_src - keys_dst
to_upd_src.sort(key=get_key) keys_to_update = keys_src & keys_dst
to_upd_dst.sort(key=get_key) keys_to_delete = keys_dst - keys_src
items_to_update = list(zip(to_upd_src, to_upd_dst))
def items_by_keys(items: List[Dict[str, Any]],
return items_to_insert, items_to_update, items_to_delete key_name: str,
keys: Set[str]) -> List[Dict[str, Any]]:
def _filter_events_to_update(self): return list(filter(lambda item: item[key_name] in keys, items))
""" filter 'to_update' events by 'updated' datetime
""" items_to_insert = items_by_keys(items_src, key, keys_to_insert)
items_to_delete = items_by_keys(items_dst, key, keys_to_delete)
def filter_updated(event_tuple):
new, old = event_tuple to_upd_src = items_by_keys(items_src, key, keys_to_update)
return dateutil.parser.parse(new['updated']) > dateutil.parser.parse(old['updated']) to_upd_dst = items_by_keys(items_dst, key, keys_to_update)
to_upd_src.sort(key=get_key)
self.to_update = list(filter(filter_updated, self.to_update)) to_upd_dst.sort(key=get_key)
items_to_update = list(zip(to_upd_src, to_upd_dst))
@staticmethod
def _filter_events_by_date(events, date, op): return items_to_insert, items_to_update, items_to_delete
""" filter events by start datetime
def _filter_events_to_update(self):
Arguments: """ filter 'to_update' events by 'updated' datetime
events -- events list """
date {datetime} -- datetime to compare
op {operator} -- comparsion operator def filter_updated(event_tuple: Tuple[Dict[str, Any], Dict[str, Any]]) -> bool:
new, old = event_tuple
Returns: new_date = dateutil.parser.parse(new['updated'])
list of filtred events old_date = dateutil.parser.parse(old['updated'])
""" return new_date > old_date
def filter_by_date(event): self.to_update = list(filter(filter_updated, self.to_update))
date_cmp = date
event_start = event['start'] @staticmethod
event_date = None def _filter_events_by_date(events: List[Dict[str, Any]],
compare_dates = False date: Union[datetime.date, datetime.datetime],
op: Callable[[Union[datetime.date, datetime.datetime],
if 'date' in event_start: Union[datetime.date, datetime.datetime]], bool]) -> List[Dict[str, Any]]:
event_date = event_start['date'] """ filter events by start datetime
compare_dates = True
elif 'dateTime' in event_start: Arguments:
event_date = event_start['dateTime'] events -- events list
date {datetime} -- datetime to compare
event_date = dateutil.parser.parse(event_date) op {operator} -- comparsion operator
if compare_dates:
date_cmp = datetime.date(date.year, date.month, date.day) Returns:
event_date = datetime.date(event_date.year, event_date.month, event_date.day) list of filtred events
"""
return op(event_date, date_cmp)
def filter_by_date(event: Dict[str, Any]) -> bool:
return list(filter(filter_by_date, events)) date_cmp = date
event_start: Dict[str, str] = event['start']
@staticmethod event_date: Union[datetime.date, datetime.datetime, str, None] = None
def _tz_aware_datetime(date): compare_dates = False
"""make tz aware datetime from datetime/date (utc if no tzinfo)
if 'date' in event_start:
Arguments: event_date = event_start['date']
date - date or datetime / with or without tzinfo compare_dates = True
elif 'dateTime' in event_start:
Returns: event_date = event_start['dateTime']
datetime with tzinfo
""" event_date = dateutil.parser.parse(event_date)
if compare_dates:
if not isinstance(date, datetime.datetime): date_cmp = datetime.date(date.year, date.month, date.day)
date = datetime.datetime(date.year, date.month, date.day) event_date = datetime.date(
if date.tzinfo is None: event_date.year, event_date.month, event_date.day)
date = date.replace(tzinfo=utc)
return date return op(event_date, date_cmp)
def prepare_sync(self, start_date): return list(filter(filter_by_date, events))
"""prepare sync lists by comparsion of events
@staticmethod
Arguments: def _tz_aware_datetime(date: Union[datetime.date, datetime.datetime]) -> datetime.datetime:
start_date -- date/datetime to start sync """make tz aware datetime from datetime/date (utc if no tzinfo)
"""
Arguments:
start_date = CalendarSync._tz_aware_datetime(start_date) date - date or datetime / with or without tzinfo
events_src = self.converter.events_to_gcal() Returns:
events_dst = self.gcalendar.list_events_from(start_date) datetime with tzinfo
"""
# divide source events by start datetime
events_src_pending = CalendarSync._filter_events_by_date( if not isinstance(date, datetime.datetime):
events_src, start_date, operator.ge) date = datetime.datetime(date.year, date.month, date.day)
events_src_past = CalendarSync._filter_events_by_date( if date.tzinfo is None:
events_src, start_date, operator.lt) date = date.replace(tzinfo=utc)
return date
events_src = None
def prepare_sync(self, start_date: Union[datetime.date, datetime.datetime]) -> None:
# first events comparsion """prepare sync lists by comparsion of events
self.to_insert, self.to_update, self.to_delete = CalendarSync._events_list_compare(
events_src_pending, events_dst) Arguments:
start_date -- date/datetime to start sync
events_src_pending, events_dst = None, None """
# find in events 'to_delete' past events from source, for update (move to past) start_date = CalendarSync._tz_aware_datetime(start_date)
_, add_to_update, self.to_delete = CalendarSync._events_list_compare(
events_src_past, self.to_delete) events_src = self.converter.events_to_gcal()
self.to_update.extend(add_to_update) events_dst = self.gcalendar.list_events_from(start_date)
events_src_past = None # divide source events by start datetime
events_src_pending = CalendarSync._filter_events_by_date(
# find if events 'to_insert' exists in gcalendar, for update them events_src, start_date, operator.ge)
add_to_update, self.to_insert = self.gcalendar.find_exists( events_src_past = CalendarSync._filter_events_by_date(
self.to_insert) events_src, start_date, operator.lt)
self.to_update.extend(add_to_update)
# first events comparsion
add_to_update = None self.to_insert, self.to_update, self.to_delete = CalendarSync._events_list_compare(
events_src_pending, events_dst)
# exclude outdated events from 'to_update' list, by 'updated' field
self._filter_events_to_update() # find in events 'to_delete' past events from source, for update (move to past)
_, add_to_update, self.to_delete = CalendarSync._events_list_compare(
self.logger.info('prepared to sync: ( insert: %d, update: %d, delete: %d )', events_src_past, self.to_delete)
len(self.to_insert), len(self.to_update), len(self.to_delete)) self.to_update.extend(add_to_update)
def apply(self): # find if events 'to_insert' exists in gcalendar, for update them
"""apply sync (insert, update, delete), using prepared lists of events add_to_update, self.to_insert = self.gcalendar.find_exists(
""" self.to_insert)
self.to_update.extend(add_to_update)
self.gcalendar.insert_events(self.to_insert)
self.gcalendar.update_events(self.to_update) # exclude outdated events from 'to_update' list, by 'updated' field
self.gcalendar.delete_events(self.to_delete) self._filter_events_to_update()
self.logger.info('sync done') self.logger.info(
'prepared to sync: ( insert: %d, update: %d, delete: %d )',
self.to_insert, self.to_update, self.to_delete = [], [], [] 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')

107
sync-calendar.py → sync_ics2gcal/sync_calendar.py

@ -1,53 +1,54 @@
import yaml from typing import Dict, Any
import dateutil.parser import yaml
import datetime
import logging import dateutil.parser
import logging.config import datetime
from gcal_sync import ( import logging
CalendarConverter, import logging.config
GoogleCalendarService, from . import (
GoogleCalendar, CalendarConverter,
CalendarSync GoogleCalendarService,
) GoogleCalendar,
CalendarSync
)
def load_config():
with open('config.yml', 'r', encoding='utf-8') as f:
result = yaml.safe_load(f) def load_config() -> Dict[str, Any]:
return result 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: def get_start_date(date_str: str) -> datetime.datetime:
result = datetime.datetime.utcnow() if 'now' == date_str:
else: result = datetime.datetime.utcnow()
result = dateutil.parser.parse(date_str) else:
return result result = dateutil.parser.parse(date_str)
return result
def main():
config = load_config() def main():
config = load_config()
if 'logging' in config:
logging.config.dictConfig(config['logging']) if 'logging' in config:
logging.config.dictConfig(config['logging'])
calendarId = config['calendar']['google_id']
ics_filepath = config['calendar']['source'] calendarId: str = config['calendar']['google_id']
srv_acc_file = config['service_account'] ics_filepath: str = config['calendar']['source']
start = get_start_date(config['start_from']) start = get_start_date(config['start_from'])
converter = CalendarConverter() converter = CalendarConverter()
converter.load(ics_filepath) converter.load(ics_filepath)
service = GoogleCalendarService.from_srv_acc_file(srv_acc_file) service = GoogleCalendarService.from_config(config)
gcalendar = GoogleCalendar(service, calendarId) gcalendar = GoogleCalendar(service, calendarId)
sync = CalendarSync(gcalendar, converter) sync = CalendarSync(gcalendar, converter)
sync.prepare_sync(start) sync.prepare_sync(start)
sync.apply() sync.apply()
if __name__ == '__main__':
main() if __name__ == '__main__':
main()

10
tests/test_converter.py

@ -1,6 +1,8 @@
from typing import Tuple
import pytest import pytest
from gcal_sync import CalendarConverter from sync_ics2gcal import CalendarConverter
uid = "UID:uisgtr8tre93wewe0yr8wqy@test.com" uid = "UID:uisgtr8tre93wewe0yr8wqy@test.com"
only_start_date = uid + """ 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) 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)) 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 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 (date_type, ics_str, start, end) = param_events_start_end
converter = CalendarConverter() converter = CalendarConverter()
converter.loads(ics_str) converter.loads(ics_str)

47
tests/test_sync.py

@ -3,15 +3,16 @@ import hashlib
import operator import operator
from copy import deepcopy from copy import deepcopy
from random import shuffle from random import shuffle
from typing import Union, List, Dict, Optional
import dateutil.parser import dateutil.parser
import pytest import pytest
from pytz import timezone, utc 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): if isinstance(string, str):
string = string.encode('utf8') string = string.encode('utf8')
h = hashlib.sha1() h = hashlib.sha1()
@ -19,55 +20,57 @@ def sha1(string):
return h.hexdigest() 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: if no_time:
start_time = datetime.date( start_time = datetime.date(
start_time.year, start_time.month, start_time.day) start_time.year, start_time.month, start_time.day)
duration = datetime.date(1, 1, 2) - datetime.date(1, 1, 1) duration: datetime.timedelta = datetime.date(1, 1, 2) - datetime.date(1, 1, 1)
date_key = "date" date_key: str = "date"
suff = '' date_end: str = ''
else: else:
start_time = utc.normalize( start_time = utc.normalize(
start_time.astimezone(utc)).replace(tzinfo=None) start_time.astimezone(utc)).replace(tzinfo=None)
duration = datetime.datetime( duration: datetime.timedelta = datetime.datetime(1, 1, 1, 2) - datetime.datetime(1, 1, 1, 1)
1, 1, 1, 2) - datetime.datetime(1, 1, 1, 1) date_key: str = "dateTime"
date_key = "dateTime" date_end: str = 'Z'
suff = 'Z'
result = [] result: List[Dict[str, Union[str, Dict[str, str]]]] = []
for i in range(start, stop): for i in range(start, stop):
event_start = start_time + (duration * i) event_start = start_time + (duration * i)
event_end = event_start + duration event_end = event_start + duration
updated = event_start updated: Union[datetime.datetime, datetime.date] = event_start
if no_time: if no_time:
updated = datetime.datetime( updated = datetime.datetime(
updated.year, updated.month, updated.day, 0, 0, 0, 1, tzinfo=utc) 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), 'summary': 'test event __ {}'.format(i),
'location': 'la la la {}'.format(i), 'location': 'la la la {}'.format(i),
'description': 'test TEST -- test event {}'.format(i), 'description': 'test TEST -- test event {}'.format(i),
"iCalUID": "{}@test.com".format(sha1("test - event {}".format(i))), "iCalUID": "{}@test.com".format(sha1("test - event {}".format(i))),
"updated": updated.isoformat() + 'Z', "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) result.append(event)
return result return result
def gen_list_to_compare(start, stop): def gen_list_to_compare(start: int, stop: int) -> List[Dict[str, str]]:
result = [] result: List[Dict[str, str]] = []
for i in range(start, stop): for i in range(start, stop):
result.append({'iCalUID': 'test{:06d}'.format(i)}) result.append({'iCalUID': 'test{:06d}'.format(i)})
return result return result
def get_start_date(event): def get_start_date(event: Dict[str, Union[str, Dict[str, str]]]) -> Union[datetime.datetime, datetime.date]:
event_start = event['start'] event_start: Dict[str, str] = event['start']
start_date = None start_date: Optional[str] = None
is_date = False is_date = False
if 'date' in event_start: if 'date' in event_start:
start_date = event_start['date'] start_date = event_start['date']
@ -113,7 +116,7 @@ def test_compare():
@pytest.mark.parametrize("no_time", [True, False], ids=['date', 'dateTime']) @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') msk = timezone('Europe/Moscow')
now = utc.localize(datetime.datetime.utcnow()) now = utc.localize(datetime.datetime.utcnow())
msk_now = msk.normalize(now.astimezone(msk)) msk_now = msk.normalize(now.astimezone(msk))