[codex] unify CLI progress with engine snapshot

docs/CLI.md

@@ -112,6 +112,16 @@ python moon_cli.py \
 The `fuckingfast.co` entries are extracted without a browser. The datanodes entry
 causes the shared Chrome instance to start; it is not one browser per worker.
 
+## Progress output
+
+While a run is active, the CLI samples the same `Engine.snapshot()` contract used
+by the GUI. The phase is explicit: `extracting [done/total]` while provider pages
+are being resolved, then `downloading [done/total]` while files are transferring.
+The aggregate speed, active-download count and byte total come from that snapshot,
+so the CLI and GUI use one calculation. Identical snapshots are not printed again,
+which keeps a slow extraction phase from filling the terminal with repeated zero
+speed lines.
+
 ## Exit codes
 
 | Code | Meaning |

moon_cli.py

@@ -7,8 +7,8 @@
     python moon_cli.py --urls links.txt --output /path/to/downloads
     python moon_cli.py --urls links.txt --output ./dl --browsers 8 --streams 24 --retries 3
 """
-import os, sys, asyncio, threading, argparse
-import time, traceback, collections
+import os, sys, asyncio, argparse
+import time, traceback
 from urllib.parse import urlparse, unquote
 
 from moon_download import (
@@ -21,6 +21,7 @@
     download_file,
     RunFatalControl,
 )
+from moon_engine import Engine
 
 # ── EXTRACTION ────────────────────────────────────────────────────────────────
 # Both host front-ends changed in 2026; the extraction layer now lives in
@@ -55,6 +56,28 @@
 def _fmt_speed(mbs: float) -> str:
     return f"{mbs:.1f} MB/s" if mbs >= 1 else f"{mbs*1024:.0f} KB/s"
 
+def _progress_line(snapshot: dict, total: int) -> str:
+    metrics = snapshot["metrics"]
+    stage = metrics["stage"]
+    if stage == "extracting":
+        phase = f"extracting {metrics['extract_done']}/{metrics['extract_total']}"
+    elif stage == "downloading":
+        phase = f"downloading {metrics['dl_done']}/{metrics['dl_total']}"
+    else:
+        phase = f"{metrics['ok']}/{total} done"
+    return (f"  [{int(metrics['elapsed_s'] // 60):02d}:{int(metrics['elapsed_s'] % 60):02d}]  "
+            f"{phase}  |  {metrics['active']} active  |  "
+            f"{_fmt_speed(metrics['speed_mbs'])}  |  "
+            f"{metrics['bytes_total']/1e9:.2f} GB")
+
+def _progress_key(snapshot: dict) -> tuple:
+    metrics = snapshot["metrics"]
+    return (
+        metrics["stage"], metrics["extract_done"], metrics["extract_total"],
+        metrics["dl_done"], metrics["dl_total"], metrics["active"],
+        metrics["speed_mbs"], metrics["bytes_total"], metrics["ok"], metrics["fail"],
+    )
+
 async def run(urls: list[str], output_dir: str, n_workers: int,
               max_dl: int, max_retries: int, proxy_path: str, is_default_proxies: bool = False):
 
@@ -70,14 +93,11 @@ async def run(urls: list[str], output_dir: str, n_workers: int,
     all_tasks     : list        = []
     tasks_lock    = asyncio.Lock()
     kill_counts   : dict[str,int] = {}
-    bytes_acc     = collections.deque(maxlen=200000)
-    lock          = threading.Lock()
-    n_done        = 0
     all_done      = asyncio.Event()
-    ok_count      = 0
-    fail_count    = 0
-    dls_active    = 0
     fatal_control = RunFatalControl()
+    progress      = Engine()
+    progress.begin_external_progress(len(urls), output_dir, n_proxies)
+    bytes_acc     = progress.progress_bytes()
 
     cfg = {"browsers": n_workers, "dl_streams": max_dl, "retries": max_retries,
            "total_links": len(urls)}
@@ -92,59 +112,45 @@ async def run(urls: list[str], output_dir: str, n_workers: int,
             print(f"  [rename] {filename} -> {rec.filename} (filename collision)")
         await q.put((url, 1, rec))
 
-    def mark_done():
-        nonlocal n_done
-        n_done += 1
-        if n_done >= len(urls): all_done.set()
-
-    t0 = time.monotonic()
-
     # Progress printer (runs every 2s)
     stop_progress = asyncio.Event()
+    last_progress_key = None
     async def progress_loop():
+        nonlocal last_progress_key
         while not stop_progress.is_set():
             try:
                 await asyncio.wait_for(stop_progress.wait(), timeout=2.0)
             except asyncio.TimeoutError:
                 pass
             if stop_progress.is_set():
                 break
-            snap = list(bytes_acc)
-            now  = time.monotonic()
-            cut  = now - 3.0
-            recent = [(t, b) for t, b in snap if t > cut]
-            mbs = 0.0
-            if len(recent) > 1:
-                span = max(min(3.0,now - t0),0.05)
-                mbs  = sum(b for _, b in recent) / span / 1_048_576
-            total_dl = sum(b for _, b in snap)
-            el = now - t0
-            with lock:
-                ok, dls = ok_count, dls_active
-            print(f"  [{int(el//60):02d}:{int(el%60):02d}]  "
-                  f"{ok}/{len(urls)} done  |  "
-                  f"{dls} active  |  "
-                  f"{_fmt_speed(mbs)}  |  "
-                  f"{total_dl/1e9:.2f} GB", flush=True)
+            snapshot = progress.snapshot()
+            key = _progress_key(snapshot)
+            if key != last_progress_key:
+                line = _progress_line(snapshot, len(urls))
+                print(line, flush=True)
+                last_progress_key = key
 
     # Kept in a local: a task with no reference can be collected mid-run.
     progress_t = asyncio.create_task(progress_loop())  # noqa: F841
 
     async def do_dl(proxy_url, cookies, filename, orig_url, rec):
-        nonlocal ok_count, fail_count, dls_active
         async with dl_sem:
             if fatal_control.is_set():
                 rec.status = "aborted"
                 return
-            with lock: dls_active += 1
+            progress.mark_download_start()
+            finalized = False
             try:
                 dest = os.path.join(output_dir, filename)
 
                 if os.path.exists(dest):
-                    with lock: ok_count += 1
                     print(f"  [exists] {filename}")
                     rec.status = "ok"; rec.dl_s = 0.0
-                    mark_done(); return
+                    if progress.mark_download_end(True):
+                        all_done.set()
+                    finalized = True
+                    return
 
                 kc       = kill_counts.get(orig_url, 0)
                 kill_evt = asyncio.Event()
@@ -155,30 +161,34 @@ async def do_dl(proxy_url, cookies, filename, orig_url, rec):
                 rec.dl_s = max(time.monotonic() - rec.dl_start, 0.001)
 
                 if ok:
-                    with lock: ok_count += 1
                     spd = f"  ({rec.avg_mbs:.1f} MB/s)" if rec.avg_mbs > 0 else ""
                     print(f"  [ok] {filename}{spd}")
-                    rec.status = "ok"; mark_done()
+                    rec.status = "ok"
+                    if progress.mark_download_end(True):
+                        all_done.set()
+                    finalized = True
                 elif msg == "stall_killed":
                     new_kc = kc + 1; kill_counts[orig_url] = new_kc
                     print(f"  [kill#{new_kc}] {filename}  ({bytes_done//(1<<20)}MB) -> re-extract")
                     rec.queued_at = time.monotonic(); rec.status = "pending"
+                    progress.mark_download_retry(); finalized = True
                     if not fatal_control.is_set():
                         await q.put((orig_url, 1, rec))
                 elif msg == "aborted_disk_full":
-                    rec.status = "aborted"
+                    rec.status = "aborted"; progress.mark_download_aborted(); finalized = True
                 else:
-                    with lock: fail_count += 1
                     failed_urls.append(orig_url)
                     rec.status = "fail"; rec.error = msg
                     if msg != "disk_full":
                         print(f"  [fail] {filename}: {msg}")
-                    mark_done()
+                    if progress.mark_download_end(False):
+                        all_done.set()
+                    finalized = True
             finally:
-                with lock: dls_active -= 1
+                if not finalized:
+                    progress.mark_download_aborted()
 
     async def browser_worker(get_browser, wid):
-        nonlocal ok_count, fail_count
         while not fatal_control.is_set():
             if all_done.is_set() and q.empty(): break
             try:
@@ -218,6 +228,7 @@ async def _task(pu=link, fn=filename, ou=url, r=rec):
                             await do_dl(pu, "", fn, ou, r)
                         t = asyncio.create_task(_task())
                         async with tasks_lock: all_tasks.append(t)
+                        progress.mark_extraction(url, True)
                         success = True
                 elif DATANODES_HOST in parsed.netloc:
                     # API key set -> single JSON GET, no browser, no captcha.
@@ -240,6 +251,7 @@ async def _task(pu=proxy_url, co=cookies, fn=filename, ou=url, r=rec):
                             await do_dl(pu, co, fn, ou, r)
                         t = asyncio.create_task(_task())
                         async with tasks_lock: all_tasks.append(t)
+                        progress.mark_extraction(url, True)
                         success = True
                 else:
                     host = parsed.hostname or parsed.netloc or "(missing host)"
@@ -271,9 +283,10 @@ async def _task(pu=proxy_url, co=cookies, fn=filename, ou=url, r=rec):
                 q.task_done(); continue
 
             if not success and not is_re and not fatal_control.is_set():
-                with lock: fail_count += 1
                 failed_urls.append(url)
-                rec.status = "fail"; mark_done()
+                rec.status = "fail"
+                if progress.mark_extraction(url, False):
+                    all_done.set()
 
             q.task_done()
 
@@ -321,17 +334,21 @@ async def _task(pu=proxy_url, co=cookies, fn=filename, ou=url, r=rec):
         # Report persistence is independent of the fatal download state.
         print(f"[warn] Log save error: {e}")
 
-    el = time.monotonic() - t0
-    total_bytes = sum(b for _, b in bytes_acc)
+    progress.finish_external_progress()
+    metrics = progress.snapshot()["metrics"]
+    el = metrics["elapsed_s"]
+    total_bytes = metrics["bytes_total"]
     disk_full = fatal_control.disk_full
     print(f"\n{'='*60}")
     if disk_full is not None:
         print(f"Run stopped: disk full in {disk_full.folder}  |  "
-              f"ok={ok_count}  fail={fail_count}")
+              f"ok={metrics['ok']}  fail={metrics['fail']}")
     else:
+        rate_mbs = total_bytes / el / 1e6 if el else 0.0
         print(f"Done in {int(el//60)}m {int(el%60)}s  |  "
-              f"ok={ok_count}  fail={fail_count}  |  "
-              f"{total_bytes/1e9:.2f} GB  @  {total_bytes/el/1e6:.1f} MB/s")
+              f"ok={metrics['ok']}  fail={metrics['fail']}  |  "
+              f"{total_bytes/1e9:.2f} GB  @  "
+              f"{rate_mbs:.1f} MB/s")
 
     if failed_urls:
         fp = os.path.join(base, "failed_links.txt")
@@ -342,7 +359,7 @@ async def _task(pu=proxy_url, co=cookies, fn=filename, ou=url, r=rec):
         except OSError as e:
             print(f"[warn] Failed links save error: {e}")
 
-    return ok_count, fail_count, disk_full is not None
+    return metrics["ok"], metrics["fail"], disk_full is not None
 
 # ── ENTRY POINT ────────────────────────────────────────────────────────────────
 def main():

moon_engine.py

@@ -122,6 +122,7 @@ def __init__(self):
         self._t0         = 0.0
         self._t_end      = 0.0
         self._proxies    = 0
+        self._progress_extracted: set[str] = set()
 
         # Live FileRecord registry: the GUI reads these objects every snapshot,
         # so a row's speed and percentage come off the download loop itself
@@ -152,6 +153,67 @@ def _inc(self, attr, delta=1):
     def _get(self, attr):
         with self._lock: return getattr(self, attr)
 
+    def begin_external_progress(self, total: int, output_folder: str = "", proxies: int = 0):
+        """Start a snapshot-backed run whose work is driven by another front-end.
+
+        The CLI owns its asyncio queue, but it must not grow a second copy of the
+        engine's progress arithmetic.  These small state transitions let it feed
+        the same snapshot contract without starting the GUI engine thread.
+        """
+        if output_folder:
+            self._cfg["out_folder"] = output_folder
+        with self._lock:
+            self._running = True; self._stop_flag = False; self._state = "running"
+            self._url_total = total; self._url_done = 0
+            self._dl_total = total; self._dl_done = 0
+            self._ok = 0; self._fail = 0; self._kills = 0
+            self._browsers = 0; self._dls = 0; self._proxies = proxies
+            self._bytes_acc.clear(); self._t0 = time.monotonic(); self._t_end = 0.0
+            self._tracked.clear(); self._progress_extracted.clear()
+        with self._log_lock:
+            self._log_ring.clear(); self._log_total = 0
+
+    def progress_bytes(self):
+        """Return the shared byte sample deque used by ``snapshot()``."""
+        return self._bytes_acc
+
+    def mark_extraction(self, url: str, success: bool):
+        """Record one URL's final extraction result at most once."""
+        with self._lock:
+            if url in self._progress_extracted:
+                return self._dl_done >= self._dl_total
+            self._progress_extracted.add(url)
+            self._url_done += 1
+            if not success:
+                self._fail += 1
+                self._dl_done += 1
+            return self._dl_done >= self._dl_total
+
+    def mark_download_start(self):
+        with self._lock:
+            self._dls += 1
+
+    def mark_download_end(self, success: bool):
+        with self._lock:
+            self._dls = max(0, self._dls - 1)
+            self._dl_done += 1
+            if success:
+                self._ok += 1
+            else:
+                self._fail += 1
+            return self._dl_done >= self._dl_total
+
+    def mark_download_retry(self):
+        with self._lock:
+            self._dls = max(0, self._dls - 1)
+
+    def mark_download_aborted(self):
+        with self._lock:
+            self._dls = max(0, self._dls - 1)
+
+    def finish_external_progress(self):
+        self._on_done()
+
     _LOG_MAX_LINES = 2000
 
     def log(self, msg, tag=""):

tests/test_cli_progress.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import time
+
+import moon_cli
+from moon_engine import Engine
+
+
+def test_cli_progress_line_reports_extraction_stage():
+    engine = Engine()
+    engine.begin_external_progress(5)
+    try:
+        first = engine.snapshot()
+        line = moon_cli._progress_line(first, 5)
+        assert "extracting 0/5" in line
+        assert "done" not in line
+        time.sleep(0.1)
+        assert moon_cli._progress_key(first) == moon_cli._progress_key(engine.snapshot())
+    finally:
+        engine.finish_external_progress()
+
+
+def test_cli_progress_line_switches_to_downloading_after_extraction():
+    engine = Engine()
+    engine.begin_external_progress(2)
+    try:
+        engine.mark_extraction("https://example.test/one", True)
+        engine.mark_extraction("https://example.test/two", True)
+        engine.mark_download_start()
+        now = time.monotonic()
+        engine.progress_bytes().extend([(now - 0.2, 1_000_000), (now - 0.1, 1_000_000)])
+
+        snapshot = engine.snapshot()
+        line = moon_cli._progress_line(snapshot, 2)
+        assert "downloading 0/2" in line
+        assert "MB/s" in line or "KB/s" in line
+        assert snapshot["metrics"]["speed_mbs"] > 0
+    finally:
+        engine.finish_external_progress()
+
+
+def test_external_progress_counts_re_extraction_once():
+    engine = Engine()
+    url = "https://example.test/retried"
+    engine.begin_external_progress(1)
+    try:
+        engine.mark_extraction(url, True)
+        engine.mark_extraction(url, True)
+
+        metrics = engine.snapshot()["metrics"]
+        assert metrics["extract_done"] == 1
+        assert metrics["extract_total"] == 1
+    finally:
+        engine.finish_external_progress()