/
calendar_handler.py
144 lines (121 loc) · 5.46 KB
/
calendar_handler.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
# Import for event dating
import datetime
# Import for preserving and restoring credentials
import pickle
import os
# Imports for authentication
from google_auth_oauthlib.flow import *
# Imports for calendar writing
import google_auth_httplib2
from apiclient import discovery
SCOPES = ["https://www.googleapis.com/auth/calendar"]
class CalendarHandler:
"""
CalendarHandler
Handles authentication and adding events to the calendar.
"""
def __init__(self, client_secrets_file, user_creds_folder, calendar_id, notifier):
"""
Authenticates the user and sets up the credentials for adding events.
Check if there are any stored user credentials, and if so refresh and
use them. Otherwise, prompt the user to authenticate via a web address.
"""
self.calendar_id = calendar_id
# Check if we have a set of pickled credentials
creds_file = os.path.join(user_creds_folder, "credentials.pkl")
try:
with open(creds_file, "rb") as pkl_file:
credentials = pickle.load(pkl_file)
except FileNotFoundError:
os.makedirs(user_creds_folder, exist_ok=True)
credentials = False
if credentials:
# Refresh the credentials to re-validate them, if possible.
request = google.auth.transport.requests.Request()
try:
credentials.refresh(request)
except google.auth.exceptions.RefreshError:
# Couldn't refresh credentials, so invalidate them
credentials = False
if not credentials or not credentials.valid:
# Credentials don't exist, or are invalid, so we need to generate some new ones
notifier.error("Calendar credentials don't exist, or are invalid.")
flow = InstalledAppFlow.\
from_client_secrets_file(client_secrets_file, scopes=SCOPES,
redirect_uri='urn:ietf:wg:oauth:2.0:oob')
credentials = flow.run_console()
self.credentials = credentials
# Pickle the credentials for next time
with open(creds_file, 'wb') as pkl_file:
pickle.dump(credentials, pkl_file)
def _add_event(self, event_details, date=None):
"""Create and add a calendar event, with specified or today's date."""
if not date:
# Retrieve today's date
date = datetime.date.today().isoformat()
event_date = {
"start": {"date": date},
"end": {"date": date},
"transparency": "transparent"
}
# Combine the event details handed to us, with the date we generated
event_data = {**event_date, **event_details}
# Dispatch it to the calendar
service = discovery.build('calendar', 'v3', credentials=self.credentials)
event = service.events().insert(calendarId=self.calendar_id, body=event_data).execute()
return event['htmlLink']
"""
Event Format:
Summary: `shortname - STATUS - issueCount/totalIssues`
Location: `#issueNo`
Description: `title`
"""
def add_shipped(self, subscription_details, scraped_info):
"""Add a 'shipped issue' event to the calendar."""
issue_no = subscription_details["start_from"] + scraped_info["shipped"] - 1
event = {
"summary": (subscription_details["short_name"] + " - SHIPPED - " +
str(scraped_info["shipped"]) + "/" + str(scraped_info["total"])),
"location": "#" + str(issue_no),
"description": scraped_info["title"]
}
return self._add_event(event)
def add_processing(self, subscription_details, scraped_info):
"""Add a 'in-processing issue' event to the calendar."""
processing_count = scraped_info["shipped"] + scraped_info["processing"]
issue_no = subscription_details["start_from"] + processing_count - 1
event = {
"summary": (subscription_details["short_name"] + " - PROCESSING - " +
str(processing_count) + "/" + str(scraped_info["total"])),
"location": "#" + str(issue_no),
"description": scraped_info["title"]
}
return self._add_event(event)
def add_estimated_arrival(self, subscription_details, scraped_info, expected_delivery_time):
"""Add a 'estimated arrival issue' event to the calendar."""
issue_no = subscription_details["start_from"] + scraped_info["shipped"] - 1
event = {
"summary": (subscription_details["short_name"] + " - ETA - " +
str(scraped_info["shipped"]) + "/" + str(scraped_info["total"])),
"location": "#" + str(issue_no),
"description": scraped_info["title"],
"reminders": {
"useDefault": False,
"overrides": [{"method": "popup", "minutes": 900}],
}
}
date = datetime.date.today() + datetime.timedelta(days=expected_delivery_time)
return self._add_event(event, date.isoformat())
class NoOpCalendarHandler(CalendarHandler):
"""
NoOpCalendarHandler
A CalendarHandler that does nothing, for use on dry-runs.
"""
def __init__(self):
pass
def add_estimated_arrival(self, subscription_details, scraped_info, expected_delivery_time):
pass
def add_processing(self, subscription_details, scraped_info):
pass
def add_shipped(self, subscription_details, scraped_info):
pass