""" Scarica gli arbitri di TUTTE le leghe dalle pagine di giornata di Transfermarkt e li importa in matches.referee (nomi completi, es. "Daniele Orsato"). Fonte: https://www.transfermarkt.com//spieltag/wettbewerb//saison_id//spieltag/ Una pagina per giornata (~2.000 pagine in tutto), con cache locale: lo script รจ interrompibile e ripartibile senza rifare i download. La mappatura nomi squadra Transfermarkt -> football-data viene imparata automaticamente accoppiando le partite per (lega, data, risultato) nei casi non ambigui (stessa tecnica di import_xg.py). Uso: python scripts/download_referees.py # tutto python scripts/download_referees.py --leagues I1 I2 # solo Italia python scripts/download_referees.py --no-download # solo parse+import dalla cache """ import argparse import random import re import sys import time from collections import Counter, defaultdict from pathlib import Path import pandas as pd import pymysql import config import requests from bs4 import BeautifulSoup PROJECT_ROOT = Path(__file__).resolve().parent.parent RAW_DIR = PROJECT_ROOT / "data" / "raw" / "transfermarkt" OUT_CSV = PROJECT_ROOT / "data" / "processed" / "referees_tm.csv" MAP_CSV = PROJECT_ROOT / "data" / "processed" / "team_mapping_tm.csv" # codice football-data -> (slug transfermarkt, codice transfermarkt, giornate max) LEAGUES = { "I1": ("serie-a", "IT1", 38), "I2": ("serie-b", "IT2", 38), "E0": ("premier-league", "GB1", 38), "SP1": ("laliga", "ES1", 38), "D1": ("bundesliga", "L1", 34), "F1": ("ligue-1", "FR1", 38), } SEASONS = list(range(2017, 2026)) # saison_id = anno di inizio HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36", "Accept-Language": "it-IT,it;q=0.9,en-US;q=0.8,en;q=0.7", "Referer": "https://www.google.com/", } RE_HOME = re.compile(r"^\(\d+\.\)\s*(.+)$") # "(5.) AC Milan" -> "AC Milan" RE_AWAY = re.compile(r"^(.+?)\s*\(\d+\.\)$") # "Lecce (17.)" -> "Lecce" RE_SCORE = re.compile(r"^(\d+):(\d+)$") def fetch_page(league: str, season: int, round_n: int, session: requests.Session, force: bool = False) -> str | None: RAW_DIR.mkdir(parents=True, exist_ok=True) cache = RAW_DIR / f"{league}_{season}_{round_n:02d}.html" if cache.exists() and not force: return cache.read_text(encoding="utf-8", errors="replace") slug, tm_code, _ = LEAGUES[league] url = (f"https://www.transfermarkt.com/{slug}/spieltag/wettbewerb/{tm_code}" f"/saison_id/{season}/spieltag/{round_n}") for attempt in range(3): try: resp = session.get(url, headers=HEADERS, timeout=30) if resp.status_code in (429, 503): wait = 30 * (attempt + 1) print(f" [rate limit] attendo {wait}s...") time.sleep(wait) continue resp.raise_for_status() cache.write_text(resp.text, encoding="utf-8") time.sleep(1.0 + random.uniform(0, 0.6)) # cortesia return resp.text except requests.RequestException as exc: print(f" [retry {attempt + 1}/3] {league} {season} g{round_n}: {exc}") time.sleep(5 * (attempt + 1)) return None def parse_page(html: str, league: str) -> list[dict]: soup = BeautifulSoup(html, "lxml") rows = [] for box in soup.select("div.box"): ref_link = box.select_one('a[href*="/profil/schiedsrichter/"]') if ref_link is None: continue date_link = box.select_one('a[href*="datum"]') score_link = next((a for a in box.select('a[href*="/spielbericht/"]') if RE_SCORE.match(a.get_text(strip=True))), None) home = away = None for td in box.select("td.hauptlink"): text = td.get_text(" ", strip=True) if home is None: m = RE_HOME.match(text) if m: home = m.group(1) continue if away is None and home is not None: m = RE_AWAY.match(text) if m: away = m.group(1) break if not (date_link and score_link and home and away): continue m = RE_SCORE.match(score_link.get_text(strip=True)) try: date = pd.to_datetime(date_link.get_text(strip=True), dayfirst=True) except ValueError: continue rows.append({"league": league, "date": date, "home_tm": home, "away_tm": away, "hg": int(m.group(1)), "ag": int(m.group(2)), "referee": ref_link.get_text(strip=True)}) return rows def download_all(leagues: list[str], force: bool = False) -> pd.DataFrame: session = requests.Session() all_rows = [] for league in leagues: _, _, max_rounds = LEAGUES[league] for season in SEASONS: n_before = len(all_rows) for round_n in range(1, max_rounds + 1): html = fetch_page(league, season, round_n, session, force=force) if html: all_rows.extend(parse_page(html, league)) print(f" {league} {season}/{str(season + 1)[2:]}: " f"{len(all_rows) - n_before} partite") df = pd.DataFrame(all_rows) df = df.drop_duplicates(subset=["league", "date", "home_tm", "away_tm"]) return df def import_to_db(df: pd.DataFrame, args) -> None: conn = pymysql.connect(host=args.host, port=args.port, user=args.user, password=args.password, database="tigertips") cur = conn.cursor() cur.execute(""" SELECT m.id, l.code, m.match_date, h.name, a.name, m.home_goals, m.away_goals FROM matches m JOIN leagues l ON l.id = m.league_id JOIN teams h ON h.id = m.home_team_id JOIN teams a ON a.id = m.away_team_id WHERE m.result IS NOT NULL """) db = pd.DataFrame(cur.fetchall(), columns=[ "id", "league", "date", "home_db", "away_db", "hg", "ag"]) db["date"] = pd.to_datetime(db["date"]) # Fase 1: impara la mappatura nomi con accoppiamenti (lega, data, risultato) univoci key = ["league", "date", "hg", "ag"] tm_unique = df.groupby(key).filter(lambda g: len(g) == 1) db_unique = db.groupby(key).filter(lambda g: len(g) == 1) paired = tm_unique.merge(db_unique, on=key) votes: dict[str, Counter] = defaultdict(Counter) for r in paired.itertuples(index=False): votes[r.home_tm][r.home_db] += 1 votes[r.away_tm][r.away_db] += 1 name_map = {} for tm_name, counter in votes.items(): best, n_best = counter.most_common(1)[0] if n_best / sum(counter.values()) < 0.9: print(f" [ATTENZIONE] mappatura ambigua '{tm_name}': {dict(counter)}") name_map[tm_name] = best pd.DataFrame(sorted(name_map.items()), columns=["transfermarkt", "football_data"]).to_csv(MAP_CSV, index=False) print(f"Mappatura nomi: {len(name_map)} squadre (in {MAP_CSV.name})") unmapped = (set(df["home_tm"]) | set(df["away_tm"])) - set(name_map) if unmapped: print(f" [ATTENZIONE] squadre senza mappatura: {sorted(unmapped)}") # Fase 2: join completo (data esatta, poi +/- 1 giorno) e update df = df.copy() df["home_db"] = df["home_tm"].map(name_map) df["away_db"] = df["away_tm"].map(name_map) df = df.dropna(subset=["home_db", "away_db"]) matched_parts = [] remaining = df.reset_index(drop=True).reset_index(names="tm_idx") db_renamed = db.rename(columns={"date": "join_date"}) for shift in (0, 1, -1): if remaining.empty: break attempt = remaining.copy() attempt["join_date"] = attempt["date"] + pd.Timedelta(days=shift) merged = attempt.merge(db_renamed, on=["league", "join_date", "home_db", "away_db"], suffixes=("_tm", "_db")) matched_parts.append(merged) remaining = remaining[~remaining["tm_idx"].isin(merged["tm_idx"])] matched = pd.concat(matched_parts, ignore_index=True) matched = matched.drop_duplicates(subset=["id"], keep="first") bad = matched[(matched["hg_tm"] != matched["hg_db"]) | (matched["ag_tm"] != matched["ag_db"])] if len(bad): print(f" [ATTENZIONE] {len(bad)} accoppiamenti con risultato diverso, scartati") matched = matched.drop(index=bad.index) rows = [(r.referee, int(r.id)) for r in matched.itertuples(index=False)] for start in range(0, len(rows), 1000): cur.executemany("UPDATE matches SET referee = %s WHERE id = %s", rows[start:start + 1000]) conn.commit() print(f"\nPartite Transfermarkt: {len(df)} | agganciate al DB: {len(matched)}") cur.execute(""" SELECT l.code, COUNT(*), SUM(m.referee IS NOT NULL) FROM matches m JOIN leagues l ON l.id = m.league_id WHERE m.result IS NOT NULL GROUP BY l.code ORDER BY l.code """) print("Copertura arbitri nel DB:") for code, tot, with_ref in cur.fetchall(): print(f" {code:<5} {int(with_ref or 0):>6} / {tot}") conn.close() def main() -> int: parser = argparse.ArgumentParser(description="Arbitri di tutte le leghe da Transfermarkt") parser.add_argument("--leagues", nargs="+", default=list(LEAGUES), choices=list(LEAGUES)) parser.add_argument("--no-download", action="store_true", help="Usa solo la cache locale (parse + import)") parser.add_argument("--force", action="store_true") config.add_db_args(parser) args = parser.parse_args() if args.no_download: rows = [] for path in sorted(RAW_DIR.glob("*.html")): league = path.name.split("_")[0] if league in args.leagues: rows.extend(parse_page( path.read_text(encoding="utf-8", errors="replace"), league)) df = pd.DataFrame(rows).drop_duplicates( subset=["league", "date", "home_tm", "away_tm"]) else: df = download_all(args.leagues, force=args.force) if df.empty: print("Nessun dato.") return 1 OUT_CSV.parent.mkdir(parents=True, exist_ok=True) df.to_csv(OUT_CSV, index=False) print(f"\nTotale partite con arbitro: {len(df)} " f"({df['referee'].nunique()} arbitri) -> {OUT_CSV.name}") import_to_db(df, args) return 0 if __name__ == "__main__": sys.exit(main())