Initial commit.

This commit is contained in:
redxef 2022-09-28 15:52:52 +02:00
commit 3b422450ac
Signed by: redxef
GPG key ID: 7DAC3AA211CBD921
6 changed files with 294 additions and 0 deletions

1
.gitignore vendored Normal file
View file

@ -0,0 +1 @@
__pycache__

11
adapters/__init__.py Normal file
View file

@ -0,0 +1,11 @@
from .google import Google
from .wordpress import Wordpress
from .wordpress import CalendarMetadata
__all__ = [
'Google',
'Wordpress',
'CalendarMetadata',
]

63
adapters/google.py Normal file
View file

@ -0,0 +1,63 @@
import tempfile
import datetime
import os.path
import json
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from googleapiclient.discovery import build
from googleapiclient.errors import HttpError
# If modifying these scopes, delete the file token.json.
SCOPES = ['https://www.googleapis.com/auth/calendar']
class Google():
def __init__(self, calendar_id, *, credentials, token_file):
_, credentials_file = tempfile.mkstemp()
with open(credentials_file, 'w') as fp:
json.dump(credentials, fp)
self.calendar_id = calendar_id
self.credentials_file = credentials_file
self.token_file = token_file
self.credentials = None
def login(self):
if os.path.exists(self.token_file):
self.credentials = Credentials.from_authorized_user_file(self.token_file, SCOPES)
if self.credentials is None or not self.credentials.valid:
if self.credentials is not None \
and self.credentials.expired \
and self.credentials.refresh_token:
self.credentials.refresh(Request())
else:
flow = InstalledAppFlow.from_client_secrets_file(
self.credentials_file,
SCOPES,
)
self.credentials = flow.run_local_server(port=0)
with open(self.token_file, 'w') as token:
token.write(self.credentials.to_json())
def get_events(self, start: datetime.datetime | None=None, until: datetime.timedelta | None=None, limit=None):
until = until if until else datetime.timedelta(days=365)
now = start if start else datetime.datetime.utcnow().astimezone()
now_365 = now + until
try:
service = build('calendar', 'v3', credentials=self.credentials)
events = service.events().list(
calendarId=self.calendar_id,
timeMin=now.isoformat(),
timeMax=now_365.isoformat(),
maxResults=limit,
).execute()
except HttpError:
print('an error occured')
raise
events = events.get('items', [])
return events

9
adapters/utils.py Normal file
View file

@ -0,0 +1,9 @@
import collections
def dict_merge(dct, merge_dct):
for k, v in merge_dct.items():
if (k in dct and isinstance(dct[k], dict) and isinstance(merge_dct[k], dict)): #noqa
dict_merge(dct[k], merge_dct[k])
else:
dct[k] = merge_dct[k]

144
adapters/wordpress.py Normal file
View file

@ -0,0 +1,144 @@
import requests
import urllib.parse
import json
import datetime
from . import utils
class CalendarMetadata():
def __init__(
self,
name: str,
id: int,
translations: dict,
):
self.name = name
self.id = id
self.translations = translations
def to_dict(self):
translations = {
f'calendar_name_translation_{k}': v for k, v in self.translations.items()
}
return {
'calendar_name': self.name,
'calendar_id': self.id,
**translations,
}
class Wordpress():
def __init__(
self,
base_url: str,
*,
calendar_metadata: CalendarMetadata,
credentials: dict,
):
self.base_url = base_url
self.session = requests.Session()
self.credentials = credentials
self.calendar_meadata = calendar_metadata
def login(self):
login_request = self.session.post(
f'{self.base_url}/wp-login.php',
data={
'log': self.credentials['user'],
'pwd': self.credentials['password'],
'wp-submit': 'Anmelden',
'redirect_to': f'{self.base_url}/wp-admin/',
'testcookie': 1,
}
)
return login_request
def _datestr_to_date(self, s, tz=None):
if s.endswith('Z'):
s = s[:-1]
return datetime.datetime.fromisoformat(s).astimezone().date()
def _generate_data_single(self, event, item_id=2):
start = event['start']
end = event['end']
summary = event['summary']
day_increment = datetime.timedelta(days=1)
if 'date' in start:
start = self._datestr_to_date(start['date'])
elif 'dateTime' in start:
start = self._datestr_to_date(start['dateTime'], tz=start['timeZone'])
else:
raise ValueError('Cannot process event')
if 'date' in end:
end = self._datestr_to_date(end['date'])
elif 'dateTime' in end:
end = self._datestr_to_date(end['dateTime'], tz=end['timeZone'])
end += day_increment # if its a time on a day, we want to add one, in order to also include it in the update
else:
raise ValueError('Cannot process event')
flow = start
dictionary = {}
while flow < end:
dictionary.setdefault(flow.year, {})
dictionary[flow.year].setdefault(flow.month, {})
dictionary[flow.year][flow.month].setdefault(flow.day, {})
dictionary[flow.year][flow.month][flow.day] = {
'description': summary,
'legend_item_id': item_id,
}
flow += day_increment
return dictionary
def _fill_empty(self, d, *, start: datetime.datetime, until: datetime.timedelta):
day_increment = datetime.timedelta(days=1)
flow = start
while flow < start + until:
d.setdefault(flow.year, {})
d[flow.year].setdefault(flow.month, {})
d[flow.year][flow.month].setdefault(flow.day, {})
d[flow.year][flow.month][flow.day].setdefault('legend_item_id', 1)
d[flow.year][flow.month][flow.day].setdefault('description', '')
flow += day_increment
return d
def _generate_data(
self,
events,
start: datetime.datetime | None=None,
until: datetime.timedelta | None=None
):
until = until if until else datetime.timedelta(days=365)
start = start if start else datetime.datetime.utcnow().astimezone()
final_dict = {}
for event in events:
data = self._generate_data_single(event)
utils.dict_merge(final_dict, data)
final_dict = self._fill_empty(
final_dict,
start=start,
until=until,
)
return final_dict
def post_events(self, events):
metadata = self.calendar_meadata.to_dict()
data = self._generate_data(events)
update_request = self.session.post(
f'{self.base_url}/wp-admin/admin-ajax.php',
auth=(self.credentials['user'], self.credentials['password']),
data={
'action': 'wpbs_save_calendar_data',
'form_data': urllib.parse.urlencode(metadata),
'calendar_data': json.dumps(data),
},
)
return update_request

66
main.py Executable file
View file

@ -0,0 +1,66 @@
#!/usr/bin/env python3
import sys
import yaml
try:
from yaml import CLoader as Loader
except ImportError:
from yaml import Loader
from adapters import *
class Config():
"""
The default configuration.
Keys:
.google.calendar_id: the id of the calendar to sync
.google.credentials: the json of the obtained credentials file
.google.token_file: where to save the login token
.wordpress.url: the base wordpress url
.wordpress.calendar.id: the id of the (wp-booking-system) calendar
.wordpress.calendar.name: the name of the calendar
.wordpress.calendar.translations: a dictionary of language <-> translation pairs (example: {"en": "Reservations"})
.wordpress.credentials.user: the user as which to log into wordpress
.wordpress.credentials.password: the users password
"""
def __init__(self, file):
self.file = file
self.config: dict | None = None
def load(self):
if self.file == '-':
config = yaml.load(sys.stdin, Loader=Loader)
else:
with open(self.file) as fp:
config = yaml.load(fp, Loader=Loader)
self.config = config
def __getitem__(self, name):
assert self.config is not None
return self.config[name]
if __name__ == '__main__':
config = Config('-')
config.load()
g = Google(
config['google']['calendar_id'],
credentials=config['google']['credentials'],
token_file=config['google'].get('token_file', '~/.wp-cal-integration-google-token.json')
)
w = Wordpress(
config['wordpress']['url'],
calendar_metadata=CalendarMetadata(
id=config['wordpress']['calendar']['id'],
name=config['wordpress']['calendar']['name'],
translations=config['wordpress']['calendar']['translations'],
),
credentials=config['wordpress']['credentials'],
)
g.login()
events = g.get_events()
w.login()
w.post_events(events)