Spaces:
Paused
Paused
File size: 4,669 Bytes
5170d08 af3b90b 5170d08 5f6decc 5170d08 af3b90b 5170d08 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 | import re
from urllib.parse import urljoin, quote
import requests
from flask import Flask, request, Response, jsonify, send_from_directory
app = Flask(__name__, static_folder="static", static_url_path="/static")
UA = (
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/124.0 Safari/537.36"
)
TIMEOUT = 20
def fetch(url, stream=False):
headers = {"User-Agent": UA, "Accept": "*/*"}
return requests.get(
url, headers=headers, timeout=TIMEOUT, stream=stream, allow_redirects=True
)
# ─────────────────────────────────────────────
# /api/parse → fetch M3U, return JSON of channels
# ─────────────────────────────────────────────
@app.route("/api/parse")
def api_parse():
url = request.args.get("url")
if not url:
return jsonify({"error": "missing url"}), 400
try:
r = fetch(url)
r.raise_for_status()
text = r.text
except Exception as e:
return jsonify({"error": f"fetch failed: {e}"}), 502
channels = parse_m3u(text)
return jsonify({"count": len(channels), "channels": channels})
def parse_m3u(text):
channels = []
lines = [ln.strip() for ln in text.splitlines() if ln.strip()]
pending = None
for line in lines:
if line.startswith("#EXTINF"):
info = {"name": "", "logo": "", "group": "", "tvg_id": ""}
m = re.search(r'tvg-logo="([^"]+)"', line)
if m: info["logo"] = m.group(1)
m = re.search(r'group-title="([^"]+)"', line)
if m: info["group"] = m.group(1)
m = re.search(r'tvg-id="([^"]+)"', line)
if m: info["tvg_id"] = m.group(1)
if "," in line:
info["name"] = line.split(",", 1)[1].strip()
pending = info
elif line.startswith("#"):
continue
else:
if pending is not None:
pending["url"] = line
channels.append(pending)
pending = None
return channels
# ─────────────────────────────────────────────
# /api/proxy → proxy HLS playlists + segments with CORS
# ─────────────────────────────────────────────
@app.route("/api/proxy")
def api_proxy():
target = request.args.get("url")
if not target:
return Response("missing url", status=400)
try:
upstream = fetch(target, stream=True)
except Exception as e:
return Response(f"fetch failed: {e}", status=502)
content_type = upstream.headers.get("Content-Type", "")
if "mpegurl" in content_type.lower() or target.endswith(".m3u8"):
body = upstream.content.decode("utf-8", errors="replace")
rewritten = rewrite_playlist(body, target)
return Response(
rewritten,
headers={
"Access-Control-Allow-Origin": "*",
"Content-Type": "application/vnd.apple.mpegurl",
},
)
def generate():
for chunk in upstream.iter_content(chunk_size=64 * 1024):
if chunk:
yield chunk
return Response(
generate(),
headers={
"Access-Control-Allow-Origin": "*",
"Content-Type": content_type or "application/octet-stream",
},
)
def rewrite_playlist(body, base_url):
out = []
for line in body.splitlines():
s = line.strip()
if not s:
out.append(line)
continue
if s.startswith("#"):
def repl(m):
u = m.group(1)
abs_u = urljoin(base_url, u)
return f'URI="/api/proxy?url={quote(abs_u, safe="")}"'
out.append(re.sub(r'URI="([^"]+)"', repl, line))
else:
abs_u = urljoin(base_url, s)
out.append(f"/api/proxy?url={quote(abs_u, safe='')}")
return "\n".join(out)
# ─────────────────────────────────────────────
# Serve the frontend
# ─────────────────────────────────────────────
@app.route("/")
def index():
return send_from_directory("static", "index.html")
if __name__ == "__main__":
app.run(host="0.0.0.0", port=7860) |