Update orchestrator.py

orchestrator.py

@@ -1,5 +1,19 @@
 """
-Orchestrator — Télécharge les listes, envoie au worker Render
+Orchestrator v2 — Envoie tout d'un coup, poll le résultat
+──────────────────────────────────────────────────────────
+
+Flow:
+1. IDLE → télécharge les listes GitHub
+2. SENDING → envoie chaque protocole au worker (1 POST chacun)
+3. POLLING → poll /job/{id} toutes les 30s
+4. COLLECTING → quand jobs finis, récupère résultats
+5. DONE → refresh pool → IDLE
+
+Requêtes HF sortantes:
+- 2× GitHub (download listes)
+- 2× POST worker (envoi proxies)
+- N× GET worker (poll status)
+= que des requêtes vers GitHub et Render = SAFE
 """
 
 import time
@@ -12,8 +26,7 @@ from enum import Enum
 logger = logging.getLogger("orchestrator")
 
 HEADERS = {"User-Agent": "Mozilla/5.0"}
-BATCH_SIZE = 20
-PAUSE_BETWEEN_BATCHES = 5
+MAX_PER_PROTO = 500  # on peut envoyer plus maintenant
 
 SOURCES = [
     (
@@ -28,13 +41,13 @@ SOURCES = [
     ),
 ]
 
-MAX_PER_PROTO = 200
-
 
 class Phase(Enum):
     IDLE = "idle"
     DOWNLOADING = "downloading"
-    TESTING = "testing"
+    SENDING = "sending"
+    POLLING = "polling"
+    COLLECTING = "collecting"
     DONE = "done"
 
 
@@ -47,12 +60,13 @@ class Orchestrator:
         self._phase = Phase.IDLE
         self._lock = threading.Lock()
 
-        self._queue = []
-        self._results = []
+        # données du cycle
+        self._proxy_lists = {}  # {protocol: [proxies]}
+        self._job_ids = {}      # {protocol: job_id}
+        self._all_results = []
         self._cycle_start = 0.0
-        self._batches_sent = 0
-        self._total_tested = 0
 
+        # stats
         self._runs = 0
         self._last_duration = 0.0
         self._last_found = 0
@@ -70,8 +84,7 @@ class Orchestrator:
             "worker_url": self.worker_url,
             "worker_ok": self._worker_ok,
             "real_ip": self.real_ip,
-            "queue_remaining": len(self._queue),
-            "results_collected": len(self._results),
+            "active_jobs": dict(self._job_ids),
         }
 
     def _detect_ip(self):
@@ -84,20 +97,18 @@ class Orchestrator:
             )
             self.real_ip = r.json().get("ip", "")
             logger.info(f"Real IP: {self.real_ip}")
-        except Exception as e:
-            logger.warning(f"IP detect failed: {e}")
+        except Exception:
+            logger.warning("IP detect failed")
 
     def _check_worker(self):
         try:
             r = requests.get(
                 f"{self.worker_url}/ping",
-                timeout=10, headers=HEADERS,
+                timeout=10,
             )
             self._worker_ok = r.status_code == 200
         except Exception:
             self._worker_ok = False
-        if not self._worker_ok:
-            logger.error("Worker unreachable!")
         return self._worker_ok
 
     def _download(self, url):
@@ -130,7 +141,8 @@ class Orchestrator:
             logger.error(f"Download failed: {e}")
             return []
 
-    def _send_batch(self, proxies, protocol):
+    def _send_to_worker(self, proxies, protocol):
+        """Envoie un gros batch au worker, retourne job_id."""
         try:
             r = requests.post(
                 f"{self.worker_url}/test",
@@ -139,20 +151,45 @@ class Orchestrator:
                     "protocol": protocol,
                     "real_ip": self.real_ip,
                 },
-                timeout=60,
+                timeout=30,
                 headers=HEADERS,
             )
             r.raise_for_status()
+            data = r.json()
             self._worker_ok = True
-            return r.json().get("working", [])
+            return data.get("job_id")
         except Exception as e:
-            logger.error(f"Batch failed: {e}")
+            logger.error(f"Send failed: {e}")
             self._worker_ok = False
-            return []
+            return None
+
+    def _poll_job(self, job_id):
+        """Poll le statut d'un job. Retourne le dict."""
+        try:
+            r = requests.get(
+                f"{self.worker_url}/job/{job_id}",
+                timeout=15, headers=HEADERS,
+            )
+            r.raise_for_status()
+            self._worker_ok = True
+            return r.json()
+        except Exception as e:
+            logger.error(f"Poll failed: {e}")
+            self._worker_ok = False
+            return None
+
+    def _cleanup_job(self, job_id):
+        """Supprime un job terminé sur le worker."""
+        try:
+            requests.delete(
+                f"{self.worker_url}/job/{job_id}",
+                timeout=10,
+            )
+        except Exception:
+            pass
 
     def tick(self):
         self._ticks += 1
-
         with self._lock:
             phase = self._phase
 
@@ -160,8 +197,12 @@ class Orchestrator:
             self._do_idle()
         elif phase == Phase.DOWNLOADING:
             self._do_download()
-        elif phase == Phase.TESTING:
-            self._do_test()
+        elif phase == Phase.SENDING:
+            self._do_send()
+        elif phase == Phase.POLLING:
+            self._do_poll()
+        elif phase == Phase.COLLECTING:
+            self._do_collect()
         elif phase == Phase.DONE:
             self._do_done()
 
@@ -172,24 +213,21 @@ class Orchestrator:
         self._detect_ip()
 
         if not self._check_worker():
-            logger.error("Skipping — worker down")
+            logger.error("Worker down, skip")
             return
 
         self._cycle_start = time.time()
-        self._queue = []
-        self._results = []
-        self._batches_sent = 0
-        self._total_tested = 0
+        self._proxy_lists = {}
+        self._job_ids = {}
+        self._all_results = []
 
         with self._lock:
             self._phase = Phase.DOWNLOADING
 
     def _do_download(self):
-        logger.info("📥 Downloading…")
+        logger.info("📥 Downloading lists…")
 
         seen = set()
-        queue = []
-
         for protocol, url in SOURCES:
             proxies = self._download(url)
             proxies = [p for p in proxies if p not in seen]
@@ -199,84 +237,127 @@ class Orchestrator:
                 random.shuffle(proxies)
                 proxies = proxies[:MAX_PER_PROTO]
 
-            for p in proxies:
-                queue.append((p, protocol))
-
+            self._proxy_lists[protocol] = proxies
             logger.info(f"   {protocol.upper()}: {len(proxies)}")
             time.sleep(1)
 
-        random.shuffle(queue)
-        self._queue = queue
-        logger.info(f"   Queue: {len(queue)}")
+        total = sum(len(v) for v in self._proxy_lists.values())
+        logger.info(f"   Total: {total}")
 
         with self._lock:
-            self._phase = Phase.TESTING
+            self._phase = Phase.SENDING
+
+    def _do_send(self):
+        """Envoie chaque protocole au worker."""
+        logger.info("📤 Sending to worker…")
+
+        for protocol, proxies in self._proxy_lists.items():
+            if not proxies:
+                continue
+
+            job_id = self._send_to_worker(proxies, protocol)
+            if job_id:
+                self._job_ids[protocol] = job_id
+                logger.info(
+                    f"   {protocol.upper()}: "
+                    f"job {job_id} ({len(proxies)} proxies)"
+                )
+            else:
+                logger.error(
+                    f"   {protocol.upper()}: send failed"
+                )
 
-    def _do_test(self):
-        if not self._queue:
+            time.sleep(1)
+
+        if not self._job_ids:
+            logger.error("No jobs created, aborting")
             with self._lock:
-                self._phase = Phase.DONE
+                self._phase = Phase.IDLE
             return
 
-        batch_items = self._queue[:BATCH_SIZE]
-        self._queue = self._queue[BATCH_SIZE:]
+        with self._lock:
+            self._phase = Phase.POLLING
+
+    def _do_poll(self):
+        """Poll les jobs en cours."""
+        all_done = True
 
-        by_proto = {}
-        for proxy, proto in batch_items:
-            by_proto.setdefault(proto, []).append(proxy)
+        for protocol, job_id in list(self._job_ids.items()):
+            data = self._poll_job(job_id)
+            if not data:
+                continue
 
-        for proto, proxies in by_proto.items():
-            working = self._send_batch(proxies, proto)
+            status = data.get("status", "unknown")
+            progress = data.get("progress", 0)
+            working = data.get("working", 0)
 
-            for w in working:
-                self._results.append({
-                    "proxy": w["proxy"],
-                    "protocol": proto,
-                    "proxy_url": f"{proto}://{w['proxy']}",
-                    "latency": w["latency"],
-                    "proxy_ip": w["proxy_ip"],
-                    "verified": w["verified"],
-                })
+            logger.info(
+                f"   {protocol.upper()} [{job_id}]: "
+                f"{status} {progress}% "
+                f"✅ {working}"
+            )
 
-            self._total_tested += len(proxies)
+            if status != "done":
+                all_done = False
 
-        self._batches_sent += 1
-        remaining = len(self._queue)
+        if all_done:
+            with self._lock:
+                self._phase = Phase.COLLECTING
 
-        logger.info(
-            f"   Batch #{self._batches_sent}: "
-            f"✅ {len(self._results)} | "
-            f"⏳ {remaining} left"
-        )
+    def _do_collect(self):
+        """Récupère les résultats de tous les jobs."""
+        logger.info("📦 Collecting results…")
 
-        if remaining > 0:
-            time.sleep(PAUSE_BETWEEN_BATCHES)
+        for protocol, job_id in self._job_ids.items():
+            data = self._poll_job(job_id)
+            if not data:
+                continue
+
+            results = data.get("results", [])
+            for r in results:
+                self._all_results.append({
+                    "proxy": r["proxy"],
+                    "protocol": protocol,
+                    "proxy_url": f"{protocol}://{r['proxy']}",
+                    "latency": r["latency"],
+                    "proxy_ip": r["proxy_ip"],
+                    "verified": r["verified"],
+                })
+
+            logger.info(
+                f"   {protocol.upper()}: {len(results)} working"
+            )
+
+            # cleanup
+            self._cleanup_job(job_id)
+
+        with self._lock:
+            self._phase = Phase.DONE
 
     def _do_done(self):
         duration = time.time() - self._cycle_start
 
-        if self._results:
-            self.pool.refresh(self._results)
+        if self._all_results:
+            self.pool.refresh(self._all_results)
 
         self._runs += 1
         self._last_duration = round(duration, 1)
-        self._last_found = len(self._results)
+        self._last_found = len(self._all_results)
 
         from collections import Counter
-        c = Counter(r["protocol"] for r in self._results)
+        c = Counter(r["protocol"] for r in self._all_results)
 
         logger.info("=" * 40)
         logger.info(f"✅ CYCLE #{self._runs}")
-        logger.info(f"   Found:   {len(self._results)}")
-        logger.info(f"   Tested:  {self._total_tested}")
-        logger.info(f"   Batches: {self._batches_sent}")
-        logger.info(f"   Time:    {self._last_duration}s")
-        logger.info(f"   SOCKS5:  {c.get('socks5', 0)}")
-        logger.info(f"   HTTP:    {c.get('http', 0)}")
+        logger.info(f"   Found:  {len(self._all_results)}")
+        logger.info(f"   Time:   {self._last_duration}s")
+        logger.info(f"   SOCKS5: {c.get('socks5', 0)}")
+        logger.info(f"   HTTP:   {c.get('http', 0)}")
         logger.info("=" * 40)
 
-        self._queue = []
-        self._results = []
+        self._proxy_lists = {}
+        self._job_ids = {}
+        self._all_results = []
 
         with self._lock:
             self._phase = Phase.IDLE
@@ -284,8 +365,7 @@ class Orchestrator:
     def ping_worker(self):
         try:
             r = requests.get(
-                f"{self.worker_url}/ping",
-                timeout=10,
+                f"{self.worker_url}/ping", timeout=10,
             )
             self._worker_ok = r.status_code == 200
         except Exception: