123 lines
4.3 KiB
Python
123 lines
4.3 KiB
Python
"""Batch runner — spawn N parallel Chrome workers to execute the active flow config."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from concurrent.futures import Future, ThreadPoolExecutor, as_completed
|
|
from typing import Any, Callable
|
|
|
|
from logger import get_logger
|
|
|
|
log = get_logger(__name__)
|
|
|
|
|
|
def _single_run(
|
|
run_index: int,
|
|
batch_id: str,
|
|
cfg: dict[str, Any],
|
|
headless: bool,
|
|
) -> tuple[bool, str | None]:
|
|
"""One isolated Chrome session. Returns (success, failure_reason)."""
|
|
from crawler.driver import driver_session
|
|
from crawler.dynamic_flow import DynamicFlow
|
|
from crawler.scenarios.fresh_session import FreshSessionScenario
|
|
from db.repositories.runs import runs_repo
|
|
|
|
identifier = f"batch-{batch_id[:8]}-{run_index:04d}"
|
|
t0 = time.monotonic()
|
|
target_url = str(cfg.get("target_url") or "").strip()
|
|
log.info(
|
|
"[batch:%s] Run %d starting — target=%s",
|
|
batch_id[:8],
|
|
run_index,
|
|
target_url or "<missing>",
|
|
)
|
|
try:
|
|
with driver_session(headless=headless, recording_name=identifier) as driver:
|
|
FreshSessionScenario(driver, target_url).run()
|
|
result = DynamicFlow(driver, cfg).run() if cfg else None
|
|
|
|
total_ms = int((time.monotonic() - t0) * 1000)
|
|
if result:
|
|
steps_data = [
|
|
{"name": s.name, "success": s.success, "error": s.error, "duration_ms": s.duration_ms}
|
|
for s in result.steps
|
|
]
|
|
asyncio.run(runs_repo.insert(
|
|
"batch", identifier, result.success,
|
|
result.failure_reason, steps_data, total_ms,
|
|
))
|
|
log.info("[batch:%s] Run %d done — success=%s (%d ms)", batch_id[:8], run_index, result.success, total_ms)
|
|
return result.success, result.failure_reason
|
|
else:
|
|
asyncio.run(runs_repo.insert("batch", identifier, True, None, [], total_ms))
|
|
log.info("[batch:%s] Run %d done (no flow config) — %d ms", batch_id[:8], run_index, total_ms)
|
|
return True, None
|
|
|
|
except Exception as exc:
|
|
total_ms = int((time.monotonic() - t0) * 1000)
|
|
asyncio.run(runs_repo.insert("batch", identifier, False, str(exc), [], total_ms))
|
|
log.error("[batch:%s] Run %d failed — %s (%d ms)", batch_id[:8], run_index, exc, total_ms)
|
|
return False, str(exc)
|
|
|
|
|
|
def run_batch(
|
|
cfg: dict[str, Any],
|
|
workers: int,
|
|
total_runs: int,
|
|
headless: bool = True,
|
|
stagger_ms: int = 0,
|
|
on_progress: Callable[[int, int, int], None] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""
|
|
Run `total_runs` crawler sessions with up to `workers` parallel Chrome processes.
|
|
|
|
`stagger_ms` delays each successive submission to the pool, spreading
|
|
Chrome startup load and reducing bot-detection fingerprint clustering.
|
|
|
|
`on_progress(completed, succeeded, failed)` is called after every finished run.
|
|
|
|
Returns {"batch_id", "total", "succeeded", "failed"}.
|
|
"""
|
|
batch_id = uuid.uuid4().hex
|
|
log.info(
|
|
"Batch %s — %d run(s), %d worker(s), stagger=%d ms, headless=%s",
|
|
batch_id[:8], total_runs, workers, stagger_ms, headless,
|
|
)
|
|
|
|
succeeded = 0
|
|
failed = 0
|
|
lock = threading.Lock()
|
|
|
|
with ThreadPoolExecutor(max_workers=workers, thread_name_prefix=f"batch-{batch_id[:8]}") as pool:
|
|
futures: list[Future[tuple[bool, str | None]]] = []
|
|
for i in range(total_runs):
|
|
futures.append(pool.submit(_single_run, i, batch_id, cfg, headless))
|
|
if stagger_ms > 0 and i < total_runs - 1:
|
|
time.sleep(stagger_ms / 1000)
|
|
|
|
for fut in as_completed(futures):
|
|
try:
|
|
success, _ = fut.result()
|
|
except Exception as exc:
|
|
log.error("[batch:%s] Unhandled thread exception: %s", batch_id[:8], exc)
|
|
success = False
|
|
|
|
with lock:
|
|
if success:
|
|
succeeded += 1
|
|
else:
|
|
failed += 1
|
|
completed = succeeded + failed
|
|
|
|
if on_progress:
|
|
on_progress(completed, succeeded, failed)
|
|
|
|
log.info(
|
|
"Batch %s finished — %d/%d succeeded, %d failed",
|
|
batch_id[:8], succeeded, total_runs, failed,
|
|
)
|
|
return {"batch_id": batch_id, "total": total_runs, "succeeded": succeeded, "failed": failed}
|