Spaces:
Paused
Paused
| 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 | |
| # ───────────────────────────────────────────── | |
| 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 | |
| # ───────────────────────────────────────────── | |
| 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 | |
| # ───────────────────────────────────────────── | |
| def index(): | |
| return send_from_directory("static", "index.html") | |
| if __name__ == "__main__": | |
| app.run(host="0.0.0.0", port=7860) |