#!/usr/bin/env python3 """Turn a rights/distribution grid PDF into an EPG workbook Crystal can read. What the input actually is -------------------------- Not an EPG. It is the grid a rights team publishes: one row per (territory, platform, broadcast partner), one COLUMN per session, and the channel name written in the cell: Spain | Linear (FTA) | RTVE | N/A | N/A | N/A | N/A | Teledeporte Crystal wants the transpose of that -- one row per telecast. So each non-N/A cell becomes a row, and the column it came from decides the telecast type. Why the parse is done on word coordinates ----------------------------------------- `extract_text()` interleaves wrapped cells: a territory whose name runs to two lines comes back with its characters woven into the neighbouring cell's, which produced entries like "Pan-Africa ( T E u x n c i l s . i A a l ...". Words are clustered by their y position instead and assigned to a column by x against the header centres, which survives wrapping. Times ----- The grid carries almost no times. Two forms appear in cells and both are the market's OWN local clock, because that is how a broadcaster publishes a slot: Eurosport 1 (Delayed 23:00) -> 23:00 local, same day, RECORDED Eurosport 1 (18/8 - 21:00) -> 21:00 local on 18 Aug Everything else is a live simultaneous session, so it has no local time in the document at all -- it has ONE moment, and each market sees it on its own clock. Those need `--sessions`, given in UTC, and are converted per territory. python3 worker/pdf_to_epg.py schedule.pdf out.xlsx \ --event "2026 London E-Prix" \ --sessions '{"FP1":"2026-08-16T08:30Z","FP2":"2026-08-16T12:00Z", "Qualifying":"2026-08-16T14:00Z","Race":"2026-08-17T14:03Z"}' There is no default. A session time invented here would travel all the way to an audience figure without anyone seeing it was a guess. """ from __future__ import annotations import argparse import json import re import sys from datetime import datetime, timedelta, timezone from pathlib import Path from zoneinfo import ZoneInfo BASE = Path(__file__).resolve().parent.parent sys.path.insert(0, str(BASE)) _vendor = BASE / "vendor" if _vendor.is_dir(): sys.path.insert(0, str(_vendor)) import pandas as pd # noqa: E402 import pdfplumber # noqa: E402 SESSIONS = ["FP1", "FP2", "Qualifying", "Race", "Highlights"] # Session column -> how Crystal should treat the broadcast. Highlights is the # only one that is not live; a cell saying "Delayed" overrides this per row. TELECAST = {"FP1": "LIVE", "FP2": "LIVE", "Qualifying": "LIVE", "Race": "LIVE", "Highlights": "HIGHLIGHTS"} # A live session happens at one instant worldwide, so a local start time can # only be produced with a zone per territory. Kept explicit rather than guessed # from the name: a wrong zone shifts a broadcast into a different daypart, and # hour-of-day is one of the measured factors in the estimate. ZONES = { "Albania & Kosovo": "Europe/Tirane", "Argentina": "America/Argentina/Buenos_Aires", "Australia": "Australia/Sydney", "Austria": "Europe/Vienna", "Bangladesh": "Asia/Dhaka", "Belgium": "Europe/Brussels", "Bosnia": "Europe/Sarajevo", "Brazil": "America/Sao_Paulo", "Bulgaria": "Europe/Sofia", "Canada": "America/Toronto", "Chile": "America/Santiago", "Croatia": "Europe/Zagreb", "Czech Republic": "Europe/Prague", "Denmark": "Europe/Copenhagen", "Estonia": "Europe/Tallinn", "Finland": "Europe/Helsinki", "France": "Europe/Paris", "Germany": "Europe/Berlin", "Greece": "Europe/Athens", "Hungary": "Europe/Budapest", "Iceland": "Atlantic/Reykjavik", "India": "Asia/Kolkata", "Indonesia": "Asia/Jakarta", "Israel": "Asia/Jerusalem", "Italy": "Europe/Rome", "Japan": "Asia/Tokyo", "Latvia": "Europe/Riga", "Lithuania": "Europe/Vilnius", "Malaysia": "Asia/Kuala_Lumpur", "Mexico": "America/Mexico_City", "Netherlands": "Europe/Amsterdam", "New Zealand": "Pacific/Auckland", "Norway": "Europe/Oslo", "Philippines": "Asia/Manila", "Poland": "Europe/Warsaw", "Portugal": "Europe/Lisbon", "Republic of Ireland": "Europe/Dublin", "Romania": "Europe/Bucharest", "Singapore": "Asia/Singapore", "Slovakia": "Europe/Bratislava", "Slovenia": "Europe/Ljubljana", "South Africa": "Africa/Johannesburg", "South Korea": "Asia/Seoul", "Spain": "Europe/Madrid", "Sweden": "Europe/Stockholm", "Switzerland": "Europe/Zurich", "Turkey": "Europe/Istanbul", "UK": "Europe/London", "Ukraine": "Europe/Kyiv", "USA": "America/New_York", "Vietnam": "Asia/Ho_Chi_Minh", "Thailand": "Asia/Bangkok", "Taiwan": "Asia/Taipei", # Pan feeds cover many markets from one signal. The zone is the one the # feed is cut for; Crystal keeps these out of single-market totals anyway. "Pan - Europe": "Europe/London", "Pan - South East Asia": "Asia/Singapore", "Pan South & Central America": "America/Bogota", "Pan Eurasia": "Asia/Almaty", "Pan-Africa": "Africa/Johannesburg", "Pan-MENA": "Asia/Dubai", "Greater China": "Asia/Shanghai", "Pan - APAC": "Asia/Singapore", "Luxembourg": "Europe/Luxembourg", "Monaco": "Europe/Monaco", "El Salvador": "America/El_Salvador", "Pakistan": "Asia/Karachi", "North America": "America/New_York", "Indian Subcontinent": "Asia/Kolkata", # The names COUNTRY_ALIAS maps onto, since the rename happens before this # lookup. Listed rather than reverse-resolved: one table to read, not two. "China": "Asia/Shanghai", "Ireland": "Europe/Dublin", "Albania": "Europe/Tirane", "Pan Africa": "Africa/Johannesburg", "Pan Middle East": "Asia/Dubai", "Pan Asia (Indian Subcontinent)": "Asia/Kolkata", "Pan C&S America (Latin America)": "America/Bogota", } # A rights grid names territories the way a rights deal does; Crystal names # them the way its reference data does. Only same-footprint pairs are mapped. # # Deliberately NOT mapped, because no equivalent footprint exists and a near # miss would inflate the TV universe rather than leave an honest gap: # Pan - Europe, Pan - APAC, Pan - South East Asia, North America. # Those stay flagged MISSING_TVU_REFUSED, which is the correct answer. COUNTRY_ALIAS = { "Greater China": "China", "Republic of Ireland": "Ireland", "Pan-Africa": "Pan Africa", "Pan South & Central America": "Pan C&S America (Latin America)", "Indian Subcontinent": "Pan Asia (Indian Subcontinent)", # Approximate: the deal covers Kosovo too, which Crystal has no separate # universe for. The overstatement is small and the alternative is nothing. "Albania & Kosovo": "Albania", # Approximate the other way: MENA includes North Africa, Crystal's # Pan Middle East does not, so this UNDER-states rather than over-states. "Pan-MENA": "Pan Middle East", } def clean_territory(t: str) -> str: """Territory name without its parenthetical exclusion list. Those lists wrap, and a wrapped line in this PDF comes back letter-spaced ("( E x c l . A l g e r ia ,"), which then reads as a separate territory. They carry nothing Crystal uses -- it matches on the country name -- so they are dropped rather than repaired. """ t = re.sub(r"\([^)]*\)?", "", t) # incl. an unclosed trailing "(" return re.sub(r"\s+", " ", t).strip(" -–,") # ------------------------------------------------------------------ parsing def parse_grid(pdf_path: Path) -> pd.DataFrame: """The PDF's rows, as they appear on the page.""" out = [] with pdfplumber.open(str(pdf_path)) as pdf: for pg in pdf.pages: words = pg.extract_words() hdr = {w["text"]: w for w in words if w["text"] in ("Platform", *SESSIONS)} if "Platform" not in hdr: continue centres = {k: (w["x0"] + w["x1"]) / 2 for k, w in hdr.items()} below = max(w["bottom"] for w in hdr.values()) # Cluster words into visual rows. 4pt buckets: tight enough to keep # adjacent rows apart, loose enough that a cell's second line lands # with its own row rather than the next one. lines: dict[int, list] = {} for w in words: if w["top"] < below: continue lines.setdefault(round(w["top"] / 4), []).append(w) # The Platform token is the anchor, not a fixed x offset. A long # territory ("Pan - Europe (excl. Germany, France Poland)") is wide # enough to reach into the Platform column, and splitting on the # header centre put its tail in Platform and dropped the row. prev = None for _, ws in sorted(lines.items()): ws.sort(key=lambda w: w["x0"]) anchor = next((w for w in ws if w["text"] in ("Linear", "Digital")), None) if anchor is None: # A wrapped continuation line, or a region banner. Fold it # into the row above rather than dropping it -- that is # where the rest of a multi-line cell lives. if prev is None or len(ws) > 12: continue for w in ws: # A wrapped line comes back letter-spaced, one character # per word ("1 3 . c l"). Joining those produced channel # names like "1 3 .cl", so they are dropped -- the cell's # first line already carries the readable name. if len(w["text"]) == 1: continue mid = (w["x0"] + w["x1"]) / 2 if mid < centres["FP1"] - 70: continue # territory/partner tails: ignore k = min(SESSIONS, key=lambda s: abs(mid - centres[s])) prev[k] = (prev[k] + " " + w["text"]).strip() continue cells = {k: [] for k in ("Territory", "Platform", "Partner", *SESSIONS)} for w in ws: mid = (w["x0"] + w["x1"]) / 2 if w["x1"] <= anchor["x0"]: cells["Territory"].append(w["text"]) elif mid < centres["FP1"] - 70: (cells["Platform"] if w["x0"] < anchor["x1"] + 30 else cells["Partner"]).append(w["text"]) else: k = min(SESSIONS, key=lambda s: abs(mid - centres[s])) cells[k].append(w["text"]) row = {k: " ".join(v).strip() for k, v in cells.items()} if not re.match(r"^(Linear|Digital)", row["Platform"]): continue t = clean_territory(row["Territory"]) row["Territory"] = COUNTRY_ALIAS.get(t, t) out.append(row) prev = row return pd.DataFrame(out) _TIME = re.compile(r"(\d{1,2})[:.](\d{2})") _DATE = re.compile(r"(\d{1,2})\s*/\s*(\d{1,2})") def _split_cell(cell: str) -> tuple[list[str], str | None, tuple[int, int] | None, bool]: """(channels, HH:MM, (day, month), delayed?) out of one grid cell.""" if not cell or cell.strip().upper() in ("N/A", "NA", "-", ""): return [], None, None, False delayed = "delay" in cell.lower() qual = re.findall(r"\(([^)]*)\)", cell) tm = dt = None for q in qual: if (m := _TIME.search(q)): tm = f"{int(m.group(1)):02d}:{m.group(2)}" if (m := _DATE.search(q)): dt = (int(m.group(1)), int(m.group(2))) name = re.sub(r"\([^)]*\)", "", cell).strip(" -–,") # "/" separates distinct services; "&" is usually one branded pair # ("Discovery+ & MAX"), so it is left intact. parts = [c.strip() for c in name.split("/") if c.strip()] # A grid writes a channel family as "SuperSport 1 / 2 / 3". Split naively # and the second and third become channels literally named "2" and "3", # which match nothing and estimate nothing. Carry the stem across. stem = re.sub(r"[\s\d]+$", "", parts[0]).strip() if parts else "" chans = [] for i, c in enumerate(parts): if i and stem and (c.isdigit() or len(c) <= 2): c = f"{stem} {c}" chans.append(c) return chans, tm, dt, delayed def to_epg(grid: pd.DataFrame, sessions: dict, event: str, default_len_min: int = 60, competition: str | None = None) -> tuple[pd.DataFrame, list[str]]: """One row per telecast. Returns (frame, warnings).""" rows, warn, unknown = [], [], set() for _, g in grid.iterrows(): terr = g["Territory"].strip() zone = ZONES.get(terr) if zone is None: unknown.add(terr) tz = ZoneInfo(zone) if zone else timezone.utc digital = g["Platform"].lower().startswith("digital") for s in SESSIONS: chans, tm, dmy, delayed = _split_cell(g.get(s, "")) if not chans: continue base = sessions.get(s) if base is None and tm is None: warn.append(f"{terr} · {s}: no session time and none in the cell") continue if base is not None: local = base.astimezone(tz) else: # Cell-only time: anchor to the nearest session we do know so # the date is right, else to the Race day. anchor = sessions.get("Race") or next(iter(sessions.values())) local = anchor.astimezone(tz) if tm: # explicit local slot overrides hh, mm = tm.split(":") local = local.replace(hour=int(hh), minute=int(mm), second=0) if dmy: local = local.replace(day=dmy[0], month=dmy[1]) ttype = "RECORDED" if delayed else TELECAST[s] for ch in chans: rows.append({ "channel_name": ch, "channel_countries": terr, "prog_title": f"{event} - {s}", "prog_st_time": local.replace(tzinfo=None), "prog_en_time": (local + timedelta(minutes=default_len_min) ).replace(tzinfo=None), "prog_date": local.date(), "telecast_type": ttype, # The COMPETITION goes here, not the round. Crystal resolves # this against its evidence base, where the series is # "FIA Formula E World Championship" -- "2026 London E-Prix" # matches nothing and every row comes back unestimable. "sports_event": competition or event, "match_level": s, "country_timezone": zone or "", # Carried through so the digital rows can be told apart -- # Crystal estimates a linear TV audience, and a streaming # service is not one. "platform": g["Platform"], "broadcast_partner": g["Partner"], }) if unknown: warn.append("no timezone for: " + ", ".join(sorted(unknown)) + " — these were left at UTC, add them to ZONES") return pd.DataFrame(rows), warn def _parse_sessions(raw: str) -> dict: out = {} for k, v in json.loads(raw).items(): if k not in SESSIONS: raise SystemExit(f"unknown session {k!r}; expected one of {SESSIONS}") out[k] = datetime.fromisoformat(v.replace("Z", "+00:00")) if out[k].tzinfo is None: raise SystemExit(f"{k}: give the time in UTC, e.g. 2026-08-17T14:03Z") return out def main(argv=None) -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("pdf", type=Path) ap.add_argument("out", type=Path) ap.add_argument("--event", required=True, help="commercial name of the round, e.g. '2026 London E-Prix'") ap.add_argument("--competition", default=None, help="the series as Crystal knows it, e.g. " "'Formula E World Championship'. Defaults to --event, " "which usually resolves to nothing.") ap.add_argument("--sessions", required=True, help='UTC start times, e.g. \'{"Race":"2026-08-17T14:03Z"}\'') ap.add_argument("--minutes", type=int, default=60, help="assumed telecast length (default 60)") ap.add_argument("--linear-only", action="store_true", help="drop Digital rows — Crystal estimates linear TV") a = ap.parse_args(argv) grid = parse_grid(a.pdf) print(f"[pdf] {len(grid):,} grid rows · {grid['Territory'].nunique()} territories") epg, warn = to_epg(grid, _parse_sessions(a.sessions), a.event, a.minutes, a.competition) if a.linear_only: before = len(epg) epg = epg[~epg["platform"].str.lower().str.startswith("digital")] print(f"[pdf] dropped {before - len(epg):,} digital rows") epg.to_excel(a.out, index=False) print(f"[pdf] wrote {a.out} — {len(epg):,} telecasts, " f"{epg['channel_name'].nunique()} channels") for w in warn: print(f"[warn] {w}") return 0 if __name__ == "__main__": raise SystemExit(main())