import logging import threading from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional from auth_handler import AuthHandler from browser_session import BrowserSession from config import HARDCODED_USERNAME, HARDCODED_PASSWORD from modify_listings_updater import update_wo_case_listing_prices logger = logging.getLogger("bot") @dataclass class DashboardState: running: bool = False progress: int = 0 current_state: str = "Idle" total_items_found: int = 0 current_page: int = 0 last_run_time: str = "" last_error: str = "" last_csv_path: str = "" logs: List[str] = field(default_factory=list) stopped: bool = False class DashboardLogHandler(logging.Handler): def __init__(self, controller: "AutomationController"): super().__init__() self.controller = controller def emit(self, record: logging.LogRecord) -> None: try: raw_message = record.getMessage() formatted_message = self.format(record) self.controller.append_log(raw_message, formatted_message) except Exception: pass class AutomationController: def __init__(self): self._lock = threading.RLock() self._state = DashboardState() self._thread: Optional[threading.Thread] = None self._stop_event = threading.Event() self._session: Optional[BrowserSession] = None def snapshot(self) -> Dict[str, Any]: with self._lock: return { "running": self._state.running, "progress": self._state.progress, "current_state": self._state.current_state, "total_items_found": self._state.total_items_found, "current_page": self._state.current_page, "last_run_time": self._state.last_run_time, "last_error": self._state.last_error, "last_csv_path": self._state.last_csv_path, "logs": list(self._state.logs), "stopped": self._state.stopped, } def append_log(self, raw_message: str, display_message: Optional[str] = None) -> None: message = display_message or raw_message with self._lock: self._state.logs.append(message) if len(self._state.logs) > 250: self._state.logs = self._state.logs[-250:] lowered = raw_message.lower() if raw_message.startswith("Phase 1") or "logging in" in lowered: self._state.current_state = "Logging in" self._state.progress = max(self._state.progress, 10) elif "login successful" in lowered: self._state.current_state = "Authenticated" self._state.progress = max(self._state.progress, 20) elif "opened" in lowered: self._state.current_state = "Page loaded" self._state.progress = max(self._state.progress, 50) elif "updating w/o case" in lowered: self._state.current_state = "Updating prices" self._state.progress = max(self._state.progress, 60) elif "update complete" in lowered: self._state.current_state = "Complete" self._state.progress = max(self._state.progress, 100) def _set_state(self, **updates: Any) -> None: with self._lock: for key, value in updates.items(): setattr(self._state, key, value) def is_running(self) -> bool: with self._lock: return self._state.running def stop(self) -> Dict[str, Any]: self._stop_event.set() self._set_state(stopped=True, current_state="Stopping...") return {"ok": True, "message": "Stop requested"} def start(self, target_url: str) -> Dict[str, Any]: if self.is_running(): return {"ok": False, "error": "A run is already in progress"} if not target_url: return {"ok": False, "error": "Target URL is required"} self._stop_event.clear() self._set_state( running=True, stopped=False, progress=1, current_state="Preparing session", total_items_found=0, current_page=0, last_error="", last_csv_path="", last_run_time=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), ) payload = {"target_url": target_url.strip()} self._thread = threading.Thread(target=self._run_worker, args=(payload,), daemon=True) self._thread.start() return {"ok": True, "message": "Automation started"} def _run_worker(self, payload: Dict[str, Any]) -> None: project_root = Path(__file__).resolve().parent.parent.parent logs_dir = project_root / "logs" session = BrowserSession(headless=True, state_dir=".browser_state", timeout=30000, logs_dir=str(logs_dir)) self._session = session log_handler = DashboardLogHandler(self) log_handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s: %(message)s")) bot_logger = logging.getLogger("bot") previous_propagate = bot_logger.propagate bot_logger.addHandler(log_handler) bot_logger.propagate = False try: session.start() self.append_log("Phase 0 ▶ Preparing session") self._set_state(progress=5, current_state="Preparing session") if self._stop_event.is_set(): self._set_state(current_state="Stopped by user") return self.append_log("Phase 1 ▶ Authentication") self._set_state(progress=10, current_state="Logging in") auth = AuthHandler(session, HARDCODED_USERNAME, HARDCODED_PASSWORD) if not auth.ensure_authenticated(): self._set_state(last_error="Login failed", current_state="Login failed") self.append_log("Login failed") return if self._stop_event.is_set(): self._set_state(current_state="Stopped by user") return target_url = payload["target_url"] self.append_log("Phase 2 ▶ Opening target page") self._set_state(progress=25, current_state="Opening page") if not session.goto(target_url): self._set_state(last_error="Could not open target page", current_state="Navigation failed") self.append_log("Could not open target page") return self.append_log(f"Opened: {target_url}") self._set_state(progress=50, current_state="Page loaded") self._set_state(progress=60, current_state="Updating prices") self.append_log("Phase 3 - Updating w/o case listing prices") result = update_wo_case_listing_prices( session.page, dashboard_url=target_url, stop_event=self._stop_event, ) self.append_log( "Update complete: " f"matched={result['matched']}, updated={result['updated']}, " f"failed={result['failed']}, pages={result['pages_scanned']}" ) self.append_log(f"CSV saved: {result['csv_path']}") self._set_state( progress=100, current_state="Complete", total_items_found=result["matched"], current_page=result["pages_scanned"], last_csv_path=result["csv_path"], ) except Exception as exc: logger.exception("Unexpected error in web runner") self._set_state(last_error=str(exc), current_state="Failed", progress=100) self.append_log(f"Error: {exc}") finally: try: bot_logger.removeHandler(log_handler) bot_logger.propagate = previous_propagate except Exception: pass try: session.stop() except Exception: pass self._set_state(running=False) self._session = None def wait_for_finish(self, timeout: float = 0.1) -> bool: thread = self._thread if not thread: return True thread.join(timeout=timeout) return not thread.is_alive()