Code:
#!/usr/bin/env python3"""Build an XMLTV EPG from tvprofil.com (Albanian TV guide) — single self-contained file.
Pure Python standard library: no browser.
Passes the site's custom firewall (resource sequence + bot-score ajax) and the
per-request 'bazinga' security token, then fetches the program grid for the
Albanian channel groups across several days and writes tvprofil-epg.xml.
Usage:
python3 tvprofil-build-epg.py [days] # days = how many days forward (default 6)
"""
import sys, re, json, time, html, os, subprocess, http.cookiejar
import urllib.request as U
# ---------------------------------------------------------------- config
LANG = "sq"
BASE = "https://tvprofil.com/al"
UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:149.0) Gecko/20100101 Firefox/149.0"
WORKER_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "epg-worker")
DAYS = int(sys.argv[1]) if len(sys.argv) > 1 and sys.argv[1].isdigit() else 6
GROUPS = {"Kanalet e mia", "Albanska televizija"} # Albanian/Kosovar channel groups
OUT = "tvprofil-epg.xml"
# ---------------------------------------------------------------- http session
jar = http.cookiejar.CookieJar()
op = U.build_opener(U.HTTPCookieProcessor(jar))
JSONP_HDRS = {
"Accept": "text/javascript, application/javascript, application/ecmascript, application/x-ecmascript, */*; q=0.01",
"X-Requested-With": "XMLHttpRequest",
"Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-origin",
"Pragma": "no-cache", "Cache-Control": "no-cache",
}
def g(url, ref=None, data=None, extra=None):
h = {"User-Agent": UA, "Accept-Language": "en-US,en;q=0.9"}
if ref: h["Referer"] = ref
if data:
h["Content-Type"] = "application/x-www-form-urlencoded"
h["X-Requested-With"] = "XMLHttpRequest"
h["Origin"] = "https://tvprofil.com"
if extra: h.update(extra)
try:
r = op.open(U.Request(url, data=data, headers=h), timeout=25)
return r.read().decode("utf-8", "replace"), r.status
except U.HTTPError as e:
return e.read().decode("utf-8", "replace"), e.code
# ---------------------------------------------------------------- firewall + security token
def clear_firewall():
"""Walk the firewall like a browser: landing -> css -> js -> logo -> bot-score ajax."""
g(f"{BASE}/")
page, _ = g(f"{BASE}/programacioni/", ref=f"{BASE}/")
m = re.search(r"/firewall/\?_t=([0-9.]+)&_r=([a-f0-9]+)", page)
if not m:
return "data-ts" in page # already cleared
t_tok, r_tok = m.group(1), m.group(2)
fw = f"https://tvprofil.com/firewall/?_t={t_tok}&_r={r_tok}"
g(fw, ref=f"{BASE}/programacioni/")
g(f"https://tvprofil.com/firewall/style.css?_t={t_tok}", ref=fw) # <- loading CSS is the key check
js, _ = g(f"https://tvprofil.com/firewall/javascript.js?_t={t_tok}", ref=fw)
T = re.search(r"T='([^']+)'", js).group(1)
R = re.search(r"R='([^']+)'", js).group(1)
K = re.search(r"K='([^']+)'", js).group(1)
g(f"https://tvprofil.com/firewall/image/logo.png?_t={t_tok}&p=1", ref=fw)
time.sleep(2)
fd = f"_t={T}&{K}={R}&score=30&moved=1&signals=ng".encode()
resp, _ = g("https://tvprofil.com/firewall/ajax", ref=fw, data=fd)
return '"ok":1' in resp
def get_bi():
page, _ = g(f"{BASE}/programacioni/", ref="https://tvprofil.com/firewall/")
m = re.search(r'"bi":(\d+)', page)
return int(m.group(1)) if m else None
def bazinga(datum, kanal, bi):
"""Replicates the page's per-request security-token function."""
b, c = 2, 5
a = datum + kanal + str(c)
i = len(a)
ua = kanal + datum or "none"
for ch in ua:
c += ord(ch)
while i:
i -= 1
b += (ord(a[i]) + c * 2 + bi) * i
b = str(b)
return "b" + str(ord(b[-1])), b # (param-name, value)
def program(datum, kanal, bi):
name, val = bazinga(datum, kanal, bi)
cb = f"tvprogram{LANG}{name}"
url = f"{BASE}/programacioni/program/?callback={cb}&datum={datum}&kanal={kanal}&{name}={val}"
txt, _ = g(url, ref=f"{BASE}/programacioni/", extra=JSONP_HDRS)
mj = re.match(r"^\w+\((.*)\)\s*$", txt, re.S)
return json.loads(mj.group(1) if mj else txt)
def esc(s):
return html.escape(str(s or ""), quote=True)
def fmt(ts):
g = time.gmtime(ts)
return time.strftime("%Y%m%d%H%M%S", g) + " +0000"
def get_channels():
txt, _ = g(BASE + "/programacioni/kanalet/?callback=cb",
ref=BASE + "/programacioni/", extra=JSONP_HDRS)
d = json.loads(re.match(r"^\w+\((.*)\)\s*$", txt, re.S).group(1))
seen, chans = set(), []
for grp in d["data"]:
if grp["name"] not in GROUPS:
continue
for c in grp.get("channels", []):
if c["urlID"] in seen:
continue
seen.add(c["urlID"])
chans.append({"id": str(c["kanalID"]), "slug": c["urlID"],
"name": c["name"], "logo": c.get("logo") or ""})
return chans
BLOCK = re.compile(r'data-ts="(\d+)"\s+data-len="(\d+)"[^>]*>(.*?)(?=<div class="row|$)', re.S)
A_RE = re.compile(r'<a href="[^"]*"\s+title="([^"]*)">(.*?)</a>(.*)', re.S)
def parse(proghtml):
out = []
for m in BLOCK.finditer(proghtml):
ts, ln, body = int(m.group(1)), int(m.group(2)), m.group(3)
am = A_RE.search(body)
if not am:
continue
orig = html.unescape(re.sub(r"<[^>]+>", "", am.group(1))).strip()
title = html.unescape(re.sub(r"<[^>]+>", "", am.group(2))).strip()
tail = am.group(3)
# episode marker like " 2, Ep. 11"
ep = ""
epm = re.match(r"\s*(\d+),\s*Ep\.\s*(\d+)", html.unescape(re.sub(r"<[^>]+>", " ", tail)))
if epm:
ep = f"S{epm.group(1)} E{epm.group(2)}"
# description: content of a <small> in the tail (genre or synopsis)
sm = re.search(r"<small>(.*?)</small>", tail, re.S)
desc = html.unescape(re.sub(r"<[^>]+>", "", sm.group(1))).strip() if sm else ""
out.append({"start": ts, "stop": ts + ln, "title": title,
"orig": orig, "ep": ep, "desc": desc})
return out
def main():
print("Clearing firewall...", flush=True)
if not clear_firewall():
print("FAILED firewall"); sys.exit(1)
bi = get_bi()
print("bi =", bi, flush=True)
chans = get_channels()
print(f"Channels: {len(chans)}", flush=True)
dates = [time.strftime("%Y-%m-%d", time.localtime(time.time() + i * 86400))
for i in range(DAYS)]
progs = {} # slug -> list
total = 0
for ci, c in enumerate(chans, 1):
rows = []
for dt in dates:
try:
d = program(dt, c["slug"], bi)
except Exception as e:
print(f" ! {c['slug']} {dt}: {e}"); continue
if d.get("data"):
rows += parse(d["data"]["program"])
time.sleep(0.25)
# dedup by start
uniq = {}
for r in rows:
uniq[r["start"]] = r
progs[c["slug"]] = sorted(uniq.values(), key=lambda r: r["start"])
total += len(progs[c["slug"]])
print(f" [{ci}/{len(chans)}] {c['name']}: {len(progs[c['slug']])} progs", flush=True)
# Build XMLTV
L = ['<?xml version="1.0" encoding="UTF-8"?>',
'<tv generator-info-name="tvprofil-epg">']
for c in chans:
L.append(f' <channel id="{esc(c["slug"])}">')
L.append(f' <display-name>{esc(c["name"])}</display-name>')
if c["logo"]:
L.append(f' <icon src="https://cdn-0.tvprofil.com/cdn/120x80/0/img/logo/{esc(c["logo"])}"/>')
L.append(" </channel>")
for c in chans:
for p in progs[c["slug"]]:
L.append(f' <programme start="{fmt(p["start"])}" stop="{fmt(p["stop"])}" channel="{esc(c["slug"])}">')
L.append(f' <title>{esc(p["title"])}</title>')
if p["orig"] and p["orig"] != p["title"]:
L.append(f' <sub-title>{esc(p["orig"])}</sub-title>')
d = p["desc"] + (f' ({p["ep"]})' if p["ep"] else "")
if d.strip():
L.append(f' <desc>{esc(d.strip())}</desc>')
L.append(" </programme>")
L.append("</tv>")
xml = "\n".join(L)
open(OUT, "w", encoding="utf-8").write(xml)
print(f"\nWrote {OUT}: {len(chans)} channels, {total} programmes, {len(xml)//1024} KB")
if "--upload" in sys.argv:
abspath = os.path.abspath(OUT)
print("Uploading to KV...", flush=True)
r = subprocess.run(
["npx", "wrangler", "kv", "key", "put", "--binding", "EPG_KV",
"tvprofil-epg.xml", "--path", abspath, "--remote"],
cwd=WORKER_DIR, capture_output=True, text=True)
print("KV upload:", "OK" if r.returncode == 0 else "FAILED\n" + r.stderr[:300])
if __name__ == "__main__":
main()
Originally Posted by Tuci-Puci
Posting Permissions
Reply With Quote