62 lines
2.7 KiB
Python
62 lines
2.7 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from datetime import datetime, time
|
||
|
|
from pathlib import Path
|
||
|
|
import json
|
||
|
|
import pytz
|
||
|
|
|
||
|
|
from aiogoogle import Aiogoogle
|
||
|
|
|
||
|
|
from lib.config import load_config
|
||
|
|
|
||
|
|
config = load_config()
|
||
|
|
|
||
|
|
class GoogleCalendarAPI():
|
||
|
|
def __init__(self, creds_path: Path = Path("token.json"), calendar_id = config.google_calendar_id):
|
||
|
|
self._creds_path = creds_path
|
||
|
|
self._calendar_id = calendar_id
|
||
|
|
|
||
|
|
def get_user_creds(self) -> dict:
|
||
|
|
with Path(self._creds_path).open("r", encoding="utf-8") as fp:
|
||
|
|
creds = json.load(fp)
|
||
|
|
output = {"access_token": creds['token'], 'refresh_token': creds['refresh_token']}
|
||
|
|
return output
|
||
|
|
|
||
|
|
def get_client_creds(self) -> dict:
|
||
|
|
"""Taken from an Installed App credentials file"""
|
||
|
|
with Path(self._creds_path).open("r", encoding="utf-8") as fp:
|
||
|
|
creds = json.load(fp)
|
||
|
|
output = {
|
||
|
|
'client_id': creds['client_id'],
|
||
|
|
'client_secret': creds['client_secret'],
|
||
|
|
'scopes': creds['scopes']
|
||
|
|
}
|
||
|
|
return output
|
||
|
|
|
||
|
|
async def get_event(self, event_id) -> dict:
|
||
|
|
async with Aiogoogle(user_creds=self.get_user_creds(), client_creds=self.get_client_creds()) as aiogoogle:
|
||
|
|
api = await aiogoogle.discover("calendar", "v3")
|
||
|
|
result = await aiogoogle.as_user(api.events.get(calendarId=self._calendar_id, eventId=event_id))
|
||
|
|
return result
|
||
|
|
|
||
|
|
async def get_events(self,
|
||
|
|
start_time = datetime.combine(datetime.now(), time.min, tzinfo=pytz.timezone(config.time_zone)).isoformat(),
|
||
|
|
end_time = datetime.combine(datetime.now(), time.max, tzinfo=pytz.timezone(config.time_zone)).isoformat()
|
||
|
|
):
|
||
|
|
async with Aiogoogle(user_creds=self.get_user_creds(), client_creds=self.get_client_creds()) as aiogoogle:
|
||
|
|
api = await aiogoogle.discover("calendar", "v3")
|
||
|
|
result = await aiogoogle.as_user(api.events.list(calendarId = self._calendar_id, timeMin = start_time, timeMax = end_time))
|
||
|
|
return result['items']
|
||
|
|
|
||
|
|
async def create_event(self, name: str, start_time: datetime, end_time: datetime):
|
||
|
|
async with Aiogoogle(user_creds=self.get_user_creds(), client_creds=self.get_client_creds()) as aiogoogle:
|
||
|
|
api = await aiogoogle.discover("calendar", "v3")
|
||
|
|
event = {
|
||
|
|
"summary": "",
|
||
|
|
"start": {},
|
||
|
|
"end": {}
|
||
|
|
}
|
||
|
|
event['summary'] = name
|
||
|
|
event['start']['dateTime'] = start_time.isoformat().strip()
|
||
|
|
event['end']['dateTime'] = end_time.isoformat().strip()
|
||
|
|
await aiogoogle.as_user(api.events.insert(calendarId=self._calendar_id, json=event))
|