add some type info

backend/config.py

@@ -1,19 +1,22 @@
 import json
-import sys
-import pprint
-import os
-import re
-import base64
 import logging
-import distutils.util
-import asyncio
-import tornado.web
-import tornado.routing
-import config
-import stream_providers
-import aiohttp
+import os
+import pprint
+import re
+import sys
 
-providers = {}
+import aiohttp
+import streamlink
+from tornado import template
+
+logging.basicConfig(
+    format="[%(filename)s:%(lineno)d] %(message)s",
+    stream=sys.stdout,
+    level=logging.INFO,
+)
+logger = logging.getLogger(__name__)
+
+providers: dict[str, str] = {}
 providers["nrk"] = "https://tv.nrk.no"
 providers["nrk_web"] = "https://nrk.no"
 providers["svt"] = "https://svtplay.se"
@@ -34,15 +37,19 @@ icecast_server = os.environ.get("ICECAST_SERVER")
 stream_server = os.environ.get("STREAM_SERVER")
 proxy_server = os.environ.get("PROXY_SERVER")
 
-class ProxyElem():
+
+class ProxyElem:
     def __init__(self, proxy, region):
         self.proxy = proxy
         self.region = region
+
     def local(self):
         timeout = aiohttp.ClientTimeout(total=5)
         return aiohttp.ClientSession(timeout=timeout)
+
     def __repr__(self):
         return f"<region=<{str(self.region)}> proxy=<{str(self.proxy)}>>"
+
     async def proxy_url(self, urls):
         clean_urls = []
         for url in urls:
@@ -85,12 +92,14 @@ class ProxyElem():
         else:
             return clean_urls
 
-proxies = {}
+
+proxies: dict[str, list[ProxyElem]] = {}
+
 new_providers = {}
 for key in providers:
-    region_expr = re.compile(f'^{key}_region(_[a-z][a-z])?[0-9]+$', re.IGNORECASE)
+    region_expr = re.compile(f"^{key}_region(_[a-z][a-z])?[0-9]+$", re.IGNORECASE)
     region_matches = list(filter(region_expr.match, os.environ.keys()))
-    proxy_expr = re.compile(f'^{key}_proxy(_[a-z][a-z])?[0-9]+$', re.IGNORECASE)
+    proxy_expr = re.compile(f"^{key}_proxy(_[a-z][a-z])?[0-9]+$", re.IGNORECASE)
     proxy_matches = list(filter(proxy_expr.match, os.environ.keys()))
     proxy_current = []
     proxy_current_keys = set()
@@ -112,10 +121,10 @@ for key in providers:
             country_temp = proxy_country_groups[pos]
             if isinstance(country_temp, str):
                 proxy_country = country_temp.strip("_")
-                proxy_current_keys.add(f'{key}_{proxy_country}')
-        proxy = os.environ.get(match)
-        if proxy is not None:
-            proxy_current.append(proxy)
+                proxy_current_keys.add(f"{key}_{proxy_country}")
+        proxy_link = os.environ.get(match)
+        if proxy_link is not None:
+            proxy_current.append(proxy_link)
             proxy_countries.append(proxy_country)
             if proxy_country is None:
                 proxy_empty = False
@@ -128,10 +137,10 @@ for key in providers:
             country_temp = region_country_groups[pos]
             if isinstance(country_temp, str):
                 region_country = country_temp.strip("_")
-                region_current_keys.add(f'{key}_{region_country}')
-        region = os.environ.get(match)
-        if region is not None:
-            region_current.append(region)
+                region_current_keys.add(f"{key}_{region_country}")
+        region_name = os.environ.get(match)
+        if region_name is not None:
+            region_current.append(region_name)
             region_countries.append(region_country)
             if region_country is None:
                 region_empty = False
@@ -139,31 +148,35 @@ for key in providers:
     for elem in proxy_current_keys:
         proxies[elem] = []
         new_providers[elem] = providers[key]
-    
-    for proxy, region, country in zip(proxy_current, region_current, proxy_countries):
+
+    for proxy_link, region_name, country in zip(proxy_current, region_current, proxy_countries):
         new_key = key
         if country is not None:
-            new_key = f'{key}_{country}'
-        proxies[new_key].append(ProxyElem(proxy, region))
+            new_key = f"{key}_{country}"
+        proxies[new_key].append(ProxyElem(proxy_link, region_name))
 
     for elem in proxy_current_keys:
         if len(proxies[elem]) == 0:
             proxies[elem].append(ProxyElem(None, None))
 
 pprint.pp(proxies)
-
 providers = new_providers
 
-proxy_keys = []
+streamlink_sessions: dict[str, streamlink.session.Streamlink] = {}
+streamlink_default_session = streamlink.Streamlink()
+
+proxy_keys: list[str] = []
 for proxy_provider in proxies.values():
-    for proxy in proxy_provider:
-        if isinstance(proxy, ProxyElem) and isinstance(proxy.proxy, str):
-            proxy_keys.append(proxy.proxy)
-stream_providers.setup(proxy_keys)
+    for proxy_elem in proxy_provider:
+        if isinstance(proxy_elem, ProxyElem) and isinstance(proxy_elem.proxy, str):
+            proxy_keys.append(proxy_elem.proxy)
+            streamlink_sessions[proxy_elem.proxy] = streamlink.Streamlink()
+            streamlink_sessions[proxy_elem.proxy].set_option("http-proxy", "socks5://" + proxy_elem.proxy)
+            streamlink_sessions[proxy_elem.proxy].set_option("https-proxy", "socks5://" + proxy_elem.proxy)
 
 if icecast_server is not None and stream_server is not None:
     try:
-        with open("/app/sources.json", "r") as f:
+        with open("/app/sources.json", "r", encoding="utf-8") as f:
             data = json.loads(f.read())
             playlist = "#EXTM3U\n"
             for key in data:
@@ -176,7 +189,7 @@ if icecast_server is not None and stream_server is not None:
                 else:
                     playlist += f'#EXTINF:0 radio="false", {name}\n'
                     playlist += stream_server + key + "\n"
-    except Exception as e:
+    except OSError as e:
         logger.info(e)
 
 template_html = None
@@ -186,21 +199,19 @@ font_awesome_version = None
 custom_style = None
 favicon = None
 try:
-    with open("/app/index.html", "r") as f:
-        template_html = tornado.template.Template(f.read().strip())
-    with open("/app/script.js", "r") as f:
-        template_script = tornado.template.Template(f.read().strip())
-    with open("/app/version/video.js.txt", "r") as f:
+    with open("/app/index.html", "r", encoding="utf-8") as f:
+        template_html = template.Template(f.read().strip())
+    with open("/app/script.js", "r", encoding="utf-8") as f:
+        template_script = template.Template(f.read().strip())
+    with open("/app/version/video.js.txt", "r", encoding="utf-8") as f:
         videojs_version = f.read().strip()
-    with open("/app/version/chromecast.txt", "r") as f:
+    with open("/app/version/chromecast.txt", "r", encoding="utf-8") as f:
         chromecast_version = f.read().strip()
-    with open("/app/version/font-awesome.txt", "r") as f:
+    with open("/app/version/font-awesome.txt", "r", encoding="utf-8") as f:
         font_awesome_version = f.read().strip()
-    with open("/app/favicon.png", "rb") as f:
-        favicon = f.read()
-    with open("/app/style.css", "r") as f:
+    with open("/app/favicon.png", "rb") as fb:
+        favicon = fb.read()
+    with open("/app/style.css", "r", encoding="utf-8") as f:
         custom_style = f.read()
-except Exception as e:
+except OSError as e:
     logger.info(e)
-
-

backend/stream.py

@@ -11,17 +11,18 @@ import tornado.routing
 import config
 import stream_providers
 import aiohttp
+from typing import Optional
 
 logging.basicConfig(format='[%(filename)s:%(lineno)d] %(message)s', stream=sys.stdout, level=logging.INFO)
 logger = logging.getLogger(__name__)
 class UpstreamHandler():
     def __init__(self):
-        self.provider = None
-        self.raw = False
-        self.valid = False
-        self.proxy = config.ProxyElem(None, None)
-        self.direct = False
-        self.upstream = None
+        self.provider: Optional[str] = None
+        self.raw: bool = False
+        self.valid: bool = False
+        self.proxy: config.ProxyElem = config.ProxyElem(None, None)
+        self.direct: bool = False
+        self.upstream: Optional[str] = None
     async def test_socks(self, proxy):
         if not hasattr(proxy, "proxy") or not isinstance(proxy.proxy, str):
             return (True, config.ProxyElem(None, None))
@@ -78,6 +79,7 @@ class UpstreamHandler():
                         if not future.done():
                             future.cancel()
 
+
 class MainHandler(tornado.web.RequestHandler):
     async def handle_any(self, redir):
         handler = UpstreamHandler()

backend/stream_providers.py

@@ -1,10 +1,6 @@
 import youtube_dl
 
-try:
-    import streamlink
-except Exception as e:
-    print(e)
-
+import streamlink
 import requests
 import asyncio
 import html.parser
@@ -15,21 +11,6 @@ import json
 import re
 import config
 
-try:
-    streamlink_sessions = {}
-    streamlink_default_session = streamlink.Streamlink()
-except Exception as e:
-    print(e)
-
-def setup(proxies):
-    try:
-        for proxy in proxies:
-            streamlink_sessions[proxy] = streamlink.Streamlink()
-            streamlink_sessions[proxy].set_option("http-proxy", "socks5://" + proxy)
-            streamlink_sessions[proxy].set_option("https-proxy", "socks5://" + proxy)
-    except Exception as e:
-        print(e)
-
 class DummyLogger():
     def debug(self, msg):
         pass
@@ -189,9 +170,9 @@ class StreamlinkRunner(StreamProvider):
         try:
             session = None
             if self.proxy is None or self.proxy.proxy is None:
-                session = streamlink_default_session
+                session = config.streamlink_default_session
             else:
-                session = streamlink_sessions.get(self.proxy.proxy)
+                session = config.streamlink_sessions.get(self.proxy.proxy)
             media = session.resolve_url(self.upstream)
             streams = None
             if isinstance(media, tuple):