"""Historical World Cup data loader (jfjelstul/worldcup, 1930-2022).""" import unicodedata from pathlib import Path import pandas as pd DATA_DIR = Path(__file__).parent / "historical" _cache: dict[str, pd.DataFrame] = {} def _load(name: str) -> pd.DataFrame: if name not in _cache: csv_path = DATA_DIR / f"{name}.csv" pq_path = DATA_DIR / f"{name}.parquet" if pq_path.exists(): _cache[name] = pd.read_parquet(pq_path) else: df = pd.read_csv(csv_path, low_memory=False) df.to_parquet(pq_path, index=False) _cache[name] = df return _cache[name] def get_matches() -> pd.DataFrame: return _load("matches") def get_goals() -> pd.DataFrame: return _load("goals") def get_squads() -> pd.DataFrame: return _load("squads") def get_players() -> pd.DataFrame: return _load("players") def get_player_appearances() -> pd.DataFrame: return _load("player_appearances") def get_team_appearances() -> pd.DataFrame: return _load("team_appearances") def _norm(name: str) -> str: """Lowercase + strip unicode accents for fuzzy name matching.""" nfkd = unicodedata.normalize("NFKD", name) return "".join(c for c in nfkd if not unicodedata.combining(c)).lower().strip() def get_h2h(team_a: str, team_b: str, limit: int = 10) -> list[dict]: """Return World Cup matches between team_a and team_b (most recent first).""" matches = get_matches() mask = ( ( (matches["home_team_name"] == team_a) & (matches["away_team_name"] == team_b) ) | ( (matches["home_team_name"] == team_b) & (matches["away_team_name"] == team_a) ) ) df = matches[mask].copy() df["match_date"] = pd.to_datetime(df["match_date"], errors="coerce") df = df.sort_values("match_date", ascending=False).head(limit) rows = [] for _, r in df.iterrows(): home, away = r["home_team_name"], r["away_team_name"] score = f"{int(r['home_team_score'])}-{int(r['away_team_score'])}" if r["penalty_shootout"]: score += f" (pen {int(r['home_team_score_penalties'])}-{int(r['away_team_score_penalties'])})" if r["home_team_win"]: winner = home elif r["away_team_win"]: winner = away else: winner = "Draw" rows.append( { "year": str(r["match_date"].year) if pd.notna(r["match_date"]) else r["tournament_name"], "stage": r["stage_name"], "home": home, "away": away, "score": score, "winner": winner, } ) return rows def get_team_wc_record(team_name: str) -> dict: """Aggregate World Cup stats for a team across all men's tournaments.""" ta = get_team_appearances() df = ta[ (ta["team_name"] == team_name) & (ta["tournament_name"].str.contains("Men", na=False)) ] if df.empty: return {} record = { "total_matches": len(df), "wins": int(df["win"].sum()), "draws": int(df["draw"].sum()), "losses": int(df["lose"].sum()), "goals_for": int(df["goals_for"].sum()), "goals_against": int(df["goals_against"].sum()), } # Tournament list (chronological, deduplicated) tournaments = ( df[["tournament_id", "tournament_name"]] .drop_duplicates("tournament_id") .sort_values("tournament_id")["tournament_name"] .tolist() ) record["tournaments"] = tournaments return record def _player_display_name(given_name: str, family_name: str) -> str: """Return display name: use family_name only when given_name is 'not applicable'.""" if str(given_name).strip().lower() in ("not applicable", "n/a", "", "nan"): return str(family_name).strip() return f"{given_name} {family_name}".strip() def get_team_top_scorers(team_name: str, limit: int = 5) -> list[dict]: """Top men's World Cup goal-scorers for a team (all-time, excl. own goals).""" goals = get_goals() df = goals[ (goals["team_name"] == team_name) & (~goals["own_goal"]) & (goals["tournament_name"].str.contains("Men", na=False)) ] if df.empty: return [] agg = ( df.groupby(["player_id", "given_name", "family_name"]) .size() .reset_index(name="goals") .sort_values("goals", ascending=False) .head(limit) ) return [ { "name": _player_display_name(r["given_name"], r["family_name"]), "goals": int(r["goals"]), } for _, r in agg.iterrows() ] def get_player_wc_stats(given_name: str, family_name: str) -> dict: """WC caps and goals for a specific player (by name).""" pa = get_player_appearances() norm_given = _norm(given_name) norm_family = _norm(family_name) mask = pa.apply( lambda r: _norm(str(r["given_name"])) == norm_given and _norm(str(r["family_name"])) == norm_family, axis=1, ) player_apps = pa[mask] caps = len(player_apps) goals = get_goals() goal_mask = goals.apply( lambda r: _norm(str(r["given_name"])) == norm_given and _norm(str(r["family_name"])) == norm_family and not r["own_goal"], axis=1, ) wc_goals = int(goal_mask.sum()) return {"caps": caps, "goals": wc_goals}