diff --git a/docs/CLI.md b/docs/CLI.md index 0f93f12..0f8ae01 100644 --- a/docs/CLI.md +++ b/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 | diff --git a/moon_cli.py b/moon_cli.py index 4feb56d..eaa68a3 100644 --- a/moon_cli.py +++ b/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,16 +112,11 @@ 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) @@ -109,42 +124,33 @@ async def progress_loop(): 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(): diff --git a/moon_engine.py b/moon_engine.py index dc02e39..f1f0198 100644 --- a/moon_engine.py +++ b/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=""): diff --git a/tests/test_cli_progress.py b/tests/test_cli_progress.py new file mode 100644 index 0000000..cfae9f9 --- /dev/null +++ b/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()