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, "", "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, }, }