"""EPG spreadsheet -> Crystal schedule frame. The dashboard's file-upload path expects a schedule with the columns `validate_schedule()` requires: event_name, country_name, channel_name, telecast_type and hour. A raw EPG export has none of those directly, so this module derives them. Nothing here writes to tvviewers. The only DB access is three read-only reference lookups (channel bridge, event vocabulary), which the caller supplies as DataFrames so this module stays testable offline. Every derived row carries a `confidence` column. Rows below "high" are meant to land in the existing analyst-review screen rather than be trusted silently -- a wrong telecast_type moves an estimate by an order of magnitude. """ from __future__ import annotations import re import pandas as pd from .epg_rules import classify # EPG columns we rely on. Anything else in the file is carried through untouched. REQUIRED_EPG_COLS = ["channel_name", "channel_countries", "prog_title", "prog_st_time"] # The same export exists under two column vocabularies: the listings feed the # adapter was first written against, and the production GSIQ export, which names # the country and start time differently and arrives with several fields already # populated. Accepting both is a matter of aliasing, not of two code paths -- # the first alias present wins, and the canonical name is added alongside so the # original columns survive into the output untouched. COLUMN_ALIASES = { "channel_name": ["channel_name", "channel", "epg_channel"], "channel_countries": ["channel_countries", "country_name", "country", "territory"], "prog_title": ["prog_title", "description", "programme", "program_title", "title"], "prog_st_time": ["prog_st_time", "country_start_time", "start_time", "prog_start_time"], "prog_en_time": ["prog_en_time", "country_end_time", "end_time", "prog_end_time"], # Fields the production export already carries. Where present they are # trusted rather than re-derived -- an upstream classification beats a # keyword guess off the title every time. "telecast_type": ["telecast_type"], "sports_event": ["sports_event", "event_name"], "sports_teams": ["sports_teams", "teams"], "match_level": ["match_level", "stage"], "sports_category": ["sports_category", "sub_genre"], "country_timezone": ["country_timezone", "timezone", "tz"], } def canon_columns(df: pd.DataFrame) -> pd.DataFrame: """Add canonical column names for whichever aliases this file happens to use. Returns a copy. Original columns are left in place, so nothing downstream that reads the file's own names breaks. """ out = df.copy() lower = {str(c).strip().lower(): c for c in df.columns} for canon, aliases in COLUMN_ALIASES.items(): if canon in out.columns: continue for a in aliases: src = lower.get(a) if src is not None: out[canon] = df[src] break return out # Broadcaster shorthand -> the spelling the event vocabulary uses. Without # this, "T20I World Cup" fails to match the alias "ICC Men's T20 World Cup" # and silently falls back to the generic alias "World Cup" -- which belongs to # the FIFA World Cup. That mislabels cricket as football. SYNONYMS = { "t20i": "t20", "wt20": "t20", "t20wc": "t20 world cup", "wc": "world cup", "cwc": "cricket world cup", "hlts": "", "hl": "", "hls": "", "spl": "", "wu": "", } # Sport families, used to veto a cross-sport match. SPORT_MARKERS = { "cricket": {"icc", "t20", "odi", "cricket", "test", "ipl"}, "football": {"fifa", "football", "soccer", "uefa", "epl", "laliga"}, "rugby": {"rugby", "nrl", "six", "nations"}, "tennis": {"tennis", "wimbledon", "atp", "wta", "roland"}, } def is_epg_export(df: pd.DataFrame) -> bool: """True if this looks like a raw EPG listings export rather than a schedule. Detected by signature columns, not filename, so either file type can be uploaded through the same box. """ return all(c in canon_columns(df).columns for c in REQUIRED_EPG_COLS) EPG_FRAMES = ("epg_channel_bridge", "epg_event_vocab", "epg_country_alias") def load_epg_reference(cfg: dict) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: """Fetch the three reference tables the adapter needs. Read-only. Prefers the local snapshot written by worker/build_cache.py. The event vocabulary alone is a GROUP BY over 5.9M rows on a remote server; querying it per run cost ~85s before a single listing had been read. """ from pathlib import Path cache = cfg.get("fixtures_dir") if cache: c = Path(cache) if all((c / f"{n}.parquet").exists() for n in EPG_FRAMES): return tuple(pd.read_parquet(c / f"{n}.parquet") for n in EPG_FRAMES) return fetch_epg_reference(cfg) def fetch_epg_reference(cfg: dict) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]: """Always go to the server. build_cache.py calls this to make the snapshot.""" from .reference import reference_engine eng = reference_engine(cfg) bridge = pd.read_sql(""" SELECT gsiq_channel, real_channel, COUNT(*) AS n FROM gl_ratings WHERE real_channel IS NOT NULL AND TRIM(real_channel) <> '' GROUP BY gsiq_channel, real_channel ORDER BY n DESC """, eng) # `weight` is how many evidence rows the canonical event actually has. # resolve_event() needs it to break ties: global_sports carries a handful of # stray season-suffixed spellings ("FIFA World Cup 2026", 4 rows) alongside # the real event ("FIFA World Cup", 25,934), and without the weight the # longest-alias rule picks the stray -- which then matches no evidence and # no sport category. vocab = pd.read_sql(""" SELECT a.event_name AS canonical, a.event_alias AS alias, COALESCE(g.n, 0) AS weight FROM event_alias a LEFT JOIN (SELECT event_name, COUNT(*) n FROM global_sports WHERE event_name IS NOT NULL GROUP BY event_name) g ON g.event_name = a.event_name WHERE a.event_alias IS NOT NULL AND TRIM(a.event_alias) <> '' UNION SELECT event_name, event_name, COUNT(*) FROM global_sports WHERE event_name IS NOT NULL AND TRIM(event_name) <> '' GROUP BY event_name """, eng) # Country spellings differ between the EPG feed and the reference tables # ("Korea Republic" vs "South Korea"). Unmapped, those rows find no TV # universe and the estimator refuses them outright. alias = pd.read_sql(""" SELECT country, alias_country FROM country_alias WHERE alias_country IS NOT NULL AND TRIM(alias_country) <> '' """, eng) return bridge, vocab, alias def adapt_epg(epg: pd.DataFrame, config_path) -> pd.DataFrame: """Convenience wrapper: load reference tables from config, then adapt.""" import json from pathlib import Path cfg = json.loads(Path(config_path).read_text()) bridge, vocab, alias = load_epg_reference(cfg) return adapt(epg, bridge, vocab, alias) def _norm(s: str) -> str: """Loose key for event matching: lowercase, alphanumeric only.""" return re.sub(r"[^a-z0-9]+", " ", str(s).lower()).strip() def _tokens(s: str) -> set[str]: out = set() for w in _norm(s).split(): w = SYNONYMS.get(w, w) out.update(x for x in w.split() if x) return out def _family(toks: set[str]) -> str | None: for fam, marks in SPORT_MARKERS.items(): if toks & marks: return fam return None EVIDENCE_FLOOR = 0.01 # a candidate needs >=1% of the best candidate's evidence def _prepare_vocab(vocab: dict[str, tuple[str, int]]) -> list: """Tokenise every alias once. resolve_event used to call _tokens() on all 5,378 aliases for every row it was given. On a 5,400-row file that is ~29 million tokenisations to answer 33 distinct questions, and it dominated the whole pipeline at 71 seconds. """ out = [] for alias, (canon, weight) in vocab.items(): a_toks = _tokens(alias) if len(a_toks) < 2: continue out.append((a_toks, canon, int(weight or 0), _family(_tokens(canon)), any(t.isdigit() for t in a_toks), _norm(alias))) return out def resolve_event(title: str, vocab: dict[str, tuple[str, int]], prepared=None) -> tuple[str | None, str]: """Map a free-text EPG title onto a canonical global_sports event_name. vocab maps normalised alias -> (canonical event name, evidence row count). Returns (event_name, confidence). Two rules decide it. The alias with the most tokens wins, so a specific event beats a generic one. But a candidate is only admissible if it carries a meaningful share of the evidence held by the best-evidenced candidate -- otherwise a 4-row misspelling of an event outranks the 25,934-row real one purely for being longer, and every row attached to it becomes unestimable. """ t_toks = _tokens(title) if not t_toks: return None, "low" t_fam = _family(t_toks) # (n_tokens, canonical, weight); an exact title match counts as the longest. cands: list[tuple[int, str, int]] = [] exact = vocab.get(_norm(title)) if exact: cands.append((99, exact[0], exact[1])) t_norm = _norm(title) for a_toks, canon, weight, a_fam, has_digit, a_norm in (prepared or _prepare_vocab(vocab)): if not a_toks <= t_toks: continue # A bare number carries no identity of its own, so token-subset matching # lets "Formula E: Tokyo E-Prix - Race 1" satisfy the alias "Formula 1", # and a whole Formula E schedule resolves to Formula 1 and Formula 2. # Where an alias contains a digit, demand the alias as a contiguous # phrase instead of a bag of words. if has_digit and not re.search(r"\b" + re.escape(a_norm) + r"\b", t_norm): continue # Veto a match from a different sport (e.g. "World Cup" -> FIFA on a # cricket listing). if t_fam and a_fam and a_fam != t_fam: continue cands.append((len(a_toks), canon, weight)) if not cands: return None, "low" best_w = max(w for _, _, w in cands) viable = [c for c in cands if best_w == 0 or c[2] >= EVIDENCE_FLOOR * best_w] n_tok, canon, _ = max(viable or cands) if n_tok == 99: return canon, "high" return canon, "high" if n_tok >= 3 else "medium" def adapt(epg: pd.DataFrame, channel_bridge: pd.DataFrame, event_vocab: pd.DataFrame, country_alias: pd.DataFrame | None = None) -> pd.DataFrame: """Convert a raw EPG frame into a Crystal schedule frame. channel_bridge: columns gsiq_channel, real_channel (from gl_ratings) event_vocab: columns canonical, alias, weight (event_alias + global_sports) country_alias: columns country, alias_country (from country_alias) """ df = canon_columns(epg) missing = [c for c in REQUIRED_EPG_COLS if c not in df.columns] if missing: raise ValueError( "EPG file is missing required columns: " + ", ".join(missing) + ". Accepted names for each: " + "; ".join(f"{c} = {'/'.join(COLUMN_ALIASES[c])}" for c in missing)) n_raw = len(df) # 1. Dedupe. Observed EPG exports repeat every row ~3x; left in place this # multiplies every downstream estimate. df = df.drop_duplicates(subset=["channel_name", "prog_title", "prog_st_time"]) n_dedup = n_raw - len(df) # 2. Channel: EPG uses GSIQ naming ("Star Sports 1 IN"); the evidence pool # uses broadcaster naming ("STAR Sports 1"). gl_ratings bridges the two. br = (channel_bridge.dropna(subset=["real_channel"]) .drop_duplicates(subset=["gsiq_channel"])) df = df.merge(br[["gsiq_channel", "real_channel"]], left_on="channel_name", right_on="gsiq_channel", how="left") # 3. Country / time. The feed's spelling is normalised first, or rows land # on a country the reference tables have never heard of and get refused for # a missing TV universe that is actually present under another name. df["country_name"] = df["channel_countries"] if country_alias is not None and len(country_alias): amap = {str(a).strip().lower(): c for a, c in zip(country_alias["alias_country"], country_alias["country"]) if isinstance(a, str) and a.strip()} df["country_name"] = df["country_name"].map( lambda v: amap.get(str(v).strip().lower(), v)) ts = pd.to_datetime(df["prog_st_time"], errors="coerce") df["prog_date"] = ts.dt.date df["hour"] = ts.dt.hour df["start_dt"] = ts # End time is needed for programme duration in the viewership report; it was # being dropped even when the feed supplied it. df["end_dt"] = (pd.to_datetime(df["prog_en_time"], errors="coerce") if "prog_en_time" in df.columns else pd.NaT) # 4. telecast_type. Where the feed already states it, that stands: an # upstream classification is better evidence than a keyword guess off the # title, and it is the reason the production export needs none of the # title-parsing rules. Only rows with no stated type go through classify(). stated = (df["telecast_type"].astype(str).str.strip() if "telecast_type" in df.columns else pd.Series("", index=df.index)) stated = stated.replace({"nan": "", "None": "", "-": ""}) # One answer per distinct title. A schedule repeats a handful of programme # names thousands of times; classifying per row does the same work again. _titles = df["prog_title"].fillna("").astype(str) _cls = {t: classify(t) for t in _titles.unique()} derived = pd.DataFrame({"_tt": _titles.map(lambda t: _cls[t][0]), "_ttc": _titles.map(lambda t: _cls[t][1])}, index=df.index) df["telecast_type"] = stated.where(stated != "", derived["_tt"]) df["tt_confidence"] = pd.Series("high", index=df.index).where( stated != "", derived["_ttc"]) # 5. event_name from the alias vocabulary. if "weight" in event_vocab.columns: vocab = {_norm(a): (c, int(w or 0)) for a, c, w in zip(event_vocab["alias"], event_vocab["canonical"], event_vocab["weight"])} else: # offline tests supply alias/canonical only vocab = {_norm(a): (c, 1) for a, c in zip(event_vocab["alias"], event_vocab["canonical"])} # Resolve from the title, then from the feed's own sports_event where the # title gave nothing. The production export names the competition directly # ("Formula E"), which the title often does not. prepared = _prepare_vocab(vocab) _res = {t: resolve_event(t, vocab, prepared) for t in _titles.unique()} ev = pd.DataFrame({"event_name": _titles.map(lambda t: _res[t][0]), "ev_confidence": _titles.map(lambda t: _res[t][1])}, index=df.index) df = pd.concat([df, ev], axis=1) if "sports_event" in df.columns: need = df["event_name"].isna() if need.any(): se = df.loc[need, "sports_event"].fillna("").astype(str) _alt = {t: resolve_event(t, vocab, prepared) for t in se.unique()} df.loc[need, "event_name"] = se.map(lambda t: _alt[t][0]) df.loc[need, "ev_confidence"] = se.map(lambda t: _alt[t][1]) # Fields the feed already supplies are carried straight through; the # estimator reads them, and _merge_fixtures only fills what is still blank. for col in ("sports_teams", "match_level", "sports_category", "country_timezone"): if col not in df.columns: df[col] = "" # 6. Overall row confidence = weakest link, plus channel resolution. rank = {"high": 2, "medium": 1, "low": 0} df["confidence"] = [ min([r_tt, r_ev, 2 if isinstance(rc, str) and rc else 0], key=lambda x: x) for r_tt, r_ev, rc in zip(df.tt_confidence.map(rank), df.ev_confidence.map(rank), df.real_channel) ] inv = {v: k for k, v in rank.items()} df["confidence"] = df["confidence"].map(inv) # The engine adds its own `confidence` column downstream, so the adapter's # is named epg_confidence to avoid a duplicate-column collision. df = df.rename(columns={"confidence": "epg_confidence"}) df["broadcast_date"] = df["prog_date"] out = df[["event_name", "country_name", "channel_name", "real_channel", "telecast_type", "prog_date", "broadcast_date", "hour", "prog_title", "epg_confidence", "tt_confidence", "ev_confidence", "sports_teams", "match_level", "sports_category", "country_timezone", "start_dt", "end_dt"]] out.attrs["n_raw"] = n_raw out.attrs["n_duplicates_removed"] = n_dedup return out.reset_index(drop=True)