-
Notifications
You must be signed in to change notification settings - Fork 0
/
api.py
65 lines (52 loc) · 2.29 KB
/
api.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
import requests as r
from collections import defaultdict
class session():
req = {"jsonrpc": "2.0", "id": 1}
sid = None
def __init__(self, server='http://raspisanie.admin.tomsk.ru/api/rpc.php'):
self.server = server
self.sid = self.request("startSession")['sid']
def request(self, method, **params):
if self.sid:
params['sid'] = self.sid
params['ok_id'] = ''
body = dict(self.req, **{"method": method, "params": params})
ans = r.post(self.server, json=body).json()
if 'result' in ans:
return ans['result']
elif 'error' in ans:
if ans['error']['code'] == -33100: # new session
print('New session')
self.__init__(self.server)
return self.request(method, **params)
else:
raise Exception(ans['error'])
else:
raise Exception(ans)
def search_stop(self, query):
result = self.request('getStopsByName', str=query)
stops = defaultdict(list)
for e in result:
e['st_id'] = [e['st_id']]
stops[e['st_title']].append(e)
for stop, obj in stops.items():
info = obj[0]
for e in obj[1:]:
info['st_id'].append(e['st_id'][0])
info.pop('st_lat')
info.pop('st_long')
stops[stop] = info
return list(stops.values())
def get_stop_arrivals(self, stop_id):
return self.request('getStopArrive', st_id=stop_id)
def get_stops_arrivals(self, stops_id):
m = {}
for stop_id in stops_id:
for bus in self.get_stop_arrivals(stop_id):
if not (bus['mr_num'], bus['rl_racetype']) in m:
m[(bus['mr_num'], bus['rl_racetype'])] = {'to': bus['laststation_title'],
'to_eng': bus['laststation_title_en'],
'units': [{'time': bus['tc_arrivetime'], 'inv': bool(int(bus['u_inv']))}]}
else:
m[(bus['mr_num'], bus['rl_racetype'])]['units'].append({'time': bus['tc_arrivetime'], 'inv': bool(int(bus['u_inv']))})
return m