From 64886b2b10654b7768bc72d81bbb1d9db6f604e0 Mon Sep 17 00:00:00 2001 From: Navid Filsaraee Date: Sat, 20 Jun 2026 17:07:48 +0330 Subject: [PATCH] updates :) --- .gitignore | 1 + Makefile | 11 +- admin/main.py | 33 ++- admin/models.py | 10 +- admin/routes.py | 103 ++++++-- admin/static/dashboard.html | 351 ++++++++++++++++++++---- config.py | 3 +- crawler/scenarios/fresh_session.py | 3 + crawler/tasks.py | 412 +++++++++++++++++++++-------- db/connection.py | 17 ++ db/repositories/__init__.py | 2 + db/repositories/batch_runs.py | 55 ++++ db/repositories/flow_configs.py | 16 +- logger.py | 20 +- main.py | 35 ++- pyproject.toml | 3 +- uv.lock | 83 +++++- 17 files changed, 896 insertions(+), 262 deletions(-) create mode 100644 db/repositories/batch_runs.py diff --git a/.gitignore b/.gitignore index 873cc70..88d5f0b 100644 --- a/.gitignore +++ b/.gitignore @@ -6,3 +6,4 @@ data/ *.db *.swp drivers/ +/logs diff --git a/Makefile b/Makefile index b340d76..8f1e61d 100644 --- a/Makefile +++ b/Makefile @@ -11,7 +11,7 @@ GREEN := $(shell tput setaf 2 2>/dev/null) CYAN := $(shell tput setaf 6 2>/dev/null) .PHONY: help install sync lock lint typecheck \ - admin otp fresh batch \ + dev admin worker otp fresh batch \ build up down logs shell \ clean @@ -50,10 +50,19 @@ patch-driver: ## Download + patch ChromeDriver once into drivers/ (run after Chr @mkdir -p drivers PYTHONPATH=. $(PYTHON) scripts/patch_driver.py +dev: ## Start admin panel + RQ worker together (Ctrl-C stops both) + @mkdir -p $(DATA_DIR) + @trap 'kill 0' INT TERM EXIT; \ + $(UV) run rq worker --with-scheduler batch & \ + $(PYTHON) main.py admin + admin: ## Start the admin panel (http://localhost:8000) @mkdir -p $(DATA_DIR) $(PYTHON) main.py admin +worker: ## Start RQ worker with scheduler (batch queue) + $(UV) run rq worker --with-scheduler batch + otp: ## Run Scenario 1 — SMS-OTP login (PHONE=+98... or all from .env) @mkdir -p $(DATA_DIR) ifdef PHONE diff --git a/admin/main.py b/admin/main.py index afdb55c..2406570 100644 --- a/admin/main.py +++ b/admin/main.py @@ -1,7 +1,9 @@ """FastAPI admin application.""" from __future__ import annotations -import threading +import os +import subprocess +import sys from pathlib import Path from fastapi import FastAPI @@ -18,24 +20,29 @@ app.include_router(router) _STATIC = Path(__file__).parent / "static" app.mount("/static", StaticFiles(directory=str(_STATIC)), name="static") +_rq_worker: subprocess.Popen[bytes] | None = None + @app.on_event("startup") def on_startup() -> None: + global _rq_worker init_db() - _start_huey_consumer() + redis_url = os.environ.get("REDIS_URL", "redis://localhost:6379/0") + _rq_worker = subprocess.Popen( + [sys.executable, "-m", "rq", "worker", "--url", redis_url, "batch"], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) -def _start_huey_consumer() -> None: - from huey.consumer import Consumer - from crawler.tasks import huey - - class _ThreadConsumer(Consumer): - def _set_signal_handlers(self) -> None: - pass # signal.signal() only works on the main thread - - consumer = _ThreadConsumer(huey, workers=1, periodic=False) - t = threading.Thread(target=consumer.run, daemon=True, name="huey-consumer") - t.start() +@app.on_event("shutdown") +def on_shutdown() -> None: + if _rq_worker is not None: + _rq_worker.terminate() + try: + _rq_worker.wait(timeout=5) + except subprocess.TimeoutExpired: + _rq_worker.kill() @app.get("/") diff --git a/admin/models.py b/admin/models.py index 66fbe10..ce75b5a 100644 --- a/admin/models.py +++ b/admin/models.py @@ -4,7 +4,7 @@ from __future__ import annotations from typing import Any from db.connection import DB_PATH as DB_PATH, init_db as init_db -from db.repositories import flow_config_repo, runs_repo +from db.repositories import flow_config_repo as flow_config_repo, runs_repo FlowConfigRow = dict[str, Any] @@ -36,10 +36,6 @@ async def get_flow_config(cfg_id: int) -> FlowConfigRow | None: return await flow_config_repo.get(cfg_id) -async def get_active_flow_config() -> FlowConfigRow | None: - return await flow_config_repo.get_active() - - async def upsert_flow_config(data: dict[str, Any], cfg_id: int | None = None) -> int: if cfg_id is None: return await flow_config_repo.create(data) @@ -49,7 +45,3 @@ async def upsert_flow_config(data: dict[str, Any], cfg_id: int | None = None) -> async def delete_flow_config(cfg_id: int) -> None: await flow_config_repo.delete(cfg_id) - - -async def activate_flow_config(cfg_id: int) -> None: - await flow_config_repo.activate(cfg_id) diff --git a/admin/routes.py b/admin/routes.py index 11a4bb7..5754e9d 100644 --- a/admin/routes.py +++ b/admin/routes.py @@ -2,7 +2,6 @@ from __future__ import annotations import secrets -import uuid from typing import Annotated, Any from fastapi import APIRouter, Depends, HTTPException, status @@ -10,7 +9,6 @@ from fastapi.security import HTTPBasic, HTTPBasicCredentials from pydantic import BaseModel from admin.models import ( - activate_flow_config, delete_flow_config, get_flow_config, get_stats, @@ -56,7 +54,6 @@ async def stats(_: Auth) -> dict[str, object]: class FlowConfigIn(BaseModel): name: str - is_active: bool = False search_texts: list[str] = [] search_box_selector: str = "" search_box_selector_type: str = "css" @@ -83,7 +80,6 @@ class FlowConfigIn(BaseModel): def _to_db_dict(body: FlowConfigIn) -> dict[str, Any]: return { "name": body.name, - "is_active": int(body.is_active), "search_texts_json": body.search_texts, "search_box_selector": body.search_box_selector, "search_box_selector_type": body.search_box_selector_type, @@ -138,15 +134,6 @@ async def update_config(cfg_id: int, body: FlowConfigIn, _: Auth) -> dict[str, o return row or {} # type: ignore[return-value] -@router.post("/api/flow-configs/{cfg_id}/activate", status_code=200) -async def activate_config(cfg_id: int, _: Auth) -> dict[str, object]: - existing = await get_flow_config(cfg_id) - if not existing: - raise HTTPException(status_code=404, detail="Not found") - await activate_flow_config(cfg_id) - return {"activated": cfg_id} - - @router.delete("/api/flow-configs/{cfg_id}", status_code=204) async def delete_config(cfg_id: int, _: Auth) -> None: existing = await get_flow_config(cfg_id) @@ -159,32 +146,42 @@ async def delete_config(cfg_id: int, _: Auth) -> None: # Batch runs (managed via Huey task queue) # --------------------------------------------------------------------------- -class BatchRunIn(BaseModel): +class BatchTaskIn(BaseModel): + config_id: int workers: int = 3 total_runs: int = 10 stagger_ms: int = 500 headless: bool = True -@router.post("/api/run-batch", status_code=202) -async def start_batch(body: BatchRunIn, _: Auth) -> dict[str, object]: - from crawler.tasks import batch_status as _status, run_batch_task, set_current +@router.post("/api/batch-queue", status_code=202) +async def add_to_batch_queue(body: BatchTaskIn, _: Auth) -> dict[str, object]: + from crawler.tasks import enqueue - current = _status() - if current.get("running"): - raise HTTPException(status_code=409, detail="A batch is already running") + cfg = await get_flow_config(body.config_id) + if not cfg: + raise HTTPException(status_code=404, detail=f"Flow config {body.config_id} not found") - from admin.models import get_active_flow_config - active_cfg = await get_active_flow_config() - if not active_cfg: - raise HTTPException(status_code=422, detail="No active flow config — activate one first") + result = enqueue(cfg, body.workers, body.total_runs, body.stagger_ms, body.headless) + log.info("Task enqueued — config=%d runs=%d workers=%d status=%s", + body.config_id, body.total_runs, body.workers, result["status"]) + return result - batch_id = uuid.uuid4().hex - result = run_batch_task(batch_id, active_cfg, body.workers, body.total_runs, body.headless, body.stagger_ms) - set_current(batch_id, result) - log.info("Batch %s enqueued — %d runs, %d workers", batch_id[:8], body.total_runs, body.workers) - return {"batch_id": batch_id, "status": "queued"} +@router.delete("/api/batch-queue/{queue_id}", status_code=200) +async def remove_from_batch_queue(queue_id: str, _: Auth) -> dict[str, object]: + from crawler.tasks import dequeue + removed = dequeue(queue_id) + if not removed: + raise HTTPException(status_code=404, detail="Queue item not found") + return {"removed": queue_id} + + +@router.post("/api/batch-queue/clear", status_code=200) +async def clear_batch_queue(_: Auth) -> dict[str, object]: + from crawler.tasks import clear_queue + count = clear_queue() + return {"cleared": count} @router.post("/api/stop-batch", status_code=200) @@ -200,3 +197,51 @@ async def stop_batch(_: Auth) -> dict[str, object]: async def batch_status(_: Auth) -> dict[str, object]: from crawler.tasks import batch_status as _status return _status() # type: ignore[return-value] + + +@router.get("/api/batch-runs") +async def list_batch_runs(_: Auth) -> list[dict[str, object]]: + from db.repositories.batch_runs import batch_runs_repo + return await batch_runs_repo.list() # type: ignore[return-value] + + +# --------------------------------------------------------------------------- +# Log viewer +# --------------------------------------------------------------------------- + +@router.get("/api/logs") +def get_logs( + _: Auth, + n: int = 200, + level: str = "", + q: str = "", +) -> dict[str, object]: + import os + from logger import _LOG_FILE + + path = _LOG_FILE + if not os.path.exists(path): + return {"lines": [], "file": path} + + with open(path, encoding="utf-8", errors="replace") as f: + all_lines = f.readlines() + + all_lines.reverse() # newest first + + level_upper = level.upper() + q_lower = q.lower() + + results: list[str] = [] + for raw in all_lines: + line = raw.rstrip() + if not line: + continue + if level_upper and level_upper not in line: + continue + if q_lower and q_lower not in line.lower(): + continue + results.append(line) + if len(results) >= n: + break + + return {"lines": results, "file": path} diff --git a/admin/static/dashboard.html b/admin/static/dashboard.html index e1cd963..86a5a53 100644 --- a/admin/static/dashboard.html +++ b/admin/static/dashboard.html @@ -56,7 +56,7 @@ .pill { display: inline-block; border-radius: 999px; padding: 2px 10px; font-size: 11px; font-weight: 600; } .pill.ok { background: rgba(34,197,94,.15); color: var(--green); } .pill.err { background: rgba(239,68,68,.15); color: var(--red); } - .pill.active-badge { background: rgba(99,102,241,.2); color: var(--accent); } + .pill.id-badge { background: rgba(99,102,241,.12); color: var(--muted); font-family: monospace; } .reason-row { display: flex; justify-content: space-between; align-items: center; padding: 8px 0; border-bottom: 1px solid var(--border); } .reason-row:last-child { border-bottom: none; } @@ -83,7 +83,7 @@ background: var(--surface); border: 1px solid var(--border); border-radius: 10px; padding: 16px 18px; display: flex; align-items: flex-start; gap: 14px; } - .config-card.is-active { border-color: var(--accent); } + .config-card { cursor: default; } .config-info { flex: 1; } .config-name { font-weight: 600; font-size: 15px; margin-bottom: 4px; display: flex; align-items: center; gap: 8px; } .config-meta { color: var(--muted); font-size: 12px; } @@ -188,12 +188,6 @@ Advanced Settings
-
- -
-
Search Step
@@ -313,6 +307,7 @@
Dashboard
Flow Configs
Batch Run
+
Logs
@@ -355,7 +350,7 @@
Flow Configurations
-
Only one config is active at a time. The crawler picks it up on every run.
+
Select a config in the Batch Run tab to use it for a run.
@@ -374,13 +369,22 @@
Batch Run
-
Launch many parallel crawler sessions using the active flow config.
+
Launch many parallel crawler sessions using a selected flow config.
-
+
+ + +
-
+
Configuration
+
+ + +
@@ -398,20 +402,37 @@
- +
+ +
+
+
Pending Queue
+ +
+
+

Queue is empty.

+
+
+ +
+ -
+
Live Status
No batch running.
+ + +
+
Run History
+
+ + + + + + + + + + + + + + + + + +
#StartedConfigWorkersTotalSucceededFailedDurationStatus
No batch runs yet.
+
+
+
+
+ + +
+
+
+
+ Log Viewer + + + + + + +
+
+ Loading… +
+
@@ -548,7 +631,9 @@ document.querySelectorAll('.tab').forEach(tab => { tab.classList.add('active'); document.getElementById(`tab-${tab.dataset.tab}`).classList.add('active'); if (tab.dataset.tab === 'flows') loadConfigs(); - if (tab.dataset.tab === 'batch') pollBatchStatus(); + if (tab.dataset.tab === 'batch') { loadConfigs(); pollBatchStatus(); loadBatchHistory(); } + if (tab.dataset.tab === 'logs') { loadLogs(); _startLogPoll(); } + else _stopLogPoll(); }); }); @@ -616,6 +701,7 @@ async function loadConfigs() { if (!res || !res.ok) return; _configs = await res.json(); renderConfigs(); + _refreshBatchConfigSelect(); } function renderConfigs() { @@ -625,11 +711,11 @@ function renderConfigs() { return; } el.innerHTML = _configs.map(c => ` -
+
${esc(c.name)} - ${c.is_active ? 'ACTIVE' : ''} + #${c.id}
Searches: ${(c.search_texts_json || []).length} text(s)  ·  @@ -644,7 +730,6 @@ function renderConfigs() {
- ${!c.is_active ? `` : ''} @@ -654,9 +739,11 @@ function renderConfigs() { `).join(''); } -async function activateConfig(id) { - const res = await apiFetch(`/api/flow-configs/${id}/activate`, { method: 'POST' }); - if (res && res.ok) await loadConfigs(); +function _refreshBatchConfigSelect() { + const sel = document.getElementById('batch-config-select'); + const cur = sel.value; + sel.innerHTML = '' + + _configs.map(c => ``).join(''); } async function deleteConfig(id) { @@ -705,7 +792,6 @@ async function duplicateConfig(id) { if (!c) return; const body = { name: c.name + ' (copy)', - is_active: false, search_texts: c.search_texts_json || [], search_box_selector: c.search_box_selector || '', search_box_selector_type: c.search_box_selector_type || 'css', @@ -743,7 +829,6 @@ document.getElementById('import-file-input').addEventListener('change', async (e try { data = JSON.parse(await file.text()); } catch { alert('Invalid JSON file'); return; } const body = { name: data.name || file.name.replace(/\.json$/, ''), - is_active: false, search_texts: data.search_texts || [], search_box_selector: data.search_box_selector || '', search_box_selector_type: data.search_box_selector_type || 'css', @@ -772,7 +857,6 @@ document.getElementById('import-file-input').addEventListener('change', async (e /* ─── Form ─── */ function _resetAdvanced() { - document.getElementById('cfg-active').checked = false; document.getElementById('cfg-search-sel').value = ''; document.getElementById('cfg-search-sel-type').value = 'css'; document.getElementById('cfg-submit-sel').value = ''; @@ -815,7 +899,6 @@ function openEditForm(id) { document.getElementById('cfg-target-url').value = c.target_url || ''; document.getElementById('cfg-search-texts').value = (c.search_texts_json || []).join('\n'); document.getElementById('cfg-item-ids').value = (c.target_item_ids_json || []).join('\n'); - document.getElementById('cfg-active').checked = !!c.is_active; document.getElementById('cfg-search-sel').value = c.search_box_selector || ''; document.getElementById('cfg-search-sel-type').value = c.search_box_selector_type || 'css'; document.getElementById('cfg-submit-sel').value = c.search_submit_selector || ''; @@ -847,7 +930,6 @@ function closeForm() { function collectForm() { return { name: document.getElementById('cfg-name').value.trim(), - is_active: document.getElementById('cfg-active').checked, search_texts: document.getElementById('cfg-search-texts').value.split('\n').map(s => s.trim()).filter(Boolean), search_box_selector: document.getElementById('cfg-search-sel').value.trim(), search_box_selector_type: document.getElementById('cfg-search-sel-type').value, @@ -939,14 +1021,16 @@ function renderBatchStatus(s) { const idle = document.getElementById('batch-idle-msg'); const body = document.getElementById('batch-status-body'); const errEl = document.getElementById('batch-error'); - const startBtn = document.getElementById('batch-start-btn'); const stopBtn = document.getElementById('batch-stop-btn'); const badge = document.getElementById('bs-active-badge'); + const banner = document.getElementById('bs-completion-banner'); + + // Always render queue + renderQueue(s.queue || []); if (!s.batch_id) { idle.style.display = ''; body.style.display = 'none'; - startBtn.style.display = ''; stopBtn.style.display = 'none'; badge.style.display = 'none'; return; @@ -956,11 +1040,16 @@ function renderBatchStatus(s) { body.style.display = ''; const pct = s.total ? Math.round(s.completed / s.total * 100) : 0; + const barColor = !s.running + ? (s.stopping ? 'var(--yellow)' : s.error ? 'var(--red)' : 'var(--green)') + : 'var(--accent)'; + document.getElementById('batch-progress-bar').style.background = barColor; document.getElementById('batch-progress-bar').style.width = pct + '%'; document.getElementById('batch-progress-label').textContent = `${s.completed} / ${s.total}`; document.getElementById('bs-completed').textContent = s.completed ?? 0; document.getElementById('bs-succeeded').textContent = s.succeeded ?? 0; document.getElementById('bs-failed').textContent = s.failed ?? 0; + document.getElementById('bs-config-name').textContent = s.config_name || '–'; document.getElementById('bs-workers').textContent = s.configured_workers ?? '–'; document.getElementById('bs-started').textContent = fmtTs(s.started_at); @@ -972,6 +1061,23 @@ function renderBatchStatus(s) { finRow.style.display = 'none'; } + // Completion banner + if (!s.running && s.finished_at) { + let msg, bg, color; + if (s.stopping) { + msg = '■ Stopped'; bg = 'rgba(234,179,8,.15)'; color = 'var(--yellow)'; + } else if (s.error) { + msg = '✕ Error: ' + s.error; bg = 'rgba(239,68,68,.12)'; color = 'var(--red)'; + } else { + msg = `✓ Batch complete — ${s.succeeded ?? 0} succeeded, ${s.failed ?? 0} failed`; + bg = 'rgba(34,197,94,.12)'; color = 'var(--green)'; + } + banner.textContent = msg; + banner.style.cssText = `display:block;border-radius:7px;padding:8px 12px;font-size:12px;font-weight:600;margin-bottom:14px;background:${bg};color:${color}`; + } else { + banner.style.display = 'none'; + } + // Active worker badge const active = s.active_workers ?? 0; if (s.running && active > 0) { @@ -992,7 +1098,7 @@ function renderBatchStatus(s) { ${w.slot} ${label} ${fmtDuration(w.duration_ms)} - ${err} + ${err} `; }).join(''); @@ -1004,16 +1110,52 @@ function renderBatchStatus(s) { } if (s.running) { - startBtn.style.display = 'none'; stopBtn.style.display = ''; stopBtn.disabled = s.stopping || false; stopBtn.textContent = s.stopping ? '■ Stopping…' : '■ Stop'; } else { - startBtn.style.display = ''; stopBtn.style.display = 'none'; } } +function renderQueue(queue) { + const list = document.getElementById('queue-list'); + const badge = document.getElementById('queue-count-badge'); + const clearBtn = document.getElementById('queue-clear-btn'); + + if (!queue.length) { + list.innerHTML = '

Queue is empty.

'; + badge.style.display = 'none'; + clearBtn.style.display = 'none'; + return; + } + + badge.textContent = queue.length; + badge.style.display = ''; + clearBtn.style.display = ''; + + list.innerHTML = queue.map((t, i) => ` +
+ ${i + 1} +
+
${esc(t.config_name || '–')}
+
${t.workers}w · ${t.total_runs} runs · ${t.stagger_ms}ms stagger
+
+ +
+ `).join('').replace(/]*>.*?<\/div>\s*$/, s => s.replace('border-bottom:1px solid var(--border)', '')); +} + +async function removeQueueItem(queueId) { + const res = await apiFetch(`/api/batch-queue/${queueId}`, { method: 'DELETE' }); + if (res && res.ok) await pollBatchStatus(); +} + +document.getElementById('queue-clear-btn').addEventListener('click', async () => { + const res = await apiFetch('/api/batch-queue/clear', { method: 'POST' }); + if (res && res.ok) await pollBatchStatus(); +}); + async function pollBatchStatus() { const res = await apiFetch('/api/batch-status'); if (!res || !res.ok) return; @@ -1032,27 +1174,31 @@ function startBatchPoll() { if (!st.running) { clearInterval(_batchPollTimer); _batchPollTimer = null; + // Delay history reload to allow the DB finish() write to complete + setTimeout(loadBatchHistory, 2000); } }, 1500); } document.getElementById('batch-start-btn').addEventListener('click', async () => { + const configId = parseInt(document.getElementById('batch-config-select').value); + if (!configId) { alert('Please select a flow config.'); return; } const workers = parseInt(document.getElementById('batch-workers').value) || 3; const runs = parseInt(document.getElementById('batch-runs').value) || 10; const stagger = parseInt(document.getElementById('batch-stagger').value) || 0; const headless = document.getElementById('batch-headless').checked; - const res = await apiFetch('/api/run-batch', { + const res = await apiFetch('/api/batch-queue', { method: 'POST', headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ workers, total_runs: runs, stagger_ms: stagger, headless }), + body: JSON.stringify({ config_id: configId, workers, total_runs: runs, stagger_ms: stagger, headless }), }); if (!res) return; - if (res.status === 409) { alert('A batch is already running.'); return; } - if (res.status === 422) { const d = await res.json(); alert(d.detail || 'No active flow config.'); return; } - if (!res.ok) { alert('Failed to start batch.'); return; } + if (res.status === 404) { const d = await res.json(); alert(d.detail || 'Flow config not found.'); return; } + if (!res.ok) { alert('Failed to add task.'); return; } + const data = await res.json(); await pollBatchStatus(); startBatchPoll(); }); @@ -1067,6 +1213,115 @@ document.getElementById('batch-stop-btn').addEventListener('click', async () => btn.textContent = '■ Stop'; } }); + +/* ──────────────────────── Batch Run History ────────────────────── */ +async function loadBatchHistory() { + const res = await apiFetch('/api/batch-runs'); + if (!res || !res.ok) return; + renderBatchHistory(await res.json()); +} + +/* ─────────────────────────── Logs ──────────────────────────── */ +let _logPollTimer = null; + +const _LEVEL_COLORS = { + DEBUG: '#8892a4', + INFO: '#6ee7b7', + WARNING: '#eab308', + ERROR: '#ef4444', + CRITICAL: '#f97316', +}; + +function _colorLine(line) { + for (const [lvl, col] of Object.entries(_LEVEL_COLORS)) { + if (line.includes(lvl)) { + return `${esc(line)}`; + } + } + return `${esc(line)}`; +} + +async function loadLogs() { + const level = document.getElementById('log-level').value; + const q = document.getElementById('log-search').value; + const n = document.getElementById('log-n').value; + const params = new URLSearchParams({ n, level, q }); + const res = await apiFetch(`/api/logs?${params}`); + if (!res || !res.ok) return; + const data = await res.json(); + document.getElementById('log-file').textContent = data.file || ''; + const out = document.getElementById('log-output'); + if (!data.lines.length) { + out.innerHTML = 'No log entries.'; + return; + } + out.innerHTML = data.lines.map(_colorLine).join('\n'); +} + +function _startLogPoll() { + _stopLogPoll(); + _logPollTimer = setInterval(() => { + if (document.getElementById('log-tail').checked) loadLogs(); + }, 5000); +} + +function _stopLogPoll() { + if (_logPollTimer) { clearInterval(_logPollTimer); _logPollTimer = null; } +} + +document.addEventListener('DOMContentLoaded', () => { + document.getElementById('log-tail').addEventListener('change', e => { + if (e.target.checked) _startLogPoll(); else _stopLogPoll(); + }); + ['log-level', 'log-n'].forEach(id => + document.getElementById(id).addEventListener('change', loadLogs) + ); + let _debounce; + document.getElementById('log-search').addEventListener('input', () => { + clearTimeout(_debounce); + _debounce = setTimeout(loadLogs, 300); + }); +}); + +/* ──────────────────────── Batch Run History ────────────────────── */ +function renderBatchHistory(rows) { + const tbody = document.getElementById('batch-history-tbody'); + if (!rows.length) { + tbody.innerHTML = 'No batch runs yet.'; + return; + } + tbody.innerHTML = rows.map(r => { + const dur = r.finished_at && r.started_at + ? fmtDuration((new Date(r.finished_at) - new Date(r.started_at))) + : '–'; + let statusLabel, statusColor; + if (!r.finished_at) { + statusLabel = '⟳ running'; statusColor = 'var(--accent)'; + } else if (r.stopped) { + statusLabel = '■ stopped'; statusColor = 'var(--yellow)'; + } else if (r.error) { + statusLabel = '✕ error'; statusColor = 'var(--red)'; + } else if (r.failed > 0 && r.succeeded === 0) { + statusLabel = 'all failed'; statusColor = 'var(--red)'; + } else { + statusLabel = '✓ done'; statusColor = 'var(--green)'; + } + const configLabel = r.config_name + ? `${esc(r.config_name)} #${r.config_id}` + : `#${r.config_id ?? '–'}`; + return ` + ${r.batch_id.slice(0,8)} + ${fmtTime(r.started_at)} + ${configLabel} + ${r.workers} + ${r.total_runs} + ${r.succeeded} + ${r.failed} + ${dur} + ${statusLabel} + `; + }).join(''); +} diff --git a/config.py b/config.py index 212fa66..bf8f429 100644 --- a/config.py +++ b/config.py @@ -2,6 +2,7 @@ from __future__ import annotations import os from dataclasses import dataclass, field + from dotenv import load_dotenv load_dotenv() @@ -12,7 +13,7 @@ class Config: admin_username: str = os.getenv("ADMIN_USERNAME", "admin") admin_password: str = os.getenv("ADMIN_PASSWORD", "changeme") admin_host: str = os.getenv("ADMIN_HOST", "127.0.0.1") - admin_port: int = int(os.getenv("ADMIN_PORT", "8000")) + admin_port: int = int(os.getenv("ADMIN_PORT", "9000")) db_path: str = os.getenv("DB_PATH", "data/tracker.db") headless: bool = os.getenv("HEADLESS", "true").lower() == "true" chrome_binary: str | None = os.getenv("CHROME_BINARY") diff --git a/crawler/scenarios/fresh_session.py b/crawler/scenarios/fresh_session.py index 116ead2..c59a46c 100644 --- a/crawler/scenarios/fresh_session.py +++ b/crawler/scenarios/fresh_session.py @@ -31,6 +31,9 @@ class FreshSessionScenario: "try { window.localStorage.clear(); window.sessionStorage.clear(); } catch(e) {}" ) + if not self.url: + return {"title": "", "url": "", "cookie_count": 0, "cookies": []} + self.driver.get(self.url) human_delay(2.0, 4.0) diff --git a/crawler/tasks.py b/crawler/tasks.py index 48121fa..293a604 100644 --- a/crawler/tasks.py +++ b/crawler/tasks.py @@ -1,137 +1,306 @@ -"""Huey task definitions and shared batch state for the admin panel.""" +"""Batch task execution via RQ (Redis Queue) — fully decoupled from FastAPI's event loop.""" from __future__ import annotations -import threading +import json +import os +import sqlite3 import time -from concurrent.futures import ThreadPoolExecutor, as_completed +import uuid from typing import Any -from huey import MemoryHuey +import redis +from rq import Queue -huey = MemoryHuey("seed", immediate=False) +from db.connection import DB_PATH -# ── Shared state (all accessed under _lock) ─────────────────────────────────── -_lock = threading.Lock() +REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379/0") -_current_batch_id: str | None = None -_current_result: Any = None # Huey Result — kept for revocation +# Redis key helpers +_K_CURRENT = "seed:current" # string: current batch_id (or empty) +_K_META = "seed:meta:{}" # JSON string: batch aggregate meta +_K_SLOTS = "seed:slots:{}" # hash: slot_key → JSON worker info +_K_STOP = "seed:stop:{}" # string: "1" if stop requested +_K_QUEUE = "seed:pending_queue" # list: JSON-encoded pending items -_batch_meta: dict[str, Any] = {} # batch_id → aggregate metadata -_worker_slots: dict[str, dict] = {} # f"{batch_id}:{slot:04d}" → worker info -_stop_events: dict[str, threading.Event] = {} + +def _r() -> redis.Redis: # type: ignore[type-arg] + return redis.from_url(REDIS_URL, decode_responses=True) + + +def _q() -> Queue: + return Queue("batch", connection=redis.from_url(REDIS_URL)) + + +# ── Sync DB helpers ──────────────────────────────────────────────────────────── + +def _db_insert_run( + scenario: str, + identifier: str, + success: bool, + failure_reason: str | None, + steps: list[dict[str, Any]], + total_ms: int, +) -> None: + conn = sqlite3.connect(DB_PATH) + try: + conn.execute( + "INSERT INTO runs (scenario, identifier, success, failure_reason, steps_json, total_ms)" + " VALUES (?,?,?,?,?,?)", + (scenario, identifier, int(success), failure_reason, json.dumps(steps), total_ms), + ) + conn.commit() + finally: + conn.close() + + +def _db_insert_batch_run( + batch_id: str, + config_id: int | None, + config_name: str, + workers: int, + total_runs: int, + stagger_ms: int, + headless: bool, +) -> None: + conn = sqlite3.connect(DB_PATH) + try: + conn.execute( + "INSERT INTO batch_runs" + " (batch_id, config_id, config_name, workers, total_runs, stagger_ms, headless)" + " VALUES (?,?,?,?,?,?,?)", + (batch_id, config_id, config_name, workers, total_runs, stagger_ms, int(headless)), + ) + conn.commit() + finally: + conn.close() + + +def _db_finish_batch_run( + batch_id: str, + succeeded: int, + failed: int, + stopped: bool, + error: str | None, +) -> None: + conn = sqlite3.connect(DB_PATH) + try: + conn.execute( + "UPDATE batch_runs" + " SET finished_at=strftime('%Y-%m-%dT%H:%M:%SZ','now')," + " succeeded=?, failed=?, stopped=?, error=?" + " WHERE batch_id=?", + (succeeded, failed, int(stopped), error, batch_id), + ) + conn.commit() + finally: + conn.close() # ── Public helpers ───────────────────────────────────────────────────────────── -def get_current_batch_id() -> str | None: - with _lock: - return _current_batch_id +def get_queue() -> list[dict[str, Any]]: + r = _r() + raw = r.lrange(_K_QUEUE, 0, -1) + items = [] + for s in raw: + try: + item = json.loads(s) + items.append({k: v for k, v in item.items() if k != "cfg"}) + except Exception: + pass + return items -def set_current(batch_id: str, result: Any) -> None: - global _current_batch_id, _current_result - with _lock: - _current_batch_id = batch_id - _current_result = result +def enqueue( + cfg: dict[str, Any], + workers: int, + total_runs: int, + stagger_ms: int, + headless: bool, +) -> dict[str, Any]: + queue_id = uuid.uuid4().hex + + item: dict[str, Any] = { + "queue_id": queue_id, + "cfg": cfg, + "config_id": cfg.get("id"), + "config_name": cfg.get("name") or "", + "workers": workers, + "total_runs": total_runs, + "stagger_ms": stagger_ms, + "headless": headless, + "added_at": time.time(), + } + + r = _r() + current = r.get(_K_CURRENT) + if current: + meta_raw = r.get(_K_META.format(current)) + if meta_raw: + meta = json.loads(meta_raw) + if meta.get("running"): + r.rpush(_K_QUEUE, json.dumps(item)) + return {"queue_id": queue_id, "status": "queued"} + + _launch(item) + return {"queue_id": queue_id, "status": "started"} + + +def dequeue(queue_id: str) -> bool: + r = _r() + raw = r.lrange(_K_QUEUE, 0, -1) + for s in raw: + try: + item = json.loads(s) + if item.get("queue_id") == queue_id: + r.lrem(_K_QUEUE, 1, s) + return True + except Exception: + pass + return False + + +def clear_queue() -> int: + r = _r() + raw = r.lrange(_K_QUEUE, 0, -1) + count = len(raw) + r.delete(_K_QUEUE) + return count def stop_current() -> bool: - """Signal the running batch to stop. Returns True if there was something to stop.""" - with _lock: - bid = _current_batch_id - result = _current_result - - if bid is None: + r = _r() + current = r.get(_K_CURRENT) + if not current: return False - - # Revoke task if it hasn't started yet (still pending in Huey queue) - if result is not None: - try: - result.revoke(revoke_once=True) - except Exception: - pass - - # Signal running workers to stop after their current step - ev = _stop_events.get(bid) - if ev: - ev.set() - - with _lock: - if bid in _batch_meta: - _batch_meta[bid]["stopping"] = True - + meta_raw = r.get(_K_META.format(current)) + if not meta_raw: + return False + meta = json.loads(meta_raw) + if not meta.get("running"): + return False + r.set(_K_STOP.format(current), "1") + meta["stopping"] = True + r.set(_K_META.format(current), json.dumps(meta)) return True def batch_status() -> dict[str, Any]: - with _lock: - bid = _current_batch_id - if bid is None: - return {"batch_id": None, "running": False} + r = _r() + bid = r.get(_K_CURRENT) + if not bid: + return {"batch_id": None, "running": False, "queue": get_queue()} - meta = dict(_batch_meta.get(bid, {})) - prefix = f"{bid}:" - workers = [dict(v) for k, v in _worker_slots.items() if k.startswith(prefix)] + meta_raw = r.get(_K_META.format(bid)) + meta = json.loads(meta_raw) if meta_raw else {} + + slot_data = r.hgetall(_K_SLOTS.format(bid)) + workers = [] + for v in slot_data.values(): + try: + workers.append(json.loads(v)) + except Exception: + pass meta["batch_id"] = bid meta["workers_detail"] = sorted(workers, key=lambda w: w.get("slot", 0)) meta["active_workers"] = sum(1 for w in workers if w.get("status") == "running") + meta["queue"] = get_queue() return meta -# ── Huey task ───────────────────────────────────────────────────────────────── +# ── Internal ─────────────────────────────────────────────────────────────────── -@huey.task(name="run_batch") -def run_batch_task( - batch_id: str, - cfg: dict[str, Any], - workers: int, - total_runs: int, - headless: bool, - stagger_ms: int, -) -> dict[str, Any]: - import asyncio +def _launch(item: dict[str, Any]) -> None: + batch_id = uuid.uuid4().hex + r = _r() + r.set(_K_CURRENT, batch_id) + + q = _q() + q.enqueue( + run_batch_job, + batch_id, + item, + job_id=batch_id, + job_timeout=-1, + result_ttl=3600, + ) + + +def _start_next_in_queue() -> None: + r = _r() + raw = r.lpop(_K_QUEUE) + if not raw: + return + try: + item = json.loads(raw) + except Exception: + return + _launch(item) + + +# ── RQ job (runs in worker process, no asyncio) ──────────────────────────────── + +def run_batch_job(batch_id: str, item: dict[str, Any]) -> None: + """Top-level RQ job function. Runs in a forked worker process.""" + import threading + from concurrent.futures import ThreadPoolExecutor, as_completed 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 from logger import get_logger log = get_logger(__name__) - stop_ev = threading.Event() - with _lock: - _stop_events[batch_id] = stop_ev - _batch_meta[batch_id] = { - "running": True, - "stopping": False, - "total": total_runs, - "completed": 0, - "succeeded": 0, - "failed": 0, - "configured_workers": workers, - "started_at": time.time(), - "finished_at": None, - "error": None, - } + r = redis.from_url(REDIS_URL, decode_responses=True) + + cfg = item["cfg"] + workers = item["workers"] + total_runs = item["total_runs"] + stagger_ms = item["stagger_ms"] + headless = item["headless"] + config_id: int | None = int(cfg["id"]) if cfg and cfg.get("id") is not None else None + config_name: str = str(cfg.get("name") or "") if cfg else "" + + _db_insert_batch_run(batch_id, config_id, config_name, workers, total_runs, stagger_ms, headless) + + meta: dict[str, Any] = { + "running": True, + "stopping": False, + "total": total_runs, + "completed": 0, + "succeeded": 0, + "failed": 0, + "configured_workers": workers, + "config_id": config_id, + "config_name": config_name, + "started_at": time.time(), + "finished_at": None, + "error": None, + } + r.set(_K_META.format(batch_id), json.dumps(meta)) + + slots_key = _K_SLOTS.format(batch_id) + stop_key = _K_STOP.format(batch_id) + meta_key = _K_META.format(batch_id) + + _slots_lock = threading.Lock() def _run_one(slot: int) -> bool: - slot_key = f"{batch_id}:{slot:04d}" + slot_key = f"{slot:04d}" - if stop_ev.is_set(): - with _lock: - _worker_slots[slot_key] = { - "slot": slot, "status": "skipped", - "started_at": None, "finished_at": None, "duration_ms": None, "error": "stopped", - } + if r.exists(stop_key): + r.hset(slots_key, slot_key, json.dumps({ + "slot": slot, "status": "skipped", + "started_at": None, "finished_at": None, "duration_ms": None, "error": "stopped", + })) return False - with _lock: - _worker_slots[slot_key] = { - "slot": slot, "status": "running", - "started_at": time.time(), "finished_at": None, "duration_ms": None, "error": None, - } + r.hset(slots_key, slot_key, json.dumps({ + "slot": slot, "status": "running", + "started_at": time.time(), "finished_at": None, "duration_ms": None, "error": None, + })) t0 = time.monotonic() identifier = f"batch-{batch_id[:8]}-{slot:04d}" @@ -150,24 +319,23 @@ def run_batch_task( {"name": s.name, "success": s.success, "error": s.error, "duration_ms": s.duration_ms} for s in (result.steps if result else []) ] - asyncio.run(runs_repo.insert("batch", identifier, success, reason, steps, total_ms)) + _db_insert_run("batch", identifier, success, reason, steps, total_ms) status = "ok" if success else "failed" - with _lock: - _worker_slots[slot_key].update({ - "status": status, "finished_at": time.time(), "duration_ms": total_ms, - }) + r.hset(slots_key, slot_key, json.dumps({ + "slot": slot, "status": status, + "finished_at": time.time(), "duration_ms": total_ms, "error": None, + })) log.info("[batch:%s] slot %d %s — %d ms", batch_id[:8], slot, status, total_ms) return success except Exception as exc: total_ms = int((time.monotonic() - t0) * 1000) - asyncio.run(runs_repo.insert("batch", identifier, False, str(exc), [], total_ms)) - with _lock: - _worker_slots[slot_key].update({ - "status": "error", "error": str(exc), - "finished_at": time.time(), "duration_ms": total_ms, - }) + _db_insert_run("batch", identifier, False, str(exc), [], total_ms) + r.hset(slots_key, slot_key, json.dumps({ + "slot": slot, "status": "error", "error": str(exc), + "finished_at": time.time(), "duration_ms": total_ms, + })) log.error("[batch:%s] slot %d error — %s", batch_id[:8], slot, exc) return False @@ -181,12 +349,17 @@ def run_batch_task( ) as pool: futures = [] for i in range(total_runs): - if stop_ev.is_set(): + if r.exists(stop_key): break futures.append(pool.submit(_run_one, i)) if stagger_ms > 0 and i < total_runs - 1: - if stop_ev.wait(timeout=stagger_ms / 1000): - break # stop was signalled during the stagger delay + deadline = time.monotonic() + stagger_ms / 1000 + while time.monotonic() < deadline: + if r.exists(stop_key): + break + time.sleep(0.1) + if r.exists(stop_key): + break for fut in as_completed(futures): try: @@ -197,18 +370,31 @@ def run_batch_task( succeeded += 1 else: failed += 1 - with _lock: - _batch_meta[batch_id]["completed"] = succeeded + failed - _batch_meta[batch_id]["succeeded"] = succeeded - _batch_meta[batch_id]["failed"] = failed + + current_meta_raw = r.get(meta_key) + if current_meta_raw: + current_meta = json.loads(current_meta_raw) + current_meta["completed"] = succeeded + failed + current_meta["succeeded"] = succeeded + current_meta["failed"] = failed + r.set(meta_key, json.dumps(current_meta)) except Exception as exc: - with _lock: - _batch_meta[batch_id]["error"] = str(exc) + current_meta_raw = r.get(meta_key) + if current_meta_raw: + current_meta = json.loads(current_meta_raw) + current_meta["error"] = str(exc) + r.set(meta_key, json.dumps(current_meta)) finally: - with _lock: - _batch_meta[batch_id]["running"] = False - _batch_meta[batch_id]["finished_at"] = time.time() + final_meta_raw = r.get(meta_key) + error_msg: str | None = None + if final_meta_raw: + final_meta = json.loads(final_meta_raw) + error_msg = final_meta.get("error") + final_meta["running"] = False + final_meta["finished_at"] = time.time() + r.set(meta_key, json.dumps(final_meta)) - return {"batch_id": batch_id, "succeeded": succeeded, "failed": failed} + _db_finish_batch_run(batch_id, succeeded, failed, bool(r.exists(stop_key)), error_msg) + _start_next_in_queue() diff --git a/db/connection.py b/db/connection.py index e7ab4fc..c728b2e 100644 --- a/db/connection.py +++ b/db/connection.py @@ -54,6 +54,23 @@ CREATE TABLE IF NOT EXISTS flow_configs ( target_item_ids_json TEXT NOT NULL DEFAULT '[]', target_url TEXT NOT NULL DEFAULT '' ); + +CREATE TABLE IF NOT EXISTS batch_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + batch_id TEXT NOT NULL UNIQUE, + config_id INTEGER, + config_name TEXT NOT NULL DEFAULT '', + workers INTEGER NOT NULL DEFAULT 1, + total_runs INTEGER NOT NULL DEFAULT 0, + stagger_ms INTEGER NOT NULL DEFAULT 0, + headless INTEGER NOT NULL DEFAULT 1, + started_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')), + finished_at TEXT, + succeeded INTEGER NOT NULL DEFAULT 0, + failed INTEGER NOT NULL DEFAULT 0, + stopped INTEGER NOT NULL DEFAULT 0, + error TEXT +); """ diff --git a/db/repositories/__init__.py b/db/repositories/__init__.py index 7e61633..a83622d 100644 --- a/db/repositories/__init__.py +++ b/db/repositories/__init__.py @@ -1,7 +1,9 @@ +from db.repositories.batch_runs import BatchRunsRepository, batch_runs_repo from db.repositories.flow_configs import FlowConfigRepository, flow_config_repo from db.repositories.runs import RunsRepository, runs_repo __all__ = [ "RunsRepository", "runs_repo", "FlowConfigRepository", "flow_config_repo", + "BatchRunsRepository", "batch_runs_repo", ] diff --git a/db/repositories/batch_runs.py b/db/repositories/batch_runs.py new file mode 100644 index 0000000..aa52fa0 --- /dev/null +++ b/db/repositories/batch_runs.py @@ -0,0 +1,55 @@ +"""Repository for the `batch_runs` table.""" +from __future__ import annotations + +from typing import Any + +from db.connection import get_db + + +class BatchRunsRepository: + async def insert( + self, + batch_id: str, + config_id: int | None, + config_name: str, + workers: int, + total_runs: int, + stagger_ms: int, + headless: bool, + ) -> None: + async with get_db() as db: + await db.execute( + """INSERT INTO batch_runs + (batch_id, config_id, config_name, workers, total_runs, stagger_ms, headless) + VALUES (?, ?, ?, ?, ?, ?, ?)""", + (batch_id, config_id, config_name, workers, total_runs, stagger_ms, int(headless)), + ) + await db.commit() + + async def finish( + self, + batch_id: str, + succeeded: int, + failed: int, + stopped: bool, + error: str | None, + ) -> None: + async with get_db() as db: + await db.execute( + """UPDATE batch_runs + SET finished_at = strftime('%Y-%m-%dT%H:%M:%SZ', 'now'), + succeeded = ?, failed = ?, stopped = ?, error = ? + WHERE batch_id = ?""", + (succeeded, failed, int(stopped), error, batch_id), + ) + await db.commit() + + async def list(self, limit: int = 100) -> list[dict[str, Any]]: + async with get_db() as db: + rows = await (await db.execute( + "SELECT * FROM batch_runs ORDER BY id DESC LIMIT ?", (limit,) + )).fetchall() + return [dict(r) for r in rows] + + +batch_runs_repo = BatchRunsRepository() diff --git a/db/repositories/flow_configs.py b/db/repositories/flow_configs.py index b753cd5..b78f02a 100644 --- a/db/repositories/flow_configs.py +++ b/db/repositories/flow_configs.py @@ -9,7 +9,7 @@ from db.connection import get_db FlowConfigRow = dict[str, Any] _FIELDS = [ - "name", "is_active", + "name", "search_texts_json", "search_box_selector", "search_box_selector_type", "search_submit_selector", "search_submit_selector_type", "search_results_selector", "search_results_selector_type", @@ -54,18 +54,16 @@ class FlowConfigRepository: )).fetchone() return _deserialise(dict(row)) if row else None - async def get_active(self) -> FlowConfigRow | None: + async def get_first(self) -> FlowConfigRow | None: async with get_db() as db: row = await (await db.execute( - "SELECT * FROM flow_configs WHERE is_active=1 ORDER BY id DESC LIMIT 1" + "SELECT * FROM flow_configs ORDER BY id ASC LIMIT 1" )).fetchone() return _deserialise(dict(row)) if row else None async def create(self, data: dict[str, Any]) -> int: data = _serialise(data) async with get_db() as db: - if data.get("is_active"): - await db.execute("UPDATE flow_configs SET is_active=0") placeholders = ", ".join("?" for _ in _FIELDS) cols = ", ".join(_FIELDS) values = [data.get(f) for f in _FIELDS] @@ -78,8 +76,6 @@ class FlowConfigRepository: async def update(self, cfg_id: int, data: dict[str, Any]) -> None: data = _serialise(data) async with get_db() as db: - if data.get("is_active"): - await db.execute("UPDATE flow_configs SET is_active=0") set_clause = ", ".join(f"{f}=?" for f in _FIELDS) values = [data.get(f) for f in _FIELDS] + [cfg_id] await db.execute( @@ -87,12 +83,6 @@ class FlowConfigRepository: ) await db.commit() - async def activate(self, cfg_id: int) -> None: - async with get_db() as db: - await db.execute("UPDATE flow_configs SET is_active=0") - await db.execute("UPDATE flow_configs SET is_active=1 WHERE id=?", (cfg_id,)) - await db.commit() - async def delete(self, cfg_id: int) -> None: async with get_db() as db: await db.execute("DELETE FROM flow_configs WHERE id=?", (cfg_id,)) diff --git a/logger.py b/logger.py index 9ba4557..c00b846 100644 --- a/logger.py +++ b/logger.py @@ -9,7 +9,7 @@ Usage anywhere in the codebase: Environment variables: LOG_LEVEL — DEBUG | INFO | WARNING | ERROR | CRITICAL (default: INFO) - LOG_FILE — path to write a rotating log file (optional) + LOG_FILE — path to write a rotating log file (default: logs/seed.log) """ from __future__ import annotations @@ -28,7 +28,7 @@ _LEVEL_MAP: Final = { } _LOG_LEVEL: int = _LEVEL_MAP.get(os.getenv("LOG_LEVEL", "INFO").upper(), logging.INFO) -_LOG_FILE: str | None = os.getenv("LOG_FILE") +_LOG_FILE: str = os.getenv("LOG_FILE", "logs/seed.log") # ── ANSI colour codes (console only) ───────────────────────────────────────── _RESET = "\033[0m" @@ -86,14 +86,14 @@ def _build_root_logger() -> logging.Logger: console.setFormatter(_ColourFormatter()) root.addHandler(console) - # Optional rotating file handler - if _LOG_FILE: - fh = logging.handlers.RotatingFileHandler( - _LOG_FILE, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8" - ) - fh.setLevel(_LOG_LEVEL) - fh.setFormatter(_FILE_FMT) - root.addHandler(fh) + # Rotating file handler (always on) + os.makedirs(os.path.dirname(_LOG_FILE), exist_ok=True) + fh = logging.handlers.RotatingFileHandler( + _LOG_FILE, maxBytes=10 * 1024 * 1024, backupCount=5, encoding="utf-8" + ) + fh.setLevel(_LOG_LEVEL) + fh.setFormatter(_FILE_FMT) + root.addHandler(fh) root.propagate = False return root diff --git a/main.py b/main.py index 5adaac3..f6e93d4 100644 --- a/main.py +++ b/main.py @@ -16,10 +16,15 @@ from logger import get_logger log = get_logger(__name__) -def _load_active_flow() -> dict[str, object]: - cfg = asyncio.run(flow_config_repo.get_active()) - if cfg is None: - log.warning("No active flow config — running without dynamic flow steps.") +def _load_flow(cfg_id: int | None = None) -> dict[str, object]: + if cfg_id is not None: + cfg = asyncio.run(flow_config_repo.get(cfg_id)) + if cfg is None: + log.error("Flow config %d not found.", cfg_id) + else: + cfg = asyncio.run(flow_config_repo.get_first()) + if cfg is None: + log.warning("No flow configs found — running without dynamic flow steps.") return cfg or {} @@ -56,7 +61,7 @@ def _record( log.info("%s OK (no flow config) — %d ms", identifier, total_ms) -def run_otp(phone: str) -> None: +def run_otp(phone: str, cfg_id: int | None = None) -> None: from crawler.driver import driver_session from crawler.dynamic_flow import DynamicFlow from crawler.scenarios.otp_login import OTPLoginScenario @@ -64,7 +69,7 @@ def run_otp(phone: str) -> None: def otp_resolver(p: str) -> str: raise NotImplementedError("Implement otp_resolver to fetch OTP from your SMS gateway") - flow_cfg = _load_active_flow() + flow_cfg = _load_flow(cfg_id) target_url: str = flow_cfg.get("target_url") or "" if flow_cfg else "" t0 = time.monotonic() try: @@ -77,12 +82,12 @@ def run_otp(phone: str) -> None: _record("otp", phone, result, None, t0) -def run_fresh() -> None: +def run_fresh(cfg_id: int | None = None) -> None: from crawler.driver import driver_session from crawler.dynamic_flow import DynamicFlow from crawler.scenarios.fresh_session import FreshSessionScenario - flow_cfg = _load_active_flow() + flow_cfg = _load_flow(cfg_id) target_url: str = flow_cfg.get("target_url") or "" if flow_cfg else "" t0 = time.monotonic() try: @@ -95,10 +100,10 @@ def run_fresh() -> None: _record("fresh", "fresh", result, None, t0) -def run_batch(workers: int, total_runs: int, stagger_ms: int, headless: bool) -> None: +def run_batch(workers: int, total_runs: int, stagger_ms: int, headless: bool, cfg_id: int | None = None) -> None: from crawler.batch import run_batch as _run_batch - flow_cfg = _load_active_flow() + flow_cfg = _load_flow(cfg_id) if not flow_cfg: log.error("No active flow config — activate one in the admin panel first.") sys.exit(1) @@ -137,10 +142,13 @@ def main() -> None: otp_p = sub.add_parser("otp", help="Run Scenario 1 (SMS-OTP login)") otp_p.add_argument("--phone", help="Single phone number (default: all from config)") + otp_p.add_argument("--config-id", type=int, default=None, help="Flow config ID (default: first config)") - sub.add_parser("fresh", help="Run Scenario 2 (fresh session)") + fresh_p = sub.add_parser("fresh", help="Run Scenario 2 (fresh session)") + fresh_p.add_argument("--config-id", type=int, default=None, help="Flow config ID (default: first config)") batch_p = sub.add_parser("batch", help="Run many parallel fresh-session runners") + batch_p.add_argument("--config-id", type=int, default=None, help="Flow config ID (default: first config)") batch_p.add_argument("--workers", type=int, default=3, help="Parallel Chrome instances (default: 3)") batch_p.add_argument("--runs", type=int, default=10, help="Total runs to complete (default: 10)") batch_p.add_argument("--stagger", type=int, default=500, help="Ms delay between worker launches (default: 500)") @@ -156,15 +164,16 @@ def main() -> None: log.error("No phone numbers configured. Set PHONE_NUMBERS in .env or pass --phone") sys.exit(1) for phone in phones: - run_otp(phone) + run_otp(phone, cfg_id=args.config_id) elif args.cmd == "fresh": - run_fresh() + run_fresh(cfg_id=args.config_id) elif args.cmd == "batch": run_batch( workers=args.workers, total_runs=args.runs, stagger_ms=args.stagger, headless=args.headless, + cfg_id=args.config_id, ) diff --git a/pyproject.toml b/pyproject.toml index e9c9330..85d9696 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -14,7 +14,8 @@ dependencies = [ "pydantic>=2.7.0", "python-dotenv>=1.0.1", "setuptools>=70.0.0", - "huey>=2.5.0", + "rq>=1.16.0", + "redis>=5.0.0", ] [tool.pyright] diff --git a/uv.lock b/uv.lock index 3209347..b8e6a22 100644 --- a/uv.lock +++ b/uv.lock @@ -42,6 +42,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/da/42/e921fccf5015463e32a3cf6ee7f980a6ed0f395ceeaa45060b61d86486c2/anyio-4.13.0-py3-none-any.whl", hash = "sha256:08b310f9e24a9594186fd75b4f73f4a4152069e3853f1ed8bfbf58369f4ad708", size = 114353, upload-time = "2026-03-24T12:59:08.246Z" }, ] +[[package]] +name = "async-timeout" +version = "5.0.1" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/a5/ae/136395dfbfe00dfc94da3f3e136d0b13f394cba8f4841120e34226265780/async_timeout-5.0.1.tar.gz", hash = "sha256:d9321a7a3d5a6a5e187e824d2fa0793ce379a202935782d555d6e9d2735677d3", size = 9274, upload-time = "2024-11-06T16:41:39.6Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/fe/ba/e2081de779ca30d473f21f5b30e0e737c438205440784c7dfc81efc2b029/async_timeout-5.0.1-py3-none-any.whl", hash = "sha256:39e3809566ff85354557ec2398b55e096c8364bacac9405a7a1fa429e77fe76c", size = 6233, upload-time = "2024-11-06T16:41:37.9Z" }, +] + [[package]] name = "attrs" version = "26.1.0" @@ -196,6 +205,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/d1/d6/3965ed04c63042e047cb6a3e6ed1a63a35087b6a609aa3a15ed8ac56c221/colorama-0.4.6-py2.py3-none-any.whl", hash = "sha256:4f1d9991f5acc0ca119f9d443620b77f9d6b33703e51011c16baf57afb285fc6", size = 25335, upload-time = "2022-10-25T02:36:20.889Z" }, ] +[[package]] +name = "croniter" +version = "6.2.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "python-dateutil" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/df/de/5832661ed55107b8a09af3f0a2e71e0957226a59eb1dcf0a445cce6daf20/croniter-6.2.2.tar.gz", hash = "sha256:ba60832a5ec8e12e51b8691c3309a113d1cf6526bdf1a48150ce8ec7a532d0ab", size = 113762, upload-time = "2026-03-15T08:43:48.112Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d0/39/783980e78cb92c2d7bdb1fc7dbc86e94ccc6d58224d76a7f1f51b6c51e30/croniter-6.2.2-py3-none-any.whl", hash = "sha256:a5d17b1060974d36251ea4faf388233eca8acf0d09cbd92d35f4c4ac8f279960", size = 45422, upload-time = "2026-03-15T08:43:46.626Z" }, +] + [[package]] name = "fastapi" version = "0.137.1" @@ -292,15 +313,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/2a/39/e50c7c3a983047577ee07d2a9e53faf5a69493943ec3f6a384bdc792deb2/httpx-0.28.1-py3-none-any.whl", hash = "sha256:d909fcccc110f8c7faf814ca82a9a4d816bc5a6dbfea25d6591d6985b8ba59ad", size = 73517, upload-time = "2024-12-06T15:37:21.509Z" }, ] -[[package]] -name = "huey" -version = "3.0.3" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/d1/cb/58f229149944602917a976d533d3fe7d54770d6ea18df5931e3f4f313fa0/huey-3.0.3.tar.gz", hash = "sha256:1a17fef95fc8432f75413f1b77439cef5f3493c1ddbfba9151756b31a1b2dad3", size = 263604, upload-time = "2026-06-12T01:53:55.49Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/e8/82/f85d8918949786420716a5e421525ab12aa084bcbe32d86c9743e50bcf3d/huey-3.0.3-py3-none-any.whl", hash = "sha256:d1c687734778b8282c035a943eead8368c1736bb28abc006596fbbc01bdc96dc", size = 94945, upload-time = "2026-06-12T01:53:53.981Z" }, -] - [[package]] name = "idna" version = "3.18" @@ -479,6 +491,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/8d/59/b4572118e098ac8e46e399a1dd0f2d85403ce8bbaad9ec79373ed6badaf9/PySocks-1.7.1-py3-none-any.whl", hash = "sha256:2725bd0a9925919b9b51739eea5f9e2bae91e83288108a9ad338b2e3a4435ee5", size = 16725, upload-time = "2019-09-20T02:06:22.938Z" }, ] +[[package]] +name = "python-dateutil" +version = "2.9.0.post0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "six" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/66/c0/0c8b6ad9f17a802ee498c46e004a0eb49bc148f2fd230864601a86dcf6db/python-dateutil-2.9.0.post0.tar.gz", hash = "sha256:37dd54208da7e1cd875388217d5e00ebd4179249f90fb72437e91a35459a0ad3", size = 342432, upload-time = "2024-03-01T18:36:20.211Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ec/57/56b9bcc3c9c6a792fcbaf139543cee77261f3651ca9da0c93f5c1221264b/python_dateutil-2.9.0.post0-py2.py3-none-any.whl", hash = "sha256:a8b2bc7bffae282281c8140a97d3aa9c14da0b136dfe83f850eea9a5f7470427", size = 229892, upload-time = "2024-03-01T18:36:18.57Z" }, +] + [[package]] name = "python-dotenv" version = "1.2.2" @@ -552,6 +576,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/f1/12/de94a39c2ef588c7e6455cfbe7343d3b2dc9d6b6b2f40c4c6565744c873d/pyyaml-6.0.3-cp314-cp314t-win_arm64.whl", hash = "sha256:ebc55a14a21cb14062aa4162f906cd962b28e2e9ea38f9b4391244cd8de4ae0b", size = 149341, upload-time = "2025-09-25T21:32:56.828Z" }, ] +[[package]] +name = "redis" +version = "8.0.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "async-timeout", marker = "python_full_version < '3.11.3'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/53/ae/ed461cca5780b5fc8b9fe8ca0ed98d89508645fb9d880c24cc42c087678f/redis-8.0.0.tar.gz", hash = "sha256:a00c5355432051ac14e593b8b197fc76c887ee12d55a0984f69328a1115fdc49", size = 5101591, upload-time = "2026-05-28T12:45:13.5Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/27/e3/b519734372d305bd547534a9f32e4ce9f98552af753dce72cf3483a0ff0b/redis-8.0.0-py3-none-any.whl", hash = "sha256:c938c18338585009f0bc310f4c7e4e4b4d37639356c4ac072cedf3af570c8dc7", size = 499870, upload-time = "2026-05-28T12:45:11.697Z" }, +] + [[package]] name = "requests" version = "2.34.2" @@ -567,6 +603,20 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/a0/f4/c67b0b3f1b9245e8d266f0f112c500d50e5b4e83cb6f3b71b6528104182a/requests-2.34.2-py3-none-any.whl", hash = "sha256:2a0d60c172f83ac6ab31e4554906c0f3b3588d37b5cb939b1c061f4907e278e0", size = 73075, upload-time = "2026-05-14T19:25:26.443Z" }, ] +[[package]] +name = "rq" +version = "2.9.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "click" }, + { name = "croniter" }, + { name = "redis" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/ca/87/d8ce4fb6b969d11454cf6f5c6e10735ffe096e4a9e208b4cf4d501bec41d/rq-2.9.1.tar.gz", hash = "sha256:2ae30c452bd1e8e26b4a213340d4b6e3957cae23176a92f9132c661d73d0c179", size = 745898, upload-time = "2026-06-06T02:48:36.949Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/77/ae/99335807dde9ce8402d31666d385122a20ecd710a1fd62e5d7014265b783/rq-2.9.1-py3-none-any.whl", hash = "sha256:092624e1bc3b3ca9d2ce7ff4049bd7667f772f84695c9dc66a7a662f67f1939a", size = 120816, upload-time = "2026-06-06T02:48:38.587Z" }, +] + [[package]] name = "seed" version = "0.1.0" @@ -575,10 +625,11 @@ dependencies = [ { name = "aiosqlite" }, { name = "fastapi" }, { name = "httpx" }, - { name = "huey" }, { name = "pydantic" }, { name = "python-dotenv" }, { name = "python-multipart" }, + { name = "redis" }, + { name = "rq" }, { name = "selenium" }, { name = "selenium-stealth" }, { name = "setuptools" }, @@ -596,10 +647,11 @@ requires-dist = [ { name = "aiosqlite", specifier = ">=0.20.0" }, { name = "fastapi", specifier = ">=0.111.0" }, { name = "httpx", specifier = ">=0.27.0" }, - { name = "huey", specifier = ">=2.5.0" }, { name = "pydantic", specifier = ">=2.7.0" }, { name = "python-dotenv", specifier = ">=1.0.1" }, { name = "python-multipart", specifier = ">=0.0.9" }, + { name = "redis", specifier = ">=5.0.0" }, + { name = "rq", specifier = ">=1.16.0" }, { name = "selenium", specifier = ">=4.18.0" }, { name = "selenium-stealth", specifier = ">=1.0.6" }, { name = "setuptools", specifier = ">=70.0.0" }, @@ -647,6 +699,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/9d/76/f789f7a86709c6b087c5a2f52f911838cad707cc613162401badc665acfe/setuptools-82.0.1-py3-none-any.whl", hash = "sha256:a59e362652f08dcd477c78bb6e7bd9d80a7995bc73ce773050228a348ce2e5bb", size = 1006223, upload-time = "2026-03-09T12:47:15.026Z" }, ] +[[package]] +name = "six" +version = "1.17.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/94/e7/b2c673351809dca68a0e064b6af791aa332cf192da575fd474ed7d6f16a2/six-1.17.0.tar.gz", hash = "sha256:ff70335d468e7eb6ec65b95b99d3a2836546063f63acc5171de367e834932a81", size = 34031, upload-time = "2024-12-04T17:35:28.174Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b7/ce/149a00dd41f10bc29e5921b496af8b574d8413afcd5e30dfa0ed46c2cc5e/six-1.17.0-py2.py3-none-any.whl", hash = "sha256:4721f391ed90541fddacab5acf947aa0d3dc7d27b2e1e8eda2be8970586c3274", size = 11050, upload-time = "2024-12-04T17:35:26.475Z" }, +] + [[package]] name = "sniffio" version = "1.3.1"