add type annotations to stream.py

backend/config.py

@@ -114,7 +114,10 @@ for key in providers:
     region_empty = True
 
     for match in proxy_matches:
-        proxy_country_groups = proxy_expr.match(match.lower()).groups()
+        proxy_match = proxy_expr.match(match.lower())
+        if proxy_match is None:
+            continue
+        proxy_country_groups = proxy_match.groups()
         proxy_country = None
         pos = len(proxy_country_groups) - 1
         if pos >= 0:
@@ -130,7 +133,10 @@ for key in providers:
                 proxy_empty = False
 
     for match in region_matches:
-        region_country_groups = region_expr.match(match.lower()).groups()
+        region_match = region_expr.match(match.lower())
+        if region_match is None:
+            continue
+        region_country_groups = region_match.groups()
         region_country = None
         pos = len(region_country_groups) - 1
         if pos >= 0:

backend/stream.py

@@ -1,8 +1,6 @@
 #!/usr/bin/env python3
 import json
 import sys
-import os
-import re
 import base64
 import logging
 import asyncio
@@ -10,11 +8,12 @@ import tornado.web
 import tornado.routing
 import config
 import stream_providers
-import aiohttp
+from typing import Optional, cast, Any
-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: Optional[str] = None
@@ -23,6 +22,7 @@ class UpstreamHandler():
         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))
@@ -35,10 +35,11 @@ class UpstreamHandler():
                 port = splitted[1]
             future = asyncio.open_connection(host=host, port=port)
             await asyncio.wait_for(future, timeout=1)
-        except Exception as e:
+        except Exception:
             return (False, proxy)
         else:
             return (True, proxy)
+
     async def setup(self, handler):
         self.provider = handler.get_query_argument("provider", None)
         raw_str = handler.get_query_argument("raw", None)
@@ -90,9 +91,10 @@ class MainHandler(tornado.web.RequestHandler):
             else:
                 await self.handle_render(handler)
         else:
-            logger.info(f'provider missing {self.request.uri}')
+            logger.info("provider missing %s", self.request.uri)
             self.set_status(404)
             self.write("Stream not found. (provider missing)")
+
     async def get_data(self, handler):
         video_info = None
         meta = None
@@ -119,13 +121,15 @@ class MainHandler(tornado.web.RequestHandler):
             meta = provider_data.meta()
             title = provider_data.title()
         return (video_info, meta, title)
+
     async def handle_raw(self, handler):
-        video_info, meta, title = await self.get_data(handler)
+        video_info = (await self.get_data(handler))[0]
         if video_info is not None:
             self.redirect(url=video_info["upstream"], status=303)
         else:
             self.set_status(404)
             self.write("HTML template missing.")
+
     async def handle_render(self, handler):
         video_info, meta, title = await self.get_data(handler)
         if video_info is not None:
@@ -142,32 +146,51 @@ class MainHandler(tornado.web.RequestHandler):
         else:
             self.set_status(404)
             self.write("HTML template missing.")
+
     async def get(self):
         await self.handle_any(True)
+
     async def head(self):
         await self.handle_any(False)
 
+    def data_received(self, _):
+        pass
+
+
 class PlaylistHandler(tornado.web.RequestHandler):
     def get(self):
         self.set_header("Content-Type", "text/plain; charset=utf-8")
         self.write(config.playlist)
 
+    def data_received(self, _):
+        pass
+
+
 class IconHandler(tornado.web.RequestHandler):
     def get(self):
         self.set_header("Content-Type", "image/png")
         self.write(config.favicon)
 
+    def data_received(self, _):
+        pass
+
+
 class StyleHandler(tornado.web.RequestHandler):
     def get(self):
         self.set_header("Content-Type", "text/css; charset=utf-8")
         self.write(config.custom_style)
+
+    def data_received(self, _):
+        pass
+
+
 try:
-    handlers = []
+    handlers: list[tuple[tornado.routing.Matcher, tornado.web.RequestHandler]] = []
-    handlers.append((tornado.routing.PathMatches("/sources.m3u8"), PlaylistHandler))
+    handlers.append((cast(tornado.routing.Matcher, tornado.routing.PathMatches("/sources.m3u8")), cast(tornado.web.RequestHandler, PlaylistHandler)))
-    handlers.append((tornado.routing.PathMatches("/favicon.ico"), IconHandler))
+    handlers.append((cast(tornado.routing.Matcher, tornado.routing.PathMatches("/favicon.ico")), cast(tornado.web.RequestHandler, IconHandler)))
-    handlers.append((tornado.routing.PathMatches("/style.css"), StyleHandler))
+    handlers.append((cast(tornado.routing.Matcher, tornado.routing.PathMatches("/style.css")), cast(tornado.web.RequestHandler, StyleHandler)))
-    handlers.append((tornado.routing.AnyMatches(), MainHandler))
+    handlers.append((cast(tornado.routing.Matcher, tornado.routing.AnyMatches()), cast(tornado.web.RequestHandler, MainHandler)))
-    app_web = tornado.web.Application(handlers)
+    app_web = tornado.web.Application(cast(Any, handlers))
     app_web.listen(8080)
     tornado.ioloop.IOLoop.current().start()
 except KeyboardInterrupt: