306 lines
11 KiB
Python
306 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from copy import deepcopy
|
|
from typing import Any, Callable
|
|
|
|
SourceResolver = Callable[[dict[str, Any]], Any]
|
|
|
|
LIST_KEYS = (
|
|
"rows", "players", "items", "data", "results", "scores", "tournaments",
|
|
"rounds", "list", "records", "entries",
|
|
)
|
|
|
|
|
|
def _safe_text(value: Any) -> str:
|
|
if value is None:
|
|
return ""
|
|
return str(value)
|
|
|
|
|
|
def get_nested_value(obj: Any, path: Any, default: Any = "") -> Any:
|
|
"""Read values by direct key, dotted path, or bracket-ish path.
|
|
|
|
Examples:
|
|
- player_id
|
|
- raw.player.id
|
|
- holes.0.score
|
|
- holes[0].score
|
|
"""
|
|
if path in (None, ""):
|
|
return default
|
|
key = str(path).strip()
|
|
if not key:
|
|
return default
|
|
|
|
if isinstance(obj, dict) and key in obj:
|
|
value = obj.get(key)
|
|
return default if value is None else value
|
|
|
|
tokens = [p for p in re.split(r"\.|\[|\]", key.replace("/", ".")) if p not in ("", None)]
|
|
cur = obj
|
|
for token in tokens:
|
|
if isinstance(cur, dict):
|
|
if token in cur:
|
|
cur = cur[token]
|
|
else:
|
|
return default
|
|
elif isinstance(cur, list):
|
|
try:
|
|
cur = cur[int(token)]
|
|
except Exception:
|
|
return default
|
|
else:
|
|
return default
|
|
return default if cur is None else cur
|
|
|
|
|
|
def flatten_dict(obj: Any, prefix: str = "") -> dict[str, Any]:
|
|
out: dict[str, Any] = {}
|
|
if isinstance(obj, dict):
|
|
for key, value in obj.items():
|
|
clean_key = str(key)
|
|
next_prefix = f"{prefix}_{clean_key}" if prefix else clean_key
|
|
if isinstance(value, dict):
|
|
out.update(flatten_dict(value, next_prefix))
|
|
elif isinstance(value, list):
|
|
if all(not isinstance(x, (dict, list)) for x in value):
|
|
out[next_prefix] = " | ".join(_safe_text(x) for x in value)
|
|
else:
|
|
out[next_prefix] = value
|
|
else:
|
|
out[next_prefix] = value
|
|
return out
|
|
|
|
|
|
def as_rows(data: Any) -> list[dict[str, Any]]:
|
|
"""Convert almost any JSON response into a list of row dictionaries."""
|
|
candidate: Any = data
|
|
if isinstance(data, dict):
|
|
for key in LIST_KEYS:
|
|
value = data.get(key)
|
|
if isinstance(value, list):
|
|
candidate = value
|
|
break
|
|
else:
|
|
# Fallback: choose the largest list of dicts in the object.
|
|
lists: list[list[Any]] = []
|
|
|
|
def walk(obj: Any) -> None:
|
|
if isinstance(obj, dict):
|
|
for value in obj.values():
|
|
walk(value)
|
|
elif isinstance(obj, list) and obj and all(isinstance(x, dict) for x in obj):
|
|
lists.append(obj)
|
|
|
|
walk(data)
|
|
if lists:
|
|
candidate = max(lists, key=len)
|
|
else:
|
|
candidate = [data]
|
|
|
|
if isinstance(candidate, dict):
|
|
candidate = [candidate]
|
|
if not isinstance(candidate, list):
|
|
return []
|
|
|
|
rows: list[dict[str, Any]] = []
|
|
for item in candidate:
|
|
if isinstance(item, dict):
|
|
flat = flatten_dict(item)
|
|
row = dict(item)
|
|
for key, value in flat.items():
|
|
row.setdefault(key, value)
|
|
rows.append(row)
|
|
else:
|
|
rows.append({"value": item})
|
|
return rows
|
|
|
|
|
|
def _normalize_match_value(value: Any, *, case_sensitive: bool = False, trim: bool = True) -> str:
|
|
text = _safe_text(value)
|
|
if trim:
|
|
text = text.strip()
|
|
if not case_sensitive:
|
|
text = text.lower()
|
|
return text
|
|
|
|
|
|
def _match_pairs(join: dict[str, Any]) -> list[dict[str, str]]:
|
|
pairs = join.get("match") or join.get("matches") or []
|
|
if isinstance(pairs, list) and pairs:
|
|
result = []
|
|
for item in pairs:
|
|
if isinstance(item, dict):
|
|
left = item.get("left") or item.get("base") or item.get("from") or item.get("local")
|
|
right = item.get("right") or item.get("source") or item.get("to") or item.get("remote")
|
|
if left and right:
|
|
result.append({"left": str(left), "right": str(right)})
|
|
if result:
|
|
return result
|
|
|
|
left_keys = join.get("left_keys") or join.get("left") or join.get("base_keys") or ""
|
|
right_keys = join.get("right_keys") or join.get("right") or join.get("source_keys") or ""
|
|
if isinstance(left_keys, str):
|
|
left_list = [x.strip() for x in re.split(r"[,;\n]+", left_keys) if x.strip()]
|
|
else:
|
|
left_list = [str(x).strip() for x in left_keys or [] if str(x).strip()]
|
|
if isinstance(right_keys, str):
|
|
right_list = [x.strip() for x in re.split(r"[,;\n]+", right_keys) if x.strip()]
|
|
else:
|
|
right_list = [str(x).strip() for x in right_keys or [] if str(x).strip()]
|
|
return [{"left": l, "right": r} for l, r in zip(left_list, right_list)]
|
|
|
|
|
|
def _field_rules(join: dict[str, Any]) -> list[dict[str, Any]]:
|
|
fields = join.get("fields") or join.get("output_fields") or []
|
|
if isinstance(fields, str):
|
|
parsed: list[dict[str, Any]] = []
|
|
for line in re.split(r"[\n;]+", fields):
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
if "=" in line:
|
|
key, source = line.split("=", 1)
|
|
elif ":" in line:
|
|
key, source = line.split(":", 1)
|
|
else:
|
|
key, source = line, line
|
|
parsed.append({"key": key.strip(), "source": source.strip()})
|
|
fields = parsed
|
|
result: list[dict[str, Any]] = []
|
|
for field in fields:
|
|
if isinstance(field, str):
|
|
result.append({"key": field, "source": field})
|
|
elif isinstance(field, dict):
|
|
key = field.get("key") or field.get("name") or field.get("target") or field.get("output")
|
|
source = field.get("source") or field.get("from") or field.get("path") or key
|
|
if key:
|
|
result.append({
|
|
"key": str(key),
|
|
"source": str(source or key),
|
|
"default": field.get("default", ""),
|
|
"enabled": field.get("enabled", True),
|
|
})
|
|
return result
|
|
|
|
|
|
def _row_key(row: dict[str, Any], fields: list[str], *, case_sensitive: bool, trim: bool) -> tuple[str, ...]:
|
|
return tuple(_normalize_match_value(get_nested_value(row, field), case_sensitive=case_sensitive, trim=trim) for field in fields)
|
|
|
|
|
|
def apply_join_rules(
|
|
base_rows: list[dict[str, Any]],
|
|
joins: list[dict[str, Any]] | None,
|
|
source_resolver: SourceResolver,
|
|
) -> list[dict[str, Any]]:
|
|
"""Merge data from other JSON sources into base rows.
|
|
|
|
Config example::
|
|
|
|
{
|
|
"enabled": true,
|
|
"title": "Подтянуть регион",
|
|
"source_type": "url", # scores | tournament | round | vmix_json | url | file
|
|
"source_key": "https://.../x.json", # or another vMix JSON key
|
|
"match": [{"left": "player_id", "right": "id"}],
|
|
"fields": [{"key": "region", "source": "region"}],
|
|
"multiple": "first", # first | last | all | count
|
|
"separator": " | ",
|
|
"prefix": "",
|
|
"case_sensitive": false,
|
|
"trim": true
|
|
}
|
|
"""
|
|
if not joins:
|
|
return base_rows
|
|
|
|
result = [dict(row) for row in base_rows]
|
|
|
|
for join in joins:
|
|
if not isinstance(join, dict) or join.get("enabled", True) is False:
|
|
continue
|
|
pairs = _match_pairs(join)
|
|
if not pairs:
|
|
continue
|
|
fields = _field_rules(join)
|
|
mode = str(join.get("multiple") or join.get("mode") or "first").lower()
|
|
separator = _safe_text(join.get("separator", " | "))
|
|
prefix = _safe_text(join.get("prefix", ""))
|
|
case_sensitive = bool(join.get("case_sensitive", False))
|
|
trim = bool(join.get("trim", True))
|
|
|
|
try:
|
|
source_rows = as_rows(source_resolver(join))
|
|
except Exception as exc:
|
|
if join.get("required"):
|
|
raise
|
|
# Keep a lightweight error field for debugging without breaking vMix.
|
|
error_key = prefix + _safe_text(join.get("error_key") or "join_error")
|
|
for row in result:
|
|
row.setdefault(error_key, str(exc))
|
|
continue
|
|
|
|
if not source_rows:
|
|
continue
|
|
|
|
right_fields = [p["right"] for p in pairs]
|
|
left_fields = [p["left"] for p in pairs]
|
|
index: dict[tuple[str, ...], list[dict[str, Any]]] = {}
|
|
for source_row in source_rows:
|
|
key = _row_key(source_row, right_fields, case_sensitive=case_sensitive, trim=trim)
|
|
# Empty composite keys are usually bad matches; skip them.
|
|
if not any(key):
|
|
continue
|
|
index.setdefault(key, []).append(source_row)
|
|
|
|
count_key = prefix + _safe_text(join.get("count_key") or "join_count")
|
|
for row in result:
|
|
key = _row_key(row, left_fields, case_sensitive=case_sensitive, trim=trim)
|
|
matches = index.get(key, []) if any(key) else []
|
|
|
|
if mode == "count":
|
|
row[count_key] = len(matches)
|
|
continue
|
|
if not matches:
|
|
for field in fields:
|
|
if field.get("enabled", True) is not False:
|
|
row.setdefault(prefix + field["key"], field.get("default", ""))
|
|
continue
|
|
|
|
selected_rows = matches
|
|
if mode == "last":
|
|
selected_rows = [matches[-1]]
|
|
elif mode not in {"all", "list"}:
|
|
selected_rows = [matches[0]]
|
|
|
|
for field in fields:
|
|
if field.get("enabled", True) is False:
|
|
continue
|
|
out_key = prefix + field["key"]
|
|
source_path = field.get("source") or field["key"]
|
|
default = field.get("default", "")
|
|
values = [get_nested_value(src, source_path, default) for src in selected_rows]
|
|
if mode in {"all", "list"}:
|
|
row[out_key] = separator.join(_safe_text(v) for v in values if v not in (None, ""))
|
|
else:
|
|
row[out_key] = values[0] if values and values[0] not in (None, "") else default
|
|
|
|
return result
|
|
|
|
|
|
def compact_join_for_storage(join: dict[str, Any]) -> dict[str, Any]:
|
|
"""Normalize UI payload before saving it to portable config."""
|
|
data = deepcopy(join)
|
|
data.setdefault("enabled", True)
|
|
data.setdefault("source_type", "scores")
|
|
data.setdefault("source_key", "")
|
|
data.setdefault("multiple", "first")
|
|
data.setdefault("separator", " | ")
|
|
data.setdefault("prefix", "")
|
|
data.setdefault("case_sensitive", False)
|
|
data.setdefault("trim", True)
|
|
data["match"] = _match_pairs(data)
|
|
data["fields"] = _field_rules(data)
|
|
return data
|