b4tman
/
sync_ics2gcal
mirror of https://github.com/b4tman/sync_ics2gcal
1
0
Fork 0
Code Issues Releases Activity

Format with black.

This commit is contained in:
Glenn Waters 2022-03-08 10:46:21 -05:00
parent 2372103807
commit 70278c1542
10 changed files with 507 additions and 328 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

140
poetry.lock generated
View File

@ -20,6 +20,29 @@ docs = ["furo", "sphinx", "zope.interface", "sphinx-notfound-page"]
tests = ["coverage[toml] (>=5.0.2)", "hypothesis", "pympler", "pytest (>=4.3.0)", "six", "mypy", "pytest-mypy-plugins", "zope.interface", "cloudpickle"]
tests_no_zope = ["coverage[toml] (>=5.0.2)", "hypothesis", "pympler", "pytest (>=4.3.0)", "six", "mypy", "pytest-mypy-plugins", "cloudpickle"]
[[package]]
name = "black"
version = "22.1.0"
description = "The uncompromising code formatter."
category = "dev"
optional = false
python-versions = ">=3.6.2"
[package.dependencies]
click = ">=8.0.0"
mypy-extensions = ">=0.4.3"
pathspec = ">=0.9.0"
platformdirs = ">=2"
tomli = ">=1.1.0"
typed-ast = {version = ">=1.4.2", markers = "python_version < \"3.8\" and implementation_name == \"cpython\""}
typing-extensions = {version = ">=3.10.0.0", markers = "python_version < \"3.10\""}
[package.extras]
colorama = ["colorama (>=0.4.3)"]
d = ["aiohttp (>=3.7.4)"]
jupyter = ["ipython (>=7.8.0)", "tokenize-rt (>=3.2.0)"]
uvloop = ["uvloop (>=0.15.2)"]
[[package]]
name = "cachetools"
version = "5.0.0"
@ -47,6 +70,18 @@ python-versions = ">=3.5.0"
[package.extras]
unicode_backport = ["unicodedata2"]
[[package]]
name = "click"
version = "8.0.4"
description = "Composable command line interface toolkit"
category = "dev"
optional = false
python-versions = ">=3.6"
[package.dependencies]
colorama = {version = "*", markers = "platform_system == \"Windows\""}
importlib-metadata = {version = "*", markers = "python_version < \"3.8\""}
[[package]]
name = "colorama"
version = "0.4.4"
@ -224,6 +259,14 @@ category = "dev"
optional = false
python-versions = "*"
[[package]]
name = "mypy-extensions"
version = "0.4.3"
description = "Experimental type system extensions for programs checked with the mypy typechecker."
category = "dev"
optional = false
python-versions = "*"
[[package]]
name = "packaging"
version = "21.3"
@ -235,6 +278,26 @@ python-versions = ">=3.6"
[package.dependencies]
pyparsing = ">=2.0.2,<3.0.5 || >3.0.5"
[[package]]
name = "pathspec"
version = "0.9.0"
description = "Utility library for gitignore style pattern matching of file paths."
category = "dev"
optional = false
python-versions = "!=3.0.*,!=3.1.*,!=3.2.*,!=3.3.*,!=3.4.*,>=2.7"
[[package]]
name = "platformdirs"
version = "2.5.1"
description = "A small Python module for determining appropriate platform-specific dirs, e.g. a \"user data dir\"."
category = "dev"
optional = false
python-versions = ">=3.7"
[package.extras]
docs = ["Sphinx (>=4)", "furo (>=2021.7.5b38)", "proselint (>=0.10.2)", "sphinx-autodoc-typehints (>=1.12)"]
test = ["appdirs (==1.4.4)", "pytest (>=6)", "pytest-cov (>=2.7)", "pytest-mock (>=3.6)"]
[[package]]
name = "pluggy"
version = "1.0.0"
@ -414,6 +477,14 @@ category = "dev"
optional = false
python-versions = ">=3.7"
[[package]]
name = "typed-ast"
version = "1.5.2"
description = "a fork of Python 2 and 3 ast modules with type comment support"
category = "dev"
optional = false
python-versions = ">=3.6"
[[package]]
name = "typing-extensions"
version = "4.1.1"
@ -458,7 +529,7 @@ testing = ["pytest (>=6)", "pytest-checkdocs (>=2.4)", "pytest-flake8", "pytest-
[metadata]
lock-version = "1.1"
python-versions = "^3.7"
content-hash = "2e3d86cebb44eec6cbff1c371b42b461fc72efa65c599c1ad5053e17f4d4943a"
content-hash = "77682dbe73e14e9803a9ee35551527645ef1d493b9b6ef454143769263f10790"
[metadata.files]
atomicwrites = [
@ -469,6 +540,31 @@ attrs = [
{file = "attrs-21.4.0-py2.py3-none-any.whl", hash = "sha256:2d27e3784d7a565d36ab851fe94887c5eccd6a463168875832a1be79c82828b4"},
{file = "attrs-21.4.0.tar.gz", hash = "sha256:626ba8234211db98e869df76230a137c4c40a12d72445c45d5f5b716f076e2fd"},
]
black = [
{file = "black-22.1.0-cp310-cp310-macosx_10_9_universal2.whl", hash = "sha256:1297c63b9e1b96a3d0da2d85d11cd9bf8664251fd69ddac068b98dc4f34f73b6"},
{file = "black-22.1.0-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:2ff96450d3ad9ea499fc4c60e425a1439c2120cbbc1ab959ff20f7c76ec7e866"},
{file = "black-22.1.0-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:0e21e1f1efa65a50e3960edd068b6ae6d64ad6235bd8bfea116a03b21836af71"},
{file = "black-22.1.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:e2f69158a7d120fd641d1fa9a921d898e20d52e44a74a6fbbcc570a62a6bc8ab"},
{file = "black-22.1.0-cp310-cp310-win_amd64.whl", hash = "sha256:228b5ae2c8e3d6227e4bde5920d2fc66cc3400fde7bcc74f480cb07ef0b570d5"},
{file = "black-22.1.0-cp36-cp36m-macosx_10_9_x86_64.whl", hash = "sha256:b1a5ed73ab4c482208d20434f700d514f66ffe2840f63a6252ecc43a9bc77e8a"},
{file = "black-22.1.0-cp36-cp36m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:35944b7100af4a985abfcaa860b06af15590deb1f392f06c8683b4381e8eeaf0"},
{file = "black-22.1.0-cp36-cp36m-win_amd64.whl", hash = "sha256:7835fee5238fc0a0baf6c9268fb816b5f5cd9b8793423a75e8cd663c48d073ba"},
{file = "black-22.1.0-cp37-cp37m-macosx_10_9_x86_64.whl", hash = "sha256:dae63f2dbf82882fa3b2a3c49c32bffe144970a573cd68d247af6560fc493ae1"},
{file = "black-22.1.0-cp37-cp37m-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:5fa1db02410b1924b6749c245ab38d30621564e658297484952f3d8a39fce7e8"},
{file = "black-22.1.0-cp37-cp37m-win_amd64.whl", hash = "sha256:c8226f50b8c34a14608b848dc23a46e5d08397d009446353dad45e04af0c8e28"},
{file = "black-22.1.0-cp38-cp38-macosx_10_9_universal2.whl", hash = "sha256:2d6f331c02f0f40aa51a22e479c8209d37fcd520c77721c034517d44eecf5912"},
{file = "black-22.1.0-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:742ce9af3086e5bd07e58c8feb09dbb2b047b7f566eb5f5bc63fd455814979f3"},
{file = "black-22.1.0-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:fdb8754b453fb15fad3f72cd9cad3e16776f0964d67cf30ebcbf10327a3777a3"},
{file = "black-22.1.0-cp38-cp38-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:f5660feab44c2e3cb24b2419b998846cbb01c23c7fe645fee45087efa3da2d61"},
{file = "black-22.1.0-cp38-cp38-win_amd64.whl", hash = "sha256:6f2f01381f91c1efb1451998bd65a129b3ed6f64f79663a55fe0e9b74a5f81fd"},
{file = "black-22.1.0-cp39-cp39-macosx_10_9_universal2.whl", hash = "sha256:efbadd9b52c060a8fc3b9658744091cb33c31f830b3f074422ed27bad2b18e8f"},
{file = "black-22.1.0-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:8871fcb4b447206904932b54b567923e5be802b9b19b744fdff092bd2f3118d0"},
{file = "black-22.1.0-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:ccad888050f5393f0d6029deea2a33e5ae371fd182a697313bdbd835d3edaf9c"},
{file = "black-22.1.0-cp39-cp39-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:07e5c049442d7ca1a2fc273c79d1aecbbf1bc858f62e8184abe1ad175c4f7cc2"},
{file = "black-22.1.0-cp39-cp39-win_amd64.whl", hash = "sha256:373922fc66676133ddc3e754e4509196a8c392fec3f5ca4486673e685a421321"},
{file = "black-22.1.0-py3-none-any.whl", hash = "sha256:3524739d76b6b3ed1132422bf9d82123cd1705086723bc3e235ca39fd21c667d"},
{file = "black-22.1.0.tar.gz", hash = "sha256:a7c0192d35635f6fc1174be575cb7915e92e5dd629ee79fdaf0dcfa41a80afb5"},
]
cachetools = [
{file = "cachetools-5.0.0-py3-none-any.whl", hash = "sha256:8fecd4203a38af17928be7b90689d8083603073622229ca7077b72d8e5a976e4"},
{file = "cachetools-5.0.0.tar.gz", hash = "sha256:486471dfa8799eb7ec503a8059e263db000cdda20075ce5e48903087f79d5fd6"},
@ -481,6 +577,10 @@ charset-normalizer = [
{file = "charset-normalizer-2.0.12.tar.gz", hash = "sha256:2857e29ff0d34db842cd7ca3230549d1a697f96ee6d3fb071cfa6c7393832597"},
{file = "charset_normalizer-2.0.12-py3-none-any.whl", hash = "sha256:6881edbebdb17b39b4eaaa821b438bf6eddffb4468cf344f09f89def34a8b1df"},
]
click = [
{file = "click-8.0.4-py3-none-any.whl", hash = "sha256:6a7a62563bbfabfda3a38f3023a1db4a35978c0abd76f6c9605ecd6554d6d9b1"},
{file = "click-8.0.4.tar.gz", hash = "sha256:8458d7b1287c5fb128c90e23381cf99dcde74beaf6c7ff6384ce84d6fe090adb"},
]
colorama = [
{file = "colorama-0.4.4-py2.py3-none-any.whl", hash = "sha256:9f47eda37229f68eee03b24b9748937c7dc3868f906e8ba69fbcbdd3bc5dc3e2"},
{file = "colorama-0.4.4.tar.gz", hash = "sha256:5941b2b48a20143d2267e95b1c2a7603ce057ee39fd88e7329b0c292aa16869b"},
@ -536,10 +636,22 @@ mccabe = [
{file = "mccabe-0.6.1-py2.py3-none-any.whl", hash = "sha256:ab8a6258860da4b6677da4bd2fe5dc2c659cff31b3ee4f7f5d64e79735b80d42"},
{file = "mccabe-0.6.1.tar.gz", hash = "sha256:dd8d182285a0fe56bace7f45b5e7d1a6ebcbf524e8f3bd87eb0f125271b8831f"},
]
mypy-extensions = [
{file = "mypy_extensions-0.4.3-py2.py3-none-any.whl", hash = "sha256:090fedd75945a69ae91ce1303b5824f428daf5a028d2f6ab8a299250a846f15d"},
{file = "mypy_extensions-0.4.3.tar.gz", hash = "sha256:2d82818f5bb3e369420cb3c4060a7970edba416647068eb4c5343488a6c604a8"},
]
packaging = [
{file = "packaging-21.3-py3-none-any.whl", hash = "sha256:ef103e05f519cdc783ae24ea4e2e0f508a9c99b2d4969652eed6a2e1ea5bd522"},
{file = "packaging-21.3.tar.gz", hash = "sha256:dd47c42927d89ab911e606518907cc2d3a1f38bbd026385970643f9c5b8ecfeb"},
]
pathspec = [
{file = "pathspec-0.9.0-py2.py3-none-any.whl", hash = "sha256:7d15c4ddb0b5c802d161efc417ec1a2558ea2653c2e8ad9c19098201dc1c993a"},
{file = "pathspec-0.9.0.tar.gz", hash = "sha256:e564499435a2673d586f6b2130bb5b95f04a3ba06f81b8f895b651a3c76aabb1"},
]
platformdirs = [
{file = "platformdirs-2.5.1-py3-none-any.whl", hash = "sha256:bcae7cab893c2d310a711b70b24efb93334febe65f8de776ee320b517471e227"},
{file = "platformdirs-2.5.1.tar.gz", hash = "sha256:7535e70dfa32e84d4b34996ea99c5e432fa29a708d0f4e394bbcb2a8faa4f16d"},
]
pluggy = [
{file = "pluggy-1.0.0-py2.py3-none-any.whl", hash = "sha256:74134bbf457f031a36d68416e1509f34bd5ccc019f0bcc952c7b909d06b37bd3"},
{file = "pluggy-1.0.0.tar.gz", hash = "sha256:4224373bacce55f955a878bf9cfa763c1e360858e330072059e10bad68531159"},
@ -684,6 +796,32 @@ tomli = [
{file = "tomli-2.0.1-py3-none-any.whl", hash = "sha256:939de3e7a6161af0c887ef91b7d41a53e7c5a1ca976325f429cb46ea9bc30ecc"},
{file = "tomli-2.0.1.tar.gz", hash = "sha256:de526c12914f0c550d15924c62d72abc48d6fe7364aa87328337a31007fe8a4f"},
]
typed-ast = [
{file = "typed_ast-1.5.2-cp310-cp310-macosx_10_9_x86_64.whl", hash = "sha256:183b183b7771a508395d2cbffd6db67d6ad52958a5fdc99f450d954003900266"},
{file = "typed_ast-1.5.2-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:676d051b1da67a852c0447621fdd11c4e104827417bf216092ec3e286f7da596"},
{file = "typed_ast-1.5.2-cp310-cp310-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:bc2542e83ac8399752bc16e0b35e038bdb659ba237f4222616b4e83fb9654985"},
{file = "typed_ast-1.5.2-cp310-cp310-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_12_x86_64.manylinux2010_x86_64.whl", hash = "sha256:74cac86cc586db8dfda0ce65d8bcd2bf17b58668dfcc3652762f3ef0e6677e76"},
{file = "typed_ast-1.5.2-cp310-cp310-win_amd64.whl", hash = "sha256:18fe320f354d6f9ad3147859b6e16649a0781425268c4dde596093177660e71a"},
{file = "typed_ast-1.5.2-cp36-cp36m-macosx_10_9_x86_64.whl", hash = "sha256:31d8c6b2df19a777bc8826770b872a45a1f30cfefcfd729491baa5237faae837"},
{file = "typed_ast-1.5.2-cp36-cp36m-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:963a0ccc9a4188524e6e6d39b12c9ca24cc2d45a71cfdd04a26d883c922b4b78"},
{file = "typed_ast-1.5.2-cp36-cp36m-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_12_x86_64.manylinux2010_x86_64.whl", hash = "sha256:0eb77764ea470f14fcbb89d51bc6bbf5e7623446ac4ed06cbd9ca9495b62e36e"},
{file = "typed_ast-1.5.2-cp36-cp36m-win_amd64.whl", hash = "sha256:294a6903a4d087db805a7656989f613371915fc45c8cc0ddc5c5a0a8ad9bea4d"},
{file = "typed_ast-1.5.2-cp37-cp37m-macosx_10_9_x86_64.whl", hash = "sha256:26a432dc219c6b6f38be20a958cbe1abffcc5492821d7e27f08606ef99e0dffd"},
{file = "typed_ast-1.5.2-cp37-cp37m-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:c7407cfcad702f0b6c0e0f3e7ab876cd1d2c13b14ce770e412c0c4b9728a0f88"},
{file = "typed_ast-1.5.2-cp37-cp37m-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_12_x86_64.manylinux2010_x86_64.whl", hash = "sha256:f30ddd110634c2d7534b2d4e0e22967e88366b0d356b24de87419cc4410c41b7"},
{file = "typed_ast-1.5.2-cp37-cp37m-win_amd64.whl", hash = "sha256:8c08d6625bb258179b6e512f55ad20f9dfef019bbfbe3095247401e053a3ea30"},
{file = "typed_ast-1.5.2-cp38-cp38-macosx_10_9_x86_64.whl", hash = "sha256:90904d889ab8e81a956f2c0935a523cc4e077c7847a836abee832f868d5c26a4"},
{file = "typed_ast-1.5.2-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:bbebc31bf11762b63bf61aaae232becb41c5bf6b3461b80a4df7e791fabb3aca"},
{file = "typed_ast-1.5.2-cp38-cp38-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:c29dd9a3a9d259c9fa19d19738d021632d673f6ed9b35a739f48e5f807f264fb"},
{file = "typed_ast-1.5.2-cp38-cp38-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_12_x86_64.manylinux2010_x86_64.whl", hash = "sha256:58ae097a325e9bb7a684572d20eb3e1809802c5c9ec7108e85da1eb6c1a3331b"},
{file = "typed_ast-1.5.2-cp38-cp38-win_amd64.whl", hash = "sha256:da0a98d458010bf4fe535f2d1e367a2e2060e105978873c04c04212fb20543f7"},
{file = "typed_ast-1.5.2-cp39-cp39-macosx_10_9_x86_64.whl", hash = "sha256:33b4a19ddc9fc551ebabca9765d54d04600c4a50eda13893dadf67ed81d9a098"},
{file = "typed_ast-1.5.2-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:1098df9a0592dd4c8c0ccfc2e98931278a6c6c53cb3a3e2cf7e9ee3b06153344"},
{file = "typed_ast-1.5.2-cp39-cp39-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:42c47c3b43fe3a39ddf8de1d40dbbfca60ac8530a36c9b198ea5b9efac75c09e"},
{file = "typed_ast-1.5.2-cp39-cp39-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_12_x86_64.manylinux2010_x86_64.whl", hash = "sha256:f290617f74a610849bd8f5514e34ae3d09eafd521dceaa6cf68b3f4414266d4e"},
{file = "typed_ast-1.5.2-cp39-cp39-win_amd64.whl", hash = "sha256:df05aa5b241e2e8045f5f4367a9f6187b09c4cdf8578bb219861c4e27c443db5"},
{file = "typed_ast-1.5.2.tar.gz", hash = "sha256:525a2d4088e70a9f75b08b3f87a51acc9cde640e19cc523c7e41aa355564ae27"},
]
typing-extensions = [
{file = "typing_extensions-4.1.1-py3-none-any.whl", hash = "sha256:21c85e0fe4b9a155d0799430b0ad741cdce7e359660ccbd8b530613e8df88ce2"},
{file = "typing_extensions-4.1.1.tar.gz", hash = "sha256:1a9462dcc3347a79b1f1c0271fbe79e844580bb598bafa1ed208b94da3cdcd42"},

1
pyproject.toml
View File

@ -29,6 +29,7 @@ fire = "0.4.0"
[tool.poetry.dev-dependencies]
pytest = "^7.0.1"
flake8 = "^4.0.1"
black = "^22.1.0"
[tool.poetry.scripts]
sync-ics2gcal = "sync_ics2gcal.sync_calendar:main"

13
sync_ics2gcal/__init__.py
View File

@ -1,18 +1,11 @@
from .ical import (
CalendarConverter,
EventConverter,
DateDateTime
)
from .ical import CalendarConverter, EventConverter, DateDateTime
from .gcal import (
GoogleCalendarService,
GoogleCalendar,
EventData,
EventList,
EventTuple
EventTuple,
)
from .sync import (
CalendarSync
)
from .sync import CalendarSync

228
sync_ics2gcal/gcal.py
View File

@ -7,7 +7,7 @@ from google.oauth2 import service_account
from googleapiclient import discovery
from pytz import utc
EventData = Dict[str, Union[str, 'EventData', None]]
EventData = Dict[str, Union[str, "EventData", None]]
EventList = List[EventData]
EventTuple = Tuple[EventData, EventData]
@ -26,24 +26,25 @@ class GoogleCalendarService:
( 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, _ = google.auth.default(scopes=scopes)
service = discovery.build(
'calendar', 'v3', credentials=credentials, cache_discovery=False)
"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)
"""
"""make service Resource from service account filename (authorize)"""
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)
service_account_file
)
scoped_credentials = credentials.with_scopes(scopes)
service = discovery.build(
'calendar', 'v3', credentials=scoped_credentials,
cache_discovery=False)
"calendar", "v3", credentials=scoped_credentials, cache_discovery=False
)
return service
@staticmethod
@ -58,9 +59,8 @@ class GoogleCalendarService:
-- 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'])
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
@ -77,18 +77,17 @@ def select_event_key(event: EventData) -> Optional[str]:
"""
key = None
if 'iCalUID' in event:
key = 'iCalUID'
elif 'id' in event:
key = 'id'
if "iCalUID" in event:
key = "iCalUID"
elif "id" in event:
key = "id"
return key
class GoogleCalendar:
"""class to interact with calendar on Google
"""
"""class to interact with calendar on Google"""
logger = logging.getLogger('GoogleCalendar')
logger = logging.getLogger("GoogleCalendar")
def __init__(self, service: discovery.Resource, calendar_id: Optional[str]):
self.service: discovery.Resource = service
@ -111,43 +110,51 @@ class GoogleCalendar:
if exception is not None:
self.logger.error(
'failed to %s event with %s: %s, exception: %s',
action, key, event.get(key), str(exception)
"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))
self.logger.info("event %s ok, %s: %s", action, key, event.get(key))
return callback
def list_events_from(self, start: datetime) -> EventList:
""" list events from calendar, where start date >= start
"""
fields = 'nextPageToken,items(id,iCalUID,updated)'
"""list events from calendar, where start date >= start"""
fields = "nextPageToken,items(id,iCalUID,updated)"
events = []
page_token = None
time_min = utc.normalize(start.astimezone(utc)).replace(
tzinfo=None).isoformat() + 'Z'
time_min = (
utc.normalize(start.astimezone(utc)).replace(tzinfo=None).isoformat() + "Z"
)
while True:
response = self.service.events().list(calendarId=self.calendar_id,
pageToken=page_token,
singleEvents=True,
timeMin=time_min,
fields=fields).execute()
if 'items' in response:
events.extend(response['items'])
page_token = response.get('nextPageToken')
response = (
self.service.events()
.list(
calendarId=self.calendar_id,
pageToken=page_token,
singleEvents=True,
timeMin=time_min,
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))
self.logger.info("%d events listed", len(events))
return events
def find_exists(self, events: List) -> Tuple[List[EventTuple], EventList]:
""" find existing events from list, by 'iCalUID' field
"""find existing events from list, by 'iCalUID' field
Arguments:
events {list} -- list of events
@ -157,7 +164,7 @@ class GoogleCalendar:
events_exist - list of tuples: (new_event, exists_event)
"""
fields = 'items(id,iCalUID,updated)'
fields = "items(id,iCalUID,updated)"
events_by_req = []
exists = []
not_found = []
@ -166,14 +173,15 @@ class GoogleCalendar:
found = False
cur_event = events_by_req[int(request_id)]
if exception is None:
found = ([] != response['items'])
found = [] != response["items"]
else:
self.logger.error(
'exception %s, while listing event with UID: %s',
str(exception), cur_event['iCalUID'])
"exception %s, while listing event with UID: %s",
str(exception),
cur_event["iCalUID"],
)
if found:
exists.append(
(cur_event, response['items'][0]))
exists.append((cur_event, response["items"][0]))
else:
not_found.append(events_by_req[int(request_id)])
@ -181,89 +189,102 @@ class GoogleCalendar:
i = 0
for event in events:
events_by_req.append(event)
batch.add(self.service.events().list(calendarId=self.calendar_id,
iCalUID=event['iCalUID'],
showDeleted=True,
fields=fields
),
request_id=str(i)
)
batch.add(
self.service.events().list(
calendarId=self.calendar_id,
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))
self.logger.info("%d events exists, %d not found", len(exists), len(not_found))
return exists, not_found
def insert_events(self, events: EventList):
""" insert list of events
"""insert list of events
Arguments:
events - events list
"""
fields = 'id'
fields = "id"
events_by_req = []
insert_callback = self._make_request_callback('insert', 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.calendar_id, body=event, fields=fields),
request_id=str(i)
batch.add(
self.service.events().insert(
calendarId=self.calendar_id, body=event, fields=fields
),
request_id=str(i),
)
i += 1
batch.execute()
def patch_events(self, event_tuples: List[EventTuple]):
""" patch (update) events
"""patch (update) events
Arguments:
event_tuples -- list of tuples: (new_event, exists_event)
"""
fields = 'id'
fields = "id"
events_by_req = []
patch_callback = self._make_request_callback('patch', 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:
if "id" not in event_old:
continue
events_by_req.append(event_new)
batch.add(self.service.events().patch(
calendarId=self.calendar_id, eventId=event_old['id'],
body=event_new), fields=fields, request_id=str(i))
batch.add(
self.service.events().patch(
calendarId=self.calendar_id, eventId=event_old["id"], body=event_new
),
fields=fields,
request_id=str(i),
)
i += 1
batch.execute()
def update_events(self, event_tuples: List[EventTuple]):
""" update events
"""update events
Arguments:
event_tuples -- list of tuples: (new_event, exists_event)
"""
fields = 'id'
fields = "id"
events_by_req = []
update_callback = self._make_request_callback('update', 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:
if "id" not in event_old:
continue
events_by_req.append(event_new)
batch.add(self.service.events().update(
calendarId=self.calendar_id, eventId=event_old['id'],
body=event_new, fields=fields), request_id=str(i))
batch.add(
self.service.events().update(
calendarId=self.calendar_id,
eventId=event_old["id"],
body=event_new,
fields=fields,
),
request_id=str(i),
)
i += 1
batch.execute()
def delete_events(self, events: EventList):
""" delete events
"""delete events
Arguments:
events -- list of events
@ -271,14 +292,17 @@ class GoogleCalendar:
events_by_req = []
delete_callback = self._make_request_callback('delete', 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.calendar_id,
eventId=event['id']), request_id=str(i))
batch.add(
self.service.events().delete(
calendarId=self.calendar_id, eventId=event["id"]
),
request_id=str(i),
)
i += 1
batch.execute()
@ -295,36 +319,33 @@ class GoogleCalendar:
calendar Resource
"""
calendar = {'summary': summary}
calendar = {"summary": summary}
if time_zone is not None:
calendar['timeZone'] = time_zone
calendar["timeZone"] = time_zone
created_calendar = self.service.calendars().insert(
body=calendar
).execute()
self.calendar_id = created_calendar['id']
created_calendar = self.service.calendars().insert(body=calendar).execute()
self.calendar_id = created_calendar["id"]
return created_calendar
def delete(self):
"""delete calendar
"""
"""delete calendar"""
self.service.calendars().delete(calendarId=self.calendar_id).execute()
def make_public(self):
"""make calendar public
"""
"""make calendar public"""
rule_public = {
'scope': {
'type': 'default',
"scope": {
"type": "default",
},
'role': 'reader'
"role": "reader",
}
return self.service.acl().insert(
calendarId=self.calendar_id,
body=rule_public
).execute()
return (
self.service.acl()
.insert(calendarId=self.calendar_id, body=rule_public)
.execute()
)
def add_owner(self, email: str):
"""add calendar owner by email
@ -334,13 +355,14 @@ class GoogleCalendar:
"""
rule_owner = {
'scope': {
'type': 'user',
'value': email,
"scope": {
"type": "user",
"value": email,
},
'role': 'owner'
"role": "owner",
}
return self.service.acl().insert(
calendarId=self.calendar_id,
body=rule_owner
).execute()
return (
self.service.acl()
.insert(calendarId=self.calendar_id, body=rule_owner)
.execute()
)

94
sync_ics2gcal/ical.py
View File

@ -20,18 +20,15 @@ def format_datetime_utc(value: DateDateTime) -> str:
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 = 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'
return utc.normalize(value.astimezone(utc)).replace(tzinfo=None).isoformat() + "Z"
def gcal_date_or_datetime(value: DateDateTime,
check_value: Optional[DateDateTime] = None) \
-> Dict[str, str]:
def gcal_date_or_datetime(
value: DateDateTime, check_value: Optional[DateDateTime] = None
) -> Dict[str, str]:
"""date or datetime to gcal (start or end dict)
Arguments:
@ -47,12 +44,12 @@ def gcal_date_or_datetime(value: DateDateTime,
result: Dict[str, str] = {}
if isinstance(check_value, datetime.datetime):
result['dateTime'] = format_datetime_utc(value)
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()
result["date"] = value.isoformat()
return result
@ -71,7 +68,7 @@ class EventConverter(Event):
string value
"""
return self.decoded(prop).decode(encoding='utf-8')
return self.decoded(prop).decode(encoding="utf-8")
def _datetime_str_prop(self, prop: str) -> str:
"""utc datetime as string from property
@ -86,7 +83,7 @@ class EventConverter(Event):
return format_datetime_utc(self.decoded(prop))
def _gcal_start(self) -> Dict[str, str]:
""" event start dict from icalendar event
"""event start dict from icalendar event
Raises:
ValueError -- if DTSTART not date or datetime
@ -95,7 +92,7 @@ class EventConverter(Event):
dict
"""
value = self.decoded('DTSTART')
value = self.decoded("DTSTART")
return gcal_date_or_datetime(value)
def _gcal_end(self) -> Dict[str, str]:
@ -108,22 +105,26 @@ class EventConverter(Event):
"""
result: Dict[str, str]
if 'DTEND' in self:
value = self.decoded('DTEND')
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')
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')
raise ValueError("no DTEND or DURATION")
return result
def _put_to_gcal(self, gcal_event: EventData,
prop: str, func: Callable[[str], str],
ics_prop: Optional[str] = None):
def _put_to_gcal(
self,
gcal_event: EventData,
prop: str,
func: Callable[[str], str],
ics_prop: Optional[str] = None,
):
"""get property from ical event if existed, and put to gcal event
Arguments:
@ -146,54 +147,47 @@ class EventConverter(Event):
"""
event = {
'iCalUID': self._str_prop('UID'),
'start': self._gcal_start(),
'end': self._gcal_end()
"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, "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, 'updated', self._datetime_str_prop, 'LAST-MODIFIED')
self._put_to_gcal(
event,
'transparency',
lambda prop: self._str_prop(prop).lower(), 'TRANSP')
event, "transparency", lambda prop: self._str_prop(prop).lower(), "TRANSP"
)
return event
class CalendarConverter:
"""Convert icalendar events to google calendar resources
"""
"""Convert icalendar events to google calendar resources"""
logger = logging.getLogger('CalendarConverter')
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:
"""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)
self.logger.info("%s loaded", filename)
def loads(self, string: str):
""" load calendar from ics string
"""
"""load calendar from ics string"""
self.calendar = Calendar.from_ical(string)
def events_to_gcal(self) -> EventList:
"""Convert events to google calendar resources
"""
"""Convert events to google calendar resources"""
ics_events = self.calendar.walk(name='VEVENT')
self.logger.info('%d events read', len(ics_events))
ics_events = self.calendar.walk(name="VEVENT")
self.logger.info("%d events read", len(ics_events))
result = list(
map(lambda event: EventConverter(event).to_gcal(), ics_events))
self.logger.info('%d events converted', len(result))
result = list(map(lambda event: EventConverter(event).to_gcal(), ics_events))
self.logger.info("%d events converted", len(result))
return result

78
sync_ics2gcal/manage_calendars.py
View File

@ -10,7 +10,7 @@ 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:
with open(filename, "r", encoding="utf-8") as f:
result = yaml.safe_load(f)
except FileNotFoundError:
pass
@ -19,24 +19,27 @@ def load_config(filename: str) -> Optional[Dict[str, Any]]:
class PropertyCommands:
""" get/set google calendar properties """
"""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
"""get calendar property
Args:
calendar_id: calendar id
property_name: property key
"""
response = self._service.calendarList().get(calendarId=calendar_id,
fields=property_name).execute()
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
"""set calendar property
Args:
calendar_id: calendar id
@ -44,53 +47,60 @@ class PropertyCommands:
property_value: property value
"""
body = {property_name: property_value}
response = self._service.calendarList().patch(body=body, calendarId=calendar_id).execute()
response = (
self._service.calendarList()
.patch(body=body, calendarId=calendar_id)
.execute()
)
print(response)
class Commands:
""" manage google calendars in service account """
"""manage google calendars in service account"""
def __init__(self, config: str = 'config.yml'):
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'])
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
"""list calendars
Args:
show_hidden: show hidden calendars
show_deleted: show deleted calendars
"""
fields: str = 'nextPageToken,items(id,summary)'
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')
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))
print("{summary}: {id}".format_map(calendar))
def create(self, summary: str, timezone: Optional[str] = None, public: bool = False) -> None:
""" create calendar
def create(
self, summary: str, timezone: Optional[str] = None, public: bool = False
) -> None:
"""create calendar
Args:
summary: new calendar summary
@ -101,10 +111,10 @@ class Commands:
calendar.create(summary, timezone)
if public:
calendar.make_public()
print('{}: {}'.format(summary, calendar.calendar_id))
print("{}: {}".format(summary, calendar.calendar_id))
def add_owner(self, calendar_id: str, email: str) -> None:
""" add owner to calendar
"""add owner to calendar
Args:
calendar_id: calendar id
@ -112,33 +122,33 @@ class Commands:
"""
calendar = GoogleCalendar(self._service, calendar_id)
calendar.add_owner(email)
print('to {} added owner: {}'.format(calendar_id, email))
print("to {} added owner: {}".format(calendar_id, email))
def remove(self, calendar_id: str) -> None:
""" remove calendar
"""remove calendar
Args:
calendar_id: calendar id
"""
calendar = GoogleCalendar(self._service, calendar_id)
calendar.delete()
print('removed: {}'.format(calendar_id))
print("removed: {}".format(calendar_id))
def rename(self, calendar_id: str, summary: str) -> None:
""" rename calendar
"""rename calendar
Args:
calendar_id: calendar id
summary:
"""
calendar = {'summary': summary}
calendar = {"summary": summary}
self._service.calendars().patch(body=calendar, calendarId=calendar_id).execute()
print('{}: {}'.format(summary, calendar_id))
print("{}: {}".format(summary, calendar_id))
def main():
fire.Fire(Commands, name='manage-ics2gcal')
fire.Fire(Commands, name="manage-ics2gcal")
if __name__ == '__main__':
if __name__ == "__main__":
main()

85
sync_ics2gcal/sync.py
View File

@ -11,10 +11,9 @@ from .ical import CalendarConverter, DateDateTime
class CalendarSync:
"""class for synchronize calendar with Google
"""
"""class for synchronize calendar with Google"""
logger = logging.getLogger('CalendarSync')
logger = logging.getLogger("CalendarSync")
def __init__(self, gcalendar: GoogleCalendar, converter: CalendarConverter):
self.gcalendar: GoogleCalendar = gcalendar
@ -24,11 +23,10 @@ class CalendarSync:
self.to_delete: EventList = []
@staticmethod
def _events_list_compare(items_src: EventList,
items_dst: EventList,
key: str = 'iCalUID') \
-> Tuple[EventList, List[EventTuple], EventList]:
""" compare list of events by key
def _events_list_compare(
items_src: EventList, items_dst: EventList, key: str = "iCalUID"
) -> Tuple[EventList, List[EventTuple], EventList]:
"""compare list of events by key
Arguments:
items_src {list of dict} -- source events
@ -41,7 +39,8 @@ class CalendarSync:
items_to_delete)
"""
def get_key(item: EventData) -> str: return item[key]
def get_key(item: EventData) -> str:
return item[key]
keys_src: Set[str] = set(map(get_key, items_src))
keys_dst: Set[str] = set(map(get_key, items_dst))
@ -50,9 +49,7 @@ class CalendarSync:
keys_to_update = keys_src & keys_dst
keys_to_delete = keys_dst - keys_src
def items_by_keys(items: EventList,
key_name: str,
keys: Set[str]) -> EventList:
def items_by_keys(items: EventList, key_name: str, keys: Set[str]) -> EventList:
return list(filter(lambda item: item[key_name] in keys, items))
items_to_insert = items_by_keys(items_src, key, keys_to_insert)
@ -67,25 +64,25 @@ class CalendarSync:
return items_to_insert, items_to_update, items_to_delete
def _filter_events_to_update(self):
""" filter 'to_update' events by 'updated' datetime
"""
"""filter 'to_update' events by 'updated' datetime"""
def filter_updated(event_tuple: EventTuple) -> bool:
new, old = event_tuple
if 'updated' not in new or 'updated' not in old:
if "updated" not in new or "updated" not in old:
return True
new_date = dateutil.parser.parse(new['updated'])
old_date = dateutil.parser.parse(old['updated'])
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: EventList,
date: DateDateTime,
op: Callable[[DateDateTime,
DateDateTime], bool]) -> EventList:
""" filter events by start datetime
def _filter_events_by_date(
events: EventList,
date: DateDateTime,
op: Callable[[DateDateTime, DateDateTime], bool],
) -> EventList:
"""filter events by start datetime
Arguments:
events -- events list
@ -98,21 +95,22 @@ class CalendarSync:
def filter_by_date(event: EventData) -> bool:
date_cmp = date
event_start: Dict[str, str] = event['start']
event_start: Dict[str, str] = event["start"]
event_date: Union[DateDateTime, str, None] = None
compare_dates = False
if 'date' in event_start:
event_date = event_start['date']
if "date" in event_start:
event_date = event_start["date"]
compare_dates = True
elif 'dateTime' in event_start:
event_date = event_start['dateTime']
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)
event_date.year, event_date.month, event_date.day
)
return op(event_date, date_cmp)
@ -149,44 +147,47 @@ class CalendarSync:
# divide source events by start datetime
events_src_pending = CalendarSync._filter_events_by_date(
events_src, start_date, operator.ge)
events_src, start_date, operator.ge
)
events_src_past = CalendarSync._filter_events_by_date(
events_src, start_date, operator.lt)
events_src, start_date, operator.lt
)
# first events comparison
self.to_insert, self.to_update, self.to_delete = CalendarSync._events_list_compare(
events_src_pending, events_dst)
(
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)
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)
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 )',
"prepared to sync: ( insert: %d, update: %d, delete: %d )",
len(self.to_insert),
len(self.to_update),
len(self.to_delete)
len(self.to_delete),
)
def clear(self) -> None:
""" clear prepared sync lists (insert, update, delete)
"""
"""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
"""
"""apply sync (insert, update, delete), using prepared lists of events"""
self.gcalendar.insert_events(self.to_insert)
self.gcalendar.update_events(self.to_update)
@ -194,4 +195,4 @@ class CalendarSync:
self.clear()
self.logger.info('sync done')
self.logger.info("sync done")

23
sync_ics2gcal/sync_calendar.py
View File

@ -6,18 +6,13 @@ import dateutil.parser
import datetime
import logging
import logging.config
from . import (
CalendarConverter,
GoogleCalendarService,
GoogleCalendar,
CalendarSync
)
from . import CalendarConverter, GoogleCalendarService, GoogleCalendar, CalendarSync
ConfigDate = Union[str, datetime.datetime]
def load_config() -> Dict[str, Any]:
with open('config.yml', 'r', encoding='utf-8') as f:
with open("config.yml", "r", encoding="utf-8") as f:
result = yaml.safe_load(f)
return result
@ -25,7 +20,7 @@ def load_config() -> Dict[str, Any]:
def get_start_date(date: ConfigDate) -> datetime.datetime:
if isinstance(date, datetime.datetime):
return date
if 'now' == date:
if "now" == date:
result = datetime.datetime.utcnow()
else:
result = dateutil.parser.parse(date)
@ -35,13 +30,13 @@ def get_start_date(date: ConfigDate) -> datetime.datetime:
def main():
config = load_config()
if 'logging' in config:
logging.config.dictConfig(config['logging'])
if "logging" in config:
logging.config.dictConfig(config["logging"])
calendar_id: str = config['calendar']['google_id']
ics_filepath: str = config['calendar']['source']
calendar_id: str = config["calendar"]["google_id"]
ics_filepath: str = config["calendar"]["source"]
start = get_start_date(config['start_from'])
start = get_start_date(config["start_from"])
converter = CalendarConverter()
converter.load(ics_filepath)
@ -54,5 +49,5 @@ def main():
sync.apply()
if __name__ == '__main__':
if __name__ == "__main__":
main()

73
tests/test_converter.py
View File

@ -5,27 +5,45 @@ import pytest
from sync_ics2gcal import CalendarConverter
uid = "UID:uisgtr8tre93wewe0yr8wqy@test.com"
only_start_date = uid + """
only_start_date = (
uid
+ """
DTSTART;VALUE=DATE:20180215
"""
date_val = only_start_date + """
)
date_val = (
only_start_date
+ """
DTEND;VALUE=DATE:20180217
"""
date_duration = only_start_date + """
)
date_duration = (
only_start_date
+ """
DURATION:P2D
"""
datetime_utc_val = uid + """
)
datetime_utc_val = (
uid
+ """
DTSTART;VALUE=DATE-TIME:20180319T092001Z
DTEND:20180321T102501Z
"""
datetime_utc_duration = uid + """
)
datetime_utc_duration = (
uid
+ """
DTSTART;VALUE=DATE-TIME:20180319T092001Z
DURATION:P2DT1H5M
"""
created_updated = date_val + """
)
created_updated = (
date_val
+ """
CREATED:20180320T071155Z
LAST-MODIFIED:20180326T120235Z
"""
)
def ics_test_cal(content: str) -> str:
@ -57,14 +75,29 @@ def test_event_no_end():
converter.events_to_gcal()
@pytest.fixture(params=[
("date", ics_test_event(date_val), '2018-02-15', '2018-02-17'),
("date", ics_test_event(date_duration), '2018-02-15', '2018-02-17'),
("dateTime", ics_test_event(datetime_utc_val),
'2018-03-19T09:20:01.000001Z', '2018-03-21T10:25:01.000001Z'),
("dateTime", ics_test_event(datetime_utc_duration), '2018-03-19T09:20:01.000001Z', '2018-03-21T10:25:01.000001Z')],
ids=['date values', 'date duration',
'datetime utc values', 'datetime utc duration']
@pytest.fixture(
params=[
("date", ics_test_event(date_val), "2018-02-15", "2018-02-17"),
("date", ics_test_event(date_duration), "2018-02-15", "2018-02-17"),
(
"dateTime",
ics_test_event(datetime_utc_val),
"2018-03-19T09:20:01.000001Z",
"2018-03-21T10:25:01.000001Z",
),
(
"dateTime",
ics_test_event(datetime_utc_duration),
"2018-03-19T09:20:01.000001Z",
"2018-03-21T10:25:01.000001Z",
),
],
ids=[
"date values",
"date duration",
"datetime utc values",
"datetime utc duration",
],
)
def param_events_start_end(request):
return request.param
@ -77,12 +110,8 @@ def test_event_start_end(param_events_start_end: Tuple[str, str, str, str]):
events = converter.events_to_gcal()
assert len(events) == 1
event = events[0]
assert event['start'] == {
date_type: start
}
assert event['end'] == {
date_type: end
}
assert event["start"] == {date_type: start}
assert event["end"] == {date_type: end}
def test_event_created_updated():
@ -91,5 +120,5 @@ def test_event_created_updated():
events = converter.events_to_gcal()
assert len(events) == 1
event = events[0]
assert event['created'] == '2018-03-20T07:11:55.000001Z'
assert event['updated'] == '2018-03-26T12:02:35.000001Z'
assert event["created"] == "2018-03-20T07:11:55.000001Z"
assert event["updated"] == "2018-03-26T12:02:35.000001Z"

100
tests/test_sync.py
View File

@ -14,28 +14,30 @@ from sync_ics2gcal import CalendarSync
def sha1(string: Union[str, bytes]) -> str:
if isinstance(string, str):
string = string.encode('utf8')
string = string.encode("utf8")
h = hashlib.sha1()
h.update(string)
return h.hexdigest()
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]]]]:
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)
start_time = datetime.date(start_time.year, start_time.month, start_time.day)
duration: datetime.timedelta = datetime.date(1, 1, 2) - datetime.date(1, 1, 1)
date_key: str = "date"
date_end: str = ''
date_end: str = ""
else:
start_time = utc.normalize(
start_time.astimezone(utc)).replace(tzinfo=None)
duration: datetime.timedelta = datetime.datetime(1, 1, 1, 2) - datetime.datetime(1, 1, 1, 1)
start_time = utc.normalize(start_time.astimezone(utc)).replace(tzinfo=None)
duration: datetime.timedelta = datetime.datetime(
1, 1, 1, 2
) - datetime.datetime(1, 1, 1, 1)
date_key: str = "dateTime"
date_end: str = 'Z'
date_end: str = "Z"
result: List[Dict[str, Union[str, Dict[str, str]]]] = []
for i in range(start, stop):
@ -45,17 +47,18 @@ def gen_events(start: int,
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)
updated.year, updated.month, updated.day, 0, 0, 0, 1, tzinfo=utc
)
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),
"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',
'start': {date_key: event_start.isoformat() + date_end},
'end': {date_key: event_end.isoformat() + date_end}
"updated": updated.isoformat() + "Z",
"created": updated.isoformat() + "Z",
"start": {date_key: event_start.isoformat() + date_end},
"end": {date_key: event_end.isoformat() + date_end},
}
result.append(event)
return result
@ -64,19 +67,21 @@ def gen_events(start: int,
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)})
result.append({"iCalUID": "test{:06d}".format(i)})
return result
def get_start_date(event: Dict[str, Union[str, Dict[str, str]]]) -> Union[datetime.datetime, datetime.date]:
event_start: Dict[str, str] = event['start']
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']
if "date" in event_start:
start_date = event_start["date"]
is_date = True
if 'dateTime' in event_start:
start_date = event_start['dateTime']
if "dateTime" in event_start:
start_date = event_start["dateTime"]
result = dateutil.parser.parse(start_date)
if is_date:
@ -90,8 +95,7 @@ def test_compare():
# [1..2n]
lst_src = gen_list_to_compare(1, 1 + part_len * 2)
# [n..3n]
lst_dst = gen_list_to_compare(
1 + part_len, 1 + part_len * 3)
lst_dst = gen_list_to_compare(1 + part_len, 1 + part_len * 3)
lst_src_rnd = deepcopy(lst_src)
lst_dst_rnd = deepcopy(lst_dst)
@ -99,15 +103,14 @@ def test_compare():
shuffle(lst_src_rnd)
shuffle(lst_dst_rnd)
to_ins, to_upd, to_del = CalendarSync._events_list_compare(
lst_src_rnd, lst_dst_rnd)
to_ins, to_upd, to_del = CalendarSync._events_list_compare(lst_src_rnd, lst_dst_rnd)
assert len(to_ins) == part_len
assert len(to_upd) == part_len
assert len(to_del) == part_len
assert sorted(to_ins, key=lambda x: x['iCalUID']) == lst_src[:part_len]
assert sorted(to_del, key=lambda x: x['iCalUID']) == lst_dst[part_len:]
assert sorted(to_ins, key=lambda x: x["iCalUID"]) == lst_src[:part_len]
assert sorted(to_del, key=lambda x: x["iCalUID"]) == lst_dst[part_len:]
to_upd_ok = list(zip(lst_src[part_len:], lst_dst[:part_len]))
assert len(to_upd) == len(to_upd_ok)
@ -115,35 +118,29 @@ def test_compare():
assert item in to_upd
@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: bool):
msk = timezone('Europe/Moscow')
msk = timezone("Europe/Moscow")
now = utc.localize(datetime.datetime.utcnow())
msk_now = msk.normalize(now.astimezone(msk))
part_len = 5
if no_time:
duration = datetime.date(
1, 1, 2) - datetime.date(1, 1, 1)
duration = datetime.date(1, 1, 2) - datetime.date(1, 1, 1)
else:
duration = datetime.datetime(
1, 1, 1, 2) - datetime.datetime(1, 1, 1, 1)
duration = datetime.datetime(1, 1, 1, 2) - datetime.datetime(1, 1, 1, 1)
date_cmp = msk_now + (duration * part_len)
if no_time:
date_cmp = datetime.date(
date_cmp.year, date_cmp.month, date_cmp.day)
date_cmp = datetime.date(date_cmp.year, date_cmp.month, date_cmp.day)
events = gen_events(
1, 1 + (part_len * 2), msk_now, no_time)
events = gen_events(1, 1 + (part_len * 2), msk_now, no_time)
shuffle(events)
events_pending = CalendarSync._filter_events_by_date(
events, date_cmp, operator.ge)
events_past = CalendarSync._filter_events_by_date(
events, date_cmp, operator.lt)
events_pending = CalendarSync._filter_events_by_date(events, date_cmp, operator.ge)
events_past = CalendarSync._filter_events_by_date(events, date_cmp, operator.lt)
assert len(events_pending) == 1 + part_len
assert len(events_past) == part_len - 1
@ -156,12 +153,11 @@ def test_filter_events_by_date(no_time: bool):
def test_filter_events_to_update():
msk = timezone('Europe/Moscow')
msk = timezone("Europe/Moscow")
now = utc.localize(datetime.datetime.utcnow())
msk_now = msk.normalize(now.astimezone(msk))
one_hour = datetime.datetime(
1, 1, 1, 2) - datetime.datetime(1, 1, 1, 1)
one_hour = datetime.datetime(1, 1, 1, 2) - datetime.datetime(1, 1, 1, 1)
date_upd = msk_now + (one_hour * 5)
count = 10
@ -196,9 +192,9 @@ def test_filter_events_no_updated():
i = 0
for event in events_new:
if 0 == i % 2:
event['updated'] = yesterday.isoformat() + 'Z'
event["updated"] = yesterday.isoformat() + "Z"
else:
del event['updated']
del event["updated"]
i += 1
sync = CalendarSync(None, None)