-
Notifications
You must be signed in to change notification settings - Fork 2
/
main.py
71 lines (56 loc) · 1.48 KB
/
main.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
#!/usr/bin/env -S python3 -u
import base64
import logging
import secrets
import sqlite3
import threading
import toml
from misc import proxy_setup
from model import execute_wrapper
import misc
from service import app
import service
import bili
import bili.token_refresh
import bili.utils
import model
# init bili utils
bili.init()
# init proxy
proxy_setup.init()
# generate global HMAC key
hmacKey = base64.b64decode(misc.config['oauth_service']['hmac_key'])
if hmacKey == b'':
hmacKey = secrets.token_bytes(64)
service.hmacKey = hmacKey
# connect database
dbCfg = misc.config['database']
dbType = dbCfg['type']
if dbType == 'sqlite3':
model.initDB(sqlite3.connect(dbCfg['path'], check_same_thread=False))
elif dbType == 'mysql':
model.initDB(execute_wrapper.WrappedMysqlConn(
host=dbCfg['host'],
port=dbCfg['port'],
db=dbCfg['db'],
user=dbCfg['user'],
passwd=dbCfg['pswd'],
))
else:
raise ValueError('unsupported database type')
# run message listener
msgThread = threading.Thread(target=bili.msg_handler.mainLoop)
msgThread.daemon = True
msgThread.start()
# enable periodic wakeup
wakerThread = threading.Thread(target=bili.msg_handler.periodicWakeup)
wakerThread.daemon = True
wakerThread.start()
# token auto refresh
refreshThread = threading.Thread(target=bili.token_refresh.autoRefreshLoop)
refreshThread.daemon = True
refreshThread.start()
if __name__ == '__main__':
host = misc.config['service']['host']
port = misc.config['service']['port']
app.run(host=host, port=port)