-
Notifications
You must be signed in to change notification settings - Fork 8
/
find_golden_tickets.py
232 lines (177 loc) · 7.79 KB
/
find_golden_tickets.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
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
from datetime import datetime, date
import numpy as np
import os
import pandas as pd
import requests
import seaborn as sns
import sys
import time
from scripts.utils.twitter import post_media, post_media_update_gt, post_quick_check
from scripts.utils.dataframes import update_csv, get_csv
from scripts.utils.github import update_no_app
from scripts.utils.sms import call_sms
from scripts.appointments_ft import get_appointment_data_gt
today = date.today()
TODAYS_DATE_IS = today.strftime("%d/%m/%Y")
SERVICE = "fast track"
IS_PROXY = False
IS_GITHUB_ACTION = True
IS_TWITTER = True
wait_mins = 5
number_of_appointments_classed_as_bulk = 10
if IS_GITHUB_ACTION:
MAIN_URL = os.environ['change_url']
else:
import config.url_credentials as url_credentials
MAIN_URL = url_credentials.change_url
session = requests.Session()
session.headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:103.0) Gecko/20100101 Firefox/103.0',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8',
'Accept-Language': 'en-US,en;q=0.5',
'Accept-Encoding': 'gzip, deflate, br',
}
def run_github_action(id):
"""
Runs the GitHub Action
:param id: <string>
Args:
id: str
The workflow id for GitHub actions
default:
The default datetime
"""
token = os.environ['access_token_github']
url = f"https://api.github.com/repos/mshodge/youshallnotpassport/actions/workflows/{id}/dispatches"
headers = {"Authorization": "bearer " + token}
json = {"ref": "main"}
r = requests.post(url, headers=headers, json=json)
print(r)
def check_if_no_apps_before():
"""
Checks if the bot has already seen a return of no appointments in the table already today
Returns:
no_app_check_date: str
The date checked last
no_appointment_check_result: str
The result checked last
"""
no_appointment_check = requests.get(
"https://raw.githubusercontent.com/mshodge/youshallnotpassport/main/data/no_apps.md").text \
.replace("\n", "").split(" ")
no_appointment_check_date = no_appointment_check[0]
no_appointment_check_result = no_appointment_check[1]
return no_appointment_check_date, no_appointment_check_result
def long_dataframe(wide_df):
"""
Make a long dataframe
Args:
wide_df: pd.dataframe
The pandas dataframe in wide format
Returns:
long_df: pd.dataframe
The pandas dataframe in long format
"""
wide_df["location"] = wide_df.index
long_df = pd.melt(wide_df, id_vars="location")
long_df.columns = ["location", "appt_date", "count"]
timestamp = datetime.now()
timestamp = timestamp.strftime("%d/%m/%Y %H:%M")
long_df["scrape_date"] = timestamp
return long_df
def make_figure(the_df):
"""
Makes a seaborn heatmap figure from appointments dataframe
Args:
the_df: pd.dataframe
The pandas dataframe
"""
days_list = list(range(0, 10))
days_list2 = list(range(10, 28))
the_df[the_df.eq(0)] = np.nan
appointments = sns.heatmap(the_df, annot=True,
cbar=False, cmap="Blues", linewidths=1, linecolor="white",
vmin=0, vmax=30, annot_kws={"fontsize": 8})
appointments.set_title("The number of Fast Track appointments \n\n")
for i in range(len(days_list)):
appointments.text(i + 0.3, -0.1, str(days_list[i]), fontsize=8)
for i in range(len(days_list2)):
appointments.text(i + 10.1, -0.1, str(days_list2[i]), fontsize=8)
appointments.text(10, -0.5, "(Days from Today)", fontsize=10)
appointments.figure.tight_layout()
fig = appointments.get_figure()
fig.savefig("out.png")
def check_diff_in_loc_counts(df):
"""
Checks the difference in the counts of appointments at each office, if an office has 10 or more new appointments
then the bot will flag this to be posted to Twitter
Args:
df: pd.DataFrame>
The pandas dataframe of the latest results
Returns:
locations_added: list
List of offices with new appointments added, is blank if None
"""
df_old = get_csv("data/fast_track_appointments_locations_gt.csv")
df_diff = df_old.copy()
df_diff['count'] = df['count'] - df_old['count']
locations_added = []
for index, row in df_diff.iterrows():
if row['count'] > number_of_appointments_classed_as_bulk:
locations_added.append(row['location'])
return locations_added
def pipeline():
"""
The main function to get the appointment data from the table at the end of the process
Args:
first: Bool
The first run after the service has gone online?
"""
if "check again later" in requests.get("https://www.passportappointment.service.gov.uk/outreach/PublicBooking.ofml").text:
nice_appointments_df = get_appointment_data_gt(IS_GITHUB_ACTION, MAIN_URL)
if nice_appointments_df is None:
time.sleep(60) # wait a minute before calling again
run_github_action("55493219") if IS_GITHUB_ACTION else None
raise Exception(f"Error. Appointments table returned was none. Will try again in one minute.")
elif nice_appointments_df is False:
time.sleep(60) # wait a minute before calling again
run_github_action("55493219") if IS_GITHUB_ACTION else None
print(f"No Appointments were available. Will try again in one minute.")
return None
print(nice_appointments_df)
appointments_per_location = nice_appointments_df.sum(axis=1).to_frame().reset_index()
appointments_per_location.columns = ['location', 'count']
locations_added_checked = check_diff_in_loc_counts(appointments_per_location)
failed = update_csv(appointments_per_location, IS_GITHUB_ACTION,
"data/fast_track_appointments_locations_gt.csv",
"updating fast track appointment location data", replace=True)
if failed:
run_github_action("55493219")
raise Exception(f"Error. Failed to return the GitHub file. Will try again.")
if len(locations_added_checked) == 0:
print(f"No new bulk appointments added. Will check again in {wait_mins} minutes.")
time.sleep(wait_mins * 60) # wait 3 mins before calling again
run_github_action("55493219") if IS_GITHUB_ACTION else None
return None
make_figure(nice_appointments_df)
# Posts a graph if new appointments have been added
if IS_TWITTER and len(locations_added_checked) > 0:
message = post_media_update_gt(IS_PROXY, IS_GITHUB_ACTION, locations_added_checked)
# call_sms(SERVICE.title(), type="app", response=message)
failed = update_no_app(IS_GITHUB_ACTION, TODAYS_DATE_IS, SERVICE, "False")
if failed:
run_github_action("55493219")
raise Exception(f"Error. Failed to post the SMS. Will try again.")
long_appointments_df = long_dataframe(nice_appointments_df)
failed = update_csv(long_appointments_df, IS_GITHUB_ACTION,
"data/fast_track_appointments_gt.csv",
"updating fast track appointment data", replace=False)
time.sleep(wait_mins * 60) # wait 3 mins before calling again
print(f"Successfully found new appointments, will check again in {wait_mins} minutes")
run_github_action("55493219") if IS_GITHUB_ACTION else None
else:
time.sleep(wait_mins * 60)
run_github_action("55493219") if IS_GITHUB_ACTION else None
raise Exception(f"It's online for everyone, will check again in {wait_mins} minutes")
if __name__ == "__main__":
pipeline()