-
Notifications
You must be signed in to change notification settings - Fork 0
/
CountiesLeaderboard.py
270 lines (224 loc) · 10.4 KB
/
CountiesLeaderboard.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
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
import logging
import requests
from concurrent.futures import ThreadPoolExecutor
from bs4 import BeautifulSoup, SoupStrainer
from typing import List, Dict, Tuple, Union, cast
from Sheets import Sheets
from Spreadsheet import Spreadsheet
Player_T = List[Union[str, int]]
# Keys are the counties to be combined, values are new counties
COMBINE_COUNTIES = {
"Tyne and Wear": "Northumberland",
"Clwyd": "North Wales",
"Gwynedd": "North Wales",
"Dyfed": "Central Wales",
"Powys": "Central Wales",
"Ayrshire": "South-West Scotland",
"Dumfries & Galloway": "South-West Scotland",
"Herefordshire": "Herefordshire & Shropshire",
"Shropshire": "Herefordshire & Shropshire",
"Armagh": "Rest of Ulster",
"Down": "Rest of Ulster",
"Tyrone & Fermanagh": "Rest of Ulster",
"Hampshire - *Including the Isle of Wight*": "Hampshire",
"Northamptonshire - *Including Rutland*": "Northamptonshire",
"Central Scotland - *Clackmannanshire, Falkirk and Stirling*": "Central Scotland",
"Grampian - *Aberdeenshire and Moray*": "Grampian",
"Highlands - *Highlands, Western Isles, Orkney & Shetland*": "Highlands",
"Dunbartonshire, Argyll & Bute": "Highlands",
"Renfrewshire - *Renfrewshire and Inverclyde*": "Renfrewshire",
"Tayside - *Angus, Perth, Dundee and Kinross*": "Tayside"
}
TRN_ID = "1-_lyvXBx89qXE8B9hoQaaT0Q1N26A1B7cHMIy6nGdxU"
LEADERBOARD_ID = "1bA_P0JPWqEodCEEI_etqj4pSnYZxkr9fIDkiZhgUWAY"
def main() -> None:
"""Builds the sheet service, collects town data, and leaderboards. Then writes them to a new spreadsheet.
"""
# Creates sheet service and attaches them to the two spreadsheets.
sheets_service = Sheets("creds.json")
trn_spreadsheet = Spreadsheet(TRN_ID, sheets_service)
leaderboard_spreadsheet = Spreadsheet(LEADERBOARD_ID, sheets_service)
# Gets all town links from the TRN spreadsheet and parses them to the correct format.
all_data = trn_spreadsheet.get_ranges_values(
["England!B2:AY48", "Scotland!B2:AD13", "Wales!B2:Z9", "Northern Ireland!B2:AA6", "Other!B2:AY3"],
value_render_option="FORMULA")["valueRanges"]
all_data = [ranges["values"] for ranges in all_data]
all_town_links = generate_town_links(all_data, 3, 6, 2, 3, 3)
country_leaderboards = get_country_leaderboards(all_town_links)
# Writes country leaderboards to BRL Counties Leaderboard spreadsheet with
# the worksheets with the provided names.
write_country_leaderboards(leaderboard_spreadsheet, country_leaderboards,
"England", "Scotland", "Wales", "Northern Ireland", "Other")
def generate_town_links(all_data: List[List[str]], *args: int) -> List[List[Tuple[str, List[str]]]]:
"""Parse trn spreadsheet data to get pure town leaderboard links attached to county names
Parameters
----------
all_data
List of countries, with a list of string representing each row of data (county)
args
List of ints, representing the start column of the links for each row of data within a worksheet.
Returns
-------
all_links
For each country, a list of tuples containing a county name, and a list of links for that county.
"""
all_links = []
for i in range(len(args)):
# Creates a list of tuples. Containing the county name, and the plain town links,
# removing Google Sheets formatting of hyperlinks.
country_town_links = [(county[0], list(map(lambda x: x.split('"')[1], county[args[i]::])))
for county in all_data[i]]
all_links.append(country_town_links)
return all_links
def get_country_leaderboards(all_town_links: List[List[Tuple[str, List[str]]]]) -> List[Dict[str, List[Player_T]]]:
"""Collects all county leaderboards for each county, also combines counties
Parameters
----------
all_town_links
For each country, a list of tuples containing a county name, and a list of town links.
Returns
-------
all_country_leaderboards
List of leaderboards containing dictionaries of each county with a list of players
"""
all_country_leaderboards = []
for country in all_town_links:
# Starts collecting players up to 10 counties at a time
with ThreadPoolExecutor(max_workers=10) as executor:
country_players = dict(executor.map(get_county_players, country))
all_country_leaderboards.append(country_players)
for country_leaderboard in all_country_leaderboards:
# Combines counties together. Remove original county after being combined.
for county in COMBINE_COUNTIES:
if county in country_leaderboard:
if COMBINE_COUNTIES[county] in country_leaderboard:
country_leaderboard[COMBINE_COUNTIES[county]] += country_leaderboard[county]
else:
country_leaderboard[COMBINE_COUNTIES[county]] = country_leaderboard[county]
del country_leaderboard[county]
# Sort all counties by player's mmr, descending order.
for county in country_leaderboard:
country_leaderboard[county].sort(key=lambda x: x[2], reverse=True)
return all_country_leaderboards
def get_county_players(county: Tuple[str, List[str]]) -> Tuple[str, List[Player_T]]:
"""For an individual county, collect all players from each town.
Parameters
----------
county
Tuple of the county name and a list of town links.
Returns
-------
county_name, players
A tuple of the county name along with a list of players from that county.
"""
# Collect players from town links, visiting up to 10 town leaderboards at once
# Yes this is nested multi-threading (didn't know it was possible)
# In theory up to 100 town links can be visited.
# Due to data format reasons I think around 60 gets visited at once max
with ThreadPoolExecutor(max_workers=10) as executor:
town_leaderboards = list(executor.map(get_town_players, county[1]))
players = [player for town in town_leaderboards for player in town]
return county[0], players
def get_town_players(town: str) -> List[Player_T]:
"""From a town, collect all players from the leaderboard.
Parameters
----------
town
A link to the town leaderboard
Returns
-------
town_players
A list of players from the town.
"""
town_players = []
try_count = 3
while try_count > 0:
try:
town_html = requests.get(town).text
leaderboard_tree = BeautifulSoup(town_html, features="lxml", parse_only=SoupStrainer("table"))
players = leaderboard_tree.findAll("tr")[2::]
for player in players:
try:
name = player.contents[3].contents[3].text.replace('"', "'")
link = "https://rocketleague.tracker.network" + player.contents[3].contents[3].attrs["href"]
mmr = int(player.contents[5].contents[1].text.splitlines()[2].replace(",", ""))
town_players.append([name, link, mmr])
except IndexError:
# Only occurs when an advert is in the leaderboard, so skips to next player
continue
except Exception as ex:
logging.exception(ex)
try_count -= 1
town_players = []
else:
try_count = 0
return town_players
def write_country_leaderboards(spreadsheet: Spreadsheet,
all_leaderboards: List[Dict[str, List[Player_T]]],
*args: str) -> None:
"""Writes leaderboard to spreadsheet, preserving format, and resizing columns to fit new data.
Parameters
----------
spreadsheet
The spreadsheet to write the leaderboards to.
all_leaderboards
The leaderboards to be writen to the spreadsheet.
args
The names of the worksheets each leaderboard will go to.
"""
def rowcol_to_a1(row: int, col: int) -> str:
"""Converts row col cell format into the A1 Google Sheets cell format.
Parameters
----------
row
Integer representing the row of a cell.
col
Integer representing the column of a cell.
Returns
-------
label
The A1 representation of the row/col number representation of a cell.
"""
row = int(row)
col = int(col)
div = col
column_label = ''
while div:
(div, mod) = divmod(div, 26)
if mod == 0:
mod = 26
div -= 1
column_label = chr(mod + 64) + column_label
label = '%s%s' % (column_label, row)
return label
all_value_ranges = []
for i in range(len(args)):
sheet_name = args[i]
start_col = 1
# Creates value range objects. Including formatting hyperlinks to player profiles.
# Each county is 3 columns wide.
for counties in sorted(all_leaderboards[i]):
value_range = {}
data = [["Rank", counties, "3's MMR"]]
rank = 1
for player in all_leaderboards[i][counties]:
data.append([rank, '=HYPERLINK("{}", "{}")'.format(player[1], player[0]), player[2]])
rank += 1
range_str = "{}!{}:{}".format(sheet_name, rowcol_to_a1(1, start_col), rowcol_to_a1(rank, start_col + 2))
value_range["range"] = range_str
value_range["values"] = data
all_value_ranges.append(value_range)
start_col += 4
# Casts args which is Tuple[str, ...] to expected List[str]
# No runtime cost, just for ide type checking system
args = cast(List[str], args)
spreadsheet.clear_ranges_values(args)
spreadsheet.update_ranges_values(all_value_ranges)
spreadsheet.autosize_all_columns()
if __name__ == "__main__":
logging.basicConfig(filename="logfile.log", level=logging.WARNING,
format="%(asctime)s %(levelname)-8s %(name)-15s %(message)s")
logging.warning("Program Started.")
main()
logging.warning("Program Finished.")
logging.warning("-" * 64)