1
0
mirror of https://github.com/b4tman/sync_ics2gcal synced 2026-02-04 15:25:04 +00:00

6 Commits

Author SHA1 Message Date
fd04de09e2 Merge branch 'develop' 2022-06-04 17:29:51 +03:00
0d862145a1 Merge branch 'develop' 2022-06-04 17:20:12 +03:00
60b0c839cc Merge branch 'develop' 2021-10-09 16:19:33 +03:00
7ac2de5824 Merge branch 'develop' 2021-05-01 18:30:29 +03:00
06f3776b1a Merge branch 'develop' 2021-05-01 18:11:56 +03:00
feba7ef258 Merge branch 'develop' 2020-02-19 13:21:56 +03:00
29 changed files with 1070 additions and 2516 deletions

1
.github/FUNDING.yml vendored
View File

@@ -1 +0,0 @@
custom: ['https://boosty.to/0xffff']

View File

@@ -4,12 +4,6 @@ updates:
directory: "/" directory: "/"
schedule: schedule:
interval: monthly interval: monthly
time: '02:00' time: '02:00'
timezone: "Europe/Moscow"
day: "saturday"
groups:
"PyPi updates":
patterns:
- "*"
open-pull-requests-limit: 10 open-pull-requests-limit: 10
target-branch: develop target-branch: develop

View File

@@ -16,7 +16,7 @@ jobs:
steps: steps:
- name: Checkout repository - name: Checkout repository
uses: actions/checkout@v4 uses: actions/checkout@v3
with: with:
# We must fetch at least the immediate parents so that if this is # We must fetch at least the immediate parents so that if this is
# a pull request then we can checkout the head. # a pull request then we can checkout the head.
@@ -24,10 +24,10 @@ jobs:
# Initializes the CodeQL tools for scanning. # Initializes the CodeQL tools for scanning.
- name: Initialize CodeQL - name: Initialize CodeQL
uses: github/codeql-action/init@v2 uses: github/codeql-action/init@v1
# Override language selection by uncommenting this and choosing your languages # Override language selection by uncommenting this and choosing your languages
with: with:
languages: python languages: python
- name: Perform CodeQL Analysis - name: Perform CodeQL Analysis
uses: github/codeql-action/analyze@v2 uses: github/codeql-action/analyze@v1

View File

@@ -15,14 +15,14 @@ jobs:
runs-on: ubuntu-latest runs-on: ubuntu-latest
strategy: strategy:
max-parallel: 3 max-parallel: 4
matrix: matrix:
python-version: ['3.11', '3.12', '3.13'] python-version: ['3.7', '3.8', '3.9', '3.10']
steps: steps:
- uses: actions/checkout@v4 - uses: actions/checkout@v3
- name: Set up Python ${{ matrix.python-version }} - name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v5 uses: actions/setup-python@v2
with: with:
python-version: ${{ matrix.python-version }} python-version: ${{ matrix.python-version }}
- name: Upgrade pip - name: Upgrade pip
@@ -30,7 +30,7 @@ jobs:
- name: Install Poetry - name: Install Poetry
uses: snok/install-poetry@v1 uses: snok/install-poetry@v1
- name: Install deps - name: Install deps
run: poetry install --with dev run: poetry install
- name: Lint with flake8 - name: Lint with flake8
run: | run: |
# stop the build if there are Python syntax errors or undefined names # stop the build if there are Python syntax errors or undefined names
@@ -39,21 +39,3 @@ jobs:
poetry run flake8 sync_ics2gcal --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics poetry run flake8 sync_ics2gcal --count --exit-zero --max-complexity=10 --max-line-length=127 --statistics
- name: Test with pytest - name: Test with pytest
run: poetry run pytest -v run: poetry run pytest -v
- name: Check type annotations with mypy
run: |
mkdir mypy_report
poetry run mypy --pretty --html-report mypy_report/ .
- name: Check type annotations with mypy strict mode (not failing)
run: |
poetry run mypy --strict --pretty . || true
- name: Check formatting with black
run: poetry run black --check --diff --color .
- name: Upload mypy report
uses: actions/upload-artifact@v4
with:
name: mypy_report-${{ matrix.python-version }}
path: mypy_report/

View File

@@ -2,15 +2,15 @@ name: Upload Python Package
on: on:
release: release:
types: [published] types: [created]
jobs: jobs:
deploy: deploy:
runs-on: ubuntu-latest runs-on: ubuntu-latest
steps: steps:
- uses: actions/checkout@v4 - uses: actions/checkout@v3
- name: Set up Python - name: Set up Python
uses: actions/setup-python@v5 uses: actions/setup-python@v2
with: with:
python-version: '3.x' python-version: '3.x'
- name: Upgrade pip - name: Upgrade pip
@@ -18,7 +18,7 @@ jobs:
- name: Install Poetry - name: Install Poetry
uses: snok/install-poetry@v1 uses: snok/install-poetry@v1
- name: Install deps - name: Install deps
run: poetry install --with dev run: poetry install
- name: Build - name: Build
run: poetry build run: poetry build
- name: Publish - name: Publish

View File

@@ -1,49 +0,0 @@
name: reviewdog
on:
pull_request:
branches:
- master
- develop
jobs:
build:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v5
with:
python-version: 3.x
- name: Upgrade pip
run: python -m pip install --upgrade pip
- name: Install Poetry
uses: snok/install-poetry@v1
- name: Install deps
run: poetry install --with dev
- name: setup mypy
run: |
mkdir tmp_bin/
echo "#!/bin/sh" > tmp_bin/mypy
echo "poetry run mypy \$@" >> tmp_bin/mypy
chmod +x tmp_bin/mypy
echo "$(pwd)/tmp_bin" >> $GITHUB_PATH
- uses: tsuyoshicho/action-mypy@v3
with:
reporter: github-pr-review
level: warning
- name: format with black
run: poetry run black .
- name: suggester / black
uses: reviewdog/action-suggester@v1
with:
tool_name: black

7
.gitignore vendored
View File

@@ -4,15 +4,8 @@ service-account.json
my-test*.ics my-test*.ics
.vscode/ .vscode/
.idea/ .idea/
.venv/
.pytest_cache/
.mypy_cache/
/dist/ /dist/
/*.egg-info/ /*.egg-info/
/build/ /build/
/.eggs/ /.eggs/
venv/ venv/
mypy_report/
tmp_bin/
docs/build/

View File

@@ -1,16 +0,0 @@
version: 2
build:
os: ubuntu-22.04
tools:
python: "3.12"
jobs:
post_create_environment:
- pip install poetry
post_install:
# VIRTUAL_ENV needs to be set manually for now.
# See https://github.com/readthedocs/readthedocs.org/pull/11152/
- VIRTUAL_ENV=$READTHEDOCS_VIRTUALENV_PATH poetry install --with docs
sphinx:
configuration: docs/source/conf.py

View File

@@ -2,7 +2,6 @@
[![PyPI version](https://badge.fury.io/py/sync-ics2gcal.svg)](https://badge.fury.io/py/sync-ics2gcal) [![PyPI version](https://badge.fury.io/py/sync-ics2gcal.svg)](https://badge.fury.io/py/sync-ics2gcal)
![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)
[![Documentation Status](https://readthedocs.org/projects/sync-ics2gcal/badge/?version=latest)](https://sync-ics2gcal.readthedocs.io/en/latest/?badge=latest)
Python scripts for sync .ics file with Google calendar Python scripts for sync .ics file with Google calendar
@@ -99,5 +98,3 @@ sync-ics2gcal
## How it works ## How it works
![How it works](how-it-works.png) ![How it works](how-it-works.png)
Documentation is available at [sync-ics2gcal.readthedocs.io](https://sync-ics2gcal.readthedocs.io).

View File

@@ -1,20 +0,0 @@
# Minimal makefile for Sphinx documentation
#
# You can set these variables from the command line, and also
# from the environment for the first two.
SPHINXOPTS ?=
SPHINXBUILD ?= poetry run sphinx-build
SOURCEDIR = source
BUILDDIR = build
# Put it first so that "make" without argument is like "make help".
help:
@$(SPHINXBUILD) -M help "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O)
.PHONY: help Makefile
# Catch-all target: route all unknown targets to Sphinx using the new
# "make mode" option. $(O) is meant as a shortcut for $(SPHINXOPTS).
%: Makefile
@$(SPHINXBUILD) -M $@ "$(SOURCEDIR)" "$(BUILDDIR)" $(SPHINXOPTS) $(O)

View File

@@ -1,35 +0,0 @@
@ECHO OFF
pushd %~dp0
REM Command file for Sphinx documentation
if "%SPHINXBUILD%" == "" (
set SPHINXBUILD=poetry run sphinx-build
)
set SOURCEDIR=source
set BUILDDIR=build
%SPHINXBUILD% >NUL 2>NUL
if errorlevel 9009 (
echo.
echo.The 'sphinx-build' command was not found. Make sure you have Sphinx
echo.installed, then set the SPHINXBUILD environment variable to point
echo.to the full path of the 'sphinx-build' executable. Alternatively you
echo.may add the Sphinx directory to PATH.
echo.
echo.If you don't have Sphinx installed, grab it from
echo.https://www.sphinx-doc.org/
exit /b 1
)
if "%1" == "" goto help
%SPHINXBUILD% -M %1 %SOURCEDIR% %BUILDDIR% %SPHINXOPTS% %O%
goto end
:help
%SPHINXBUILD% -M help %SOURCEDIR% %BUILDDIR% %SPHINXOPTS% %O%
:end
popd

View File

@@ -1,39 +0,0 @@
# Configuration file for the Sphinx documentation builder.
#
# For the full list of built-in configuration values, see the documentation:
# https://www.sphinx-doc.org/en/master/usage/configuration.html
# -- Project information -----------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#project-information
import importlib
from typing import List
project = "sync_ics2gcal"
copyright = "2023, b4tman"
author = "b4tman"
version = importlib.metadata.version("sync_ics2gcal")
release = version
# -- General configuration ---------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#general-configuration
extensions = [
"myst_parser",
"sphinx.ext.autodoc",
"sphinx_design",
"sphinx.ext.viewcode",
"sphinx_copybutton",
"sphinx.ext.githubpages",
]
templates_path = ["_templates"]
exclude_patterns: List[str] = []
source_suffix = [".rst", ".md"]
# -- Options for HTML output -------------------------------------------------
# https://www.sphinx-doc.org/en/master/usage/configuration.html#options-for-html-output
html_theme = "sphinx_rtd_theme"
html_static_path = ["_static"]

Binary file not shown.

Before

Width:  |  Height:  |  Size: 98 KiB

View File

@@ -1,22 +0,0 @@
.. sync_ics2gcal documentation master file, created by
sphinx-quickstart on Sat Aug 20 22:19:59 2023.
You can adapt this file completely to your liking, but it should at least
contain the root `toctree` directive.
Welcome to sync_ics2gcal's documentation!
=========================================
.. toctree::
:maxdepth: 2
:caption: Contents:
readme_link
reference
Indices and tables
==================
* :ref:`genindex`
* :ref:`modindex`
* :ref:`search`

View File

@@ -1,7 +0,0 @@
sync_ics2gcal
=============
.. toctree::
:maxdepth: 4
sync_ics2gcal

View File

@@ -1,3 +0,0 @@
```{include} ../../README.md
```

View File

@@ -1,6 +0,0 @@
-------------
Reference
-------------
.. include:: modules.rst

View File

@@ -1,53 +0,0 @@
sync\_ics2gcal package
======================
Submodules
----------
sync\_ics2gcal.gcal module
--------------------------
.. automodule:: sync_ics2gcal.gcal
:members:
:undoc-members:
:show-inheritance:
sync\_ics2gcal.ical module
--------------------------
.. automodule:: sync_ics2gcal.ical
:members:
:undoc-members:
:show-inheritance:
sync\_ics2gcal.manage\_calendars module
---------------------------------------
.. automodule:: sync_ics2gcal.manage_calendars
:members:
:undoc-members:
:show-inheritance:
sync\_ics2gcal.sync module
--------------------------
.. automodule:: sync_ics2gcal.sync
:members:
:undoc-members:
:show-inheritance:
sync\_ics2gcal.sync\_calendar module
------------------------------------
.. automodule:: sync_ics2gcal.sync_calendar
:members:
:undoc-members:
:show-inheritance:
Module contents
---------------
.. automodule:: sync_ics2gcal
:members:
:undoc-members:
:show-inheritance:

2704
poetry.lock generated

File diff suppressed because it is too large Load Diff

View File

@@ -1,6 +1,6 @@
[tool.poetry] [tool.poetry]
name = "sync_ics2gcal" name = "sync_ics2gcal"
version = "0.1.5" version = "0.1.4"
description = "Sync ics file with Google calendar" description = "Sync ics file with Google calendar"
authors = ["Dmitry Belyaev <b4tm4n@mail.ru>"] authors = ["Dmitry Belyaev <b4tm4n@mail.ru>"]
license = "MIT" license = "MIT"
@@ -11,42 +11,25 @@ keywords = ["icalendar", "sync", "google", "calendar"]
classifiers = [ classifiers = [
'License :: OSI Approved :: MIT License', 'License :: OSI Approved :: MIT License',
'Operating System :: OS Independent', 'Operating System :: OS Independent',
'Programming Language :: Python :: 3.11', 'Programming Language :: Python :: 3.7',
'Programming Language :: Python :: 3.12', 'Programming Language :: Python :: 3.8',
'Programming Language :: Python :: 3.13', 'Programming Language :: Python :: 3.9',
'Programming Language :: Python :: 3.10',
] ]
[tool.poetry.dependencies] [tool.poetry.dependencies]
python = "^3.11" python = "^3.7"
google-auth = "2.48.0" google-auth = "2.6.6"
google-api-python-client = "2.188.0" google-api-python-client = "2.49.0"
icalendar = "6.3.2" icalendar = "4.0.9"
pytz = "2025.2" pytz = "2022.1"
PyYAML = "6.0.3" PyYAML = "6.0"
fire = "0.7.1" fire = "0.4.0"
[tool.poetry.group.dev] [tool.poetry.dev-dependencies]
optional = true pytest = "^7.1.2"
flake8 = "^4.0.1"
[tool.poetry.group.docs] black = "^22.3.0"
optional = true
[tool.poetry.group.dev.dependencies]
pytest = ">=8.1,<10.0"
flake8 = ">=7.0.4,<8.0.0"
black = ">=25.0,<27.0"
mypy = ">=1.16.1"
types-python-dateutil = ">=2.9.0.20250516"
types-pytz = ">=2025.2.0.20250516"
types-PyYAML = "^6.0.12.20250516"
lxml = ">=5.4.0,<7.0.0"
[tool.poetry.group.docs.dependencies]
sphinx = ">=8.2,<9.0"
myst-parser = ">=4,<6"
sphinx-rtd-theme = ">=3.0.2,<4.0.0"
sphinx-copybutton = "^0.5.2"
sphinx-design = ">=0.6,<0.8"
[tool.poetry.scripts] [tool.poetry.scripts]
sync-ics2gcal = "sync_ics2gcal.sync_calendar:main" sync-ics2gcal = "sync_ics2gcal.sync_calendar:main"
@@ -55,12 +38,3 @@ manage-ics2gcal = "sync_ics2gcal.manage_calendars:main"
[build-system] [build-system]
requires = ["poetry-core>=1.0.0"] requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api" build-backend = "poetry.core.masonry.api"
[[tool.mypy.overrides]]
module = [
'icalendar',
'google.*',
'googleapiclient',
'fire'
]
ignore_missing_imports = true

View File

@@ -6,39 +6,6 @@ from .gcal import (
EventData, EventData,
EventList, EventList,
EventTuple, EventTuple,
EventDataKey,
EventDateOrDateTime,
EventDate,
EventDateTime,
EventsSearchResults,
ACLRule,
ACLScope,
CalendarData,
BatchRequestCallback,
) )
from .sync import CalendarSync, ComparedEvents from .sync import CalendarSync
__all__ = [
"ical",
"gcal",
"sync",
"CalendarConverter",
"EventConverter",
"DateDateTime",
"GoogleCalendarService",
"GoogleCalendar",
"EventData",
"EventList",
"EventTuple",
"EventDataKey",
"EventDateOrDateTime",
"EventDate",
"EventDateTime",
"EventsSearchResults",
"ACLRule",
"ACLScope",
"CalendarData",
"CalendarSync",
"ComparedEvents",
]

View File

@@ -1,97 +1,17 @@
import logging import logging
from datetime import datetime from datetime import datetime
from typing import ( from typing import List, Dict, Any, Callable, Tuple, Optional, Union
List,
Dict,
Any,
Callable,
Tuple,
Optional,
Union,
TypedDict,
Literal,
NamedTuple,
)
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
EventData = Dict[str, Union[str, "EventData", None]]
class EventDate(TypedDict, total=False):
date: str
timeZone: str
class EventDateTime(TypedDict, total=False):
dateTime: str
timeZone: str
EventDateOrDateTime = Union[EventDate, EventDateTime]
class ACLScope(TypedDict, total=False):
type: str
value: str
class ACLRule(TypedDict, total=False):
scope: ACLScope
role: str
class CalendarData(TypedDict, total=False):
id: str
summary: str
description: str
timeZone: str
class EventData(TypedDict, total=False):
id: str
summary: str
description: str
start: EventDateOrDateTime
end: EventDateOrDateTime
iCalUID: str
location: str
status: str
created: str
updated: str
sequence: int
transparency: str
visibility: str
EventDataKey = Union[
Literal["id"],
Literal["summary"],
Literal["description"],
Literal["start"],
Literal["end"],
Literal["iCalUID"],
Literal["location"],
Literal["status"],
Literal["created"],
Literal["updated"],
Literal["sequence"],
Literal["transparency"],
Literal["visibility"],
]
EventList = List[EventData] EventList = List[EventData]
EventTuple = Tuple[EventData, EventData] EventTuple = Tuple[EventData, EventData]
class EventsSearchResults(NamedTuple):
exists: List[EventTuple]
new: List[EventData]
BatchRequestCallback = Callable[[str, Any, Optional[Exception]], None]
class GoogleCalendarService: class GoogleCalendarService:
"""class for make google calendar service Resource """class for make google calendar service Resource
@@ -100,7 +20,7 @@ class GoogleCalendarService:
""" """
@staticmethod @staticmethod
def default() -> discovery.Resource: def default():
"""make service Resource from default credentials (authorize) """make service Resource from default credentials (authorize)
( https://developers.google.com/identity/protocols/application-default-credentials ) ( https://developers.google.com/identity/protocols/application-default-credentials )
( https://googleapis.dev/python/google-auth/latest/reference/google.auth.html#google.auth.default ) ( https://googleapis.dev/python/google-auth/latest/reference/google.auth.html#google.auth.default )
@@ -114,7 +34,7 @@ class GoogleCalendarService:
return service return service
@staticmethod @staticmethod
def from_srv_acc_file(service_account_file: str) -> discovery.Resource: 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"]
@@ -128,23 +48,19 @@ class GoogleCalendarService:
return service return service
@staticmethod @staticmethod
def from_config(config: Optional[Dict[str, str]] = None) -> discovery.Resource: def from_config(config: Optional[Dict[str, Optional[str]]] = None):
"""make service Resource from config dict """make service Resource from config dict
Arguments: Arguments:
config -- config with keys:
**config** -- config with keys: (optional) service_account: - service account filename
if key not in dict then default credentials will be used
(optional) service_account: - service account filename ( https://developers.google.com/identity/protocols/application-default-credentials )
if key not in dict then default credentials will be used -- None: default credentials will be used
( https://developers.google.com/identity/protocols/application-default-credentials )
-- **None**: default credentials will be used
""" """
if config is not None and "service_account" in config: if config is not None and "service_account" in config:
service_account_filename: str = config["service_account"] service = GoogleCalendarService.from_srv_acc_file(config["service_account"])
service = GoogleCalendarService.from_srv_acc_file(service_account_filename)
else: else:
service = GoogleCalendarService.default() service = GoogleCalendarService.default()
return service return service
@@ -160,7 +76,7 @@ def select_event_key(event: EventData) -> Optional[str]:
key name or None if no key found key name or None if no key found
""" """
key: Optional[str] = None key = None
if "iCalUID" in event: if "iCalUID" in event:
key = "iCalUID" key = "iCalUID"
elif "id" in event: elif "id" in event:
@@ -175,11 +91,9 @@ class GoogleCalendar:
def __init__(self, service: discovery.Resource, calendar_id: Optional[str]): def __init__(self, service: discovery.Resource, calendar_id: Optional[str]):
self.service: discovery.Resource = service self.service: discovery.Resource = service
self.calendar_id: str = str(calendar_id) self.calendar_id: str = calendar_id
def _make_request_callback( def _make_request_callback(self, action: str, events_by_req: EventList) -> Callable:
self, action: str, events_by_req: EventList
) -> BatchRequestCallback:
"""make callback for log result of batch request """make callback for log result of batch request
Arguments: Arguments:
@@ -190,12 +104,9 @@ class GoogleCalendar:
callback function callback function
""" """
def callback( def callback(request_id, response, exception):
request_id: str, response: Any, exception: Optional[Exception] event = events_by_req[int(request_id)]
) -> None: key = select_event_key(event)
event: EventData = events_by_req[int(request_id)]
event_key: Optional[str] = select_event_key(event)
key: str = event_key if event_key is not None else ""
if exception is not None: if exception is not None:
self.logger.error( self.logger.error(
@@ -206,7 +117,7 @@ class GoogleCalendar:
str(exception), str(exception),
) )
else: else:
resp_key: Optional[str] = select_event_key(response) resp_key = select_event_key(response)
if resp_key is not None: if resp_key is not None:
event = response event = response
key = resp_key key = resp_key
@@ -216,10 +127,10 @@ class GoogleCalendar:
def list_events_from(self, start: datetime) -> EventList: def list_events_from(self, start: datetime) -> EventList:
"""list events from calendar, where start date >= start""" """list events from calendar, where start date >= start"""
fields: str = "nextPageToken,items(id,iCalUID,updated)" fields = "nextPageToken,items(id,iCalUID,updated)"
events: EventList = [] events = []
page_token: Optional[str] = None page_token = None
time_min: str = ( time_min = (
utc.normalize(start.astimezone(utc)).replace(tzinfo=None).isoformat() + "Z" utc.normalize(start.astimezone(utc)).replace(tzinfo=None).isoformat() + "Z"
) )
while True: while True:
@@ -242,27 +153,25 @@ class GoogleCalendar:
self.logger.info("%d events listed", len(events)) self.logger.info("%d events listed", len(events))
return events return events
def find_exists(self, events: EventList) -> EventsSearchResults: 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: Arguments:
events {list} -- list of events events {list} -- list of events
Returns: Returns:
EventsSearchResults -- (events_exist, events_not_found) tuple -- (events_exist, events_not_found)
events_exist - list of tuples: (new_event, exists_event) events_exist - list of tuples: (new_event, exists_event)
""" """
fields: str = "items(id,iCalUID,updated)" fields = "items(id,iCalUID,updated)"
events_by_req: EventList = [] events_by_req = []
exists: List[EventTuple] = [] exists = []
not_found: EventList = [] not_found = []
def list_callback( def list_callback(request_id, response, exception):
request_id: str, response: Any, exception: Optional[Exception] found = False
) -> None: cur_event = events_by_req[int(request_id)]
found: bool = False
cur_event: EventData = events_by_req[int(request_id)]
if exception is None: if exception is None:
found = [] != response["items"] found = [] != response["items"]
else: else:
@@ -277,7 +186,7 @@ class GoogleCalendar:
not_found.append(events_by_req[int(request_id)]) not_found.append(events_by_req[int(request_id)])
batch = self.service.new_batch_http_request(callback=list_callback) batch = self.service.new_batch_http_request(callback=list_callback)
i: int = 0 i = 0
for event in events: for event in events:
events_by_req.append(event) events_by_req.append(event)
batch.add( batch.add(
@@ -292,21 +201,21 @@ class GoogleCalendar:
i += 1 i += 1
batch.execute() 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 EventsSearchResults(exists, not_found) return exists, not_found
def insert_events(self, events: EventList) -> None: def insert_events(self, events: EventList):
"""insert list of events """insert list of events
Arguments: Arguments:
events - events list events - events list
""" """
fields: str = "id" fields = "id"
events_by_req: EventList = [] 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) batch = self.service.new_batch_http_request(callback=insert_callback)
i: int = 0 i = 0
for event in events: for event in events:
events_by_req.append(event) events_by_req.append(event)
batch.add( batch.add(
@@ -318,19 +227,19 @@ class GoogleCalendar:
i += 1 i += 1
batch.execute() batch.execute()
def patch_events(self, event_tuples: List[EventTuple]) -> None: def patch_events(self, event_tuples: List[EventTuple]):
"""patch (update) events """patch (update) events
Arguments: Arguments:
event_tuples -- list of tuples: (new_event, exists_event) event_tuples -- list of tuples: (new_event, exists_event)
""" """
fields: str = "id" fields = "id"
events_by_req: EventList = [] 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) batch = self.service.new_batch_http_request(callback=patch_callback)
i: int = 0 i = 0
for event_new, event_old in event_tuples: for event_new, event_old in event_tuples:
if "id" not in event_old: if "id" not in event_old:
continue continue
@@ -345,19 +254,19 @@ class GoogleCalendar:
i += 1 i += 1
batch.execute() batch.execute()
def update_events(self, event_tuples: List[EventTuple]) -> None: def update_events(self, event_tuples: List[EventTuple]):
"""update events """update events
Arguments: Arguments:
event_tuples -- list of tuples: (new_event, exists_event) event_tuples -- list of tuples: (new_event, exists_event)
""" """
fields: str = "id" fields = "id"
events_by_req: EventList = [] 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) batch = self.service.new_batch_http_request(callback=update_callback)
i: int = 0 i = 0
for event_new, event_old in event_tuples: for event_new, event_old in event_tuples:
if "id" not in event_old: if "id" not in event_old:
continue continue
@@ -374,18 +283,18 @@ class GoogleCalendar:
i += 1 i += 1
batch.execute() batch.execute()
def delete_events(self, events: EventList) -> None: def delete_events(self, events: EventList):
"""delete events """delete events
Arguments: Arguments:
events -- list of events events -- list of events
""" """
events_by_req: EventList = [] 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) batch = self.service.new_batch_http_request(callback=delete_callback)
i: int = 0 i = 0
for event in events: for event in events:
events_by_req.append(event) events_by_req.append(event)
batch.add( batch.add(
@@ -410,7 +319,7 @@ class GoogleCalendar:
calendar Resource calendar Resource
""" """
calendar = CalendarData(summary=summary) calendar = {"summary": summary}
if time_zone is not None: if time_zone is not None:
calendar["timeZone"] = time_zone calendar["timeZone"] = time_zone
@@ -418,27 +327,42 @@ class GoogleCalendar:
self.calendar_id = created_calendar["id"] self.calendar_id = created_calendar["id"]
return created_calendar return created_calendar
def delete(self) -> None: def delete(self):
"""delete calendar""" """delete calendar"""
self.service.calendars().delete(calendarId=self.calendar_id).execute() self.service.calendars().delete(calendarId=self.calendar_id).execute()
def make_public(self) -> None: def make_public(self):
"""make calendar public""" """make calendar public"""
rule_public = ACLRule(scope=ACLScope(type="default"), role="reader") rule_public = {
self.service.acl().insert( "scope": {
calendarId=self.calendar_id, body=rule_public "type": "default",
).execute() },
"role": "reader",
}
return (
self.service.acl()
.insert(calendarId=self.calendar_id, body=rule_public)
.execute()
)
def add_owner(self, email: str) -> None: def add_owner(self, email: str):
"""add calendar owner by email """add calendar owner by email
Arguments: Arguments:
email -- email to add email -- email to add
""" """
rule_owner = ACLRule(scope=ACLScope(type="user", value=email), role="owner") rule_owner = {
self.service.acl().insert( "scope": {
calendarId=self.calendar_id, body=rule_owner "type": "user",
).execute() "value": email,
},
"role": "owner",
}
return (
self.service.acl()
.insert(calendarId=self.calendar_id, body=rule_owner)
.execute()
)

View File

@@ -1,18 +1,11 @@
import datetime import datetime
import logging import logging
from typing import Union, Dict, Callable, Optional, Mapping, TypedDict from typing import Union, Dict, Callable, Optional
from icalendar import Calendar, Event from icalendar import Calendar, Event
from pytz import utc from pytz import utc
from .gcal import ( from .gcal import EventData, EventList
EventData,
EventList,
EventDateOrDateTime,
EventDateTime,
EventDate,
EventDataKey,
)
DateDateTime = Union[datetime.date, datetime.datetime] DateDateTime = Union[datetime.date, datetime.datetime]
@@ -35,7 +28,7 @@ def format_datetime_utc(value: DateDateTime) -> str:
def gcal_date_or_datetime( def gcal_date_or_datetime(
value: DateDateTime, check_value: Optional[DateDateTime] = None value: DateDateTime, check_value: Optional[DateDateTime] = None
) -> EventDateOrDateTime: ) -> Dict[str, str]:
"""date or datetime to gcal (start or end dict) """date or datetime to gcal (start or end dict)
Arguments: Arguments:
@@ -49,18 +42,18 @@ def gcal_date_or_datetime(
if check_value is None: if check_value is None:
check_value = value check_value = value
result: EventDateOrDateTime result: Dict[str, str] = {}
if isinstance(check_value, datetime.datetime): if isinstance(check_value, datetime.datetime):
result = EventDateTime(dateTime=format_datetime_utc(value)) result["dateTime"] = format_datetime_utc(value)
else: else:
if isinstance(check_value, datetime.date): if isinstance(check_value, datetime.date):
if isinstance(value, datetime.datetime): if isinstance(value, datetime.datetime):
value = datetime.date(value.year, value.month, value.day) value = datetime.date(value.year, value.month, value.day)
result = EventDate(date=value.isoformat()) result["date"] = value.isoformat()
return result return result
class EventConverter(Event): # type: ignore class EventConverter(Event):
"""Convert icalendar event to google calendar resource """Convert icalendar event to google calendar resource
( https://developers.google.com/calendar/v3/reference/events#resource-representations ) ( https://developers.google.com/calendar/v3/reference/events#resource-representations )
""" """
@@ -75,7 +68,7 @@ class EventConverter(Event): # type: ignore
string value string value
""" """
return str(self.decoded(prop).decode(encoding="utf-8")) return self.decoded(prop).decode(encoding="utf-8")
def _datetime_str_prop(self, prop: str) -> str: def _datetime_str_prop(self, prop: str) -> str:
"""utc datetime as string from property """utc datetime as string from property
@@ -89,7 +82,7 @@ class EventConverter(Event): # type: ignore
return format_datetime_utc(self.decoded(prop)) return format_datetime_utc(self.decoded(prop))
def _gcal_start(self) -> EventDateOrDateTime: def _gcal_start(self) -> Dict[str, str]:
"""event start dict from icalendar event """event start dict from icalendar event
Raises: Raises:
@@ -102,7 +95,7 @@ class EventConverter(Event): # type: ignore
value = self.decoded("DTSTART") value = self.decoded("DTSTART")
return gcal_date_or_datetime(value) return gcal_date_or_datetime(value)
def _gcal_end(self) -> EventDateOrDateTime: def _gcal_end(self) -> Dict[str, str]:
"""event end dict from icalendar event """event end dict from icalendar event
Raises: Raises:
@@ -111,7 +104,7 @@ class EventConverter(Event): # type: ignore
dict dict
""" """
result: EventDateOrDateTime result: Dict[str, str]
if "DTEND" in self: if "DTEND" in self:
value = self.decoded("DTEND") value = self.decoded("DTEND")
result = gcal_date_or_datetime(value) result = gcal_date_or_datetime(value)
@@ -128,10 +121,10 @@ class EventConverter(Event): # type: ignore
def _put_to_gcal( def _put_to_gcal(
self, self,
gcal_event: EventData, gcal_event: EventData,
prop: EventDataKey, prop: str,
func: Callable[[str], str], func: Callable[[str], str],
ics_prop: Optional[str] = None, ics_prop: Optional[str] = None,
) -> None: ):
"""get property from ical event if existed, and put to gcal event """get property from ical event if existed, and put to gcal event
Arguments: Arguments:
@@ -146,18 +139,18 @@ class EventConverter(Event): # type: ignore
if ics_prop in self: if ics_prop in self:
gcal_event[prop] = func(ics_prop) gcal_event[prop] = func(ics_prop)
def convert(self) -> EventData: def to_gcal(self) -> EventData:
"""Convert """Convert
Returns: Returns:
dict - google calendar#event resource dict - google calendar#event resource
""" """
event: EventData = EventData( event = {
iCalUID=self._str_prop("UID"), "iCalUID": self._str_prop("UID"),
start=self._gcal_start(), "start": self._gcal_start(),
end=self._gcal_end(), "end": self._gcal_end(),
) }
self._put_to_gcal(event, "summary", self._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, "description", self._str_prop)
@@ -179,23 +172,22 @@ class CalendarConverter:
def __init__(self, calendar: Optional[Calendar] = None): def __init__(self, calendar: Optional[Calendar] = None):
self.calendar: Optional[Calendar] = calendar self.calendar: Optional[Calendar] = calendar
def load(self, filename: str) -> None: def load(self, filename: str):
"""load calendar from ics file""" """load calendar from ics file"""
with open(filename, "r", encoding="utf-8") as f: with open(filename, "r", encoding="utf-8") as f:
self.calendar = Calendar.from_ical(f.read()) self.calendar = Calendar.from_ical(f.read())
self.logger.info("%s loaded", filename) self.logger.info("%s loaded", filename)
def loads(self, string: str) -> None: def loads(self, string: str):
"""load calendar from ics string""" """load calendar from ics string"""
self.calendar = Calendar.from_ical(string) self.calendar = Calendar.from_ical(string)
def events_to_gcal(self) -> EventList: def events_to_gcal(self) -> EventList:
"""Convert events to google calendar resources""" """Convert events to google calendar resources"""
calendar: Calendar = self.calendar ics_events = self.calendar.walk(name="VEVENT")
ics_events = calendar.walk(name="VEVENT")
self.logger.info("%d events read", len(ics_events)) self.logger.info("%d events read", len(ics_events))
result = list(map(lambda event: EventConverter(event).convert(), ics_events)) result = list(map(lambda event: EventConverter(event).to_gcal(), ics_events))
self.logger.info("%d events converted", len(result)) self.logger.info("%d events converted", len(result))
return result return result

View File

@@ -8,7 +8,7 @@ from . import GoogleCalendar, GoogleCalendarService
def load_config(filename: str) -> Optional[Dict[str, Any]]: def load_config(filename: str) -> Optional[Dict[str, Any]]:
result: Optional[Dict[str, Any]] = None result = None
try: try:
with open(filename, "r", encoding="utf-8") as f: with open(filename, "r", encoding="utf-8") as f:
result = yaml.safe_load(f) result = yaml.safe_load(f)
@@ -21,7 +21,7 @@ def load_config(filename: str) -> Optional[Dict[str, Any]]:
class PropertyCommands: class PropertyCommands:
"""get/set google calendar properties""" """get/set google calendar properties"""
def __init__(self, _service: Any) -> None: def __init__(self, _service):
self._service = _service self._service = _service
def get(self, calendar_id: str, property_name: str) -> None: def get(self, calendar_id: str, property_name: str) -> None:
@@ -146,7 +146,7 @@ class Commands:
print("{}: {}".format(summary, calendar_id)) print("{}: {}".format(summary, calendar_id))
def main() -> None: def main():
fire.Fire(Commands, name="manage-ics2gcal") fire.Fire(Commands, name="manage-ics2gcal")

View File

View File

@@ -1,31 +1,15 @@
import datetime import datetime
import logging import logging
import operator import operator
from typing import List, Dict, Set, Tuple, Union, Callable, NamedTuple from typing import List, Dict, Set, Tuple, Union, Callable
import dateutil.parser import dateutil.parser
from pytz import utc from pytz import utc
from .gcal import ( from .gcal import GoogleCalendar, EventData, EventList, EventTuple
GoogleCalendar,
EventData,
EventList,
EventTuple,
EventDataKey,
EventDateOrDateTime,
EventDate,
)
from .ical import CalendarConverter, DateDateTime from .ical import CalendarConverter, DateDateTime
class ComparedEvents(NamedTuple):
"""Compared events"""
added: EventList
changed: List[EventTuple]
deleted: EventList
class CalendarSync: class CalendarSync:
"""class for synchronize calendar with Google""" """class for synchronize calendar with Google"""
@@ -40,8 +24,8 @@ class CalendarSync:
@staticmethod @staticmethod
def _events_list_compare( def _events_list_compare(
items_src: EventList, items_dst: EventList, key: EventDataKey = "iCalUID" items_src: EventList, items_dst: EventList, key: str = "iCalUID"
) -> ComparedEvents: ) -> Tuple[EventList, List[EventTuple], EventList]:
"""compare list of events by key """compare list of events by key
Arguments: Arguments:
@@ -50,11 +34,13 @@ class CalendarSync:
key {str} -- name of key to compare (default: {'iCalUID'}) key {str} -- name of key to compare (default: {'iCalUID'})
Returns: Returns:
ComparedEvents -- (added, changed, deleted) tuple -- (items_to_insert,
items_to_update,
items_to_delete)
""" """
def get_key(item: EventData) -> str: def get_key(item: EventData) -> str:
return str(item[key]) return item[key]
keys_src: Set[str] = set(map(get_key, items_src)) keys_src: Set[str] = set(map(get_key, items_src))
keys_dst: Set[str] = set(map(get_key, items_dst)) keys_dst: Set[str] = set(map(get_key, items_dst))
@@ -63,21 +49,21 @@ class CalendarSync:
keys_to_update = keys_src & keys_dst keys_to_update = keys_src & keys_dst
keys_to_delete = keys_dst - keys_src keys_to_delete = keys_dst - keys_src
def items_by_keys(items: EventList, keys: Set[str]) -> EventList: def items_by_keys(items: EventList, key_name: str, keys: Set[str]) -> EventList:
return list(filter(lambda item: get_key(item) in keys, items)) return list(filter(lambda item: item[key_name] in keys, items))
items_to_insert = items_by_keys(items_src, keys_to_insert) items_to_insert = items_by_keys(items_src, key, keys_to_insert)
items_to_delete = items_by_keys(items_dst, keys_to_delete) items_to_delete = items_by_keys(items_dst, key, keys_to_delete)
to_upd_src = items_by_keys(items_src, keys_to_update) to_upd_src = items_by_keys(items_src, key, keys_to_update)
to_upd_dst = items_by_keys(items_dst, keys_to_update) to_upd_dst = items_by_keys(items_dst, key, keys_to_update)
to_upd_src.sort(key=get_key) to_upd_src.sort(key=get_key)
to_upd_dst.sort(key=get_key) to_upd_dst.sort(key=get_key)
items_to_update = list(zip(to_upd_src, to_upd_dst)) items_to_update = list(zip(to_upd_src, to_upd_dst))
return ComparedEvents(items_to_insert, items_to_update, items_to_delete) return items_to_insert, items_to_update, items_to_delete
def _filter_events_to_update(self) -> None: 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: def filter_updated(event_tuple: EventTuple) -> bool:
@@ -109,17 +95,17 @@ class CalendarSync:
def filter_by_date(event: EventData) -> bool: def filter_by_date(event: EventData) -> bool:
date_cmp = date date_cmp = date
event_start: EventDateOrDateTime = event["start"] event_start: Dict[str, str] = event["start"]
event_date: Union[DateDateTime, str, None] = None event_date: Union[DateDateTime, str, None] = None
compare_dates = False compare_dates = False
if "date" in event_start: if "date" in event_start:
event_date = event_start["date"] # type: ignore event_date = event_start["date"]
compare_dates = True compare_dates = True
elif "dateTime" in event_start: elif "dateTime" in event_start:
event_date = event_start["dateTime"] # type: ignore event_date = event_start["dateTime"]
event_date = dateutil.parser.parse(str(event_date)) event_date = dateutil.parser.parse(event_date)
if compare_dates: if compare_dates:
date_cmp = datetime.date(date.year, date.month, date.day) date_cmp = datetime.date(date.year, date.month, date.day)
event_date = datetime.date( event_date = datetime.date(

View File

@@ -14,20 +14,20 @@ ConfigDate = Union[str, datetime.datetime]
def load_config() -> Dict[str, Any]: 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) result = yaml.safe_load(f)
return result # type: ignore return result
def get_start_date(date: ConfigDate) -> datetime.datetime: def get_start_date(date: ConfigDate) -> datetime.datetime:
if isinstance(date, datetime.datetime): if isinstance(date, datetime.datetime):
return date return date
if "now" == date: if "now" == date:
result = datetime.datetime.now(datetime.UTC) result = datetime.datetime.utcnow()
else: else:
result = dateutil.parser.parse(date) result = dateutil.parser.parse(date)
return result return result
def main() -> None: def main():
config = load_config() config = load_config()
if "logging" in config: if "logging" in config:

View File

@@ -1,5 +1,5 @@
import datetime import datetime
from typing import Tuple, Any from typing import Tuple
import pytest import pytest
from pytz import timezone, utc from pytz import timezone, utc
@@ -8,27 +8,45 @@ from sync_ics2gcal import CalendarConverter
from sync_ics2gcal.ical import format_datetime_utc from sync_ics2gcal.ical import format_datetime_utc
uid = "UID:uisgtr8tre93wewe0yr8wqy@test.com" uid = "UID:uisgtr8tre93wewe0yr8wqy@test.com"
only_start_date = uid + """ only_start_date = (
uid
+ """
DTSTART;VALUE=DATE:20180215 DTSTART;VALUE=DATE:20180215
""" """
date_val = only_start_date + """ )
date_val = (
only_start_date
+ """
DTEND;VALUE=DATE:20180217 DTEND;VALUE=DATE:20180217
""" """
date_duration = only_start_date + """ )
date_duration = (
only_start_date
+ """
DURATION:P2D DURATION:P2D
""" """
datetime_utc_val = uid + """ )
datetime_utc_val = (
uid
+ """
DTSTART;VALUE=DATE-TIME:20180319T092001Z DTSTART;VALUE=DATE-TIME:20180319T092001Z
DTEND:20180321T102501Z DTEND:20180321T102501Z
""" """
datetime_utc_duration = uid + """ )
datetime_utc_duration = (
uid
+ """
DTSTART;VALUE=DATE-TIME:20180319T092001Z DTSTART;VALUE=DATE-TIME:20180319T092001Z
DURATION:P2DT1H5M DURATION:P2DT1H5M
""" """
created_updated = date_val + """ )
created_updated = (
date_val
+ """
CREATED:20180320T071155Z CREATED:20180320T071155Z
LAST-MODIFIED:20180326T120235Z LAST-MODIFIED:20180326T120235Z
""" """
)
def ics_test_cal(content: str) -> str: def ics_test_cal(content: str) -> str:
@@ -39,21 +57,21 @@ 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))
def test_empty_calendar() -> None: def test_empty_calendar():
converter = CalendarConverter() converter = CalendarConverter()
converter.loads(ics_test_cal("")) converter.loads(ics_test_cal(""))
evnts = converter.events_to_gcal() evnts = converter.events_to_gcal()
assert evnts == [] assert evnts == []
def test_empty_event() -> None: def test_empty_event():
converter = CalendarConverter() converter = CalendarConverter()
converter.loads(ics_test_event("")) converter.loads(ics_test_event(""))
with pytest.raises(KeyError): with pytest.raises(KeyError):
converter.events_to_gcal() converter.events_to_gcal()
def test_event_no_end() -> None: def test_event_no_end():
converter = CalendarConverter() converter = CalendarConverter()
converter.loads(ics_test_event(only_start_date)) converter.loads(ics_test_event(only_start_date))
with pytest.raises(ValueError): with pytest.raises(ValueError):
@@ -84,12 +102,12 @@ def test_event_no_end() -> None:
"datetime utc duration", "datetime utc duration",
], ],
) )
def param_events_start_end(request: Any) -> Any: def param_events_start_end(request):
return request.param return request.param
def test_event_start_end(param_events_start_end: Tuple[str, str, str, str]) -> None: 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)
events = converter.events_to_gcal() events = converter.events_to_gcal()
@@ -99,7 +117,7 @@ def test_event_start_end(param_events_start_end: Tuple[str, str, str, str]) -> N
assert event["end"] == {date_type: end} assert event["end"] == {date_type: end}
def test_event_created_updated() -> None: def test_event_created_updated():
converter = CalendarConverter() converter = CalendarConverter()
converter.loads(ics_test_event(created_updated)) converter.loads(ics_test_event(created_updated))
events = converter.events_to_gcal() events = converter.events_to_gcal()
@@ -124,5 +142,5 @@ def test_event_created_updated() -> None:
], ],
ids=["utc", "with timezone", "date"], ids=["utc", "with timezone", "date"],
) )
def test_format_datetime_utc(value: datetime.datetime, expected_str: str) -> None: def test_format_datetime_utc(value: datetime.datetime, expected_str: str):
assert format_datetime_utc(value) == expected_str assert format_datetime_utc(value) == expected_str

View File

@@ -3,93 +3,95 @@ 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, AnyStr 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 sync_ics2gcal import CalendarSync, DateDateTime from sync_ics2gcal import CalendarSync
from sync_ics2gcal.gcal import EventDateOrDateTime, EventData, EventList
def sha1(s: AnyStr) -> str: def sha1(string: Union[str, bytes]) -> str:
if isinstance(string, str):
string = string.encode("utf8")
h = hashlib.sha1() h = hashlib.sha1()
h.update(str(s).encode("utf8") if isinstance(s, str) else s) h.update(string)
return h.hexdigest() return h.hexdigest()
def gen_events( def gen_events(
start: int, start: int,
stop: int, stop: int,
start_time: DateDateTime, start_time: Union[datetime.datetime, datetime.date],
no_time: bool = False, no_time: bool = False,
) -> EventList: ) -> List[Dict[str, Union[str, Dict[str, str]]]]:
duration: datetime.timedelta
date_key: str
date_end: str
if no_time: 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.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"
date_end = "" date_end: str = ""
else: else:
start_time = utc.normalize(start_time.astimezone(utc)).replace(tzinfo=None) # type: ignore start_time = utc.normalize(start_time.astimezone(utc)).replace(tzinfo=None)
duration = datetime.datetime(1, 1, 1, 2) - datetime.datetime(1, 1, 1, 1) duration: datetime.timedelta = datetime.datetime(
date_key = "dateTime" 1, 1, 1, 2
date_end = "Z" ) - datetime.datetime(1, 1, 1, 1)
date_key: str = "dateTime"
date_end: str = "Z"
result: EventList = [] 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: DateDateTime = 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: EventData = { 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}, # type: ignore "start": {date_key: event_start.isoformat() + date_end},
"end": {date_key: event_end.isoformat() + date_end}, # type: ignore "end": {date_key: event_end.isoformat() + date_end},
} }
result.append(event) result.append(event)
return result return result
def gen_list_to_compare(start: int, stop: int) -> EventList: def gen_list_to_compare(start: int, stop: int) -> List[Dict[str, str]]:
result: EventList = [] 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: EventData) -> DateDateTime: def get_start_date(
event_start: EventDateOrDateTime = event["start"] 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 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"] # type: ignore start_date = event_start["date"]
is_date = True is_date = True
if "dateTime" in event_start: if "dateTime" in event_start:
start_date = event_start["dateTime"] # type: ignore start_date = event_start["dateTime"]
result: DateDateTime = dateutil.parser.parse(str(start_date)) result = dateutil.parser.parse(start_date)
if is_date: if is_date:
result = datetime.date(result.year, result.month, result.day) result = datetime.date(result.year, result.month, result.day)
return result return result
def test_compare() -> None: def test_compare():
part_len: int = 20 part_len = 20
# [1..2n] # [1..2n]
lst_src = gen_list_to_compare(1, 1 + part_len * 2) lst_src = gen_list_to_compare(1, 1 + part_len * 2)
# [n..3n] # [n..3n]
@@ -117,9 +119,9 @@ def test_compare() -> None:
@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) -> None: def test_filter_events_by_date(no_time: bool):
msk = timezone("Europe/Moscow") msk = timezone("Europe/Moscow")
now = datetime.datetime.now(datetime.UTC) now = utc.localize(datetime.datetime.utcnow())
msk_now = msk.normalize(now.astimezone(msk)) msk_now = msk.normalize(now.astimezone(msk))
part_len = 5 part_len = 5
@@ -129,7 +131,7 @@ def test_filter_events_by_date(no_time: bool) -> None:
else: 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: DateDateTime = msk_now + (duration * part_len) date_cmp = msk_now + (duration * part_len)
if no_time: 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)
@@ -150,9 +152,9 @@ def test_filter_events_by_date(no_time: bool) -> None:
assert get_start_date(event) < date_cmp assert get_start_date(event) < date_cmp
def test_filter_events_to_update() -> None: def test_filter_events_to_update():
msk = timezone("Europe/Moscow") msk = timezone("Europe/Moscow")
now = datetime.datetime.now(datetime.UTC) now = utc.localize(datetime.datetime.utcnow())
msk_now = msk.normalize(now.astimezone(msk)) 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)
@@ -162,11 +164,11 @@ def test_filter_events_to_update() -> None:
events_old = gen_events(1, 1 + count, msk_now) events_old = gen_events(1, 1 + count, msk_now)
events_new = gen_events(1, 1 + count, date_upd) events_new = gen_events(1, 1 + count, date_upd)
sync1 = CalendarSync(None, None) # type: ignore sync1 = CalendarSync(None, None)
sync1.to_update = list(zip(events_new, events_old)) sync1.to_update = list(zip(events_new, events_old))
sync1._filter_events_to_update() sync1._filter_events_to_update()
sync2 = CalendarSync(None, None) # type: ignore sync2 = CalendarSync(None, None)
sync2.to_update = list(zip(events_old, events_new)) sync2.to_update = list(zip(events_old, events_new))
sync2._filter_events_to_update() sync2._filter_events_to_update()
@@ -174,12 +176,12 @@ def test_filter_events_to_update() -> None:
assert sync2.to_update == [] assert sync2.to_update == []
def test_filter_events_no_updated() -> None: def test_filter_events_no_updated():
""" """
test filtering events that not have 'updated' field test filtering events that not have 'updated' field
such events should always pass the filter such events should always pass the filter
""" """
now = datetime.datetime.now() now = datetime.datetime.utcnow()
yesterday = now - datetime.timedelta(days=-1) yesterday = now - datetime.timedelta(days=-1)
count = 10 count = 10
@@ -195,7 +197,7 @@ def test_filter_events_no_updated() -> None:
del event["updated"] del event["updated"]
i += 1 i += 1
sync = CalendarSync(None, None) # type: ignore sync = CalendarSync(None, None)
sync.to_update = list(zip(events_old, events_new)) sync.to_update = list(zip(events_old, events_new))
sync._filter_events_to_update() sync._filter_events_to_update()
assert len(sync.to_update) == count // 2 assert len(sync.to_update) == count // 2