import json, re, urllib.parse, urllib.request from pathlib import Path from concurrent.futures import ThreadPoolExecutor, as_completed SRC = Path("/var/www/html/Extract/series_full.json") OUT = Path("/var/www/html/Extract/series_list.m3u") JSON_OUT = Path("/var/www/html/Extract/series_list.json") EXTRACT = "http://127.0.0.1:9393/extract?url=" MAX_WORKERS = 16 TIMEOUT = 15 MIN_HEIGHT = 720 SAVE_EVERY = 100 MAX_SOURCES_PER_EPISODE = 2 UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/138.0 Safari/537.36" def clean(s): return re.sub(r"\s+", " ", str(s or "")).strip() def esc(s): return str(s or "").replace('"', "'") def fetch_text(url, timeout=10, referer="", useragent=""): headers = { "User-Agent": useragent or UA, "Accept": "application/vnd.apple.mpegurl,*/*" } if referer: headers["Referer"] = referer headers["Origin"] = referer.rstrip("/") req = urllib.request.Request(url, headers=headers) with urllib.request.urlopen(req, timeout=timeout) as r: return r.status, r.read(120000).decode("utf-8", "ignore") def get_height(line): m = re.search(r"RESOLUTION=\d+x(\d+)", line) return int(m.group(1)) if m else 0 def resolve_best_quality(master_url, referer="", useragent=""): try: status, text = fetch_text(master_url, 10, referer, useragent) if status != 200 or "#EXTM3U" not in text: return "" base = master_url.rsplit("/", 1)[0] + "/" variants = [] lines = text.splitlines() for i, line in enumerate(lines): if line.startswith("#EXT-X-STREAM-INF"): h = get_height(line) if h >= MIN_HEIGHT and i + 1 < len(lines): u = lines[i + 1].strip() if u.startswith("/"): parsed = urllib.parse.urlparse(master_url) u = parsed.scheme + "://" + parsed.netloc + u elif not u.startswith("http"): u = urllib.parse.urljoin(base, u) variants.append((h, u)) if variants: variants.sort(reverse=True) return variants[0][1] return master_url except: return "" def score(u): s = 0 ul = str(u or "").lower() if ".m3u8" in ul: s += 100 if "master.m3u8" in ul: s += 100 if "index-" in ul: s += 70 if "1080" in ul: s += 90 if "720" in ul: s += 80 if "strm" in ul: s += 50 if "uqload" in ul: s += 30 return s def extract_best(embed, referer="", useragent=""): try: api = EXTRACT + urllib.parse.quote(embed, safe="") headers = {"User-Agent": useragent or UA} if referer: headers["Referer"] = referer req = urllib.request.Request(api, headers=headers) with urllib.request.urlopen(req, timeout=TIMEOUT) as r: d = json.loads(r.read().decode("utf-8", "ignore")) urls = [] best = d.get("best") or {} bu = best.get("url", "") if ".m3u8" in bu: urls.append(bu) for x in d.get("results", []): u = x.get("url", "") if ".m3u8" in u: urls.append(u) urls = list(dict.fromkeys(urls)) urls.sort(key=score, reverse=True) for u in urls[:3]: final = resolve_best_quality(u, referer, useragent) if final: return final except: pass return "" def load_items(): raw = json.loads(SRC.read_text(encoding="utf-8")) if raw and isinstance(raw, list) and isinstance(raw[0], dict) and "videos" in raw[0]: return raw items = [] for serie in raw: serie_name = clean(serie.get("name") or serie.get("original_name") or "No Series Name") logo = clean(serie.get("poster_path") or serie.get("image") or serie.get("logo") or "") subtitle = clean(serie.get("subtitle") or "") for season in serie.get("seasons", []): season_no = season.get("season_number") or "" for ep in season.get("episodes", []): ep_no = ep.get("episode_number") or "" ep_name = clean(ep.get("name") or f"الحلقة {ep_no}") title = f"{serie_name} - S{season_no}E{ep_no}" if ep_name: title += f" - {ep_name}" items.append({ "id": ep.get("id"), "title": title, "name": title, "subtitle": subtitle, "logo": logo, "image": logo, "poster_path": logo, "group": "Series", "category": "Series", "videos": ep.get("videos") or [] }) return items def process(item): item_id = item.get("id", "") name = clean(item.get("title") or item.get("name") or item.get("id") or "No Name") logo = clean(item.get("logo") or item.get("image") or item.get("poster_path") or item.get("still_path") or "") group = clean(item.get("group") or item.get("category") or "Series") subtitle = clean(item.get("subtitle") or item.get("quality") or "") videos = item.get("videos", [])[:MAX_SOURCES_PER_EPISODE] for v in videos: link = clean(v.get("link") or v.get("url") or v.get("m3u8") or "") referer = clean(v.get("header") or "") useragent = clean(v.get("useragent") or "") if not link: continue if ".m3u8" in link.lower(): final = resolve_best_quality(link, referer, useragent) if final: return {"ok": True, "row": (name, logo, group, final, subtitle)} final = extract_best(link, referer, useragent) if final: return {"ok": True, "row": (name, logo, group, final, subtitle)} return {"ok": False, "id": item_id, "title": name} def sort_key(row): name = row[0] m = re.search(r"S(\d+)E(\d+)", name, re.I) if m: season = int(m.group(1)) episode = int(m.group(2)) else: season = 999999 episode = 999999 series_name = re.split(r"\s*-\s*S\d+E\d+", name, flags=re.I)[0].strip().lower() return (series_name, season, episode, name.lower()) def write_files(rows): rows = sorted(rows, key=sort_key) lines = ["#EXTM3U"] arr = [] for name, logo, group, link, subtitle in rows: lines.append(f'#EXTINF:-1 tvg-name="{esc(name)}" tvg-logo="{esc(logo)}" group-title="{esc(group)}",{name}') lines.append(link) arr.append({ "name": name, "title": name, "subtitle": subtitle, "quality": subtitle, "image": logo, "img": logo, "logo": logo, "category": group, "group": group, "link": link, "url": link, "alt": link, "urlobj": [{"playUrl": link}] }) OUT.with_suffix(".m3u.tmp").write_text("\n".join(lines) + "\n", encoding="utf-8") JSON_OUT.with_suffix(".json.tmp").write_text(json.dumps(arr, ensure_ascii=False, indent=2), encoding="utf-8") OUT.with_suffix(".m3u.tmp").replace(OUT) JSON_OUT.with_suffix(".json.tmp").replace(JSON_OUT) items = load_items() rows = [] ok = 0 fail = 0 OUT.write_text("#EXTM3U\n", encoding="utf-8") JSON_OUT.write_text("[]", encoding="utf-8") print("TOTAL ITEMS:", len(items), flush=True) with ThreadPoolExecutor(max_workers=MAX_WORKERS) as ex: futs = [ex.submit(process, item) for item in items] for fut in as_completed(futs): try: r = fut.result() except Exception: fail += 1 continue if not r or not r.get("ok"): fail += 1 continue ok += 1 rows.append(r["row"]) print("OK", ok, r["row"][0], flush=True) if ok % SAVE_EVERY == 0: write_files(rows) print("SAVED:", ok, "items", "FAIL:", fail, flush=True) write_files(rows) print("DONE OK:", ok, "FAIL:", fail) print("M3U:", OUT) print("JSON:", JSON_OUT)