283 lines
9.1 KiB
Python
283 lines
9.1 KiB
Python
import nasio
|
||
import re
|
||
import xml.etree.ElementTree as ET
|
||
import io
|
||
import platform
|
||
import os
|
||
from pathlib import Path
|
||
from urllib.parse import urlparse
|
||
from dotenv import load_dotenv
|
||
from synology_drive_api.drive import SynologyDrive
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
||
|
||
|
||
def get_env_path() -> Path:
|
||
"""Возвращает тот же .env, который использует systemd на сервере.
|
||
|
||
На production deploy.sh подключает /mnt/wfl/.env через EnvironmentFile.
|
||
Для локальной разработки остаётся <project>/.env. Путь можно явно
|
||
переопределить переменной WFL_ENV_FILE.
|
||
"""
|
||
configured = str(os.getenv("WFL_ENV_FILE") or "").strip()
|
||
if configured:
|
||
return Path(configured)
|
||
|
||
production_env = Path("/mnt/wfl/.env")
|
||
if production_env.exists():
|
||
return production_env
|
||
|
||
return PROJECT_ROOT / ".env"
|
||
|
||
|
||
def _refresh_env() -> None:
|
||
# ВАЖНО: не используем load_dotenv() без пути. Иначе /root/WFL/.env
|
||
# может перезаписать значения из systemd EnvironmentFile=/mnt/wfl/.env.
|
||
env_path = get_env_path()
|
||
if env_path.exists():
|
||
load_dotenv(env_path, override=True)
|
||
|
||
|
||
_refresh_env()
|
||
|
||
|
||
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()
|
||
env_path = get_env_path()
|
||
|
||
if not syno_url:
|
||
raise RuntimeError(f"Не задан SYNO_URL (env: {env_path})")
|
||
if not syno_username:
|
||
raise RuntimeError(f"Не задан SYNO_USERNAME (env: {env_path})")
|
||
if not syno_password:
|
||
raise RuntimeError(f"Не задан SYNO_PASSWORD (env: {env_path})")
|
||
|
||
# nasio ожидает имя хоста/IP отдельно от порта, поэтому убираем случайно
|
||
# добавленную схему https:// и завершающий slash.
|
||
normalized_syno_url = str(syno_url).strip().rstrip("/")
|
||
if normalized_syno_url.startswith(("http://", "https://")):
|
||
normalized_syno_url = urlparse(normalized_syno_url).hostname or normalized_syno_url
|
||
|
||
print(
|
||
f"[vmix] source_key={source_key or '-'} preset_path={vmix_preset_path} "
|
||
f"nas={normalized_syno_url}:443 env={env_path}"
|
||
)
|
||
|
||
with SynologyDrive(
|
||
username=syno_username,
|
||
password=syno_password,
|
||
nas_domain=normalized_syno_url,
|
||
port=443,
|
||
https=True,
|
||
dsm_version="7",
|
||
) as nas:
|
||
vmix_bio = nas.download_file(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
|