Create app.py

app.py

@@ -0,0 +1,183 @@
+"""
+CroxyProxy Rotating Proxy API - HuggingFace Spaces
+90 servers: Paris, LA, Dallas, Warsaw, Amsterdam...
+
+GET  /health         - Status
+GET  /servers        - List 90 servers
+POST /proxy/fetch    - Rotating proxy
+POST /proxy/random   - Random server
+POST /proxy/batch    - Multiple URLs
+"""
+
+import json, base64, re, random, time, threading
+from datetime import datetime, timezone
+from flask import Flask, request, jsonify
+from bs4 import BeautifulSoup
+import cloudscraper
+from html import unescape
+import warnings
+warnings.filterwarnings("ignore")
+
+BASE = "https://www.croxyproxy.com"
+app = Flask(__name__)
+
+class S:
+    servers = []
+    idx = 0
+    lock = threading.Lock()
+    last = None
+    stats = {"req": 0, "ok": 0, "fail": 0}
+
+def dec(e):
+    try:
+        return json.loads(bytes.fromhex(base64.b64decode(e).decode()).decode())
+    except:
+        return None
+
+def fetch(url, sid=None):
+    sc = cloudscraper.create_scraper(browser={"browser":"chrome","platform":"windows","desktop":True})
+    S.stats["req"] += 1
+    try:
+        # 1. GET / → csrf
+        r1 = sc.get(BASE, timeout=30)
+        if r1.status_code != 200:
+            S.stats["fail"] += 1
+            return {"success": False, "error": f"Homepage {r1.status_code}"}
+        
+        s1 = BeautifulSoup(r1.text, "lxml")
+        ci = s1.find("input", {"name": "csrf"})
+        if not ci:
+            S.stats["fail"] += 1
+            return {"success": False, "error": "No CSRF"}
+        
+        # 2. POST /servers → selector page
+        r2 = sc.post(f"{BASE}/servers", data={
+            "url": url, "proxyServerId": "274", "csrf": ci["value"],
+            "demo": "0", "frontOrigin": BASE,
+        }, headers={
+            "Content-Type": "application/x-www-form-urlencoded",
+            "Origin": BASE, "Referer": BASE + "/",
+        }, allow_redirects=True, timeout=30)
+        
+        if r2.status_code != 200:
+            S.stats["fail"] += 1
+            return {"success": False, "error": f"Servers {r2.status_code}"}
+        
+        s2 = BeautifulSoup(r2.text, "lxml")
+        sel = s2.find("script", {"id": "serverSelectorScript"})
+        if not sel:
+            S.stats["fail"] += 1
+            return {"success": False, "error": "No selector"}
+        
+        # 3. Parse servers + csrf2
+        ss = [x for x in (dec(i) for i in json.loads(unescape(sel.get("data-ss","")))) if x and x.get("id")]
+        csrf2 = unescape(sel.get("data-csrf", "")).strip('"')
+        fo = unescape(sel.get("data-fo", "")).strip('"')
+        
+        if not ss:
+            S.stats["fail"] += 1
+            return {"success": False, "error": "No servers"}
+        
+        S.servers = ss
+        S.last = datetime.now(timezone.utc).isoformat()
+        
+        # Choose server
+        ch = None
+        if sid:
+            ch = next((x for x in ss if x["id"] == sid), None)
+        if not ch:
+            with S.lock:
+                ch = ss[S.idx % len(ss)]
+                S.idx += 1
+        
+        # 4. POST /requests?fso= → 302
+        r3 = sc.post(f"{BASE}/requests?fso=", data={
+            "url": url, "proxyServerId": str(ch["id"]),
+            "csrf": csrf2, "demo": "0", "frontOrigin": fo,
+        }, headers={
+            "Content-Type": "application/x-www-form-urlencoded",
+            "Origin": BASE, "Referer": f"{BASE}/servers",
+        }, allow_redirects=False, timeout=30)
+        
+        loc = r3.headers.get("Location") or r3.headers.get("location")
+        if not loc:
+            S.stats["fail"] += 1
+            return {"success": False, "error": f"No redirect ({r3.status_code})", "server": ch.get("name")}
+        
+        # 5. GET redirect → launching page → data-r
+        r4 = sc.get(loc, timeout=30, allow_redirects=True)
+        dr = re.search(r'data-r="([^"]+)"', r4.text)
+        if not dr:
+            S.stats["fail"] += 1
+            return {"success": False, "error": "No data-r", "server": ch.get("name")}
+        
+        # 6. decode data-r → GET final
+        final = base64.b64decode(dr.group(1)).decode()
+        r5 = sc.get(final, timeout=30, allow_redirects=True)
+        
+        S.stats["ok"] += 1
+        return {
+            "success": True, "status": r5.status_code,
+            "headers": dict(r5.headers), "body": r5.text, "url": url,
+            "proxy": {
+                "server_id": ch["id"], "server_name": ch.get("name"),
+                "ip": ch.get("url","").replace("https://","").replace("/__cp2.php",""),
+            },
+            "servers_available": len(ss),
+        }
+    except Exception as e:
+        S.stats["fail"] += 1
+        return {"success": False, "error": str(e)}
+
+
+@app.route("/")
+def index():
+    return jsonify({
+        "name": "CroxyProxy Rotating Proxy API",
+        "endpoints": {
+            "GET /health": "Status + stats",
+            "GET /servers": "List all servers",
+            "POST /proxy/fetch": "Rotating proxy {url, server_id?}",
+            "POST /proxy/random": "Random server {url}",
+            "POST /proxy/batch": "Multiple URLs {urls: [...]}",
+        },
+        "example": 'curl -X POST /proxy/fetch -H "Content-Type: application/json" -d \'{"url":"https://httpbin.org/ip"}\'',
+    })
+
+@app.route("/health")
+def health():
+    return jsonify({"status": "ready", "servers": len(S.servers), "last_refresh": S.last, "stats": S.stats})
+
+@app.route("/servers")
+def servers():
+    return jsonify({"count": len(S.servers), "servers": [
+        {"id": s.get("id"), "name": s.get("name"), "ip": s.get("url","").replace("https://","").replace("/__cp2.php","")}
+        for s in S.servers
+    ]})
+
+@app.route("/proxy/fetch", methods=["POST"])
+def pf():
+    d = request.get_json() or {}
+    if not d.get("url"): return jsonify({"error": "url required"}), 400
+    return jsonify(fetch(d["url"], d.get("server_id")))
+
+@app.route("/proxy/random", methods=["POST"])
+def pr():
+    d = request.get_json() or {}
+    if not d.get("url"): return jsonify({"error": "url required"}), 400
+    sid = random.choice(S.servers)["id"] if S.servers else None
+    return jsonify(fetch(d["url"], sid))
+
+@app.route("/proxy/batch", methods=["POST"])
+def pb():
+    d = request.get_json() or {}
+    urls = d.get("urls", [])
+    if not urls: return jsonify({"error": "urls required"}), 400
+    res = []
+    for u in urls:
+        res.append(fetch(u))
+        time.sleep(0.5)
+    return jsonify({"results": res})
+
+if __name__ == "__main__":
+    app.run(host="0.0.0.0", port=7860)