Update app.py

app.py

@@ -1,124 +1,109 @@
 import json
-import requests
 import gzip
-import io
-import xml.etree.ElementTree as ET
-import gradio as gr
 import re
 import difflib
+import requests
+import xml.etree.ElementTree as ET
+import gradio as gr
 
-# URLs
-CHANNELS_JSON_URL = "https://raw.githubusercontent.com/pigzillaaa/daddylive/main/channels.json"
-EPG_US_URL = "https://epgshare01.online/epgshare01/epg_ripper_US2.xml.gz"
-EPG_CA_URL = "https://epgshare01.online/epgshare01/epg_ripper_CA2.xml.gz"
-
-# ---------------- Fetch + Parse Helpers ----------------
-def fetch_json(url):
-    r = requests.get(url, timeout=15)
-    r.raise_for_status()
-    return r.json()
-
-def fetch_epg(url):
-    r = requests.get(url, timeout=20)
-    r.raise_for_status()
-    buf = io.BytesIO(r.content)
-    with gzip.open(buf, "rb") as f:
-        xml_data = f.read()
-    root = ET.fromstring(xml_data)
-    epg_map = {}
-    for channel in root.findall("channel"):
-        cid = channel.get("id")
-        display_name = channel.findtext("display-name")
-        if cid:
-            epg_map[cid.strip()] = cid.strip()
-        if display_name:
-            epg_map[display_name.strip()] = cid.strip() if cid else None
-    return epg_map
-
-# ---------------- Matching Helpers ----------------
+# --- Sources ---
+CHANNELS_URL = "https://raw.githubusercontent.com/pigzillaaa/daddylive/main/channels.json"
+US_EPG_URL = "https://epgshare01.online/epgshare01/epg_ripper_US2.xml.gz"
+US_LOCALS_URL = "https://epgshare01.online/epgshare01/epg_ripper_US_LOCALS2.xml.gz"
+CA_EPG_URL = "https://epgshare01.online/epgshare01/epg_ripper_CA2.xml.gz"
+
+# --- Manual Rules (learned fixes) ---
+RULES = {
+    "A&E USA": "A.and.E.HD.East.us2",
+    "AMC USA": "AMC.HD.us2",
+    "Adult Swim": "AdultSwim.com.Cartoon.Network.us2",
+    "BBC America (BBCA)": "BBC.America.HD.us2",
+    "BBC News Channel HD": "BBC.News.(North.America).HD.us2",
+    "BET USA": "BET.HD.us2",
+}
+
+# --- Helpers ---
 def normalize(s):
     return re.sub(r'[^a-z0-9]', '', s.lower())
 
-def find_best_match(old_id, ch_name, epg_map):
-    if not old_id and not ch_name:
-        return None
+def load_epg_map(url):
+    try:
+        r = requests.get(url, timeout=20)
+        r.raise_for_status()
+        with gzip.decompress(r.content) as f:
+            pass
+    except Exception:
+        f = gzip.decompress(r.content)
+    root = ET.fromstring(f)
+    return {ch.attrib.get("id", ""): ch.attrib.get("id", "") for ch in root.findall("channel")}
+
+def build_epg():
+    us = load_epg_map(US_EPG_URL)
+    us_locals = load_epg_map(US_LOCALS_URL)
+    ca = load_epg_map(CA_EPG_URL)
+    # Merge maps
+    epg = {**us, **us_locals, **ca}
+    return epg
 
-    # Build normalized map
-    epg_norm_map = {normalize(k): v for k, v in epg_map.items() if k}
+def find_best_match(old_id, ch_name, epg_map):
+    # Rule-based override first
+    if ch_name in RULES:
+        return RULES[ch_name]
+
+    # Normalize input
+    old_norm, name_norm = normalize(old_id), normalize(ch_name)
+    norm_map = {normalize(k): v for k, v in epg_map.items() if k}
+
+    # Try old_id
+    if old_norm in norm_map:
+        return norm_map[old_norm]
+    # Try channel name
+    if name_norm in norm_map:
+        return norm_map[name_norm]
+
+    # Fuzzy matching
+    all_norm_keys = list(norm_map.keys())
+    for candidate in difflib.get_close_matches(name_norm, all_norm_keys, n=1, cutoff=0.8):
+        return norm_map[candidate]
 
-    old_norm = normalize(old_id)
-    name_norm = normalize(ch_name)
+    return None
 
-    # Try normalized id
-    if old_norm in epg_norm_map:
-        return epg_norm_map[old_norm]
+# --- Main Update ---
+def update_channels():
+    r = requests.get(CHANNELS_URL)
+    data = json.loads(r.text)
 
-    # Try normalized name
-    if name_norm in epg_norm_map:
-        return epg_norm_map[name_norm]
+    epg_map = build_epg()
 
-    # Fuzzy match on normalized values
-    all_norm_keys = list(epg_norm_map.keys())
-    for candidate in difflib.get_close_matches(name_norm, all_norm_keys, n=1, cutoff=0.85):
-        return epg_norm_map[candidate]
+    log = []
+    for ch_name, info in data.items():
+        group = info.get("group_title", "")
+        if group not in ["UNITED STATES", "CANADA"]:
+            continue
 
-    return None
+        old_id = info.get("tvg_id", "")
+        new_id = find_best_match(old_id, ch_name, epg_map)
 
-# ---------------- Main Processing ----------------
-def fix_channels():
-    # Load sources
-    channels = fetch_json(CHANNELS_JSON_URL)
-    us_epg = fetch_epg(EPG_US_URL)
-    ca_epg = fetch_epg(EPG_CA_URL)
+        if new_id:
+            if new_id != old_id:
+                info["tvg_id"] = new_id
+                log.append(f"✅ {group} | {ch_name} → {new_id}")
+        else:
+            log.append(f"⚠️ {group} | {ch_name} (no match, kept {old_id})")
 
-    updated = {}
-    log_lines = []
+    # Save updated JSON
+    with open("channels_updated.json", "w") as f:
+        json.dump(data, f, indent=4)
 
-    for ch_name, ch_data in channels.items():
-        group = ch_data.get("group_title", "").upper()
-        old_id = ch_data.get("tvg_id", "").strip()
-        new_id = None
+    return "\n".join(log), "channels_updated.json"
 
-        if group == "UNITED STATES":
-            new_id = find_best_match(old_id, ch_name, us_epg)
+# --- Gradio UI ---
+with gr.Blocks() as demo:
+    gr.Markdown("## Project 2: Update Channels JSON with New EPG IDs")
+    run_btn = gr.Button("Run Update")
+    log_out = gr.Textbox(label="Update Log", lines=25)
+    file_out = gr.File(label="Download Updated JSON")
 
-        elif group == "CANADA":
-            new_id = find_best_match(old_id, ch_name, ca_epg)
+    run_btn.click(fn=update_channels, outputs=[log_out, file_out])
 
-        # Apply update if found
-        if new_id:
-            ch_data["tvg_id"] = new_id
-            if group in ["UNITED STATES", "CANADA"]:
-                log_lines.append(f"✅ {group} | {ch_name} → {new_id}")
-        else:
-            if group in ["UNITED STATES", "CANADA"]:
-                log_lines.append(f"⚠️ {group} | {ch_name} (no match, kept {old_id})")
-
-        updated[ch_name] = ch_data
-
-    # Write output
-    updated_file = "/tmp/channels_updated.json"
-    with open(updated_file, "w", encoding="utf-8") as f:
-        json.dump(updated, f, indent=4, ensure_ascii=False)
-
-    # Sort log: show all ✅ first, ⚠️ unmatched at bottom
-    fixed = [l for l in log_lines if l.startswith("✅")]
-    unmatched = [l for l in log_lines if l.startswith("⚠️")]
-    log_text = "\n".join(fixed + ["", "---- Unmatched ----"] + unmatched if unmatched else fixed)
-
-    return updated_file, log_text
-
-# ---------------- Gradio UI ----------------
-demo = gr.Interface(
-    fn=fix_channels,
-    inputs=[],
-    outputs=[
-        gr.File(label="Download Updated channels.json"),
-        gr.Textbox(label="Log (US & Canada only)", lines=25, interactive=False)
-    ],
-    title="Project 2: Channels.json Updater",
-    description="Updates tvg_id for UNITED STATES and CANADA channels using new EPG XMLs. Uses fuzzy matching to handle naming differences (e.g., TSN1 ⇔ TSN.1). Outputs a single updated file + log."
-)
-
-if __name__ == "__main__":
-    demo.launch()
+demo.launch()