Tivimate EPG

#!/usr/bin/env python3
from datetime import datetime, timedelta, timezone
import time, uuid, json, requests, xml.etree.ElementTree as ET, os, re, platform

# pwd und grp gibt es nur unter Linux → auf Windows ignorieren
if platform.system() != "Windows":
    import pwd, grp

def login(data, credentials, headers):
    x = 0
    while x < 120:
        mac = str(uuid.uuid4())
        ter = str(uuid.uuid4())
        auth_url = 'https://api.prod.sngtv.magentatv.de/EPG/JSON/Authenticate'
        auth_data = (
            '{"areaid":"1","cnonce":"c4b11948545fb3089720dd8b12c81f8e",'
            f'"mac":"{mac}","preSharedKeyID":"NGTV000001","subnetId":"4901",'
            '"templatename":"NGTV","terminalid":"' + ter + '","terminaltype":"WEB-MTV",'
            '"terminalvendor":"WebTV","timezone":"UTC","usergroup":"-1",'
            '"userType":3,"utcEnable":1}'
        )
        s = requests.Session()
        try:
            r = s.post(auth_url, timeout=10, data=auth_data, headers=headers)
            r.raise_for_status()
            if r.json().get("retcode") == "-2":
                time.sleep(0.2)
                x += 1
                continue
            print("Login erfolgreich.")
            return True, {"cookies": s.cookies.get_dict(), "data": None}
        except Exception as e:
            print(f"Login-Fehler (Versuch {x+1}): {e}")
            time.sleep(1)
            x += 1
    print("Login nach 120 Versuchen fehlgeschlagen.")
    return False, None


def channels(data, session, headers=None):
    if headers is None:
        headers = {}
    url = 'https://api.prod.sngtv.magentatv.de/EPG/JSON/AllChannel'
    payload = ('{"properties":[{"name":"logicalChannel",'
               '"include":"/channellist/logicalChannel/contentId,'
               '/channellist/logicalChannel/name"}],'
               '"metaDataVer":"Channel/1.1","channelNamespace":"2",'
               '"filterlist":[{"key":"IsHide","value":"-1"}],"returnSatChannel":0}')
    
    headers.update({"X_csrftoken": session["cookies"].get("CSRFSESSION", "")})
    
    try:
        r = requests.post(url, data=payload, headers=headers, cookies=session["cookies"], timeout=10)
        r.raise_for_status()
        ch_list = r.json().get("channellist", [])
        return {ch["contentId"]: ch.get("name", "") for ch in ch_list}
    except Exception as e:
        print(f"Fehler beim Laden der Kanalliste: {e}")
        return {}


def epg_main_links(data, channels, settings, session, headers=None):
    if headers is None:
        headers = {}
    headers.update({"X_csrftoken": session["cookies"].get("CSRFSESSION", "")})
    
    today = datetime.today()
    links = []
    days = int(settings.get("days", 7))
    
    for day in range(days):
        sd = today.replace(hour=6, minute=0, second=0) + timedelta(days=day)
        ed = today.replace(hour=5, minute=59, second=0) + timedelta(days=day + 1)
        
        ts = sd.astimezone(timezone.utc).strftime("%Y%m%d%H%M%S")
        te = ed.astimezone(timezone.utc).strftime("%Y%m%d%H%M%S")
        
        p = {
            "type": 2, "isFiltrate": 0, "orderType": 4, "isFillProgram": 1,
            "channelNamespace": "2", "offset": 0, "count": -1,
            "properties": [{"name": "playbill", "include": "subName,id,name,starttime,endtime,channelid,ratingid,genres,introduce,cast,country,pictures,producedate,seasonNum,subNum"}],
            "endtime": te, "begintime": ts
        }
        
        links.append({
            "url": "https://api.prod.sngtv.magentatv.de/EPG/JSON/PlayBillList",
            "d": json.dumps(p),
            "h": headers.copy(),
            "cc": session["cookies"]
        })
    return links


def epg_main_converter(data, channels, settings, ch_id=None):
    try:
        item = json.loads(data)
    except:
        return []
    
    pl = item.get("playbilllist") or []
    air = []
    
    def pt(s):
        try:
            dt = datetime.strptime(s.replace(" UTC+00:00", ""), "%Y-%m-%d %H:%M:%S")
            return str(int(dt.replace(tzinfo=timezone.utc).timestamp()))
        except:
            return None
    
    for prog in pl:
        cid = prog.get("channelid")
        if cid not in channels:
            continue
        start = pt(prog.get("starttime", ""))
        end = pt(prog.get("endtime", ""))
        if not start or not end:
            continue
            
        air.append({
            "c_id": cid,
            "start": start,
            "end": end,
            "title": prog.get("name", ""),
            "subtitle": prog.get("subName"),
            "desc": prog.get("introduce"),
        })
    return air


def write_epg_xml(airings, channels, filename="myteamepg.xml"):
    regex = re.compile(r"^Sport (\d+) - myTeamTV$")
    pairs = []

    for cid, name in channels.items():
        m = regex.match(name)
        if not m:
            continue
        num = m.group(1)
        xml_id = f"Sport{num}-myTeamTV.de"
        pairs.append((cid, int(num), xml_id))

    pairs.sort(key=lambda x: x[1])

    tv = ET.Element("tv", {"generator-info-name": "generate_epg.py"})
    idmap = {}

    for cid, _, xml_id in pairs:
        idmap[cid] = xml_id
        chan = ET.SubElement(tv, "channel", id=xml_id)
        ET.SubElement(chan, "display-name").text = xml_id

    for p in airings:
        cid = p["c_id"]
        if cid not in idmap:
            continue

        st = datetime.utcfromtimestamp(int(p["start"])).strftime("%Y%m%d%H%M%S") + " +0000"
        sp = datetime.utcfromtimestamp(int(p["end"])).strftime("%Y%m%d%H%M%S") + " +0000"

        prog = ET.SubElement(
            tv,
            "programme",
            {"start": st, "stop": sp, "channel": idmap[cid]}
        )

        ET.SubElement(prog, "title", {"lang": "de"}).text = p["title"]
        if p.get("subtitle"):
            ET.SubElement(prog, "sub-title").text = p["subtitle"]
        if p.get("desc"):
            ET.SubElement(prog, "desc").text = p["desc"]

    # XML schön formatieren
    rough = ET.tostring(tv, "utf-8")
    pretty = ET.ElementTree(ET.fromstring(rough)).getroot()  # einfacher Weg für Pretty-Print
    rough = ET.tostring(tv, "utf-8")
    from xml.dom import minidom
    pretty = minidom.parseString(rough).toprettyxml(indent="  ", encoding="UTF-8")
    header = b'<?xml version="1.0" encoding="UTF-8"?>\n'
    body = pretty.split(b"\n", 1)[1] if b"\n" in pretty else pretty

    base = os.path.dirname(os.path.abspath(__file__))
    epg_dir = os.path.join(base, "temp")
    os.makedirs(epg_dir, exist_ok=True)
    out = os.path.join(epg_dir, filename)

    with open(out, "wb") as f:
        f.write(header + body)

    print(f"✅ EPG-Datei erfolgreich erstellt: {out}")
    print(f"   Anzahl Kanäle: {len(pairs)}")
    print(f"   Anzahl Sendungen: {len(airings)}")


if __name__ == "__main__":
    print("Starte myTeamTV EPG-Generator für MagentaTV...")
    
    headers = {}
    success, session = login(None, None, headers)
    
    if not success:
        print("Skript wird beendet.")
        exit(1)
    
    chlist = channels(None, session, headers)
    print(f"{len(chlist)} Kanäle geladen.")
    
    settings = {"days": 7}
    links = epg_main_links(None, chlist, settings, session, headers)
    
    all_air = []
    print(f"Lade EPG für {len(links)} Tage...")
    
    for i, req in enumerate(links, 1):
        try:
            r = requests.post(req["url"], data=req["d"], headers=req["h"], cookies=req["cc"], timeout=15)
            r.raise_for_status()
            air = epg_main_converter(r.text, chlist, settings)
            all_air.extend(air)
            print(f"  Tag {i}: {len(air)} Sendungen geladen")
        except Exception as e:
            print(f"  Fehler bei Tag {i}: {e}")
    
    write_epg_xml(all_air, chlist)