import argparse import sys from concurrent.futures import ThreadPoolExecutor from getpass import getpass import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry AGENT = "IPTVSmarters/1.0" def build_session(): s = requests.Session() s.headers["User-Agent"] = AGENT retry = Retry(total=4, backoff_factor=1.0, status_forcelist=(429, 500, 502, 503, 504)) ad = HTTPAdapter(max_retries=retry, pool_connections=32, pool_maxsize=32) s.mount("http://", ad) s.mount("https://", ad) return s def call(sess, base, user, pwd, action=None, **extra): q = {"username": user, "password": pwd} if action: q["action"] = action q.update(extra) r = sess.get(base + "/player_api.php", params=q, timeout=(15, 180)) r.raise_for_status() try: return r.json() except ValueError: return None def clean(v): if v is None: return "" return str(v).replace("\r", " ").replace("\n", " ").strip() def cats(sess, base, user, pwd, kind): data = call(sess, base, user, pwd, "get_" + kind + "_categories") or [] out = {} for c in data: out[str(c.get("category_id"))] = clean(c.get("category_name")) or "Other" return out def entry(fh, url, name, group="", tvg="", logo=""): name = clean(name) or "No name" fh.write('#EXTINF:-1 tvg-id="%s" tvg-name="%s" tvg-logo="%s" group-title="%s",%s\n' % (clean(tvg), name, clean(logo), clean(group), name)) fh.write(url + "\n") def live(sess, base, user, pwd, fh, ext): groups = cats(sess, base, user, pwd, "live") items = call(sess, base, user, pwd, "get_live_streams") or [] for it in items: sid = it.get("stream_id") if sid is None: continue entry(fh, "%s/live/%s/%s/%s.%s" % (base, user, pwd, sid, ext), it.get("name"), group=groups.get(str(it.get("category_id")), "Live"), tvg=it.get("epg_channel_id"), logo=it.get("stream_icon")) return len(items) def movies(sess, base, user, pwd, fh): groups = cats(sess, base, user, pwd, "vod") items = call(sess, base, user, pwd, "get_vod_streams") or [] for it in items: sid = it.get("stream_id") if sid is None: continue ext = it.get("container_extension") or "mp4" entry(fh, "%s/movie/%s/%s/%s.%s" % (base, user, pwd, sid, ext), it.get("name"), group="Movies - " + groups.get(str(it.get("category_id")), "VOD"), logo=it.get("stream_icon")) return len(items) def series(sess, base, user, pwd, fh, workers): groups = cats(sess, base, user, pwd, "series") shows = call(sess, base, user, pwd, "get_series") or [] def grab(show): sid = show.get("series_id") if sid is None: return show, None try: return show, call(sess, base, user, pwd, "get_series_info", series_id=sid) except Exception: return show, None count = 0 with ThreadPoolExecutor(max_workers=workers) as pool: for show, info in pool.map(grab, shows): if not info: continue seasons = info.get("episodes") if not isinstance(seasons, dict): continue title = clean(show.get("name")) or "Series" logo = clean(show.get("cover")) group = "Series - " + groups.get(str(show.get("category_id")), "Series") for season, eps in seasons.items(): if not isinstance(eps, list): continue try: sn = int(season) except (TypeError, ValueError): sn = 0 for ep in eps: eid = ep.get("id") if eid is None: continue ext = ep.get("container_extension") or "mp4" try: en = int(ep.get("episode_num") or 0) except (TypeError, ValueError): en = 0 tag = "S%02dE%02d" % (sn, en) entry(fh, "%s/series/%s/%s/%s.%s" % (base, user, pwd, eid, ext), "%s %s" % (title, tag), group=group, logo=logo) count += 1 return count def normalize(server): server = server.strip() if not server.startswith("http://") and not server.startswith("https://"): server = "http://" + server return server.rstrip("/") def prompt(value, text, secret=False): if value: return value return getpass(text) if secret else input(text).strip() def main(): p = argparse.ArgumentParser(description="Extract an M3U playlist from an Xtream Codes server.") p.add_argument("-s", "--server") p.add_argument("-u", "--username") p.add_argument("-p", "--password") p.add_argument("-o", "--output", default="playlist.m3u") p.add_argument("--live-ext", default="m3u8") p.add_argument("--no-live", action="store_true") p.add_argument("--no-vod", action="store_true") p.add_argument("--no-series", action="store_true") p.add_argument("-w", "--workers", type=int, default=8) a = p.parse_args() base = normalize(prompt(a.server, "Server (host:port): ")) user = prompt(a.username, "Username: ") pwd = prompt(a.password, "Password: ", secret=True) sess = build_session() info = call(sess, base, user, pwd) or {} if str(info.get("user_info", {}).get("auth")) != "1": print("Login failed. Check the server, username and password.", file=sys.stderr) return 1 print("Connected as %s (%s)." % (user, info.get("user_info", {}).get("status", "?")), file=sys.stderr) nc = nm = ns = 0 with open(a.output, "w", encoding="utf-8", newline="\n") as fh: fh.write("#EXTM3U\n") if not a.no_live: print("Reading live channels...", file=sys.stderr) nc = live(sess, base, user, pwd, fh, a.live_ext) if not a.no_vod: print("Reading movies...", file=sys.stderr) nm = movies(sess, base, user, pwd, fh) if not a.no_series: print("Reading series (this one takes a while)...", file=sys.stderr) ns = series(sess, base, user, pwd, fh, a.workers) print("Saved %s" % a.output, file=sys.stderr) print(" channels: %d" % nc, file=sys.stderr) print(" movies: %d" % nm, file=sys.stderr) print(" episodes: %d" % ns, file=sys.stderr) return 0 if __name__ == "__main__": sys.exit(main())