from pathlib import Path from urllib.request import Request, urlopen import gzip import io import re import difflib import xml.etree.ElementTree as ET from collections import defaultdict MEDIA = Path(r"C:\Media") PLAYLIST = MEDIA / "Channels.m3u" OUTPUT_XML = MEDIA / "epg.xml" OUTPUT_GZ = MEDIA / "epg.xml.gz" # Current EPGShare country guides covering almost all of our curated list. EPG_SOURCES = { "ca": "https://epgshare01.online/epgshare01/epg_ripper_CA2.xml.gz", "us": "https://epgshare01.online/epgshare01/epg_ripper_US2.xml.gz", "fr": "https://epgshare01.online/epgshare01/epg_ripper_FR1.xml.gz", "uk": "https://epgshare01.online/epgshare01/epg_ripper_UK1.xml.gz", "ie": "https://epgshare01.online/epgshare01/epg_ripper_IE1.xml.gz", "ch": "https://epgshare01.online/epgshare01/epg_ripper_CH1.xml.gz", "au": "https://epgshare01.online/epgshare01/epg_ripper_AU1.xml.gz", "nz": "https://epgshare01.online/epgshare01/epg_ripper_NZ1.xml.gz", } COUNTRY_MAP = { "ca": "ca", "us": "us", "fr": "fr", "uk": "uk", "ie": "ie", "ch": "ch", "au": "au", "nz": "nz", } # A few common title differences between our M3U and guide listings. ALIASES = { "cf to dt": "ctv toronto", "ctv 2 atlantic": "ctv two atlantic", "ici radio canada tele": "ici radio canada tele", "ici rdi": "ici rdi", "abc eastern": "abc national feed", "abc central": "abc national feed", "abc pacific": "abc national feed pacific", "cbs eastern": "cbs streaming sd east feed", "cbs pacific": "cbs west", "fox eastern": "fox", "fox pacific": "fox", "nbc eastern": "nbc", "nbc pacific": "nbc", "pbs eastern": "pbs", "pbs pacific": "pbs", } def norm(s): s = s.lower() s = s.replace("&", " and ") s = re.sub(r"\[[^\]]+\]", " ", s) s = re.sub(r"\b(uhd|fhd|hd|sd|hevc|4k)\b", " ", s) s = re.sub(r"\b(channel|network|television|tv)\b", " ", s) s = re.sub(r"\([^)]*\)", " ", s) s = re.sub(r"[^a-z0-9]+", " ", s) return " ".join(s.split()) def parse_playlist(): text = PLAYLIST.read_text(encoding="utf-8-sig", errors="replace").splitlines() channels = [] for line in text: if not line.startswith("#EXTINF:"): continue m = re.search(r'tvg-id="([^"]+)"', line) if not m: continue tvg_id = m.group(1) title = line.split(",", 1)[1].strip() if "," in line else tvg_id base = tvg_id.split("@", 1)[0] country = base.rsplit(".", 1)[-1].lower() if "." in base else "" channels.append({ "tvg_id": tvg_id, "base_id": base, "country": country, "title": title, "norm": norm(title), }) return channels def download_gzip_xml(url): print(f"Downloading {url}") req = Request(url, headers={"User-Agent": "Mozilla/5.0"}) with urlopen(req, timeout=90) as r: data = r.read() try: return gzip.decompress(data) except OSError: return data def build_guide_index(xml_bytes): root = ET.fromstring(xml_bytes) guide_channels = {} name_index = defaultdict(list) for ch in root.findall("channel"): cid = ch.get("id", "") names = [] for dn in ch.findall("display-name"): if dn.text: names.append(dn.text.strip()) guide_channels[cid] = ch keys = {norm(cid)} for n in names: keys.add(norm(n)) for key in keys: if key: name_index[key].append(cid) programmes = defaultdict(list) for prog in root.findall("programme"): cid = prog.get("channel") if cid: programmes[cid].append(prog) return guide_channels, name_index, programmes def score_match(target, candidate): if not target or not candidate: return 0 if target == candidate: return 1.0 return difflib.SequenceMatcher(None, target, candidate).ratio() def best_match(item, name_index): targets = [item["norm"]] alias = ALIASES.get(item["norm"]) if alias: targets.insert(0, norm(alias)) # Time-zone labels in our list often correspond to a generic network EPG. stripped = re.sub(r"\b(eastern|central|mountain|pacific|hawaii)\b", " ", item["norm"]) stripped = " ".join(stripped.split()) if stripped and stripped not in targets: targets.append(stripped) for target in targets: if target in name_index: return name_index[target][0], 1.0 # Fuzzy fallback. Keep the threshold fairly strict to avoid bad guide matches. best_id = None best_score = 0.0 for key, ids in name_index.items(): for target in targets: score = score_match(target, key) if score > best_score: best_score = score best_id = ids[0] if best_score >= 0.84: return best_id, best_score return None, best_score def clone_with_new_id(element, new_id): copy = ET.fromstring(ET.tostring(element, encoding="utf-8")) copy.set("id", new_id) return copy def clone_programme(element, new_id): copy = ET.fromstring(ET.tostring(element, encoding="utf-8")) copy.set("channel", new_id) return copy playlist_channels = parse_playlist() print(f"Playlist channels: {len(playlist_channels)}") by_source = defaultdict(list) for item in playlist_channels: source_key = COUNTRY_MAP.get(item["country"]) if source_key: by_source[source_key].append(item) output_root = ET.Element("tv") matched = 0 unmatched = [] for source_key, items in by_source.items(): url = EPG_SOURCES[source_key] try: xml_bytes = download_gzip_xml(url) guide_channels, name_index, programmes = build_guide_index(xml_bytes) except Exception as exc: print(f"FAILED {source_key}: {exc}") unmatched.extend((x["title"], "source download failed") for x in items) continue print(f"{source_key.upper()}: matching {len(items)} playlist channels") for item in items: epg_id, score = best_match(item, name_index) if not epg_id: unmatched.append((item["title"], f"no match ({score:.2f})")) continue source_channel = guide_channels.get(epg_id) if source_channel is None: unmatched.append((item["title"], "channel element missing")) continue output_root.append(clone_with_new_id(source_channel, item["tvg_id"])) for prog in programmes.get(epg_id, []): output_root.append(clone_programme(prog, item["tvg_id"])) matched += 1 tree = ET.ElementTree(output_root) ET.indent(tree, space=" ") tree.write(OUTPUT_XML, encoding="utf-8", xml_declaration=True) with open(OUTPUT_XML, "rb") as source_file, gzip.open(OUTPUT_GZ, "wb", compresslevel=6) as gz: gz.write(source_file.read()) print() print(f"EPG channels matched: {matched}") print(f"EPG channels unmatched: {len(unmatched)}") print(f"Saved: {OUTPUT_XML}") print(f"Saved: {OUTPUT_GZ}") if unmatched: report = MEDIA / "epg_unmatched.txt" report.write_text( "\n".join(f"{name} - {reason}" for name, reason in unmatched) + "\n", encoding="utf-8", ) print(f"Unmatched report: {report}")