243 lines
No EOL
7.4 KiB
Python
Executable file
243 lines
No EOL
7.4 KiB
Python
Executable file
import asyncio
|
|
import os.path
|
|
import re
|
|
import subprocess
|
|
import json
|
|
import logging
|
|
import time
|
|
from datetime import date
|
|
|
|
import uvicorn
|
|
from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect
|
|
from fastapi.staticfiles import StaticFiles
|
|
from fastapi.responses import HTMLResponse
|
|
from fastapi.templating import Jinja2Templates
|
|
from starlette.websockets import WebSocketState
|
|
|
|
import video_class as vc
|
|
from ffmpeg_class import FFCmd as Fc
|
|
from config_class import Config as Cfg
|
|
|
|
# Globale Variablen
|
|
config = Cfg().config
|
|
|
|
ffmpeg = Fc(config)
|
|
queue = asyncio.Queue()
|
|
semaphore = asyncio.Semaphore(config["convert"]["max_process"])
|
|
|
|
thread_output = None
|
|
video_files = {}
|
|
date = date.today()
|
|
|
|
# Prozess Objects
|
|
active_process = set()
|
|
active_tasks = set()
|
|
clients = set()
|
|
|
|
if not os.path.exists("./logs"):
|
|
os.mkdir("./logs")
|
|
|
|
logging.basicConfig(
|
|
level=logging.DEBUG,
|
|
format="%(asctime)s - %(levelname)s - %(message)s",
|
|
handlers=[
|
|
logging.StreamHandler(),
|
|
logging.FileHandler(f"./logs/{date}.log")
|
|
]
|
|
)
|
|
|
|
app = FastAPI()
|
|
app.mount("/webs", StaticFiles(directory="app/webs"), name="webs")
|
|
templates = Jinja2Templates(directory="app/templates")
|
|
|
|
def obj_list():
|
|
list_json = []
|
|
for obj_video in video_files.values():
|
|
list_json.append(obj_video.get_vars())
|
|
|
|
return json.dumps({"list_video":list_json})
|
|
|
|
# Media ----------------------------------------------------------------------------------------------------------------
|
|
def get_video_information(media_path):
|
|
pattern = r"(?<=\.mkv\s|\.mp4\s|\.avi\s)|(?<=\.webm\s)"
|
|
|
|
file_paths = media_path.get("files", [])[0]
|
|
var_list_file = re.split(pattern, file_paths)
|
|
|
|
try:
|
|
for source_file in var_list_file:
|
|
source_file = source_file.strip()
|
|
|
|
video_streams, video_format = get_ffprobe("v", source_file)
|
|
audio_streams = get_ffprobe("a",source_file)
|
|
subtitle_streams = get_ffprobe("s", source_file)
|
|
|
|
obj = vc.Video(source_file, video_streams, video_format, audio_streams, subtitle_streams)
|
|
video_files.update({source_file: obj})
|
|
|
|
logging.info(obj)
|
|
return 1
|
|
except Exception as e:
|
|
logging.error(f"Get Video Information: {e}")
|
|
return 0
|
|
|
|
def get_ffprobe(select, source_file):
|
|
command = ffmpeg.video_info(select, source_file)
|
|
|
|
result = subprocess.run(command, stdout=subprocess.PIPE, text=True)
|
|
json_data = json.loads(result.stdout)
|
|
|
|
if select == "v":
|
|
return json_data.get("streams", []),json_data.get("format", [])
|
|
elif select == "a":
|
|
return json_data.get("streams", [])
|
|
elif select == "s":
|
|
return json_data.get("streams", [])
|
|
|
|
# Convert Process ------------------------------------------------------------------------------------------------------
|
|
async def queue_video():
|
|
async with semaphore:
|
|
for key, obj in video_files.items():
|
|
if obj.finished == 0:
|
|
obj.task = await asyncio.create_task(video_convert(obj))
|
|
active_tasks.add(obj)
|
|
logging.info(f"Warteschlange started Auftrag - {obj.task}")
|
|
|
|
async def video_convert(obj):
|
|
"""Startet die Videokonvertierung asynchron."""
|
|
global active_process, active_tasks
|
|
|
|
obj.convert_start = time.time()
|
|
command = ffmpeg.video_convert(obj)
|
|
|
|
logging.info(f"Starte Konvertierung: {command}")
|
|
|
|
try:
|
|
# Starte den Subprozess asynchron
|
|
obj.process = await asyncio.create_subprocess_exec(
|
|
*command,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE
|
|
)
|
|
|
|
active_process.add(obj)
|
|
obj.finished = 3
|
|
await queue.put(await video_list())
|
|
|
|
while obj.process.returncode is None:
|
|
logging.info(f"{obj.file_name} ... läuft noch")
|
|
await asyncio.sleep(10)
|
|
|
|
# Prozess beendet, Status auswerten
|
|
if obj.process.returncode == 0:
|
|
obj.finished = 1
|
|
result = "Finished"
|
|
else:
|
|
obj.finished = 2
|
|
result = "Failure"
|
|
|
|
logging.info(f"Prozess {result}({obj.process.returncode}): {obj.file_name}")
|
|
|
|
await queue.put(obj.to_dict())
|
|
active_process.discard(obj)
|
|
active_tasks.discard(obj)
|
|
|
|
obj.convert_end = time.time()
|
|
|
|
except Exception as e:
|
|
obj.finished = 2
|
|
logging.error(f"Fehler in video_convert(): {e}")
|
|
|
|
#UviCorn WebServer Teil
|
|
#----------------------------------------------------------------------------------------------------------------------
|
|
async def read_output():
|
|
"""Thread-Funktion, die Prozessausgaben in die Queue legt."""
|
|
global active_process
|
|
|
|
while True:
|
|
if not len(active_process):
|
|
await asyncio.sleep(30)
|
|
await queue.put(video_list())
|
|
continue
|
|
|
|
for obj in list(active_process):
|
|
line = await obj.process.stderr.read(1024)
|
|
line_error = line.decode()
|
|
|
|
obj.extract_convert_data(line_error)
|
|
|
|
await queue.put(obj.to_dict())
|
|
await asyncio.sleep(3)
|
|
|
|
async def video_list():
|
|
vlist = []
|
|
for video in video_files.values():
|
|
vlist.append(video.get_vars())
|
|
|
|
return json.dumps({"video_list": vlist})
|
|
|
|
|
|
@app.post("/")
|
|
async def receive_video_file(data: dict):
|
|
if get_video_information(data):
|
|
await queue_video()
|
|
await queue.put(await video_list())
|
|
else:
|
|
logging.error(f"Videos konnten nicht verarbeitet werden! Warteschleife wurde nicht gestarted")
|
|
|
|
@app.get("/command", response_class=HTMLResponse)
|
|
async def display_paths(request: Request):
|
|
list_command: list = []
|
|
|
|
for video in video_files.values():
|
|
if video.finished == 3:
|
|
list_command.append(ffmpeg.command_as_string(ffmpeg.video_convert(video)))
|
|
|
|
return templates.TemplateResponse("command.html", {"request": request, "videos": video_files, "commands": list_command})
|
|
|
|
@app.get("/webs-ui", response_class=HTMLResponse)
|
|
async def display_paths(request: Request):
|
|
return templates.TemplateResponse("webs-ui.html", {"request": request, "videos": video_files})
|
|
|
|
@app.websocket("/ws")
|
|
async def websocket_v(websocket: WebSocket):
|
|
"""WebSocket-Verbindung für die Kommunikation mit Clients."""
|
|
global thread_output, clients
|
|
|
|
clients.add(websocket)
|
|
await websocket.accept()
|
|
|
|
if thread_output is None:
|
|
# Startet den Thread zum Verarbeiten der Prozessausgaben
|
|
thread_output = asyncio.create_task(read_output())
|
|
|
|
try:
|
|
var_first_sending = 0
|
|
|
|
while True:
|
|
if not var_first_sending:
|
|
await queue.put(await video_list())
|
|
var_first_sending = True
|
|
|
|
message = await queue.get()
|
|
|
|
# Alle Clients benachrichtigen
|
|
for client in list(clients):
|
|
await client.send_text(message)
|
|
|
|
|
|
except WebSocketDisconnect:
|
|
logging.info("WebSocket disconnected")
|
|
except Exception as e:
|
|
logging.error(f"WebSocket error: {e}")
|
|
finally:
|
|
clients.discard(websocket)
|
|
if websocket.client_state == WebSocketState.DISCONNECTED:
|
|
await websocket.close()
|
|
|
|
@app.get("/clients")
|
|
async def get_clients_count():
|
|
return {"active_clients": len(clients), "active_processes": len(active_process), "active_tasks": len(active_tasks)}
|
|
|
|
if __name__ == "__main__":
|
|
uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=False, log_level="debug") |