231 lines
7.4 KiB
Python
231 lines
7.4 KiB
Python
from __future__ import annotations
|
||
|
||
from urllib.parse import urljoin
|
||
|
||
from repositories.project_settings_repository import (
|
||
DEFAULT_APP_SETTINGS,
|
||
DEFAULT_PARSER_SOURCES,
|
||
ensure_project_settings_tables,
|
||
get_app_setting,
|
||
get_parser_source_from_db,
|
||
list_parser_sources_from_db,
|
||
)
|
||
|
||
|
||
def _fallback_sources() -> list[dict]:
|
||
return [dict(item) for item in DEFAULT_PARSER_SOURCES]
|
||
|
||
|
||
def _fallback_sources_dict() -> dict[str, dict]:
|
||
return {item["key"]: dict(item) for item in DEFAULT_PARSER_SOURCES}
|
||
|
||
|
||
def _fallback_default_source_key() -> str:
|
||
return DEFAULT_APP_SETTINGS["default_parser_source_key"]
|
||
|
||
|
||
def _fallback_base_url() -> str:
|
||
return DEFAULT_APP_SETTINGS["rfs_base_url"].rstrip("/")
|
||
|
||
|
||
def _get_base_url() -> str:
|
||
try:
|
||
ensure_project_settings_tables()
|
||
return get_app_setting("rfs_base_url", _fallback_base_url()).rstrip("/")
|
||
except Exception:
|
||
return _fallback_base_url()
|
||
|
||
|
||
RFS_BASE_URL = _get_base_url()
|
||
|
||
|
||
def get_rfs_base_url() -> str:
|
||
return _get_base_url()
|
||
|
||
|
||
def get_default_source_key() -> str:
|
||
try:
|
||
ensure_project_settings_tables()
|
||
value = get_app_setting("default_parser_source_key", _fallback_default_source_key())
|
||
value = (value or _fallback_default_source_key()).upper().strip()
|
||
known_keys = {source["key"] for source in list_parser_sources()}
|
||
return value if value in known_keys else _fallback_default_source_key()
|
||
except Exception:
|
||
return _fallback_default_source_key()
|
||
|
||
|
||
def get_parser_sources() -> dict[str, dict]:
|
||
try:
|
||
ensure_project_settings_tables()
|
||
sources = list_parser_sources_from_db(active_only=True)
|
||
if not sources:
|
||
return _fallback_sources_dict()
|
||
return {source["key"]: source for source in sources}
|
||
except Exception:
|
||
return _fallback_sources_dict()
|
||
|
||
|
||
def list_parser_sources() -> list[dict]:
|
||
sources = get_parser_sources()
|
||
return list(sources.values())
|
||
|
||
|
||
def get_parser_source(source_key: str | None = None) -> dict:
|
||
"""Возвращает настройки источника из БД. Если БД недоступна, используется fallback."""
|
||
source_key = (source_key or get_default_source_key() or "SUPERLEAGUE").upper().strip()
|
||
try:
|
||
ensure_project_settings_tables()
|
||
source = get_parser_source_from_db(source_key)
|
||
if source:
|
||
return source
|
||
except Exception:
|
||
pass
|
||
|
||
fallback = _fallback_sources_dict()
|
||
if source_key in fallback:
|
||
return fallback[source_key]
|
||
raise ValueError(f"Неизвестный источник парсинга: {source_key}")
|
||
|
||
|
||
def source_absolute_url(source: dict, path_or_url: str) -> str:
|
||
value = (path_or_url or "").strip()
|
||
if value.startswith(("http://", "https://")):
|
||
return value
|
||
base_url = (source.get("base_url") or get_rfs_base_url()).rstrip("/")
|
||
return urljoin(base_url + "/", value.lstrip("/"))
|
||
|
||
|
||
def source_match_url(source: dict, match_external_id: str | int) -> str:
|
||
match_base_url = (source.get("match_base_url") or f"{get_rfs_base_url()}/match/").rstrip("/") + "/"
|
||
return f"{match_base_url}{str(match_external_id).strip()}"
|
||
|
||
|
||
def extract_logo_filename(value: str | None) -> str:
|
||
"""Возвращает только имя файла логотипа из старого полного пути или нового значения."""
|
||
value = str(value or "").strip().strip('"').strip("'")
|
||
if not value:
|
||
return ""
|
||
|
||
normalized = value.replace("/", "\\")
|
||
filename = normalized.split("\\")[-1].strip()
|
||
return filename
|
||
|
||
|
||
def ensure_logo_extension(filename: str) -> str:
|
||
filename = extract_logo_filename(filename)
|
||
if not filename:
|
||
return ""
|
||
if "." not in filename.rsplit("\\", 1)[-1]:
|
||
return f"{filename}.png"
|
||
return filename
|
||
|
||
|
||
def get_logo_base_path(source_key: str | None = None) -> str:
|
||
source = get_parser_source(source_key)
|
||
return str(source.get("logo_base_path") or "").rstrip("\\/")
|
||
|
||
|
||
def build_logo_path(source_key: str | None, filename: str | None) -> str:
|
||
"""Собирает полный путь к логотипу для vMix по источнику турнира."""
|
||
logo_file = ensure_logo_extension(filename)
|
||
if not logo_file:
|
||
return ""
|
||
|
||
base_path = get_logo_base_path(source_key)
|
||
if not base_path:
|
||
return logo_file
|
||
|
||
return f"{base_path}\\{logo_file}"
|
||
|
||
|
||
def build_logo_variant_path(
|
||
source_key: str | None,
|
||
filename: str | None,
|
||
variant: str | None = None,
|
||
) -> str:
|
||
"""
|
||
Собирает путь к варианту логотипа.
|
||
variant="white" -> Динамо_Белый / Зенит_Белый
|
||
variant="blue" -> Динамо_Синий / Зенит_Синий
|
||
Остальные команды остаются без изменения.
|
||
"""
|
||
logo_file = ensure_logo_extension(filename)
|
||
if not logo_file:
|
||
return ""
|
||
|
||
suffix = ""
|
||
if variant == "white":
|
||
suffix = "_Белый"
|
||
elif variant == "blue":
|
||
suffix = "_Синий"
|
||
|
||
if suffix:
|
||
lower = logo_file.lower()
|
||
if "динамо" in lower or "зенит" in lower:
|
||
if "." in logo_file:
|
||
stem, ext = logo_file.rsplit(".", 1)
|
||
ext = "." + ext
|
||
else:
|
||
stem, ext = logo_file, ".png"
|
||
|
||
for old_suffix in ("_Белый", "_Синий"):
|
||
if stem.endswith(old_suffix):
|
||
stem = stem[: -len(old_suffix)]
|
||
logo_file = f"{stem}{suffix}{ext}"
|
||
|
||
return build_logo_path(source_key, logo_file)
|
||
|
||
|
||
|
||
def extract_photo_filename(value: str | None) -> str:
|
||
"""Возвращает имя файла/относительный путь фото из старого полного пути или нового значения."""
|
||
value = str(value or "").strip().strip('"').strip("'")
|
||
if not value:
|
||
return ""
|
||
|
||
normalized = value.replace("/", "\\")
|
||
lower = normalized.lower()
|
||
|
||
# Для старых полных путей пытаемся сохранить относительную часть от папки Photo.
|
||
marker = "\\photo\\"
|
||
if marker in lower:
|
||
idx = lower.rfind(marker)
|
||
return normalized[idx + len(marker):].strip("\\")
|
||
|
||
parts = [part for part in normalized.split("\\") if part]
|
||
if len(parts) >= 2 and ":" in parts[0]:
|
||
return parts[-1].strip()
|
||
return normalized.strip("\\")
|
||
|
||
|
||
def ensure_photo_extension(filename: str) -> str:
|
||
filename = extract_photo_filename(filename)
|
||
if not filename:
|
||
return ""
|
||
if "." not in filename.rsplit("\\", 1)[-1]:
|
||
return f"{filename}.png"
|
||
return filename
|
||
|
||
|
||
def get_photo_base_path(source_key: str | None = None) -> str:
|
||
source = get_parser_source(source_key)
|
||
return str(source.get("photo_base_path") or "").rstrip("\\/")
|
||
|
||
|
||
def build_photo_path(source_key: str | None, filename: str | None) -> str:
|
||
"""Собирает полный путь к фото игрока для vMix по источнику турнира."""
|
||
photo_file = ensure_photo_extension(filename)
|
||
if not photo_file:
|
||
return ""
|
||
|
||
base_path = get_photo_base_path(source_key)
|
||
if not base_path:
|
||
return photo_file
|
||
|
||
return f"{base_path}\\{photo_file}"
|
||
|
||
|
||
def build_empty_photo_path(source_key: str | None = None) -> str:
|
||
"""Путь к пустой картинке фото для выбранного источника."""
|
||
return build_photo_path(source_key, "EMPTY.png")
|