-
Notifications
You must be signed in to change notification settings - Fork 0
/
app.py
160 lines (130 loc) · 5.75 KB
/
app.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
#!/usr/bin/env python3
import requests
import json
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed
import threading
from flask import Flask, render_template, make_response
import time, datetime, os, pickle
from dotenv import load_dotenv
load_dotenv()
logging.basicConfig(filename='app.log', filemode='w', format='%(message)s')
app = Flask(__name__)
# Edit this function to choose what appends when new modules are added
def on_new_modules(new_modules):
print('new modules found')
logging.error('new modules found')
try:
headers = {'Access-Token': os.getenv("ACCESS_TOKEN"),
'Content-Type': 'application/json'}
text = 'New modules have been added\n\n'
discord_text = 'New modules have been added\n\n'
for module in new_modules:
public_url = module['url'].replace(os.getenv("AUTOLOGIN"), 'https://intra.epitech.eu')
text += module['title'] + '\n' + module['url'] + '\n'
discord_text += module['title'] + '\n' + public_url + '\n'
requests.post('https://api.pushbullet.com/v2/pushes', json={'body': text, 'type': 'note'},
headers=headers)
requests.post(os.getenv("DISCORD_URL"),json={'username': 'flyingBot', 'avatar.url': "", "content": discord_text})
except:
print('new modules action failed')
logging.error('new modules action failed')
def init():
if not os.path.exists('save/modules.save'):
print('first start may take about 15s please wait...')
refresh_modules(old_dates)
threading.Timer(10, refresh_job).start()
def refresh_job():
thread = threading.Timer(60 * 10, refresh_job)
thread.setDaemon(True)
thread.start()
print("'.'\n<" + datetime.datetime.now().strftime("%H:%M:%S") + '>')
logging.warning("'.'\n:" + datetime.datetime.now().strftime("%H:%M:%S") + '>')
refresh_modules()
def to_time(string):
try:
return time.mktime(datetime.datetime.strptime(string,
"%Y-%m-%d").timetuple())
except:
try:
return time.mktime(datetime.datetime.strptime(string,
"%Y-%m-%d %H:%M:%S").timetuple())
except:
pass
def merge_activities(item, module_data, url):
item['activites'] = list(map(
lambda activity: {'chart_array': [activity['title'], activity['begin'], activity['end']],
'url': url + '/' + activity['codeacti'] + '/project'}, module_data['activites']))
def retrieve_module_data(item):
item['url'] = os.getenv("AUTOLOGIN") + '/module/' + str(item['scolaryear']) + '/' + item['code'] + '/' + item[
'codeinstance']
res = requests.get(item['url'] + '/?format=json')
module_data = res.json()
merge_activities(item, module_data, item['url'])
item['project_start'] = module_data['activites'][0]['start'] if len(module_data['activites']) else item['begin']
item['project_end'] = module_data['activites'][0]['end'] if len(module_data['activites']) else item['end']
if to_time(item['project_start']) > to_time(item['end']):
item['end'] = item['project_end']
item['registered'] = True if module_data['student_registered'] else False
return 'done ' + item['title']
def find_old_module(module, saved_modules):
for i, saved in enumerate(saved_modules):
if saved['id'] == module['id']:
return i
return -1
def finding_additions(modules):
saved_modules = load_modules()
new_modules = []
for module in modules:
index = find_old_module(module, saved_modules)
if index == -1:
module['created'] = time.time()
new_modules.append(module)
else:
module['created'] = saved_modules[index]['created']
if len(new_modules) > 0:
threading.Thread(target=on_new_modules, args=[new_modules]).start()
def old_dates(modules):
for module in modules:
module['created'] = 0
def refresh_modules(postprocess=finding_additions):
print('Refreshing data...')
logging.warning('Refreshing data...')
res = requests.get(os.getenv(
"AUTOLOGIN") + '/course/filter?format=json&preload=1&location%5B%5D=FR&location%5B%5D=FR%2FPAR&course%5B%5D=master%2Fclassic&scolaryear%5B%5D=2020')
modules = res.json()['items']
with ThreadPoolExecutor(max_workers=15) as executor:
futures = map(lambda module: executor.submit(retrieve_module_data, item=module), modules)
for future in as_completed(futures):
future.result()
print('done')
logging.warning('done')
modules.sort(key=lambda module: to_time(module['project_start']))
data = map(lambda module: {'chart_array': [module['title'], module['project_start'], module['end']],
'activities': module['activites'],
'url': module['url'],
'title': module['title'],
'id': module['id'],
'registered': module['registered']}, modules)
data = list(data)
postprocess(data)
with open('save/modules.save', 'wb') as module_save:
pickle.dump(data, module_save)
print('data saved')
logging.warning('data saved')
return data
def load_modules():
with open('save/modules.save', 'rb') as module_save:
return pickle.load(module_save)
@app.route('/modules')
def modules_route():
response = make_response(json.dumps(load_modules()))
response.headers['Content-Type'] = 'application/json'
return response
@app.route('/')
def timeline():
threading.Thread(target=refresh_modules).start()
return render_template('index.html')
init()
if __name__ == "__main__":
app.run(host='0.0.0.0')