WS server template

backend/main.py

@@ -0,0 +1,61 @@
+import asyncio
+import json
+from typing import Any
+
+import secrets
+import time
+
+import websockets
+
+import response
+
+
+# TODO: config
+HOST = '127.0.0.1'
+PORT = 5678
+
+# TODO: make dict with functions creating YoutubeDL for each site instead of this set
+SITES = {'youtube', 'yandex'}
+
+sessions = {}
+
+
+def generate_key() -> str:
+    return hex(time.time_ns()) + secrets.token_hex(2)
+
+
+async def handler(socket: websockets.WebSocketServerProtocol) -> None:
+    data = json.loads(await socket.recv())
+    if 'action' not in data:
+        await socket.send(response.error_field('action'))
+        return
+    match data['action']:
+        case 'init':  # create session
+            if data.get('site') not in SITES:
+                await socket.send(response.error_field('site'))
+                return
+            key = generate_key()
+            sessions[key] = ""  # TODO
+            await socket.send(response.ok_init(key))
+        case 'list':  # list tracks in album
+            if 'url' not in data:
+                await socket.send(response.error_field('url'))
+                return
+            await socket.send(response.ok_playlist(['title 1', 'title 2']))  # TODO
+        case 'download':  # download by URL
+            if 'url' not in data:
+                await socket.send(response.error_field('url'))
+                return
+            # TODO: pass command to the thread started in `init`
+            await socket.send(response.OK)
+        case _:
+            await socket.send(response.error_field('action'))
+
+
+async def main() -> None:
+    async with websockets.serve(handler, HOST, PORT):
+        await asyncio.Future()
+
+
+if __name__ == '__main__':
+    asyncio.run(main())

backend/response.py

@@ -0,0 +1,22 @@
+import json
+
+OK = '{"ok":true}'
+
+def ok_init(key: str) -> str:
+    return json.dumps({
+        "ok": True,
+        "data": key,
+    })
+
+def ok_playlist(items: list[str]) -> str:
+    return json.dumps({
+        "ok": True,
+        "data": items,
+    })
+
+def error_field(field: str) -> str:
+    return json.dumps({
+        "ok": False,
+        "error": "field",
+        "data": field,
+    })