238 lines
7.3 KiB
Python
238 lines
7.3 KiB
Python
import nasio
|
||
import re
|
||
import xml.etree.ElementTree as ET
|
||
import io
|
||
import platform
|
||
import os
|
||
from urllib.parse import urlparse
|
||
from dotenv import load_dotenv
|
||
|
||
load_dotenv(override=True)
|
||
|
||
|
||
def _refresh_env() -> None:
|
||
# Читаем .env заново перед скачиванием, чтобы изменение SYNO_PATH_VMIX_2
|
||
# применялось без пересборки кода.
|
||
load_dotenv(override=True)
|
||
|
||
|
||
def get_syno_settings() -> tuple[str | None, str | None, str | None]:
|
||
_refresh_env()
|
||
return (
|
||
os.getenv("SYNO_URL"),
|
||
os.getenv("SYNO_USERNAME"),
|
||
os.getenv("SYNO_PASSWORD"),
|
||
)
|
||
|
||
|
||
def normalize_source_key(source_key: str | None) -> str:
|
||
return str(source_key or "").upper().strip()
|
||
|
||
|
||
def get_vmix_preset_path(source_key: str | None = None) -> str | None:
|
||
"""Выбирает NAS-путь к vMix-пресету по источнику матча.
|
||
|
||
SUPERLEAGUE -> SYNO_PATH_VMIX
|
||
RUSSIAN_CUP -> SYNO_PATH_VMIX_2, если он заполнен
|
||
|
||
Значения читаются из .env прямо перед скачиванием.
|
||
"""
|
||
_refresh_env()
|
||
key = normalize_source_key(source_key)
|
||
primary_path = os.getenv("SYNO_PATH_VMIX")
|
||
cup_path = os.getenv("SYNO_PATH_VMIX_2")
|
||
|
||
if key == "RUSSIAN_CUP" and cup_path:
|
||
return cup_path
|
||
|
||
return primary_path
|
||
|
||
|
||
def get_fqdn():
|
||
system_name = platform.system()
|
||
if system_name == "Linux":
|
||
hostname = platform.node().lower()
|
||
return f"https://{hostname}.tvstart.ru"
|
||
return "http://127.0.0.1:8000"
|
||
|
||
|
||
FQDN = get_fqdn()
|
||
|
||
|
||
def rebuild_vmix_url(old_url: str, new_base_url: str, session_token: str) -> str:
|
||
"""
|
||
Меняет URL вида:
|
||
http://127.0.0.1:8000/vmix/session/OLDTOKEN/home-lineup
|
||
на:
|
||
https://hostname.tvstart.ru/vmix/session/NEWTOKEN/home-lineup
|
||
"""
|
||
old_url = (old_url or "").strip()
|
||
parsed = urlparse(old_url)
|
||
path = parsed.path or ""
|
||
|
||
match = re.match(r"^/vmix/session/[^/]+/(.+)$", path)
|
||
if match:
|
||
endpoint = match.group(1)
|
||
return f"{new_base_url}/vmix/session/{session_token}/{endpoint}"
|
||
|
||
# Старые пресеты могли хранить ссылки без session-токена:
|
||
# /vmix/home-lineup, /vmix/info.json и т.п.
|
||
old_style_match = re.match(r"^/vmix/(?!session/)(.+)$", path)
|
||
if old_style_match:
|
||
endpoint = old_style_match.group(1).strip("/")
|
||
if endpoint.endswith(".json"):
|
||
endpoint = endpoint[:-5]
|
||
return f"{new_base_url}/vmix/session/{session_token}/{endpoint}"
|
||
|
||
# fallback: если структура другая, просто меняем host
|
||
return re.sub(r"https?://[^/]+", new_base_url, old_url)
|
||
|
||
|
||
def change_vmix_datasource_urls(
|
||
xml_data,
|
||
new_base_url: str,
|
||
session_token: str,
|
||
match_id: str | int | None = None,
|
||
operator_login: str | None = None,
|
||
source_key: str | None = None,
|
||
) -> bytes:
|
||
if isinstance(xml_data, dict):
|
||
candidate = None
|
||
|
||
for key in ("content", "data", "body", "text", "file", "bio"):
|
||
if key in xml_data and xml_data[key] is not None:
|
||
candidate = xml_data[key]
|
||
break
|
||
|
||
if candidate is None:
|
||
raise TypeError(
|
||
f"Unsupported xml_data dict structure. Keys: {list(xml_data.keys())}"
|
||
)
|
||
|
||
if isinstance(candidate, (bytes, bytearray)):
|
||
raw_bytes = bytes(candidate)
|
||
elif isinstance(candidate, str):
|
||
raw_bytes = candidate.encode("utf-8")
|
||
elif isinstance(candidate, io.IOBase) or hasattr(candidate, "read"):
|
||
raw_bytes = candidate.read()
|
||
try:
|
||
candidate.seek(0)
|
||
except Exception:
|
||
pass
|
||
else:
|
||
raise TypeError(
|
||
f"Unsupported xml_data payload inside dict: {type(candidate)}; "
|
||
f"available keys: {list(xml_data.keys())}"
|
||
)
|
||
|
||
elif isinstance(xml_data, (bytes, bytearray)):
|
||
raw_bytes = bytes(xml_data)
|
||
|
||
elif isinstance(xml_data, str):
|
||
raw_bytes = xml_data.encode("utf-8")
|
||
|
||
elif isinstance(xml_data, io.IOBase) or hasattr(xml_data, "read"):
|
||
raw_bytes = xml_data.read()
|
||
try:
|
||
xml_data.seek(0)
|
||
except Exception:
|
||
pass
|
||
|
||
else:
|
||
raise TypeError(f"Unsupported xml_data type: {type(xml_data)}")
|
||
|
||
text = raw_bytes.decode("utf-8", errors="replace")
|
||
root = ET.fromstring(text)
|
||
url_tags = root.findall(
|
||
".//datasource[@friendlyName='JSON']//instance//state/xml/url"
|
||
) + root.findall(".//datasource[@friendlyName='Text']//instance//state/xml/url")
|
||
|
||
for url_tag in url_tags:
|
||
old_url = (url_tag.text or "").strip()
|
||
url_tag.text = rebuild_vmix_url(old_url, new_base_url, session_token)
|
||
|
||
dynamic_settings = root.find(".//DynamicSettings")
|
||
if dynamic_settings is None:
|
||
dynamic_settings = ET.SubElement(root, "DynamicSettings")
|
||
|
||
dynamic = dynamic_settings.find("Dynamic")
|
||
if dynamic is None:
|
||
dynamic = ET.SubElement(dynamic_settings, "Dynamic")
|
||
|
||
def get_or_create_dynamic_value(index: int):
|
||
values = dynamic.findall("DynamicValue")
|
||
|
||
# гарантируем нужное количество
|
||
while len(values) <= index:
|
||
ET.SubElement(dynamic, "DynamicInput")
|
||
ET.SubElement(dynamic, "DynamicValue")
|
||
values = dynamic.findall("DynamicValue")
|
||
|
||
return values[index]
|
||
|
||
# value1 = session
|
||
get_or_create_dynamic_value(0).text = str(session_token)
|
||
|
||
# value2 = match_id
|
||
get_or_create_dynamic_value(1).text = "" if match_id is None else str(match_id)
|
||
|
||
# value3 = source_key (SUPERLEAGUE / RUSSIAN_CUP)
|
||
get_or_create_dynamic_value(2).text = (
|
||
"" if not source_key else normalize_source_key(source_key)
|
||
)
|
||
|
||
# value4 = operator_login
|
||
get_or_create_dynamic_value(3).text = (
|
||
"" if not operator_login else str(operator_login)
|
||
)
|
||
|
||
return ET.tostring(root, encoding="utf-8", method="xml")
|
||
|
||
|
||
def build_vmix_project_bytes(
|
||
session_token: str,
|
||
match_id: str | int | None = None,
|
||
operator_login: str | None = None,
|
||
source_key: str | None = None,
|
||
) -> bytes:
|
||
source_key = normalize_source_key(source_key)
|
||
vmix_preset_path = get_vmix_preset_path(source_key)
|
||
if not vmix_preset_path:
|
||
raise RuntimeError(
|
||
"Не задан путь к vMix-пресету: SYNO_PATH_VMIX или SYNO_PATH_VMIX_2"
|
||
)
|
||
|
||
syno_url, syno_username, syno_password = get_syno_settings()
|
||
print(f"[vmix] source_key={source_key or '-'} preset_path={vmix_preset_path}")
|
||
|
||
vmix_bio = nasio.load_bio(
|
||
user=syno_username,
|
||
password=syno_password,
|
||
nas_ip=syno_url,
|
||
nas_port="443",
|
||
path=vmix_preset_path,
|
||
)
|
||
|
||
edited_vmix = change_vmix_datasource_urls(
|
||
vmix_bio,
|
||
FQDN,
|
||
session_token,
|
||
match_id,
|
||
operator_login,
|
||
source_key,
|
||
)
|
||
|
||
if isinstance(edited_vmix, str):
|
||
edited_vmix = edited_vmix.encode("utf-8")
|
||
|
||
return edited_vmix
|
||
|
||
|
||
def build_vmix_filename(session_row) -> str:
|
||
home = session_row[14].replace("«", "").replace("»", "")
|
||
away = session_row[18].replace("«", "").replace("»", "")
|
||
tour = session_row[10]
|
||
date_str = session_row[9].strftime("%d%m%Y")
|
||
filename = f"ЖФЛ_{home} VS {away}_{tour}_{date_str}.vmix"
|
||
return filename
|