Spaces:
Sleeping
Sleeping
| 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") | |
| 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() | |