#!/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)