python.video_converter_v3/app/main_server.py

import asyncio

import websockets
import json
import logging
import shlex

from asyncio import CancelledError
from websockets import InvalidUpgrade, ConnectionClosed, ConnectionClosedError
from aiohttp import web

from app.class_settings import Settings
from app.class_file_path import Path
from app.class_file_convert import Convert
from app.class_media_file_stat import Stat

var_convert_active = False

class Server:
    def __init__(self):
        self.websocket = None
        self.clients = set()
        obj_settings = Settings()
        obj_settings.set_logging()

        self.yaml = obj_settings.yaml
        self.obj_path = Path(self.yaml)
        self.obj_convert = Convert(self, self.yaml, self.obj_path)

    async def start_convert(self):
        global var_convert_active

        try:
            if not var_convert_active and self.yaml['autostart']:
                await self.obj_convert.snake_waiting()
                var_convert_active = True
            else:
                self.obj_convert.snake_update()
        finally:
            self.set_var_convert_active(False)

    async def send_websocket(self, message):
        message_json = json.dumps(message)

        for client in self.clients.copy():
            try:
                await client.send(message_json)
            except websockets.exceptions.ConnectionClosed:
                self.clients.remove(client)

    async def handle_client(self, websocket):
        self.websocket = websocket

        global var_convert_active

        if websocket not in self.clients:
            self.clients.add(websocket)
            await self.send_websocket(self.obj_path.active_path_to_dict())
            await self.send_websocket(self.obj_path.queue_path_to_dict())

        try:
            async for message in websocket:
                data = json.loads(message)

                if data.get("data_path"):
                    self.obj_path.receive_paths(data.get("data_path"))
                    await self.send_websocket(self.obj_path.queue_path_to_dict())
                    await self.start_convert()

                elif data.get("data_command"):
                    if data["data_command"]["cmd"] == "delete":
                        self.obj_path.delete_path(data["data_command"]["id"])
                        await self.send_websocket(self.obj_path.queue_path_to_dict())
                elif data.get("data_message"):
                    logging.info(f"Server hat Empfangen: {data.get('data_message')}")

        except (ConnectionClosedError, ConnectionClosed):
            logging.warning("Client Verbindung geschlossen")
        except InvalidUpgrade:
            logging.warning("Ungültiger Websocket Upgrade versuch")
        except Exception as e:
            logging.error(f"Unerwarteter Fehler {e}")
        finally:
            self.clients.discard(websocket)

    @staticmethod
    def set_var_convert_active(value: bool):
        global var_convert_active

        var_convert_active = value

    async def server_websocket(self):
        self.obj_path.read_paths()
        asyncio.create_task(self.start_convert())

        server = await websockets.serve(self.handle_client, self.yaml['server_ip'], self.yaml['websocket_port'])
        logging.info(f"Websocket Server läuft auf IP: {self.yaml['server_ip']} Port: {self.yaml['websocket_port']}")
        await server.wait_closed()

    # WebServer --------------------------------------------------------------------------------------------------------

    # noinspection PyUnusedLocal
    @staticmethod
    async def handle_index(request):
        return web.FileResponse("./client/index.html")

    # noinspection PyUnusedLocal
    async def handle_ip(self, request):
        url = self.yaml["extern_websocket_url"]
        port = self.yaml["websocket_port"]
        websocket_https = self.yaml["websocket_https"]

        return web.json_response({"extern_websocket_url": url, "websocket_port": port, "websocket_https": websocket_https})

    # noinspection PyUnusedLocal
    @staticmethod
    async def handle_stat(request):
        obj_stat = Stat()

        return web.json_response(obj_stat.read_stat())

    # noinspection PyUnusedLocal
    async def handle_cmd(self, request):
        command = self.obj_convert.active_tasks # deine oben gepostete Funktion

        html = ""

        for obj in command:
            # ffmpeg-Befehl als String (mit Shell-Escapes, falls Leerzeichen etc.)
            cmd_str = " ".join([shlex.quote(arg) for arg in obj.cmd])

            html += f"{cmd_str}<br /><br />"

        return web.Response(text=html, content_type="text/html")

    async def server_http(self):
        app = web.Application()
        app.router.add_get("/", self.handle_index)
        app.router.add_get("/api/ip", self.handle_ip)
        app.router.add_get("/api/stats", self.handle_stat)
        app.router.add_get("/api/cmd", self.handle_cmd)
        app.router.add_static("/client/", path="./client", name="client")
        runner = web.AppRunner(app)
        await runner.setup()
        site = web.TCPSite(runner, "0.0.0.0", self.yaml["webserver_port"])
        await site.start()
        logging.info(f"HTTP Server läuft auf Port {self.yaml['webserver_port']}")

    # Start Server -----------------------------------------------------------------------------------------------------

    async def start_server(self):
        try:
            await asyncio.gather(
                self.server_websocket(),
                self.server_http()
            )
        except CancelledError:
            logging.warning("Server wurde durch Keyboard Interrupt gestoppt.")
        except Exception as e:
            logging.error(f"Unerwarteter Fehler beim Starten des Servers {e}")