2122 lines
82 KiB
Python
2122 lines
82 KiB
Python
from __future__ import annotations
|
||
|
||
import builtins
|
||
import io
|
||
import re
|
||
import tokenize
|
||
import warnings
|
||
from datetime import datetime
|
||
from typing import Any, Callable
|
||
|
||
from .joins import apply_join_rules
|
||
|
||
|
||
ROUND_KEYS = {"round", "rounds", "round_id", "roundId", "roundTitle", "round_title"}
|
||
SCORE_KEYS = {
|
||
"score", "scores", "total", "to_par", "toPar", "par", "holes", "hole", "thru", "position", "place", "rank"
|
||
}
|
||
PLAYER_KEYS = {
|
||
"player", "players", "athlete", "athletes", "participant", "participants", "name", "fullName", "lastName", "firstName"
|
||
}
|
||
|
||
|
||
def as_list(data: Any) -> list[Any]:
|
||
if isinstance(data, list):
|
||
return data
|
||
if isinstance(data, dict):
|
||
for key in ("data", "items", "tournaments", "results", "rows", "scores", "list"):
|
||
value = data.get(key)
|
||
if isinstance(value, list):
|
||
return value
|
||
return []
|
||
|
||
|
||
def get_any(d: dict[str, Any], *keys: str, default: Any = "") -> Any:
|
||
for key in keys:
|
||
if key in d and d[key] not in (None, ""):
|
||
return d[key]
|
||
return default
|
||
|
||
|
||
def get_scalar_any(d: dict[str, Any], *keys: str, default: Any = "") -> Any:
|
||
for key in keys:
|
||
if key in d and d[key] not in (None, "") and not isinstance(d[key], (dict, list)):
|
||
return d[key]
|
||
return default
|
||
|
||
|
||
def normalize_title(item: dict[str, Any]) -> str:
|
||
return str(get_any(item, "title", "name", "tournamentTitle", "competitionTitle", "eventTitle", "shortTitle", default="Без названия"))
|
||
|
||
|
||
def normalize_tournaments(data: Any) -> list[dict[str, Any]]:
|
||
items = as_list(data)
|
||
result: list[dict[str, Any]] = []
|
||
|
||
for item in items:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
|
||
tournament_id = get_any(item, "id", "tournamentId", "competitionId", "eventId")
|
||
if tournament_id == "":
|
||
continue
|
||
|
||
result.append({
|
||
"id": tournament_id,
|
||
"title": normalize_title(item),
|
||
"date_start": get_any(item, "dateStart", "date_start", "startDate", "start_date", "dateFrom", default=""),
|
||
"date_end": get_any(item, "dateEnd", "date_end", "endDate", "end_date", "dateTo", default=""),
|
||
"place": get_any(item, "place", "venue", "club", "city", "location", default=""),
|
||
"status": get_any(item, "status", "state", "isActive", "active", default=""),
|
||
"raw": item,
|
||
})
|
||
|
||
return result
|
||
|
||
|
||
def _walk(obj: Any, path: str = "root"):
|
||
yield path, obj
|
||
if isinstance(obj, dict):
|
||
for key, value in obj.items():
|
||
yield from _walk(value, f"{path}.{key}")
|
||
elif isinstance(obj, list):
|
||
for index, value in enumerate(obj):
|
||
yield from _walk(value, f"{path}[{index}]")
|
||
|
||
|
||
def _score_dict_for_round_candidate(item: dict[str, Any], path: str = "") -> int:
|
||
keys = set(item.keys())
|
||
text = " ".join(map(str, keys)).lower()
|
||
path_text = path.lower()
|
||
score = 0
|
||
if keys & ROUND_KEYS:
|
||
score += 8
|
||
if "round" in text or "раунд" in text:
|
||
score += 8
|
||
if "round" in path_text or "раунд" in path_text:
|
||
score += 10
|
||
if any(k.lower() in text for k in ("title", "name", "date", "id", "number", "no")):
|
||
score += 3
|
||
if get_any(item, "id", "roundId", "round_id") != "":
|
||
score += 4
|
||
return score
|
||
|
||
|
||
def extract_rounds(data: Any) -> list[dict[str, Any]]:
|
||
"""Find round-like objects inside any tournament JSON."""
|
||
candidates: list[dict[str, Any]] = []
|
||
|
||
for path, obj in _walk(data):
|
||
if not isinstance(obj, list):
|
||
continue
|
||
if not obj or not all(isinstance(x, dict) for x in obj):
|
||
continue
|
||
|
||
scored = []
|
||
for item in obj:
|
||
s = _score_dict_for_round_candidate(item, path)
|
||
if s >= 8:
|
||
scored.append((s, item, path))
|
||
|
||
if scored:
|
||
for _, item, found_path in scored:
|
||
round_id = get_any(item, "id", "roundId", "round_id")
|
||
title = get_any(
|
||
item,
|
||
"title", "name", "roundTitle", "round_title", "label", "caption",
|
||
default=f"Раунд {get_any(item, 'number', 'roundNumber', 'round_number', default=round_id)}",
|
||
)
|
||
if round_id != "":
|
||
candidates.append({
|
||
"id": round_id,
|
||
"title": str(title),
|
||
"number": get_any(item, "number", "roundNumber", "round_number", "no", default=""),
|
||
"date": get_any(item, "date", "dateStart", "startDate", "startedAt", default=""),
|
||
"status": get_any(item, "status", "state", "isActive", "active", default=""),
|
||
"source_path": found_path,
|
||
"raw": item,
|
||
})
|
||
|
||
unique: dict[str, dict[str, Any]] = {}
|
||
for item in candidates:
|
||
unique[str(item["id"])] = item
|
||
|
||
return list(unique.values())
|
||
|
||
|
||
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):
|
||
# Keep compact representation for small primitive lists, flatten hole arrays separately later.
|
||
if all(not isinstance(x, (dict, list)) for x in value):
|
||
out[next_prefix] = " | ".join(map(str, value))
|
||
else:
|
||
out[next_prefix] = value
|
||
else:
|
||
out[next_prefix] = value
|
||
else:
|
||
out[prefix or "value"] = obj
|
||
|
||
return out
|
||
|
||
|
||
def _candidate_score_for_scores_list(items: list[dict[str, Any]]) -> int:
|
||
if not items:
|
||
return 0
|
||
keys: set[str] = set()
|
||
for item in items[:5]:
|
||
keys |= set(map(str, item.keys()))
|
||
text = " ".join(keys).lower()
|
||
score = 0
|
||
if keys & SCORE_KEYS:
|
||
score += 10
|
||
if keys & PLAYER_KEYS:
|
||
score += 10
|
||
if any(word in text for word in ("score", "hole", "thru", "total", "par", "player", "golfer", "participant", "result")):
|
||
score += 8
|
||
score += min(len(items), 100) // 10
|
||
return score
|
||
|
||
|
||
def find_scores_list(data: Any) -> list[dict[str, Any]]:
|
||
"""Find the most likely list of score rows inside raw scores JSON."""
|
||
best_score = -1
|
||
best_items: list[dict[str, Any]] = []
|
||
|
||
direct = as_list(data)
|
||
if direct and all(isinstance(x, dict) for x in direct):
|
||
best_items = direct # good fallback
|
||
best_score = _candidate_score_for_scores_list(direct)
|
||
|
||
for _, obj in _walk(data):
|
||
if not isinstance(obj, list):
|
||
continue
|
||
if not obj or not all(isinstance(x, dict) for x in obj):
|
||
continue
|
||
score = _candidate_score_for_scores_list(obj)
|
||
if score > best_score:
|
||
best_score = score
|
||
best_items = obj
|
||
|
||
return best_items
|
||
|
||
|
||
def _join_name(*parts: Any) -> str:
|
||
cleaned = [str(p).strip() for p in parts if p not in (None, "") and str(p).strip()]
|
||
return " ".join(cleaned)
|
||
|
||
|
||
def _extract_player_name(flat: dict[str, Any]) -> str:
|
||
# Common direct fields
|
||
direct = get_any(
|
||
flat,
|
||
"player", "player_name", "player_fullName", "player_full_name", "participant_name",
|
||
"athlete_name", "name", "fullName", "full_name", default=""
|
||
)
|
||
if direct:
|
||
return str(direct)
|
||
|
||
# Common first/last combinations
|
||
combos = [
|
||
("lastName", "firstName"),
|
||
("last_name", "first_name"),
|
||
("player_lastName", "player_firstName"),
|
||
("player_last_name", "player_first_name"),
|
||
("participant_lastName", "participant_firstName"),
|
||
]
|
||
for last_key, first_key in combos:
|
||
joined = _join_name(flat.get(last_key), flat.get(first_key))
|
||
if joined:
|
||
return joined
|
||
|
||
# Fuzzy fallback
|
||
last = ""
|
||
first = ""
|
||
for key, value in flat.items():
|
||
lk = key.lower()
|
||
if (not last) and ("lastname" in lk or "last_name" in lk):
|
||
last = str(value)
|
||
if (not first) and ("firstname" in lk or "first_name" in lk):
|
||
first = str(value)
|
||
return _join_name(last, first)
|
||
|
||
|
||
def _to_number(value: Any) -> float | None:
|
||
if value in (None, ""):
|
||
return None
|
||
if isinstance(value, (int, float)):
|
||
return float(value)
|
||
text = str(value).strip().replace(",", ".")
|
||
match = re.search(r"[-+]?\d+(?:\.\d+)?", text)
|
||
if not match:
|
||
return None
|
||
try:
|
||
return float(match.group(0))
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def _get_first_like(flat: dict[str, Any], needles: tuple[str, ...], *, prefer_id: bool = False) -> Any:
|
||
"""Find first value in flattened dict by fuzzy key matching."""
|
||
for key, value in flat.items():
|
||
if value in (None, "") or isinstance(value, (dict, list)):
|
||
continue
|
||
lk = str(key).lower()
|
||
if all(n in lk for n in needles):
|
||
if prefer_id and not (lk.endswith("id") or lk.endswith("_id") or lk.endswith("id") or "id" in lk):
|
||
continue
|
||
return value
|
||
return ""
|
||
|
||
|
||
def _player_name_parts(flat: dict[str, Any]) -> tuple[str, str]:
|
||
last = get_any(flat, "lastName", "last_name", "surname", "familyName", "family_name", "player_lastName", "player_last_name", "participant_lastName", "participant_last_name", default="")
|
||
first = get_any(flat, "firstName", "first_name", "givenName", "given_name", "player_firstName", "player_first_name", "participant_firstName", "participant_first_name", default="")
|
||
if not last:
|
||
last = _get_first_like(flat, ("last", "name")) or _get_first_like(flat, ("surname",))
|
||
if not first:
|
||
first = _get_first_like(flat, ("first", "name")) or _get_first_like(flat, ("given",))
|
||
return str(last or "").strip(), str(first or "").strip()
|
||
|
||
|
||
def _clean_hole_no(value: Any) -> str:
|
||
if value in (None, ""):
|
||
return ""
|
||
num = _to_number(value)
|
||
if num is None:
|
||
return ""
|
||
n = int(num)
|
||
if 1 <= n <= 18:
|
||
return str(n)
|
||
return ""
|
||
|
||
|
||
def _classify_hole_to_par(value: Any) -> tuple[str, str]:
|
||
num = _to_number(value)
|
||
if num is None:
|
||
return "", ""
|
||
n = int(num)
|
||
if n <= -3:
|
||
return "albatros", "Albatros"
|
||
if n == -2:
|
||
return "eagle", "Eagle"
|
||
if n == -1:
|
||
return "birdie", "Birdie"
|
||
if n == 0:
|
||
return "par", "Par"
|
||
if n == 1:
|
||
return "bogey", "Bogey"
|
||
if n == 2:
|
||
return "double-bogey", "Dbl (+2)"
|
||
return "triple-bogey", f"+{n}"
|
||
|
||
|
||
def _hole_score_from_flat(flat: dict[str, Any]) -> Any:
|
||
return get_any(
|
||
flat,
|
||
"score", "strokes", "stroke", "gross", "grossScore", "gross_score",
|
||
"result", "value", "count", "total", default="",
|
||
)
|
||
|
||
|
||
def _hole_par_from_flat(flat: dict[str, Any]) -> Any:
|
||
return get_any(flat, "par", "holePar", "hole_par", "parValue", "par_value", default="")
|
||
|
||
|
||
def _hole_to_par_from_flat(flat: dict[str, Any], score: Any = "", par: Any = "") -> Any:
|
||
direct = get_any(
|
||
flat,
|
||
"toPar", "to_par", "relativeToPar", "relative_to_par", "scoreToPar", "score_to_par",
|
||
"diff", "difference", "parDiff", "par_diff", default="",
|
||
)
|
||
if direct not in (None, ""):
|
||
return direct
|
||
score_num = _to_number(score)
|
||
par_num = _to_number(par)
|
||
if score_num is not None and par_num is not None:
|
||
return int(score_num - par_num)
|
||
return ""
|
||
|
||
|
||
def _hole_points_from_flat(flat: dict[str, Any]) -> Any:
|
||
return get_any(
|
||
flat,
|
||
"points", "point", "stableford", "stablefordPoints", "stableford_points",
|
||
"scorePoints", "score_points", default="",
|
||
)
|
||
|
||
|
||
def _looks_like_holes_array(key: str, items: list[Any]) -> bool:
|
||
key_lower = str(key).lower()
|
||
if any(word in key_lower for word in ("hole", "holes", "score", "scores", "card", "scorecard")):
|
||
return True
|
||
sample_keys: set[str] = set()
|
||
for item in items[:5]:
|
||
if isinstance(item, dict):
|
||
sample_keys |= set(k.lower() for k in map(str, item.keys()))
|
||
return bool(sample_keys & {"hole", "holenumber", "hole_number", "number", "no"}) and bool(sample_keys & {"score", "strokes", "gross", "par", "points", "value"})
|
||
|
||
|
||
def _extract_holes_recursive(obj: Any, prefix: str = "") -> dict[str, dict[str, Any]]:
|
||
"""Extract hole-by-hole data from many possible API shapes.
|
||
|
||
RusGolf pages can expose hole data as:
|
||
- a list of dicts: [{hole: 1, score: 4, par: 4, points: 2}, ...]
|
||
- a dict keyed by hole number: {"1": 4, "2": 5} or {"1": {score: 4, par: 4}}
|
||
- flat/array values: scores: [4,5,...], points: [2,1,...], par: [4,4,...]
|
||
|
||
This parser keeps score/par/to_par/points separately so the visual table can show
|
||
the full scorecard even if the endpoint does not use the exact `hole_1` names.
|
||
"""
|
||
holes: dict[str, dict[str, Any]] = {}
|
||
|
||
def merge(hole_no: str, data: dict[str, Any]) -> None:
|
||
hole_no = _clean_hole_no(hole_no)
|
||
if not hole_no:
|
||
return
|
||
dest = holes.setdefault(hole_no, {})
|
||
for k, v in data.items():
|
||
if v not in (None, ""):
|
||
dest[k] = v
|
||
|
||
def kind_from_key(key: str) -> str:
|
||
lk = str(key).lower()
|
||
if "point" in lk or "stableford" in lk:
|
||
return "points"
|
||
if "par" in lk and "topar" not in lk and "to_par" not in lk:
|
||
return "par"
|
||
if "topar" in lk or "to_par" in lk or "diff" in lk or "relative" in lk:
|
||
return "to_par"
|
||
if "class" in lk:
|
||
return "class"
|
||
if "label" in lk or "type" in lk:
|
||
return "label"
|
||
return "score"
|
||
|
||
if isinstance(obj, dict):
|
||
for key, value in obj.items():
|
||
key_s = str(key)
|
||
# Shape: {"1": 4, "2": 5} or {"h1": 4}
|
||
direct_match = re.fullmatch(r"(?:h|hole)?[_\- ]?(\d{1,2})", key_s, flags=re.IGNORECASE)
|
||
if direct_match and not isinstance(value, (dict, list)):
|
||
merge(direct_match.group(1), {"score": value})
|
||
continue
|
||
|
||
# Shape: {"1": {score/par/points...}}
|
||
if direct_match and isinstance(value, dict):
|
||
flat_item = flatten_dict(value)
|
||
score = _hole_score_from_flat(flat_item)
|
||
par = _hole_par_from_flat(flat_item)
|
||
to_par = _hole_to_par_from_flat(flat_item, score, par)
|
||
points = _hole_points_from_flat(flat_item)
|
||
css, label = _classify_hole_to_par(to_par)
|
||
merge(direct_match.group(1), {"score": score, "par": par, "to_par": to_par, "points": points, "class": css, "label": label})
|
||
|
||
# Shape: holeScores: [4, 5, ...] / points: [2, 1, ...] / par: [4, 4, ...]
|
||
if isinstance(value, list) and value and all(not isinstance(x, (dict, list)) for x in value):
|
||
lk = key_s.lower()
|
||
if len(value) >= 9 and any(word in lk for word in ("hole", "score", "stroke", "gross", "point", "stableford", "par")):
|
||
kind = kind_from_key(key_s)
|
||
for idx, item_value in enumerate(value[:18], start=1):
|
||
merge(str(idx), {kind: item_value})
|
||
|
||
# Shape: holes: [{...}, {...}]
|
||
if isinstance(value, list) and value and all(isinstance(x, dict) for x in value):
|
||
if _looks_like_holes_array(str(key), value):
|
||
for idx, item in enumerate(value, start=1):
|
||
flat_item = flatten_dict(item)
|
||
hole_no = get_any(flat_item, "hole", "holeNumber", "hole_number", "holeNo", "hole_no", "number", "no", default="")
|
||
hole_no = _clean_hole_no(hole_no) or _clean_hole_no(idx)
|
||
if not hole_no:
|
||
continue
|
||
score = _hole_score_from_flat(flat_item)
|
||
par = _hole_par_from_flat(flat_item)
|
||
to_par = _hole_to_par_from_flat(flat_item, score, par)
|
||
points = _hole_points_from_flat(flat_item)
|
||
css, label = _classify_hole_to_par(to_par)
|
||
merge(hole_no, {"score": score, "par": par, "to_par": to_par, "points": points, "class": css, "label": label})
|
||
|
||
if isinstance(value, (dict, list)):
|
||
nested = _extract_holes_recursive(value, f"{prefix}.{key}" if prefix else str(key))
|
||
for hole_no, data in nested.items():
|
||
merge(hole_no, data)
|
||
elif isinstance(obj, list):
|
||
# Shape: [4,5,4,...] directly under a score-like prefix.
|
||
if obj and all(not isinstance(x, (dict, list)) for x in obj):
|
||
kind = kind_from_key(prefix)
|
||
if len(obj) >= 9 and any(word in prefix.lower() for word in ("hole", "score", "stroke", "gross", "point", "stableford", "par")):
|
||
for idx, item_value in enumerate(obj[:18], start=1):
|
||
merge(str(idx), {kind: item_value})
|
||
for item in obj:
|
||
nested = _extract_holes_recursive(item, prefix)
|
||
for hole_no, data in nested.items():
|
||
merge(hole_no, data)
|
||
|
||
# Fill labels/classes when score and par exist but endpoint did not provide explicit to_par.
|
||
for hole_no, data in holes.items():
|
||
if data.get("to_par") in (None, ""):
|
||
data["to_par"] = _hole_to_par_from_flat({}, data.get("score", ""), data.get("par", ""))
|
||
if data.get("class") in (None, "") or data.get("label") in (None, ""):
|
||
css, label = _classify_hole_to_par(data.get("to_par"))
|
||
data.setdefault("class", css)
|
||
data.setdefault("label", label)
|
||
return holes
|
||
|
||
|
||
def _extract_flat_holes(flat: dict[str, Any]) -> dict[str, dict[str, Any]]:
|
||
holes: dict[str, dict[str, Any]] = {}
|
||
for key, value in flat.items():
|
||
lk = str(key).lower()
|
||
match = re.search(r"(?:^|_)(?:hole|h)[_\- ]?(\d{1,2})(?:_|$)", lk)
|
||
if not match:
|
||
match = re.search(r"(?:^|_)(\d{1,2})[_\- ]?(?:score|strokes|par|points)(?:_|$)", lk)
|
||
if not match:
|
||
continue
|
||
hole_no = _clean_hole_no(match.group(1))
|
||
if not hole_no:
|
||
continue
|
||
holes.setdefault(hole_no, {})
|
||
if "par" in lk and "to_par" not in lk and "topar" not in lk:
|
||
holes[hole_no]["par"] = value
|
||
elif "point" in lk or "stableford" in lk:
|
||
holes[hole_no]["points"] = value
|
||
elif "to_par" in lk or "topar" in lk or "diff" in lk:
|
||
holes[hole_no]["to_par"] = value
|
||
elif "class" in lk:
|
||
holes[hole_no]["class"] = value
|
||
elif "label" in lk or "type" in lk:
|
||
holes[hole_no]["label"] = value
|
||
else:
|
||
holes[hole_no]["score"] = value
|
||
return holes
|
||
|
||
|
||
def _sum_numeric(values: list[Any]) -> Any:
|
||
nums = [_to_number(v) for v in values]
|
||
nums = [n for n in nums if n is not None]
|
||
if not nums:
|
||
return ""
|
||
total = sum(nums)
|
||
return int(total) if float(total).is_integer() else round(total, 3)
|
||
|
||
|
||
def _row_matches_category(flat: dict[str, Any], category_id: Any) -> bool:
|
||
if category_id in (None, ""):
|
||
return True
|
||
needle = str(category_id).strip()
|
||
category_values: list[str] = []
|
||
for key, value in flat.items():
|
||
lk = str(key).lower()
|
||
if "cat" in lk or "category" in lk or "division" in lk or "group" in lk:
|
||
if value not in (None, "") and not isinstance(value, (dict, list)):
|
||
category_values.append(str(value).strip())
|
||
if not category_values:
|
||
# If API already filtered by ?cat=... and rows have no category field, keep rows.
|
||
return True
|
||
return needle in category_values
|
||
|
||
|
||
|
||
|
||
def _score_dict_for_player_candidate(item: dict[str, Any], path: str = "") -> int:
|
||
"""Score a dict/list as a likely player profile object, not just a score row."""
|
||
flat = flatten_dict(item)
|
||
keys = list(map(str, flat.keys()))
|
||
text = " ".join(keys + [path]).lower()
|
||
score = 0
|
||
|
||
strong_words = (
|
||
"player", "golfer", "participant", "athlete", "member", "person",
|
||
"lastname", "last_name", "surname", "firstname", "first_name", "fullname", "full_name",
|
||
)
|
||
extra_words = (
|
||
"club", "team", "country", "region", "city", "gender", "sex", "handicap", "hcp",
|
||
"birth", "birthday", "dob", "photo", "avatar", "image", "category", "division", "group",
|
||
)
|
||
for word in strong_words:
|
||
if word in text:
|
||
score += 7
|
||
for word in extra_words:
|
||
if word in text:
|
||
score += 3
|
||
|
||
# Player-like paths are strong; scoring paths are weaker unless they contain actual first/last/name fields.
|
||
path_l = path.lower()
|
||
if any(word in path_l for word in ("players", "golfers", "participants", "athletes", "members")):
|
||
score += 18
|
||
if any(word in path_l for word in ("score", "scores", "result", "leaderboard")):
|
||
score -= 5
|
||
|
||
if get_scalar_any(flat, "id", "playerId", "player_id", "participantId", "participant_id", "golferId", "golfer_id", "memberId", "member_id", default="") != "":
|
||
score += 4
|
||
if _extract_player_name(flat):
|
||
score += 10
|
||
if _player_name_parts(flat) != ("", ""):
|
||
score += 8
|
||
return score
|
||
|
||
|
||
def _extract_photo(flat: dict[str, Any]) -> Any:
|
||
for key, value in flat.items():
|
||
if value in (None, "") or isinstance(value, (dict, list)):
|
||
continue
|
||
lk = str(key).lower()
|
||
if any(word in lk for word in ("photo", "avatar", "image", "picture", "portrait")):
|
||
return value
|
||
return ""
|
||
|
||
|
||
def _extract_handicap(flat: dict[str, Any]) -> Any:
|
||
return get_scalar_any(
|
||
flat,
|
||
"handicap", "hcp", "player_handicap", "player_hcp", "exactHandicap", "exact_handicap",
|
||
default=_get_first_like(flat, ("handicap",)) or _get_first_like(flat, ("hcp",)),
|
||
)
|
||
|
||
|
||
def _extract_birth_date(flat: dict[str, Any]) -> Any:
|
||
return get_scalar_any(
|
||
flat,
|
||
"birthDate", "birth_date", "birthday", "dob", "dateOfBirth", "date_of_birth",
|
||
default=_get_first_like(flat, ("birth",)) or _get_first_like(flat, ("dob",)),
|
||
)
|
||
|
||
|
||
def _normalize_player_profile(item: dict[str, Any], *, source_path: str = "") -> dict[str, Any]:
|
||
flat = flatten_dict(item)
|
||
last_name, first_name = _player_name_parts(flat)
|
||
full_name = _extract_player_name(flat) or _join_name(last_name, first_name)
|
||
player_id = get_scalar_any(
|
||
flat,
|
||
"player_id", "playerId", "participant_id", "participantId", "golferId", "golfer_id",
|
||
"memberId", "member_id", "personId", "person_id", "id",
|
||
default="",
|
||
)
|
||
category_id = get_scalar_any(
|
||
flat,
|
||
"category_id", "categoryId", "cat_id", "catId", "divisionId", "division_id", "groupId", "group_id",
|
||
default="",
|
||
)
|
||
category_title = get_scalar_any(
|
||
flat,
|
||
"categoryTitle", "category_title", "category_name", "categoryName", "division", "divisionTitle", "group", "groupTitle",
|
||
default="",
|
||
)
|
||
|
||
out = {
|
||
"player_id": player_id,
|
||
"player": full_name,
|
||
"player_full_name": full_name,
|
||
"player_last_name": last_name,
|
||
"player_first_name": first_name,
|
||
"player_middle_name": get_scalar_any(flat, "middleName", "middle_name", "patronymic", default=""),
|
||
"player_gender": get_scalar_any(flat, "gender", "sex", "player_gender", default=""),
|
||
"player_birth_date": _extract_birth_date(flat),
|
||
"player_handicap": _extract_handicap(flat),
|
||
"player_photo": _extract_photo(flat),
|
||
"country": get_scalar_any(flat, "country", "countryName", "country_name", "countryCode", "country_code", "countryISO", "player_country", default=""),
|
||
"club": get_scalar_any(flat, "club", "clubName", "club_name", "team", "teamName", "region", "regionName", default=""),
|
||
"city": get_scalar_any(flat, "city", "town", "location", default=""),
|
||
"category_id": category_id,
|
||
"category_title": category_title,
|
||
"source_path": source_path,
|
||
}
|
||
|
||
for key, value in flat.items():
|
||
if isinstance(value, (dict, list)):
|
||
continue
|
||
safe_key = str(key).replace(".", "_")
|
||
out.setdefault(safe_key, value)
|
||
out.setdefault(f"player_api_{safe_key}", value)
|
||
return out
|
||
|
||
|
||
def _dedupe_players(players: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
result: dict[str, dict[str, Any]] = {}
|
||
for item in players:
|
||
pid = str(item.get("player_id") or item.get("id") or "").strip()
|
||
name_key = str(item.get("player_full_name") or item.get("player") or "").strip().lower()
|
||
key = f"id:{pid}" if pid else f"name:{name_key}"
|
||
if not key or key == "name:":
|
||
key = f"row:{len(result)}"
|
||
if key not in result:
|
||
result[key] = dict(item)
|
||
else:
|
||
# Merge non-empty values, keeping the first source as primary.
|
||
for k, v in item.items():
|
||
if result[key].get(k) in (None, "") and v not in (None, ""):
|
||
result[key][k] = v
|
||
return list(result.values())
|
||
|
||
|
||
|
||
|
||
def _player_matches_category(profile: dict[str, Any], category_id: int | str | None) -> bool:
|
||
if category_id in (None, ""):
|
||
return True
|
||
needle = str(category_id).strip()
|
||
id_values: list[str] = []
|
||
for key, value in profile.items():
|
||
lk = str(key).lower()
|
||
if ("cat" in lk or "category" in lk or "division" in lk or "group" in lk) and ("id" in lk or lk.endswith("_id")):
|
||
if value not in (None, "") and not isinstance(value, (dict, list)):
|
||
id_values.append(str(value).strip())
|
||
# If the API profile does not expose category id, do not drop the player.
|
||
if not id_values:
|
||
return True
|
||
return needle in id_values
|
||
|
||
def extract_players(data: Any, category_id: int | str | None = None) -> list[dict[str, Any]]:
|
||
"""Find and normalize player profile data inside tournament/round JSON.
|
||
|
||
RusGolf tournament JSON can keep players in different nested arrays depending on event type.
|
||
This extractor is deliberately fuzzy: it scans the whole JSON, finds likely player arrays,
|
||
normalizes the common fields, and keeps raw fields with player_api_* prefixes.
|
||
"""
|
||
candidates: list[tuple[int, dict[str, Any]]] = []
|
||
|
||
for path, obj in _walk(data):
|
||
if isinstance(obj, list) and obj and all(isinstance(x, dict) for x in obj[: min(len(obj), 10)]):
|
||
scored: list[tuple[int, dict[str, Any]]] = []
|
||
for item in obj:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
score = _score_dict_for_player_candidate(item, path)
|
||
if score >= 18:
|
||
scored.append((score, item))
|
||
if scored:
|
||
for score, item in scored:
|
||
normalized = _normalize_player_profile(item, source_path=path)
|
||
if _player_matches_category(normalized, category_id):
|
||
candidates.append((score, normalized))
|
||
|
||
# Fallback: if the root itself is a player-like object/list.
|
||
direct = as_list(data)
|
||
if direct and all(isinstance(x, dict) for x in direct):
|
||
for item in direct:
|
||
score = _score_dict_for_player_candidate(item, "root")
|
||
if score >= 18:
|
||
normalized = _normalize_player_profile(item, source_path="root")
|
||
if _player_matches_category(normalized, category_id):
|
||
candidates.append((score, normalized))
|
||
|
||
candidates.sort(key=lambda x: -x[0])
|
||
return _dedupe_players([item for _, item in candidates])
|
||
|
||
|
||
def enrich_rows_with_players(rows: list[dict[str, Any]], players: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
"""Merge player profile fields into score rows by id or normalized full name."""
|
||
if not rows or not players:
|
||
return rows
|
||
|
||
by_id: dict[str, dict[str, Any]] = {}
|
||
by_name: dict[str, dict[str, Any]] = {}
|
||
for player in players:
|
||
pid = str(player.get("player_id") or player.get("id") or "").strip()
|
||
if pid:
|
||
by_id[pid] = player
|
||
name = str(player.get("player_full_name") or player.get("player") or "").strip().lower()
|
||
if name:
|
||
by_name[name] = player
|
||
|
||
enriched: list[dict[str, Any]] = []
|
||
for row in rows:
|
||
out = dict(row)
|
||
pid = str(out.get("player_id") or out.get("id") or "").strip()
|
||
name = str(out.get("player_full_name") or out.get("player") or "").strip().lower()
|
||
profile = by_id.get(pid) if pid else None
|
||
if not profile and name:
|
||
profile = by_name.get(name)
|
||
if profile:
|
||
# Add profile fields with explicit prefix and fill empty normalized fields.
|
||
for key, value in profile.items():
|
||
if isinstance(value, (dict, list)):
|
||
continue
|
||
out.setdefault(f"profile_{key}", value)
|
||
if out.get(key) in (None, "") and value not in (None, ""):
|
||
out[key] = value
|
||
out["player_profile_found"] = True
|
||
out["player_profile_source"] = profile.get("source_path", "")
|
||
else:
|
||
out.setdefault("player_profile_found", False)
|
||
out.setdefault("player_profile_source", "")
|
||
enriched.append(out)
|
||
return enriched
|
||
|
||
|
||
def extract_categories(*sources: Any) -> list[dict[str, Any]]:
|
||
"""Find category-like objects in tournament/round JSON."""
|
||
found: dict[str, dict[str, Any]] = {}
|
||
for source in sources:
|
||
for path, obj in _walk(source):
|
||
if isinstance(obj, list) and obj and all(isinstance(x, dict) for x in obj[: min(len(obj), 10)]):
|
||
path_l = path.lower()
|
||
if not any(word in path_l for word in ("cat", "categor", "division", "group")):
|
||
sample_keys = " ".join(str(k).lower() for item in obj[:5] for k in (item.keys() if isinstance(item, dict) else []))
|
||
if not any(word in sample_keys for word in ("category", "cat", "division", "group")):
|
||
continue
|
||
for item in obj:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
flat = flatten_dict(item)
|
||
category_id = get_any(flat, "id", "categoryId", "category_id", "catId", "cat_id", "divisionId", "groupId", default="")
|
||
title = get_any(flat, "title", "name", "caption", "label", "category", "categoryTitle", "category_title", default="")
|
||
if category_id == "" or title == "":
|
||
continue
|
||
key = str(category_id)
|
||
found[key] = {
|
||
"id": category_id,
|
||
"title": str(title),
|
||
"source_path": path,
|
||
"raw": item,
|
||
}
|
||
return list(found.values())
|
||
|
||
|
||
def _extract_holes(row: dict[str, Any], flat: dict[str, Any]) -> dict[str, Any]:
|
||
holes = _extract_holes_recursive(row)
|
||
flat_holes = _extract_flat_holes(flat)
|
||
for hole_no, data in flat_holes.items():
|
||
holes.setdefault(hole_no, {}).update({k: v for k, v in data.items() if v not in (None, "")})
|
||
|
||
result: dict[str, Any] = {}
|
||
stroke_values: list[Any] = []
|
||
point_values: list[Any] = []
|
||
par_values: list[Any] = []
|
||
|
||
for n in range(1, 19):
|
||
key = str(n)
|
||
data = holes.get(key, {})
|
||
score = data.get("score", "")
|
||
par = data.get("par", "")
|
||
to_par = data.get("to_par", "")
|
||
if to_par in (None, ""):
|
||
to_par = _hole_to_par_from_flat({}, score, par)
|
||
css, label = _classify_hole_to_par(to_par)
|
||
css = data.get("class") or css
|
||
label = data.get("label") or label
|
||
points = data.get("points", "")
|
||
|
||
result[f"hole_{n}"] = score
|
||
result[f"hole_{n}_par"] = par
|
||
result[f"hole_{n}_to_par"] = to_par
|
||
result[f"hole_{n}_points"] = points
|
||
result[f"hole_{n}_class"] = css
|
||
result[f"hole_{n}_label"] = label
|
||
|
||
if score not in (None, ""):
|
||
stroke_values.append(score)
|
||
if points not in (None, ""):
|
||
point_values.append(points)
|
||
if par not in (None, ""):
|
||
par_values.append(par)
|
||
|
||
result["out"] = _sum_numeric([result.get(f"hole_{n}") for n in range(1, 10)])
|
||
result["in"] = _sum_numeric([result.get(f"hole_{n}") for n in range(10, 19)])
|
||
result["scorecard_total"] = _sum_numeric([result.get(f"hole_{n}") for n in range(1, 19)])
|
||
result["out_par"] = _sum_numeric([result.get(f"hole_{n}_par") for n in range(1, 10)])
|
||
result["in_par"] = _sum_numeric([result.get(f"hole_{n}_par") for n in range(10, 19)])
|
||
result["round_par"] = _sum_numeric([result.get(f"hole_{n}_par") for n in range(1, 19)])
|
||
result["points_total"] = _sum_numeric([result.get(f"hole_{n}_points") for n in range(1, 19)])
|
||
total_score = _to_number(result.get("scorecard_total"))
|
||
total_par = _to_number(result.get("round_par"))
|
||
if total_score is not None and total_par is not None:
|
||
result["scorecard_to_par"] = int(total_score - total_par)
|
||
result["scorecard_to_par_text"] = _plus_minus_value(result["scorecard_to_par"], zero="E")
|
||
else:
|
||
result["scorecard_to_par"] = ""
|
||
result["scorecard_to_par_text"] = ""
|
||
return result
|
||
|
||
|
||
def normalize_scores(data: Any, category_id: int | str | None = None) -> list[dict[str, Any]]:
|
||
rows = find_scores_list(data)
|
||
normalized: list[dict[str, Any]] = []
|
||
|
||
for index, row in enumerate(rows, start=1):
|
||
if not isinstance(row, dict):
|
||
continue
|
||
|
||
flat = flatten_dict(row)
|
||
if not _row_matches_category(flat, category_id):
|
||
continue
|
||
player = _extract_player_name(flat)
|
||
last_name, first_name = _player_name_parts(flat)
|
||
|
||
position = get_scalar_any(flat, "position", "place", "rank", "pos", "standing", default=index)
|
||
total = get_scalar_any(flat, "total", "totalScore", "total_score", "score_total", "result", "strokes", default="")
|
||
to_par = get_scalar_any(flat, "toPar", "to_par", "par", "relativeToPar", "scoreToPar", default="")
|
||
thru = get_scalar_any(flat, "thru", "through", "holesPlayed", "playedHoles", default="")
|
||
today = get_scalar_any(flat, "today", "roundScore", "round_score", "currentRound", default="")
|
||
country = get_scalar_any(flat, "country", "countryName", "country_code", "countryISO", "player_country", default="")
|
||
club = get_scalar_any(flat, "club", "clubName", "team", "region", default="")
|
||
player_id = get_scalar_any(flat, "player_id", "playerId", "participant_id", "participantId", "golferId", "golfer_id", "memberId", "member_id", "id", default="")
|
||
category_value = get_scalar_any(flat, "category_id", "categoryId", "cat_id", "catId", "category", "categoryTitle", "category_title", default="")
|
||
|
||
out = {
|
||
"position": position,
|
||
"player": player,
|
||
"player_full_name": player,
|
||
"player_last_name": last_name,
|
||
"player_first_name": first_name,
|
||
"player_id": player_id,
|
||
"category_id": category_value,
|
||
"country": country,
|
||
"club": club,
|
||
"total": total,
|
||
"to_par": to_par,
|
||
"today": today,
|
||
"thru": thru,
|
||
}
|
||
out.update(_extract_holes(row, flat))
|
||
if out.get("scorecard_total") not in (None, "") and out.get("total") in (None, ""):
|
||
out["total"] = out.get("scorecard_total")
|
||
if out.get("scorecard_to_par") not in (None, "") and out.get("to_par") in (None, ""):
|
||
out["to_par"] = out.get("scorecard_to_par")
|
||
if out.get("points_total") not in (None, ""):
|
||
out.setdefault("points", out.get("points_total"))
|
||
|
||
# Add flat raw fields after normalized fields, without overwriting normalized names.
|
||
for key, value in flat.items():
|
||
if isinstance(value, (dict, list)):
|
||
continue
|
||
safe_key = str(key).replace(".", "_")
|
||
if safe_key not in out:
|
||
out[safe_key] = value
|
||
|
||
normalized.append(out)
|
||
|
||
return normalized
|
||
|
||
|
||
def sort_leaderboard(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
def key_func(row: dict[str, Any]):
|
||
pos = _to_number(row.get("position"))
|
||
total = _to_number(row.get("total"))
|
||
to_par = _to_number(row.get("to_par"))
|
||
return (
|
||
9999 if pos is None else pos,
|
||
9999 if to_par is None else to_par,
|
||
9999 if total is None else total,
|
||
str(row.get("player", "")),
|
||
)
|
||
|
||
return sorted(rows, key=key_func)
|
||
|
||
|
||
def select_columns(rows: list[dict[str, Any]], columns: str | None) -> list[dict[str, Any]]:
|
||
if not columns or columns.strip().lower() == "all":
|
||
return rows
|
||
|
||
wanted = [c.strip() for c in columns.split(",") if c.strip()]
|
||
return [{key: row.get(key, "") for key in wanted} for row in rows]
|
||
|
||
|
||
def _get_nested_value(row: dict[str, Any], source: str, default: Any = "") -> Any:
|
||
"""Read a value by plain key or dotted path. Falls back to default."""
|
||
if not source:
|
||
return default
|
||
if source in row:
|
||
return row.get(source, default)
|
||
|
||
current: Any = row
|
||
for part in str(source).split("."):
|
||
if isinstance(current, dict) and part in current:
|
||
current = current[part]
|
||
else:
|
||
return default
|
||
return current
|
||
|
||
|
||
def _render_template(template: str, row: dict[str, Any]) -> str:
|
||
result = template
|
||
for key, value in row.items():
|
||
if isinstance(value, (dict, list)):
|
||
continue
|
||
result = result.replace("{" + str(key) + "}", str(value if value is not None else ""))
|
||
return result
|
||
|
||
|
||
|
||
|
||
def _safe_text(value: Any) -> str:
|
||
if value is None:
|
||
return ""
|
||
if isinstance(value, bool):
|
||
return "true" if value else "false"
|
||
return str(value)
|
||
|
||
|
||
def _normalize_search_text(value: Any, ignore_case: bool = True) -> str:
|
||
text = _safe_text(value)
|
||
return text.lower() if ignore_case else text
|
||
|
||
|
||
def contains_value(value: Any, needle: Any, ignore_case: bool = True) -> bool:
|
||
search = _normalize_search_text(needle, ignore_case)
|
||
return bool(search) and search in _normalize_search_text(value, ignore_case)
|
||
|
||
|
||
def equals_value(value: Any, expected: Any, ignore_case: bool = True) -> bool:
|
||
return _normalize_search_text(value, ignore_case).strip() == _normalize_search_text(expected, ignore_case).strip()
|
||
|
||
|
||
def regex_value(value: Any, pattern: Any, ignore_case: bool = True) -> bool:
|
||
try:
|
||
return re.search(_safe_text(pattern), _safe_text(value), re.IGNORECASE if ignore_case else 0) is not None
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def regex_replace_value(value: Any, pattern: Any, replacement: Any = "", count: int = 0, ignore_case: bool = True) -> str:
|
||
try:
|
||
return re.sub(_safe_text(pattern), _safe_text(replacement), _safe_text(value), count=int(count or 0), flags=re.IGNORECASE if ignore_case else 0)
|
||
except Exception:
|
||
return _safe_text(value)
|
||
|
||
|
||
def regex_extract_value(value: Any, pattern: Any, group: Any = 1, default: Any = "", ignore_case: bool = True) -> Any:
|
||
try:
|
||
match = re.search(_safe_text(pattern), _safe_text(value), re.IGNORECASE if ignore_case else 0)
|
||
if not match:
|
||
return default
|
||
if group in (None, ""):
|
||
return match.group(0)
|
||
try:
|
||
return match.group(int(group))
|
||
except Exception:
|
||
return match.group(_safe_text(group))
|
||
except Exception:
|
||
return default
|
||
|
||
|
||
def first_match_value(value: Any, rules: Any, default: Any = "", ignore_case: bool = True, regex: bool = False) -> Any:
|
||
if not rules:
|
||
return default
|
||
iterable = rules.items() if isinstance(rules, dict) else rules
|
||
for item in iterable:
|
||
try:
|
||
needle, result = item
|
||
except Exception:
|
||
continue
|
||
matched = regex_value(value, needle, ignore_case) if regex else contains_value(value, needle, ignore_case)
|
||
if matched:
|
||
return result
|
||
return default
|
||
|
||
|
||
def template_value(template: Any, row: dict[str, Any] | None = None) -> str:
|
||
row = row or {}
|
||
class SafeDict(dict):
|
||
def __missing__(self, key):
|
||
return ""
|
||
try:
|
||
return str(template).format_map(SafeDict({str(k): _safe_text(v) for k, v in row.items()}))
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def _number_value(value: Any, default: Any = 0) -> Any:
|
||
try:
|
||
return float(_safe_text(value).replace(",", ".").strip())
|
||
except Exception:
|
||
return default
|
||
|
||
|
||
def _format_distance(value: Any) -> str:
|
||
text = _safe_text(value)
|
||
return regex_replace_value(text, r"(\d)(?=(\d{3})+\b)", r"\1.")
|
||
|
||
|
||
def _parse_time_seconds(value: Any, default: Any = "") -> Any:
|
||
if value in (None, ""):
|
||
return default
|
||
if isinstance(value, (int, float)) and not isinstance(value, bool):
|
||
return float(value)
|
||
text = _safe_text(value).strip().replace(",", ".")
|
||
match = re.search(r"-?\d+(?::\d{1,2}){1,2}(?:\.\d+)?", text)
|
||
if match:
|
||
token = match.group(0)
|
||
sign = -1 if token.startswith("-") else 1
|
||
parts = token.lstrip("-").split(":")
|
||
try:
|
||
seconds = float(parts[-1])
|
||
multiplier = 60.0
|
||
for part in reversed(parts[:-1]):
|
||
seconds += float(part) * multiplier
|
||
multiplier *= 60.0
|
||
return sign * seconds
|
||
except Exception:
|
||
return default
|
||
match = re.search(r"-?\d+(?:\.\d+)?", text)
|
||
if match:
|
||
try:
|
||
return float(match.group(0))
|
||
except Exception:
|
||
return default
|
||
return default
|
||
|
||
|
||
def _format_time(value: Any, fmt: str = "auto", decimals: Any = None, default: Any = "") -> Any:
|
||
seconds = _parse_time_seconds(value, default=None)
|
||
if seconds is None:
|
||
return default
|
||
seconds = float(seconds)
|
||
negative = seconds < 0
|
||
seconds = abs(seconds)
|
||
fmt_text = _safe_text(fmt or "auto").lower()
|
||
if decimals not in (None, ""):
|
||
try:
|
||
precision = int(decimals)
|
||
except Exception:
|
||
precision = 2
|
||
elif "tt" in fmt_text or ".2" in fmt_text:
|
||
precision = 2
|
||
elif "_t" in fmt_text or ".t" in fmt_text or ".1" in fmt_text:
|
||
precision = 1
|
||
else:
|
||
precision = 2
|
||
precision = max(0, min(6, precision))
|
||
sign = "-" if negative else ""
|
||
if fmt_text.startswith("s") or seconds < 60 and fmt_text.startswith("auto"):
|
||
return f"{sign}{seconds:.{precision}f}" if precision else f"{sign}{round(seconds):.0f}"
|
||
total_int = int(seconds)
|
||
frac = seconds - total_int
|
||
minutes = total_int // 60
|
||
sec_value = (total_int % 60) + frac
|
||
if precision:
|
||
return f"{sign}{minutes}:{sec_value:0{3 + precision}.{precision}f}"
|
||
return f"{sign}{minutes}:{int(round(sec_value)):02d}"
|
||
|
||
|
||
def _join_values(sep: Any, *values: Any) -> str:
|
||
return _safe_text(sep).join(_safe_text(value) for value in values if value not in (None, ""))
|
||
|
||
|
||
def _split_value(value: Any, sep: Any = None, index: Any = None) -> Any:
|
||
parts = _safe_text(value).split(None if sep is None else _safe_text(sep))
|
||
if index in (None, ""):
|
||
return parts
|
||
try:
|
||
return parts[int(index)]
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def _left_value(value: Any, count: Any) -> str:
|
||
try:
|
||
return _safe_text(value)[:int(count)]
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def _right_value(value: Any, count: Any) -> str:
|
||
try:
|
||
count = int(count)
|
||
except Exception:
|
||
return ""
|
||
if count <= 0:
|
||
return ""
|
||
return _safe_text(value)[-count:]
|
||
|
||
|
||
def _mid_value(value: Any, start: Any, count: Any = None) -> str:
|
||
try:
|
||
start = int(start)
|
||
text = _safe_text(value)
|
||
if count in (None, ""):
|
||
return text[start:]
|
||
return text[start:start + int(count)]
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def _coalesce_value(*values: Any) -> Any:
|
||
for value in values:
|
||
if value not in (None, ""):
|
||
return value
|
||
return ""
|
||
|
||
|
||
def _bool_value(value: Any) -> bool:
|
||
if isinstance(value, bool):
|
||
return value
|
||
if isinstance(value, (int, float)):
|
||
return value != 0
|
||
text = _safe_text(value).strip().lower()
|
||
if text in ("", "0", "false", "no", "none", "null", "нет", "ложь", "off"):
|
||
return False
|
||
return True
|
||
|
||
|
||
def startswith_value(value: Any, prefix: Any, ignore_case: bool = True) -> bool:
|
||
text = _normalize_search_text(value, ignore_case)
|
||
search = _normalize_search_text(prefix, ignore_case)
|
||
return bool(search) and text.startswith(search)
|
||
|
||
|
||
def endswith_value(value: Any, suffix: Any, ignore_case: bool = True) -> bool:
|
||
text = _normalize_search_text(value, ignore_case)
|
||
search = _normalize_search_text(suffix, ignore_case)
|
||
return bool(search) and text.endswith(search)
|
||
|
||
|
||
def regex_replace_first_value(value: Any, pattern: Any, replacement: Any = "", ignore_case: bool = True) -> str:
|
||
return regex_replace_value(value, pattern, replacement, 1, ignore_case)
|
||
|
||
|
||
def replace_if_contains_value(value: Any, needle: Any, new_value: Any, default: Any = None, ignore_case: bool = True) -> Any:
|
||
if contains_value(value, needle, ignore_case):
|
||
return new_value
|
||
return value if default is None else default
|
||
|
||
|
||
def _empty_value(value: Any, *also_empty: Any) -> bool:
|
||
empties = {""}
|
||
empties.update(_safe_text(x) for x in also_empty)
|
||
if value is None:
|
||
return True
|
||
if isinstance(value, str):
|
||
return value.strip() in empties
|
||
return _safe_text(value) in empties
|
||
|
||
|
||
def _filled_value(value: Any, *also_empty: Any) -> bool:
|
||
return not _empty_value(value, *also_empty)
|
||
|
||
|
||
def _if_empty_value(value: Any, true_value: Any, false_value: Any = "", *also_empty: Any) -> Any:
|
||
return true_value if _empty_value(value, *also_empty) else false_value
|
||
|
||
|
||
def _if_filled_value(value: Any, true_value: Any, false_value: Any = "", *also_empty: Any) -> Any:
|
||
return true_value if _filled_value(value, *also_empty) else false_value
|
||
|
||
|
||
def _format_thousands_value(value: Any, separator: Any = ".") -> str:
|
||
text = _safe_text(value).strip()
|
||
if not text:
|
||
return ""
|
||
sep = _safe_text(separator or ".")
|
||
sign = ""
|
||
if text.startswith(("+", "-")):
|
||
sign, text = text[0], text[1:]
|
||
# keep decimal part
|
||
decimal = ""
|
||
if "," in text and "." not in text:
|
||
integer, decimal = text.split(",", 1)
|
||
decimal = "," + decimal
|
||
elif "." in text:
|
||
integer, decimal = text.split(".", 1)
|
||
decimal = "." + decimal
|
||
else:
|
||
integer = text
|
||
integer = re.sub(r"\D", "", integer)
|
||
if not integer:
|
||
return sign + text
|
||
groups = []
|
||
while integer:
|
||
groups.append(integer[-3:])
|
||
integer = integer[:-3]
|
||
return sign + sep.join(reversed(groups)) + decimal
|
||
|
||
|
||
def _format_distance(value: Any, separator: Any = ".") -> str:
|
||
return _format_thousands_value(value, separator)
|
||
|
||
|
||
def _time_diff_value(value: Any, previous: Any, fmt: str = "auto_tt", default: Any = "") -> Any:
|
||
left = _parse_time_seconds(value, default=None)
|
||
right = _parse_time_seconds(previous, default=None)
|
||
if left is None or right is None:
|
||
return default
|
||
return _format_time(float(left) - float(right), fmt=fmt, default=default)
|
||
|
||
|
||
def _parse_date_value(value: Any, default: Any = None) -> Any:
|
||
if value in (None, ""):
|
||
return default
|
||
if hasattr(value, "year") and hasattr(value, "month") and hasattr(value, "day"):
|
||
return value
|
||
text = _safe_text(value).strip()
|
||
if not text:
|
||
return default
|
||
if "T" in text:
|
||
text = text.split("T", 1)[0]
|
||
for fmt in ("%Y-%m-%d", "%d.%m.%Y", "%d/%m/%Y", "%Y/%m/%d", "%Y%m%d"):
|
||
try:
|
||
return datetime.strptime(text, fmt).date()
|
||
except Exception:
|
||
pass
|
||
match = re.search(r"(\d{4})[-./]?(\d{2})[-./]?(\d{2})", text)
|
||
if match:
|
||
try:
|
||
return datetime(int(match.group(1)), int(match.group(2)), int(match.group(3))).date()
|
||
except Exception:
|
||
pass
|
||
match = re.search(r"(\d{2})[./-](\d{2})[./-](\d{4})", text)
|
||
if match:
|
||
try:
|
||
return datetime(int(match.group(3)), int(match.group(2)), int(match.group(1))).date()
|
||
except Exception:
|
||
pass
|
||
return default
|
||
|
||
|
||
def _today_value(fmt: Any = "%Y-%m-%d") -> str:
|
||
try:
|
||
return datetime.now().strftime(_safe_text(fmt or "%Y-%m-%d"))
|
||
except Exception:
|
||
return datetime.now().strftime("%Y-%m-%d")
|
||
|
||
|
||
def _age_value(birth_date: Any, on_date: Any = None, default: Any = "") -> Any:
|
||
born = _parse_date_value(birth_date)
|
||
if not born:
|
||
return default
|
||
at_date = _parse_date_value(on_date) if on_date not in (None, "") else datetime.now().date()
|
||
if not at_date:
|
||
return default
|
||
try:
|
||
years = int(at_date.year) - int(born.year)
|
||
if (int(at_date.month), int(at_date.day)) < (int(born.month), int(born.day)):
|
||
years -= 1
|
||
return max(0, years)
|
||
except Exception:
|
||
return default
|
||
|
||
|
||
def _birth_year_value(birth_date: Any, default: Any = "") -> Any:
|
||
parsed = _parse_date_value(birth_date)
|
||
return parsed.year if parsed else default
|
||
|
||
|
||
def _plus_minus_value(value: Any, zero: Any = "0", default: Any = "") -> str:
|
||
number = _number_value(value, default=None)
|
||
if number is None:
|
||
return _safe_text(default if default not in (None, "") else value)
|
||
if number == 0:
|
||
return _safe_text(zero)
|
||
if number > 0:
|
||
if float(number).is_integer():
|
||
return f"+{int(number)}"
|
||
return f"+{number:g}"
|
||
if float(number).is_integer():
|
||
return str(int(number))
|
||
return f"{number:g}"
|
||
|
||
|
||
def _zero_as_value(value: Any, zero_text: Any = "E") -> Any:
|
||
number = _number_value(value, default=None)
|
||
if number == 0:
|
||
return zero_text
|
||
return value
|
||
|
||
|
||
def _golf_score_value(value: Any, even_text: Any = "E") -> str:
|
||
return _plus_minus_value(value, zero=even_text, default="")
|
||
|
||
|
||
|
||
def _extract_python_string_token(token_text: Any):
|
||
match = re.match(r"(?is)^([rubf]*)(\'\'\'|\"\"\"|\'|\")", str(token_text or ""))
|
||
if not match:
|
||
return None
|
||
prefix = match.group(1) or ""
|
||
quote = match.group(2)
|
||
start = len(prefix) + len(quote)
|
||
end = -len(quote)
|
||
if not str(token_text).endswith(quote):
|
||
return None
|
||
return prefix, quote, str(token_text)[start:end]
|
||
|
||
|
||
def _literal_text_from_string_inner(inner: Any, quote: str) -> str:
|
||
text = str(inner or "")
|
||
quote_char = quote[0] if quote else "'"
|
||
out: list[str] = []
|
||
i = 0
|
||
while i < len(text):
|
||
ch = text[i]
|
||
if ch == "\\" and i + 1 < len(text):
|
||
nxt = text[i + 1]
|
||
if nxt == quote_char or nxt == "\\":
|
||
out.append(nxt)
|
||
i += 2
|
||
continue
|
||
out.append("\\")
|
||
out.append(nxt)
|
||
i += 2
|
||
continue
|
||
out.append(ch)
|
||
i += 1
|
||
return "".join(out)
|
||
|
||
|
||
def _repair_trailing_backslash_before_quote(source: Any) -> str:
|
||
text = str(source or "")
|
||
out: list[str] = []
|
||
in_quote = None
|
||
triple = False
|
||
i = 0
|
||
while i < len(text):
|
||
ch = text[i]
|
||
if not in_quote:
|
||
if ch in ("'", '"'):
|
||
if text[i:i + 3] == ch * 3:
|
||
in_quote = ch
|
||
triple = True
|
||
out.append(ch * 3)
|
||
i += 3
|
||
continue
|
||
in_quote = ch
|
||
triple = False
|
||
out.append(ch)
|
||
i += 1
|
||
continue
|
||
if triple:
|
||
if text[i:i + 3] == in_quote * 3:
|
||
out.append(in_quote * 3)
|
||
i += 3
|
||
in_quote = None
|
||
triple = False
|
||
continue
|
||
out.append(ch)
|
||
i += 1
|
||
continue
|
||
if ch == "\\" and i + 1 < len(text) and text[i + 1] == in_quote:
|
||
j = i + 2
|
||
while j < len(text) and text[j].isspace():
|
||
j += 1
|
||
if j >= len(text) or text[j] in ",)]}:+-*/%<>=!&|":
|
||
out.append("\\\\")
|
||
out.append(in_quote)
|
||
i += 2
|
||
in_quote = None
|
||
continue
|
||
if ch == in_quote:
|
||
out.append(ch)
|
||
i += 1
|
||
in_quote = None
|
||
continue
|
||
out.append(ch)
|
||
i += 1
|
||
return "".join(out)
|
||
|
||
|
||
def _normalize_vmix_expression_string_literals(expr: Any) -> str:
|
||
source = _repair_trailing_backslash_before_quote(str(expr or ""))
|
||
try:
|
||
tokens = []
|
||
stream = io.StringIO(source).readline
|
||
for tok in tokenize.generate_tokens(stream):
|
||
if tok.type == tokenize.STRING:
|
||
parsed = _extract_python_string_token(tok.string)
|
||
if parsed:
|
||
prefix, quote, inner = parsed
|
||
prefix_lower = prefix.lower()
|
||
if "r" not in prefix_lower and "f" not in prefix_lower and "b" not in prefix_lower:
|
||
literal_text = _literal_text_from_string_inner(inner, quote)
|
||
tok = tokenize.TokenInfo(tok.type, repr(literal_text), tok.start, tok.end, tok.line)
|
||
tokens.append(tok)
|
||
return tokenize.untokenize(tokens)
|
||
except Exception:
|
||
return source
|
||
|
||
def _safe_eval_expr(expr: str, row: dict[str, Any], all_rows: list[dict[str, Any]] | None = None, row_index: int = 0) -> Any:
|
||
"""Expression engine compatible with eTiming vmixTransforms + project extensions."""
|
||
expr = str(expr or "").strip()
|
||
if not expr:
|
||
return ""
|
||
|
||
def get_value(name: Any, default: Any = "") -> Any:
|
||
return _get_nested_value(row, _safe_text(name), default)
|
||
|
||
def field_empty(name: Any, *also_empty: Any) -> bool:
|
||
return _empty_value(get_value(name), *also_empty)
|
||
|
||
def field_filled(name: Any, *also_empty: Any) -> bool:
|
||
return _filled_value(get_value(name), *also_empty)
|
||
|
||
rows_context = all_rows or []
|
||
|
||
def column_values(name: Any, numeric: Any = True) -> list[Any]:
|
||
key = _safe_text(name)
|
||
values: list[Any] = []
|
||
for item in rows_context:
|
||
raw = _get_nested_value(item, key, "")
|
||
if _empty_value(raw):
|
||
continue
|
||
if _bool_value(numeric):
|
||
num = _number_value(raw, default=None)
|
||
if num is not None:
|
||
values.append(num)
|
||
else:
|
||
values.append(raw)
|
||
return values
|
||
|
||
def column_avg(name: Any, default: Any = "") -> Any:
|
||
values = column_values(name, True)
|
||
return (sum(values) / len(values)) if values else default
|
||
|
||
def column_max(*args: Any) -> Any:
|
||
if len(args) == 1 and isinstance(args[0], str):
|
||
values = column_values(args[0], True)
|
||
return builtins.max(values) if values else ""
|
||
if len(args) == 1:
|
||
try:
|
||
return builtins.max(args[0])
|
||
except Exception:
|
||
return args[0]
|
||
return builtins.max(args)
|
||
|
||
def column_min(*args: Any) -> Any:
|
||
if len(args) == 1 and isinstance(args[0], str):
|
||
values = column_values(args[0], True)
|
||
return builtins.min(values) if values else ""
|
||
if len(args) == 1:
|
||
try:
|
||
return builtins.min(args[0])
|
||
except Exception:
|
||
return args[0]
|
||
return builtins.min(args)
|
||
|
||
def first_value(name: Any, default: Any = "") -> Any:
|
||
key = _safe_text(name)
|
||
if not rows_context:
|
||
return default
|
||
return _get_nested_value(rows_context[0], key, default)
|
||
|
||
def diff_from_first(name: Any, decimals: Any = None, default: Any = "") -> Any:
|
||
current = _number_value(get_value(name), default=None)
|
||
first = _number_value(first_value(name), default=None)
|
||
if current is None or first is None:
|
||
return default
|
||
value = current - first
|
||
if decimals not in (None, ""):
|
||
try:
|
||
return round(value, int(decimals))
|
||
except Exception:
|
||
return value
|
||
return value
|
||
|
||
def gap_from_first(name: Any, decimals: Any = 2, zero: Any = "0", default: Any = "") -> Any:
|
||
value = diff_from_first(name, decimals, default=None)
|
||
if value is None:
|
||
return default
|
||
return _plus_minus_value(value, zero=zero, default=default)
|
||
|
||
safe_globals: dict[str, Any] = {
|
||
"__builtins__": {},
|
||
"str": str,
|
||
"int": int,
|
||
"float": float,
|
||
"len": len,
|
||
"min": column_min,
|
||
"max": column_max,
|
||
"avg": column_avg,
|
||
"average": column_avg,
|
||
"mean": column_avg,
|
||
"sum": sum,
|
||
"round": round,
|
||
"abs": abs,
|
||
"True": True,
|
||
"False": False,
|
||
"None": None,
|
||
"get": get_value,
|
||
"val": get_value,
|
||
"concat": lambda *parts: "".join(_safe_text(p) for p in parts if p is not None),
|
||
"join": _join_values,
|
||
"split": _split_value,
|
||
"part": _split_value,
|
||
"left": _left_value,
|
||
"right": _right_value,
|
||
"mid": _mid_value,
|
||
"replace": lambda v, old, new="": _safe_text(v).replace(_safe_text(old), _safe_text(new)),
|
||
"swap": lambda v, sep=" ": _safe_text(sep).join(reversed([p for p in _safe_text(v).split(_safe_text(sep)) if p])),
|
||
"upper": lambda v: _safe_text(v).upper(),
|
||
"lower": lambda v: _safe_text(v).lower(),
|
||
"title_case": lambda v: _safe_text(v).title(),
|
||
"to_title": lambda v: _safe_text(v).title(),
|
||
"strip": lambda v: _safe_text(v).strip(),
|
||
"coalesce": _coalesce_value,
|
||
"tpl": lambda template: template_value(template, row),
|
||
"template": lambda template: template_value(template, row),
|
||
"number": _number_value,
|
||
"num": _number_value,
|
||
"today": _today_value,
|
||
"date_today": _today_value,
|
||
"age": _age_value,
|
||
"age_at": _age_value,
|
||
"years_old": _age_value,
|
||
"birth_year": _birth_year_value,
|
||
"time_seconds": _parse_time_seconds,
|
||
"parse_time": _parse_time_seconds,
|
||
"seconds": _parse_time_seconds,
|
||
"format_time": _format_time,
|
||
"time_format": _format_time,
|
||
"fmt_time": _format_time,
|
||
"time_diff": _time_diff_value,
|
||
"format_time_diff": _time_diff_value,
|
||
"thousands": _format_thousands_value,
|
||
"format_thousands": _format_thousands_value,
|
||
"format_distance": _format_distance,
|
||
"distance": _format_distance,
|
||
"bool": _bool_value,
|
||
"case": lambda condition, yes, no="": yes if _bool_value(condition) else no,
|
||
"iif": lambda condition, yes, no="": yes if _bool_value(condition) else no,
|
||
"empty": _empty_value,
|
||
"blank": _empty_value,
|
||
"is_empty": _empty_value,
|
||
"is_blank": _empty_value,
|
||
"filled": _filled_value,
|
||
"not_empty": _filled_value,
|
||
"is_filled": _filled_value,
|
||
"if_empty": _if_empty_value,
|
||
"if_blank": _if_empty_value,
|
||
"if_filled": _if_filled_value,
|
||
"if_not_empty": _if_filled_value,
|
||
"field_empty": field_empty,
|
||
"field_blank": field_empty,
|
||
"field_filled": field_filled,
|
||
"field_not_empty": field_filled,
|
||
"if_field_empty": lambda key, yes, no="", *also_empty: yes if field_empty(key, *also_empty) else no,
|
||
"if_field_blank": lambda key, yes, no="", *also_empty: yes if field_empty(key, *also_empty) else no,
|
||
"if_field_filled": lambda key, yes, no="", *also_empty: yes if field_filled(key, *also_empty) else no,
|
||
"if_field_not_empty": lambda key, yes, no="", *also_empty: yes if field_filled(key, *also_empty) else no,
|
||
"contains": contains_value,
|
||
"not_contains": lambda v, n, ignore_case=True: not contains_value(v, n, ignore_case),
|
||
"starts": startswith_value,
|
||
"startswith": startswith_value,
|
||
"ends": endswith_value,
|
||
"endswith": endswith_value,
|
||
"equals": equals_value,
|
||
"eq": equals_value,
|
||
"regex": regex_value,
|
||
"match": regex_value,
|
||
"re_replace": regex_replace_value,
|
||
"regex_replace": regex_replace_value,
|
||
"regex_sub": regex_replace_value,
|
||
"rx_replace": regex_replace_value,
|
||
"re_replace_first": regex_replace_first_value,
|
||
"regex_replace_first": regex_replace_first_value,
|
||
"re_remove": lambda v, p, count=0, ignore_case=True: regex_replace_value(v, p, "", count, ignore_case),
|
||
"regex_remove": lambda v, p, count=0, ignore_case=True: regex_replace_value(v, p, "", count, ignore_case),
|
||
"rx_remove": lambda v, p, count=0, ignore_case=True: regex_replace_value(v, p, "", count, ignore_case),
|
||
"re_extract": regex_extract_value,
|
||
"regex_extract": regex_extract_value,
|
||
"rx_extract": regex_extract_value,
|
||
"if_contains": lambda v, n, yes, no="", ignore_case=True: yes if contains_value(v, n, ignore_case) else no,
|
||
"if_not_contains": lambda v, n, yes, no="", ignore_case=True: yes if not contains_value(v, n, ignore_case) else no,
|
||
"if_regex": lambda v, p, yes, no="", ignore_case=True: yes if regex_value(v, p, ignore_case) else no,
|
||
"if_match": lambda v, p, yes, no="", ignore_case=True: yes if regex_value(v, p, ignore_case) else no,
|
||
"if_equals": lambda v, e, yes, no="", ignore_case=True: yes if equals_value(v, e, ignore_case) else no,
|
||
"if_eq": lambda v, e, yes, no="", ignore_case=True: yes if equals_value(v, e, ignore_case) else no,
|
||
"first_match": first_match_value,
|
||
"map_contains": first_match_value,
|
||
"replace_if_contains": replace_if_contains_value,
|
||
"plus_minus": _plus_minus_value,
|
||
"zero_as": _zero_as_value,
|
||
"golf_score": _golf_score_value,
|
||
"values": column_values,
|
||
"column_values": column_values,
|
||
"first_value": first_value,
|
||
"first": first_value,
|
||
"row_index": lambda: row_index,
|
||
"row_number": lambda: row_index + 1,
|
||
"diff_first": diff_from_first,
|
||
"diff_from_first": diff_from_first,
|
||
"gap_first": gap_from_first,
|
||
"gap_from_first": gap_from_first,
|
||
}
|
||
|
||
class SafeEvalLocals(dict):
|
||
def __missing__(self, key: str) -> Any:
|
||
if key in safe_globals:
|
||
raise KeyError(key)
|
||
return ""
|
||
|
||
reserved = set(safe_globals.keys())
|
||
safe_locals = SafeEvalLocals({
|
||
str(k): ("" if v is None else v)
|
||
for k, v in row.items()
|
||
if re.fullmatch(r"[A-Za-z_А-Яа-яЁё][\wА-Яа-яЁё]*", str(k)) and str(k) not in reserved
|
||
})
|
||
try:
|
||
normalized_expr = _normalize_vmix_expression_string_literals(expr)
|
||
with warnings.catch_warnings():
|
||
warnings.simplefilter("ignore", SyntaxWarning)
|
||
code = compile(normalized_expr, "<vmix-json-expr>", "eval")
|
||
return eval(code, safe_globals, safe_locals)
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def _apply_expression_columns(row: dict[str, Any], columns_config: list[dict[str, Any]] | None, all_rows: list[dict[str, Any]] | None = None, row_index: int = 0) -> dict[str, Any]:
|
||
working = dict(row)
|
||
for column in columns_config or []:
|
||
if not column.get("enabled", True):
|
||
continue
|
||
mode = str(column.get("mode") or "").lower()
|
||
expr = str(column.get("expr") or "").strip()
|
||
source = str(column.get("source") or "").strip()
|
||
if not expr and source.startswith("expr:"):
|
||
expr = source[5:].strip()
|
||
elif not expr and source.startswith("="):
|
||
expr = source[1:].strip()
|
||
if mode == "expr" or expr:
|
||
key = str(column.get("key") or "").strip()
|
||
if key:
|
||
value = _safe_eval_expr(expr or source, working, all_rows=all_rows, row_index=row_index)
|
||
if value in (None, ""):
|
||
value = column.get("default", "")
|
||
working[key] = value
|
||
return working
|
||
|
||
def apply_column_config(rows: list[dict[str, Any]], columns_config: list[dict[str, Any]] | None) -> list[dict[str, Any]]:
|
||
"""
|
||
Build rows for vMix by configurable column rules.
|
||
|
||
Column format:
|
||
- key: output field name
|
||
- source: input field name/dotted path, template like "{position} {player}", or expression if mode="expr"
|
||
- expr: formula for computed columns
|
||
- mode: "field" | "template" | "expr". Empty mode is auto.
|
||
- default: fallback value
|
||
- enabled: false hides the column
|
||
"""
|
||
if not columns_config:
|
||
return rows
|
||
|
||
enabled_columns = [c for c in columns_config if c.get("enabled", True) and c.get("key")]
|
||
out_rows: list[dict[str, Any]] = []
|
||
|
||
for row_index, base_row in enumerate(rows):
|
||
row = _apply_expression_columns(base_row, enabled_columns, all_rows=rows, row_index=row_index)
|
||
out: dict[str, Any] = {}
|
||
for column in enabled_columns:
|
||
key = str(column.get("key", "")).strip()
|
||
source = str(column.get("source", key) or key).strip()
|
||
default = column.get("default", "")
|
||
mode = str(column.get("mode") or "").lower()
|
||
expr = str(column.get("expr") or "").strip()
|
||
|
||
if mode == "expr" or expr or source.startswith("expr:") or source.startswith("="):
|
||
# Expression columns have already been evaluated into row[key].
|
||
value = row.get(key, default)
|
||
elif mode == "template" or ("{" in source and "}" in source):
|
||
value = _render_template(source, row)
|
||
else:
|
||
value = _get_nested_value(row, source, default)
|
||
|
||
if value in (None, ""):
|
||
value = default
|
||
out[key] = value
|
||
out_rows.append(out)
|
||
|
||
return out_rows
|
||
|
||
|
||
def apply_configured_json(
|
||
rows: list[dict[str, Any]],
|
||
config: dict[str, Any],
|
||
limit: int | None = None,
|
||
source_resolver: Callable[[dict[str, Any]], Any] | None = None,
|
||
) -> list[dict[str, Any]]:
|
||
# 1) First merge additional JSON sources, so formulas/columns can use joined fields.
|
||
joins = config.get("joins") or []
|
||
if joins and source_resolver is not None:
|
||
rows = apply_join_rules(rows, joins, source_resolver)
|
||
|
||
# 2) Then apply limit. URL ?limit=... has priority; config.default_limit works silently.
|
||
configured_limit = int(config.get("default_limit") or 0)
|
||
final_limit = configured_limit if limit is None else limit
|
||
if final_limit:
|
||
rows = rows[:final_limit]
|
||
|
||
# 3) Finally build output columns/formulas.
|
||
columns = config.get("columns") or []
|
||
output_mode = str(config.get("output_mode") or "columns").lower()
|
||
|
||
if output_mode == "all":
|
||
return [_apply_expression_columns(row, columns, all_rows=rows, row_index=index) for index, row in enumerate(rows)]
|
||
|
||
return apply_column_config(rows, columns)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Full RusGolf data aggregation helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _merge_rows_by_identity(*row_groups: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
"""Merge rows from scores/results/profile sources by player id or normalized name."""
|
||
merged: dict[str, dict[str, Any]] = {}
|
||
order: list[str] = []
|
||
|
||
def identity(row: dict[str, Any]) -> str:
|
||
pid = str(row.get("player_id") or row.get("id") or row.get("profile_player_id") or "").strip()
|
||
if pid:
|
||
return f"id:{pid}"
|
||
name = str(row.get("player_full_name") or row.get("player") or "").strip().lower()
|
||
if name:
|
||
return f"name:{re.sub(r'\\s+', ' ', name)}"
|
||
number = str(row.get("number") or row.get("player_number") or row.get("bib") or "").strip()
|
||
return f"row:{number}:{len(order)}"
|
||
|
||
for rows in row_groups:
|
||
for row in rows or []:
|
||
if not isinstance(row, dict):
|
||
continue
|
||
key = identity(row)
|
||
if key not in merged:
|
||
merged[key] = {}
|
||
order.append(key)
|
||
dest = merged[key]
|
||
for field, value in row.items():
|
||
if isinstance(value, (dict, list)):
|
||
continue
|
||
if dest.get(field) in (None, "") and value not in (None, ""):
|
||
dest[field] = value
|
||
elif field not in dest:
|
||
dest[field] = value
|
||
# raw-prefixed fields are always preserved if unique enough.
|
||
for field, value in row.items():
|
||
if isinstance(value, (dict, list)):
|
||
continue
|
||
if str(field).startswith(("score_", "result_", "profile_")) and value not in (None, ""):
|
||
dest.setdefault(field, value)
|
||
|
||
return [merged[key] for key in order]
|
||
|
||
|
||
def _extract_number(flat: dict[str, Any]) -> Any:
|
||
return get_scalar_any(
|
||
flat,
|
||
"number", "no", "bib", "startNumber", "start_number", "playerNumber", "player_number",
|
||
"participantNumber", "participant_number", default="",
|
||
)
|
||
|
||
|
||
def _extract_points(flat: dict[str, Any]) -> Any:
|
||
return get_scalar_any(
|
||
flat,
|
||
"points_total", "totalPoints", "total_points", "points", "scorePoints", "score_points",
|
||
"stablefordPoints", "stableford_points", "resultPoints", "result_points", default="",
|
||
)
|
||
|
||
|
||
def _last_name_for_sort(row: dict[str, Any]) -> str:
|
||
last = str(row.get("player_last_name") or "").strip()
|
||
if last:
|
||
return last.lower()
|
||
player = str(row.get("player") or row.get("player_full_name") or "").strip()
|
||
return player.split()[0].lower() if player else ""
|
||
|
||
|
||
def sort_full_rows(rows: list[dict[str, Any]], sort_by: str = "total_points", sort_dir: str = "desc") -> list[dict[str, Any]]:
|
||
sort_by = str(sort_by or "total_points").lower()
|
||
reverse = str(sort_dir or "desc").lower() == "desc"
|
||
|
||
def numeric(value: Any) -> float | None:
|
||
return _to_number(value)
|
||
|
||
def key(row: dict[str, Any]):
|
||
if sort_by in {"number", "player_number", "bib"}:
|
||
n = numeric(row.get("number") or row.get("player_number") or row.get("bib"))
|
||
return (999999 if n is None else n, _last_name_for_sort(row))
|
||
if sort_by in {"lastname", "last_name", "surname", "player_last_name", "name"}:
|
||
return (_last_name_for_sort(row), str(row.get("player_first_name") or "").lower())
|
||
if sort_by in {"total", "score", "strokes"}:
|
||
n = numeric(row.get("scorecard_total") or row.get("total"))
|
||
return (-999999 if n is None else n, _last_name_for_sort(row))
|
||
# default: points desc, then score asc if points missing/equal.
|
||
points = numeric(row.get("points_total") or row.get("points") or row.get("total_points"))
|
||
total = numeric(row.get("scorecard_total") or row.get("total"))
|
||
# reverse=True outside; use total as negative to keep lower score better when points equal.
|
||
return (-999999 if points is None else points, 999999 if total is None else -total, _last_name_for_sort(row))
|
||
|
||
return sorted(rows, key=key, reverse=reverse)
|
||
|
||
|
||
def normalize_full_individual_rows(
|
||
tournament: Any,
|
||
round_data: Any,
|
||
scores: Any,
|
||
results: Any | None = None,
|
||
players: list[dict[str, Any]] | None = None,
|
||
category_id: int | str | None = None,
|
||
sort_by: str = "total_points",
|
||
sort_dir: str = "desc",
|
||
) -> list[dict[str, Any]]:
|
||
"""Build one rich player table from tournament, round, scores and results JSON."""
|
||
score_rows = normalize_scores(scores, category_id=category_id)
|
||
for row in score_rows:
|
||
for key, value in list(row.items()):
|
||
if key not in row and value not in (None, ""):
|
||
row[f"score_{key}"] = value
|
||
|
||
result_rows = normalize_scores(results, category_id=category_id) if results is not None else []
|
||
prefixed_results: list[dict[str, Any]] = []
|
||
for row in result_rows:
|
||
prefixed = dict(row)
|
||
for key, value in row.items():
|
||
if not isinstance(value, (dict, list)):
|
||
prefixed.setdefault(f"result_{key}", value)
|
||
prefixed_results.append(prefixed)
|
||
|
||
profile_rows = players or []
|
||
rows = _merge_rows_by_identity(score_rows, prefixed_results)
|
||
rows = enrich_rows_with_players(rows, profile_rows)
|
||
|
||
# Final field cleanup / derived columns.
|
||
normalized: list[dict[str, Any]] = []
|
||
for index, row in enumerate(rows, start=1):
|
||
out = dict(row)
|
||
flat = flatten_dict(out)
|
||
out.setdefault("position", get_scalar_any(flat, "position", "place", "rank", default=index))
|
||
out.setdefault("number", _extract_number(flat))
|
||
out.setdefault("player_number", out.get("number", ""))
|
||
out.setdefault("points_total", _extract_points(flat) or out.get("points_total", ""))
|
||
out.setdefault("total_points", out.get("points_total", ""))
|
||
if out.get("player_last_name") in (None, "") or out.get("player_first_name") in (None, ""):
|
||
last, first = _player_name_parts(flat)
|
||
if out.get("player_last_name") in (None, ""):
|
||
out["player_last_name"] = last
|
||
if out.get("player_first_name") in (None, ""):
|
||
out["player_first_name"] = first
|
||
if out.get("player") in (None, ""):
|
||
out["player"] = _join_name(out.get("player_last_name"), out.get("player_first_name")) or _extract_player_name(flat)
|
||
out["player_full_name"] = out.get("player_full_name") or out.get("player") or ""
|
||
if out.get("scorecard_total") not in (None, ""):
|
||
out.setdefault("total", out.get("scorecard_total"))
|
||
if out.get("points_total") not in (None, ""):
|
||
out.setdefault("points", out.get("points_total"))
|
||
out["leaderboard_type"] = "individual"
|
||
normalized.append(out)
|
||
|
||
return sort_full_rows(normalized, sort_by=sort_by, sort_dir=sort_dir)
|
||
|
||
|
||
def _candidate_score_for_team_list(items: list[dict[str, Any]], path: str = "") -> int:
|
||
if not items:
|
||
return 0
|
||
keys: set[str] = set()
|
||
for item in items[:10]:
|
||
if isinstance(item, dict):
|
||
keys |= set(map(str, flatten_dict(item).keys()))
|
||
text = " ".join(keys).lower() + " " + path.lower()
|
||
score = 0
|
||
for word in ("team", "teams", "match", "matches", "club", "points", "wins", "loss", "draw"):
|
||
if word in text:
|
||
score += 7
|
||
if "team" in path.lower():
|
||
score += 20
|
||
if any(word in text for word in ("player", "hole", "scorecard")):
|
||
score -= 3
|
||
return score + min(len(items), 50)
|
||
|
||
|
||
def find_team_like_list(data: Any, preferred: str = "team") -> list[dict[str, Any]]:
|
||
direct = as_list(data)
|
||
best_items = direct if direct and all(isinstance(x, dict) for x in direct) else []
|
||
best_score = _candidate_score_for_team_list(best_items, "root") if best_items else -1
|
||
for path, obj in _walk(data):
|
||
if isinstance(obj, list) and obj and all(isinstance(x, dict) for x in obj[: min(len(obj), 10)]):
|
||
score = _candidate_score_for_team_list(obj, path)
|
||
if preferred and preferred.lower() in path.lower():
|
||
score += 10
|
||
if score > best_score:
|
||
best_score = score
|
||
best_items = obj
|
||
return best_items
|
||
|
||
|
||
def _extract_team_name(flat: dict[str, Any]) -> str:
|
||
return str(get_scalar_any(
|
||
flat,
|
||
"team", "teamName", "team_name", "name", "title", "club", "clubName", "club_name", default="",
|
||
) or "").strip()
|
||
|
||
|
||
def _extract_team_id(flat: dict[str, Any]) -> Any:
|
||
return get_scalar_any(flat, "team_id", "teamId", "id", "club_id", "clubId", default="")
|
||
|
||
|
||
def _team_identity(row: dict[str, Any]) -> str:
|
||
tid = str(row.get("team_id") or row.get("id") or "").strip()
|
||
if tid:
|
||
return f"id:{tid}"
|
||
name = str(row.get("team") or row.get("team_name") or "").strip().lower()
|
||
return f"name:{name}" if name else f"row:{id(row)}"
|
||
|
||
|
||
def normalize_team_rows(teams_data: Any, team_matches_data: Any | None = None, category_id: int | str | None = None, sort_by: str = "points_total", sort_dir: str = "desc") -> list[dict[str, Any]]:
|
||
"""Normalize team standings and team match rows into one team leaderboard."""
|
||
team_items = find_team_like_list(teams_data, preferred="team")
|
||
match_items = find_team_like_list(team_matches_data, preferred="match") if team_matches_data is not None else []
|
||
|
||
teams: dict[str, dict[str, Any]] = {}
|
||
order: list[str] = []
|
||
|
||
def ensure_team(flat: dict[str, Any], prefix: str = "team") -> dict[str, Any]:
|
||
team_id = _extract_team_id(flat)
|
||
team_name = _extract_team_name(flat)
|
||
key = f"id:{team_id}" if team_id not in (None, "") else f"name:{team_name.lower()}"
|
||
if not key or key == "name:":
|
||
key = f"row:{len(order)}"
|
||
if key not in teams:
|
||
teams[key] = {
|
||
"position": get_scalar_any(flat, "position", "place", "rank", default=len(order) + 1),
|
||
"team_id": team_id,
|
||
"team": team_name,
|
||
"team_name": team_name,
|
||
"club": get_scalar_any(flat, "club", "clubName", "region", default=""),
|
||
"points_total": _extract_points(flat),
|
||
"total_points": _extract_points(flat),
|
||
"matches_count": 0,
|
||
"wins": get_scalar_any(flat, "wins", "win", "w", default=""),
|
||
"draws": get_scalar_any(flat, "draws", "draw", "d", default=""),
|
||
"losses": get_scalar_any(flat, "losses", "loss", "l", default=""),
|
||
"leaderboard_type": "team",
|
||
}
|
||
order.append(key)
|
||
dest = teams[key]
|
||
for k, v in flat.items():
|
||
if isinstance(v, (dict, list)):
|
||
continue
|
||
safe = str(k).replace(".", "_")
|
||
dest.setdefault(safe, v)
|
||
dest.setdefault(f"{prefix}_{safe}", v)
|
||
return dest
|
||
|
||
for item in team_items:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
flat = flatten_dict(item)
|
||
if not _row_matches_category(flat, category_id):
|
||
continue
|
||
ensure_team(flat, prefix="team")
|
||
|
||
# Attach match data to teams when possible, and keep a compact match list.
|
||
for item in match_items:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
flat = flatten_dict(item)
|
||
if not _row_matches_category(flat, category_id):
|
||
continue
|
||
# Try every team-like subobject first.
|
||
attached = False
|
||
for path, obj in _walk(item):
|
||
if isinstance(obj, dict):
|
||
f = flatten_dict(obj)
|
||
if _extract_team_name(f) or _extract_team_id(f):
|
||
dest = ensure_team(f, prefix="match_team")
|
||
dest["matches_count"] = int(_to_number(dest.get("matches_count")) or 0) + 1
|
||
attached = True
|
||
if not attached:
|
||
dest = ensure_team(flat, prefix="match")
|
||
dest["matches_count"] = int(_to_number(dest.get("matches_count")) or 0) + 1
|
||
# Store common match-level info on all matching teams if keys were direct.
|
||
match_no = get_scalar_any(flat, "match", "matchNo", "match_no", "round", "roundNumber", default="")
|
||
for dest in teams.values():
|
||
dest.setdefault("last_match_no", match_no)
|
||
|
||
rows = list(teams.values())
|
||
for idx, row in enumerate(rows, start=1):
|
||
row.setdefault("position", idx)
|
||
if row.get("points_total") in (None, ""):
|
||
row["points_total"] = _extract_points(row)
|
||
row["total_points"] = row.get("points_total", "")
|
||
return sort_team_rows(rows, sort_by=sort_by, sort_dir=sort_dir)
|
||
|
||
|
||
def sort_team_rows(rows: list[dict[str, Any]], sort_by: str = "points_total", sort_dir: str = "desc") -> list[dict[str, Any]]:
|
||
sort_by = str(sort_by or "points_total").lower()
|
||
reverse = str(sort_dir or "desc").lower() == "desc"
|
||
|
||
def key(row: dict[str, Any]):
|
||
if sort_by in {"team", "team_name", "name"}:
|
||
return str(row.get("team") or row.get("team_name") or "").lower()
|
||
if sort_by in {"number", "team_id", "id"}:
|
||
n = _to_number(row.get("team_id") or row.get("id"))
|
||
return -999999 if n is None else n
|
||
points = _to_number(row.get("points_total") or row.get("total_points") or row.get("points"))
|
||
pos = _to_number(row.get("position"))
|
||
return (-999999 if points is None else points, -999999 if pos is None else -pos)
|
||
|
||
return sorted(rows, key=key, reverse=reverse)
|
||
|
||
|
||
def build_unified_full_json(
|
||
*,
|
||
competition_id: int,
|
||
round_id: int | None,
|
||
category_id: int | str | None = None,
|
||
tournament: Any,
|
||
round_data: Any | None = None,
|
||
scores: Any | None = None,
|
||
results: Any | None = None,
|
||
players: list[dict[str, Any]] | None = None,
|
||
team_matches: Any | None = None,
|
||
teams: Any | None = None,
|
||
sort_by: str = "total_points",
|
||
sort_dir: str = "desc",
|
||
mode: str = "auto",
|
||
) -> dict[str, Any]:
|
||
"""Build a single rich JSON for both individual and team competitions."""
|
||
individual_rows = normalize_full_individual_rows(
|
||
tournament=tournament,
|
||
round_data=round_data,
|
||
scores=scores or [],
|
||
results=results,
|
||
players=players or [],
|
||
category_id=category_id,
|
||
sort_by=sort_by,
|
||
sort_dir=sort_dir,
|
||
) if scores is not None else []
|
||
team_rows = normalize_team_rows(teams, team_matches, category_id=category_id, sort_by=sort_by, sort_dir=sort_dir) if (teams is not None or team_matches is not None) else []
|
||
|
||
detected = "team" if team_rows and (str(mode).lower() == "team" or len(team_rows) >= max(1, len(individual_rows) // 4)) else "individual"
|
||
if str(mode).lower() in {"individual", "team"}:
|
||
detected = str(mode).lower()
|
||
|
||
primary_rows = team_rows if detected == "team" else individual_rows
|
||
return {
|
||
"competition_id": competition_id,
|
||
"round_id": round_id,
|
||
"category_id": category_id,
|
||
"competition_type": detected,
|
||
"sort_by": sort_by,
|
||
"sort_dir": sort_dir,
|
||
"counts": {
|
||
"players": len(individual_rows),
|
||
"teams": len(team_rows),
|
||
"primary": len(primary_rows),
|
||
},
|
||
"sources": {
|
||
"tournament": f"/api/livescoring/{competition_id}",
|
||
"round": f"/api/livescoring/{competition_id}/rounds/{round_id}" if round_id else "",
|
||
"scores": f"/api/livescoring/{competition_id}/rounds/{round_id}/scores" if round_id else "",
|
||
"results": f"/api/livescoring/{competition_id}/rounds/{round_id}/results" if round_id else f"/api/livescoring/{competition_id}/results",
|
||
"team_matches": f"/api/livescoring/{competition_id}/results/team-matches",
|
||
"teams": f"/api/livescoring/{competition_id}/results/teams",
|
||
},
|
||
"tournament": tournament,
|
||
"round": round_data,
|
||
"players": individual_rows,
|
||
"teams": team_rows,
|
||
"leaderboard": primary_rows,
|
||
"raw": {
|
||
"scores": scores,
|
||
"results": results,
|
||
"team_matches": team_matches,
|
||
"teams": teams,
|
||
},
|
||
}
|