-
-
Notifications
You must be signed in to change notification settings - Fork 7
/
app.py
123 lines (99 loc) · 4.01 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
#!/usr/bin/python3
import os
from http.server import HTTPServer, BaseHTTPRequestHandler
from socketserver import ThreadingMixIn
from urllib.parse import urlparse, parse_qsl
import requests
PORT = 80
REGION = os.getenv('REGION', 'us').strip().lower()
CHUNKSIZE = int(os.getenv('CHUNK_SIZE', 64 * 1024))
PLAYLIST_URL = 'playlist.m3u'
EPG_URL = 'epg.xml'
STATUS_URL = ''
APP_URL = 'https://i.mjh.nz/SamsungTVPlus/app.json'
class Handler(BaseHTTPRequestHandler):
def __init__(self, *args, **kwargs):
self._params = {}
super().__init__(*args, **kwargs)
def _error(self, message):
self.send_response(500)
self.end_headers()
self.wfile.write(f'Error: {message}'.encode('utf8'))
raise
def do_GET(self):
routes = {
PLAYLIST_URL: self._playlist,
EPG_URL: self._epg,
STATUS_URL: self._status,
}
parsed = urlparse(self.path)
func = parsed.path.split('/')[1]
self._params = dict(parse_qsl(parsed.query, keep_blank_values=True))
if func not in routes:
self.send_response(404)
self.end_headers()
return
try:
routes[func]()
except Exception as e:
self._error(e)
def _playlist(self):
regions = requests.get(APP_URL).json()['regions']
channels = {}
if REGION == 'all':
for key, region in regions.items():
channels.update(region.get('channels', {}))
else:
channels = regions[REGION].get('channels', {})
start_chno = int(self._params['start_chno']) if 'start_chno' in self._params else None
sort = self._params.get('sort', 'chno')
include = [x for x in self._params.get('include', '').split(',') if x]
exclude = [x for x in self._params.get('exclude', '').split(',') if x]
self.send_response(200)
self.end_headers()
self.wfile.write(b'#EXTM3U\n')
for key in sorted(channels.keys(), key=lambda x: channels[x]['chno'] if sort == 'chno' else channels[x]['name'].strip().lower()):
channel = channels[key]
logo = channel['logo']
group = channel['group']
name = channel['name']
url = channel['url']
channel_id = f'samsung-{key}'
# skip widevine channels
if channel.get('license_url'):
continue
if (include and channel_id not in include) or (exclude and channel_id in exclude):
print(f"Skipping {channel_id} due to include / exclude")
continue
chno = ''
if start_chno is not None:
if start_chno > 0:
chno = f' tvg-chno="{start_chno}"'
start_chno += 1
elif channel.get('chno') is not None:
chno = ' tvg-chno="{}"'.format(channel['chno'])
self.wfile.write(f'#EXTINF:-1 channel-id="{channel_id}" tvg-id="{key}" tvg-logo="{logo}" group-title="{group}"{chno},{name}\n{url}\n'.encode('utf8'))
def _epg(self):
self._proxy(f'https://i.mjh.nz/SamsungTVPlus/{REGION}.xml')
def _proxy(self, url):
resp = requests.get(url)
self.send_response(resp.status_code)
self.send_header('content-type', resp.headers.get('content-type'))
self.end_headers()
if resp.ok:
for chunk in resp.iter_content(CHUNKSIZE):
self.wfile.write(chunk)
else:
self.wfile.write(f'{url} returned error {resp.status_code}\nCheck your REGION is correct'.encode('utf8'))
def _status(self):
self.send_response(200)
self.end_headers()
host = self.headers.get('Host')
self.wfile.write(f'Playlist URL: http://{host}/{PLAYLIST_URL}\nEPG URL: http://{host}/{EPG_URL}'.encode('utf8'))
class ThreadingSimpleServer(ThreadingMixIn, HTTPServer):
pass
def run():
server = ThreadingSimpleServer(('0.0.0.0', PORT), Handler)
server.serve_forever()
if __name__ == '__main__':
run()