diff --git a/.env.example b/.env.example index f44ca5e..2af4972 100644 --- a/.env.example +++ b/.env.example @@ -18,6 +18,12 @@ HEADLESS=true # CHROME_BINARY=/usr/bin/google-chrome-stable # optional: explicit Chrome binary path # CHROMEDRIVER_PATH=drivers/chromedriver # pre-patched driver (run: make patch-driver) +# Worker browser-session recordings (Docker Compose enables these by default) +RECORD_SESSIONS=false +RECORDING_FPS=12 +RECORDING_MAX_FILES=3 +RECORDINGS_DIR=recordings + # Scenario 1 — comma-separated phone numbers PHONE_NUMBERS=+989100000001,+989100000002,+989100000003 diff --git a/Dockerfile b/Dockerfile index 1b77109..b267198 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,8 +1,8 @@ # ── Stage 1: dependency resolver ───────────────────────────────────────────── FROM python:3.11-slim AS builder -# Install uv -COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv +# Keep the build independent of GHCR; dependencies already use official PyPI. +RUN pip install --no-cache-dir uv==0.11.31 WORKDIR /app @@ -17,12 +17,9 @@ FROM python:3.11-slim AS runtime # ── Chromium + matching system ChromeDriver ─────────────────────────────────── RUN apt-get update && apt-get install -y --no-install-recommends \ - ca-certificates curl chromium chromium-driver \ + ca-certificates curl chromium chromium-driver ffmpeg xvfb \ && rm -rf /var/lib/apt/lists/* -# ── uv + venv from builder ──────────────────────────────────────────────────── -COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv - WORKDIR /app # Make the venv's binaries the default Python @@ -43,7 +40,7 @@ COPY --from=builder /app/.venv /app/.venv # Patch Debian's version-matched driver at build time. Runtime jobs require no # ChromeDriver network download. -RUN PYTHONPATH=. python scripts/patch_driver.py && mkdir -p data logs +RUN PYTHONPATH=. python scripts/patch_driver.py && mkdir -p data logs recordings # Non-root user for safety RUN useradd -m -u 1001 seed && chown -R seed:seed /app diff --git a/README.md b/README.md index af8c83c..ddb9cf1 100644 --- a/README.md +++ b/README.md @@ -169,6 +169,9 @@ make down ``` The SQLite database is stored in the `seed-data` Docker volume and persists across rebuilds. +Worker browser sessions are recorded at 12 FPS into the shared `seed-recordings` +volume. The admin panel's **Recordings** tab lists and downloads the newest three +completed sessions. --- @@ -186,6 +189,10 @@ Copy `.env.example` to `.env` and set values before running anything. | `DB_PATH` | `data/tracker.db` | Path to the SQLite file | | `HEADLESS` | `true` | Set to `false` to watch the browser (local only) | | `CHROME_BINARY` | _(auto)_ | Explicit path to Chrome, e.g. `/usr/bin/google-chrome-stable` | +| `RECORD_SESSIONS` | `false` | Record browser sessions; Docker workers override this to `true` | +| `RECORDING_FPS` | `12` | Recording frame rate | +| `RECORDING_MAX_FILES` | `3` | Number of completed recordings retained across all workers | +| `RECORDINGS_DIR` | `recordings` | Recording directory; Docker uses the shared `/app/recordings` volume | | `PHONE_NUMBERS` | _(empty)_ | Comma-separated list of phone numbers for Scenario 1 | | `FLOW_STEPS` | `step_home` | Comma-separated step names for the legacy step-registry runner | diff --git a/admin/routes.py b/admin/routes.py index 19eab4d..37949f3 100644 --- a/admin/routes.py +++ b/admin/routes.py @@ -3,10 +3,13 @@ from __future__ import annotations import secrets +from datetime import UTC, datetime +from pathlib import Path from typing import Annotated, Any import speedtest # type: ignore[import-untyped] from fastapi import APIRouter, Depends, HTTPException, status +from fastapi.responses import FileResponse from fastapi.security import HTTPBasic, HTTPBasicCredentials from pydantic import BaseModel @@ -232,6 +235,63 @@ async def list_batch_runs(_: Auth) -> list[dict[str, object]]: return await batch_runs_repo.list() # type: ignore[return-value] +# --------------------------------------------------------------------------- +# Browser-session recordings +# --------------------------------------------------------------------------- + + +def _recording_path(filename: str) -> Path: + if filename != Path(filename).name or Path(filename).suffix.lower() != ".mp4": + raise HTTPException(status_code=404, detail="Recording not found") + directory = Path(config.recordings_dir).resolve() + path = directory / filename + if path.is_symlink() or not path.is_file(): + raise HTTPException(status_code=404, detail="Recording not found") + return path + + +@router.get("/api/recordings") +def list_recordings(_: Auth) -> list[dict[str, object]]: + directory = Path(config.recordings_dir) + if not directory.is_dir(): + return [] + + rows: list[dict[str, object]] = [] + for path in directory.glob("*.mp4"): + if path.is_symlink(): + continue + try: + stat_result = path.stat() + except FileNotFoundError: + continue + rows.append( + { + "name": path.name, + "size_bytes": stat_result.st_size, + "created_at": datetime.fromtimestamp( + stat_result.st_mtime, tz=UTC + ).isoformat(), + } + ) + rows.sort(key=lambda row: str(row["created_at"]), reverse=True) + return rows + + +@router.get("/api/recordings/{filename}") +def download_recording(filename: str, _: Auth) -> FileResponse: + path = _recording_path(filename) + return FileResponse(path, media_type="video/mp4", filename=path.name) + + +@router.delete("/api/recordings/{filename}", status_code=204) +def delete_recording(filename: str, _: Auth) -> None: + path = _recording_path(filename) + try: + path.unlink() + except FileNotFoundError: + raise HTTPException(status_code=404, detail="Recording not found") from None + + # --------------------------------------------------------------------------- # Log viewer # --------------------------------------------------------------------------- diff --git a/admin/static/dashboard.html b/admin/static/dashboard.html index caa7238..9e9d357 100644 --- a/admin/static/dashboard.html +++ b/admin/static/dashboard.html @@ -348,6 +348,7 @@
Dashboard
Flow Configs
Batch Run
+
Recordings
Logs
@@ -589,6 +590,29 @@ + +
+
+
+
+
+
Browser Session Recordings
+
The newest three completed worker sessions are retained at 12 FPS.
+
+ +
+
+ + + + + +
RecordedSessionSizeActions
No recordings yet.
+
+
+
+
+
@@ -712,6 +736,7 @@ document.querySelectorAll('.tab').forEach(tab => { document.getElementById(`tab-${tab.dataset.tab}`).classList.add('active'); if (tab.dataset.tab === 'flows') loadConfigs(); if (tab.dataset.tab === 'batch') { loadConfigs(); pollBatchStatus(); loadBatchHistory(); } + if (tab.dataset.tab === 'recordings') loadRecordings(); if (tab.dataset.tab === 'logs') { loadLogs(); _startLogPoll(); } else _stopLogPoll(); }); @@ -1421,6 +1446,56 @@ async function loadBatchHistory() { renderBatchHistory(await res.json()); } +/* ─────────────────────────── Recordings ─────────────────────── */ +function fmtBytes(bytes) { + if (!bytes) return '0 B'; + const units = ['B', 'KB', 'MB', 'GB']; + const index = Math.min(Math.floor(Math.log(bytes) / Math.log(1024)), units.length - 1); + return `${(bytes / Math.pow(1024, index)).toFixed(index ? 1 : 0)} ${units[index]}`; +} + +async function loadRecordings() { + const res = await apiFetch('/api/recordings'); + if (!res || !res.ok) return; + const rows = await res.json(); + const tbody = document.getElementById('recordings-tbody'); + if (!rows.length) { + tbody.innerHTML = 'No recordings yet.'; + return; + } + tbody.innerHTML = rows.map(row => { + const encodedName = encodeURIComponent(row.name); + return ` + ${fmtTime(row.created_at)} + ${esc(row.name)} + ${fmtBytes(row.size_bytes)} + + + + + `; + }).join(''); +} + +async function downloadRecording(encodedName) { + const res = await apiFetch(`/api/recordings/${encodedName}`); + if (!res || !res.ok) return; + const blobUrl = URL.createObjectURL(await res.blob()); + const link = document.createElement('a'); + link.href = blobUrl; + link.download = decodeURIComponent(encodedName); + document.body.appendChild(link); + link.click(); + link.remove(); + URL.revokeObjectURL(blobUrl); +} + +async function deleteRecording(encodedName) { + if (!confirm(`Delete recording ${decodeURIComponent(encodedName)}?`)) return; + const res = await apiFetch(`/api/recordings/${encodedName}`, { method: 'DELETE' }); + if (res && res.ok) await loadRecordings(); +} + /* ─────────────────────────── Logs ──────────────────────────── */ let _logPollTimer = null; diff --git a/config.py b/config.py index bf8f429..92ddfbf 100644 --- a/config.py +++ b/config.py @@ -18,6 +18,10 @@ class Config: headless: bool = os.getenv("HEADLESS", "true").lower() == "true" chrome_binary: str | None = os.getenv("CHROME_BINARY") chromedriver_path: str = os.getenv("CHROMEDRIVER_PATH", "drivers/chromedriver") + record_sessions: bool = os.getenv("RECORD_SESSIONS", "false").lower() == "true" + recording_fps: int = int(os.getenv("RECORDING_FPS", "12")) + recording_max_files: int = int(os.getenv("RECORDING_MAX_FILES", "3")) + recordings_dir: str = os.getenv("RECORDINGS_DIR", "recordings") # Scenario 1 — SMS-OTP: list of phone numbers (one per line in env or comma-separated) phone_numbers: list[str] = field(default_factory=list) diff --git a/crawler/batch.py b/crawler/batch.py index c8433f4..79ef517 100644 --- a/crawler/batch.py +++ b/crawler/batch.py @@ -30,7 +30,7 @@ def _single_run( log.info("[batch:%s] Run %d starting", batch_id[:8], run_index) try: target_url: str = cfg.get("target_url") or "" - with driver_session(headless=headless) as driver: + with driver_session(headless=headless, recording_name=identifier) as driver: FreshSessionScenario(driver, target_url).run() result = DynamicFlow(driver, cfg).run() if cfg else None diff --git a/crawler/driver.py b/crawler/driver.py index 8e37fc0..07bcf50 100644 --- a/crawler/driver.py +++ b/crawler/driver.py @@ -20,6 +20,7 @@ from selenium.webdriver.chrome.options import Options from selenium_stealth import stealth from config import config +from crawler.recording import SessionRecorder from logger import get_logger log = get_logger(__name__) @@ -88,6 +89,7 @@ def _build_options( headless: bool, chrome_binary: str, profile_dir: str | None = None, + display: str | None = None, ) -> Options: opts = Options() opts.binary_location = chrome_binary @@ -96,6 +98,9 @@ def _build_options( opts.add_argument("--headless=new") if profile_dir: opts.add_argument(f"--user-data-dir={profile_dir}") + if display: + opts.add_argument(f"--display={display}") + opts.add_argument("--ozone-platform=x11") opts.add_argument(f"--user-agent={user_agent}") opts.add_argument("--no-sandbox") opts.add_argument("--disable-gpu") @@ -109,12 +114,13 @@ def _build_options( def make_driver( headless: bool | None = None, profile_dir: str | None = None, + display: str | None = None, ) -> uc.Chrome: """Return a stealthed undetected Chrome instance.""" use_headless = config.headless if headless is None else headless chrome_binary, major = _detect_chrome() ua = random.choice(_USER_AGENTS) - opts = _build_options(ua, use_headless, chrome_binary, profile_dir) + opts = _build_options(ua, use_headless, chrome_binary, profile_dir, display) log.debug("Starting ChromeDriver (headless=%s, Chrome %d)", use_headless, major) @@ -125,7 +131,9 @@ def make_driver( use_subprocess=True, ) - if not use_headless: + if display: + driver.set_window_size(1920, 1080) + elif not use_headless: driver.maximize_window() else: driver.set_window_size(1920, 1080) @@ -180,11 +188,26 @@ def _terminate_profile_processes(profile_dir: str) -> None: @contextmanager -def driver_session(headless: bool | None = None) -> Generator[uc.Chrome, None, None]: +def driver_session( + headless: bool | None = None, + recording_name: str | None = None, +) -> Generator[uc.Chrome, None, None]: profile_dir = tempfile.mkdtemp(prefix="seed-chrome-") driver: uc.Chrome | None = None + recorder: SessionRecorder | None = None + display: str | None = None + effective_headless = config.headless if headless is None else headless try: - driver = make_driver(headless, profile_dir=profile_dir) + if config.record_sessions: + recorder = SessionRecorder(recording_name) + if recorder.start(): + display = recorder.display + effective_headless = False + driver = make_driver( + effective_headless, + profile_dir=profile_dir, + display=display, + ) yield driver finally: _terminate_profile_processes(profile_dir) @@ -194,6 +217,8 @@ def driver_session(headless: bool | None = None) -> Generator[uc.Chrome, None, N except Exception as exc: log.warning("Driver shutdown error: %s", exc) _terminate_profile_processes(profile_dir) + if recorder is not None: + recorder.stop() shutil.rmtree(profile_dir, ignore_errors=True) log.debug("Driver session closed") diff --git a/crawler/recording.py b/crawler/recording.py new file mode 100644 index 0000000..0b8c9fe --- /dev/null +++ b/crawler/recording.py @@ -0,0 +1,186 @@ +"""Per-session Xvfb/FFmpeg recording with shared-volume retention.""" + +from __future__ import annotations + +import fcntl +import os +import re +import signal +import subprocess +import threading +import time +import uuid +from datetime import UTC, datetime +from pathlib import Path + +from config import config +from logger import get_logger + +log = get_logger(__name__) + +_DISPLAY_LOCK = threading.Lock() +_SAFE_NAME = re.compile(r"[^A-Za-z0-9_.-]+") +_WIDTH = 1920 +_HEIGHT = 1080 + + +def _safe_label(label: str | None) -> str: + cleaned = _SAFE_NAME.sub("-", label or "session").strip("-._") + return cleaned[:80] or "session" + + +def _stop_process(process: subprocess.Popen[bytes] | None, *, interrupt: bool = False) -> None: + if process is None or process.poll() is not None: + return + try: + process.send_signal(signal.SIGINT if interrupt else signal.SIGTERM) + process.wait(timeout=10) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=5) + + +def _prune_recordings(directory: Path) -> None: + keep = max(1, config.recording_max_files) + lock_path = directory / ".retention.lock" + with lock_path.open("a+b") as lock_file: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + recordings = sorted( + directory.glob("*.mp4"), + key=lambda path: path.stat().st_mtime_ns, + reverse=True, + ) + for stale in recordings[keep:]: + try: + stale.unlink() + log.info("Removed expired recording %s", stale.name) + except FileNotFoundError: + pass + + +class SessionRecorder: + """Own one virtual display and one FFmpeg recorder.""" + + def __init__(self, label: str | None = None) -> None: + self.label = _safe_label(label) + self.display: str | None = None + self._xvfb: subprocess.Popen[bytes] | None = None + self._ffmpeg: subprocess.Popen[bytes] | None = None + self._temp_path: Path | None = None + self._final_path: Path | None = None + + def start(self) -> bool: + directory = Path(config.recordings_dir) + directory.mkdir(parents=True, exist_ok=True) + stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%S.%fZ") + token = uuid.uuid4().hex[:8] + self._final_path = directory / f"{self.label}-{stamp}-{token}.mp4" + self._temp_path = directory / f".{self._final_path.name}.part.mp4" + + try: + with _DISPLAY_LOCK: + self._start_xvfb() + self._start_ffmpeg() + except Exception as exc: + log.error("Could not start session recording: %s", exc) + self.stop(publish=False) + return False + + log.info( + "Recording session to %s at %d FPS", + self._final_path.name, + config.recording_fps, + ) + return True + + def _start_xvfb(self) -> None: + for display_number in range(100, 1000): + socket_path = Path(f"/tmp/.X11-unix/X{display_number}") + if socket_path.exists(): + continue + display = f":{display_number}" + process = subprocess.Popen( + [ + "Xvfb", + display, + "-screen", + "0", + f"{_WIDTH}x{_HEIGHT}x24", + "-nolisten", + "tcp", + "-ac", + ], + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + ) + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + if process.poll() is not None: + stderr = (process.stderr.read() if process.stderr else b"").decode( + "utf-8", errors="replace" + ) + raise RuntimeError(f"Xvfb exited early: {stderr.strip()}") + if socket_path.exists(): + self.display = display + self._xvfb = process + return + time.sleep(0.05) + _stop_process(process) + raise RuntimeError("no free X display was available") + + def _start_ffmpeg(self) -> None: + if self.display is None or self._temp_path is None: + raise RuntimeError("virtual display is not ready") + self._ffmpeg = subprocess.Popen( + [ + "ffmpeg", + "-y", + "-nostdin", + "-loglevel", + "error", + "-f", + "x11grab", + "-framerate", + str(config.recording_fps), + "-video_size", + f"{_WIDTH}x{_HEIGHT}", + "-i", + f"{self.display}.0", + "-c:v", + "libx264", + "-preset", + "ultrafast", + "-threads", + "1", + "-crf", + "28", + "-pix_fmt", + "yuv420p", + str(self._temp_path), + ], + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + ) + time.sleep(0.2) + if self._ffmpeg.poll() is not None: + stderr = (self._ffmpeg.stderr.read() if self._ffmpeg.stderr else b"").decode( + "utf-8", errors="replace" + ) + raise RuntimeError(f"FFmpeg exited early: {stderr.strip()}") + + def stop(self, *, publish: bool = True) -> Path | None: + _stop_process(self._ffmpeg, interrupt=True) + _stop_process(self._xvfb) + self._ffmpeg = None + self._xvfb = None + + if self._temp_path is None or self._final_path is None: + return None + if not publish or not self._temp_path.exists() or self._temp_path.stat().st_size == 0: + self._temp_path.unlink(missing_ok=True) + return None + + os.replace(self._temp_path, self._final_path) + _prune_recordings(self._final_path.parent) + log.info("Recording saved: %s", self._final_path.name) + return self._final_path diff --git a/crawler/tasks.py b/crawler/tasks.py index a852668..16bcbdb 100644 --- a/crawler/tasks.py +++ b/crawler/tasks.py @@ -309,7 +309,7 @@ def run_batch_job(batch_id: str, item: dict[str, Any]) -> None: try: target_url: str = cfg.get("target_url") or "" scenario: str = cfg.get("scenario") or "dynamic" - with driver_session(headless=headless) as driver: + with driver_session(headless=headless, recording_name=identifier) as driver: FreshSessionScenario(driver, target_url).run() if cfg: if scenario == "digipay": diff --git a/docker-compose.yml b/docker-compose.yml index 6084c13..2794a33 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -33,9 +33,11 @@ services: HEADLESS: "true" CHROME_BINARY: /usr/bin/chromium CHROMEDRIVER_PATH: /usr/bin/chromedriver + RECORDINGS_DIR: /app/recordings volumes: - seed-data:/app/data - seed-logs:/app/logs + - seed-recordings:/app/recordings depends_on: redis: condition: service_healthy @@ -50,8 +52,9 @@ services: worker: build: . image: seed:latest - container_name: seed-worker restart: unless-stopped + deploy: + replicas: 3 command: ["rq", "worker", "--url", "redis://redis:6379/0", "batch"] env_file: - .env @@ -62,9 +65,14 @@ services: HEADLESS: "true" CHROME_BINARY: /usr/bin/chromium CHROMEDRIVER_PATH: /usr/bin/chromedriver + RECORD_SESSIONS: "true" + RECORDING_FPS: "12" + RECORDING_MAX_FILES: "3" + RECORDINGS_DIR: /app/recordings volumes: - seed-data:/app/data - seed-logs:/app/logs + - seed-recordings:/app/recordings depends_on: redis: condition: service_healthy @@ -80,5 +88,7 @@ volumes: driver: local seed-logs: driver: local + seed-recordings: + driver: local redis-data: driver: local