-
Notifications
You must be signed in to change notification settings - Fork 0
/
sessioncontroller.py
107 lines (84 loc) · 3.11 KB
/
sessioncontroller.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
import platform
import json
import requests
from typing import Literal
import os
# Global request Timeout Option in seconds, leaving it here for now but may make it into a setting in the future
TIMEOUT = 5
# TODO: Refactor to read from external file
VERSION = '0.0.1'
CLIENT = 'Jellyfin Music Player'
CLIENTID = '12345'
# Use device name as Jellyfin device name
DEVICE = platform.node()
# Load authentication info from previous session(s)
try:
with open('./data/auth.json', 'r', encoding='utf8') as authFile:
auth = json.load(authFile)
except:
# Load file failed: blank token and IP, start over
serverIp: str = ''
_token: str = ''
else:
# Use the items present to authenticate
if 'serverIp' in auth:
serverIp = auth['serverIp']
if 'token' in auth:
_token = auth['token']
def loginUsername(server: str, username: str, password: str) -> Literal['Success', 'BadCredentials', 'UnknownError', 'InvalidUrl']:
server = server.strip('/')
if not server.startswith('http'):
server = 'http://' + server
try:
res = requests.post(f'{server}/Users/AuthenticateByName', headers=buildHeader(), timeout=TIMEOUT, json={
'Username': username,
'Pw': password
}
)
except requests.exceptions.ConnectTimeout:
return 'UnknownError'
except requests.exceptions.InvalidURL:
return 'InvalidUrl'
# print(res)
match res.status_code:
case 200:
global serverIp, _token
serverIp = server
_token = res.json()['AccessToken']
try:
if not os.path.exists('./data'):
os.mkdir('./data')
with open('./data/auth.json', 'w', encoding='utf8') as authFile:
body = json.dumps({
'serverIp': serverIp,
'token': _token
})
authFile.write(body)
return 'Success'
except:
pass
case 401:
return 'BadCredentials'
case 404:
return 'InvalidUrl'
case _:
return 'UnknownError'
def buildHeader():
headers = {
"Authorization": f'MediaBrowser Client="{CLIENT}", Device="{DEVICE}", DeviceId="{CLIENTID}", Version="{VERSION}"'
}
if _token:
headers["Authorization"] += f', Token="{_token}"'
return headers
def validateCurrentSession():
try:
res = requests.get(f'{serverIp}/Users/Me', headers=buildHeader(), timeout=TIMEOUT)
return res.status_code == 200
except:
return False
def get(endpoint: str, params: dict = None, rjson: dict = None) -> requests.models.Response:
endpoint = endpoint.strip('/')
return requests.get(f'{serverIp}/{endpoint}', params=params, json=rjson, headers=buildHeader(), timeout=TIMEOUT)
def post(endpoint: str, params: dict = None, rjson: dict = None) -> requests.models.Response:
endpoint = endpoint.strip('/')
return requests.post(f'{serverIp}/{endpoint}', params=params, json=rjson, headers=buildHeader(), timeout=TIMEOUT)