import logging import threading import time from dataclasses import dataclass, field from datetime import datetime from pathlib import Path from typing import Any, Dict, List, Optional from urllib.parse import urlparse from auth_handler import AuthHandler from browser_session import BrowserSession from listing_scraper import ListingScraper from product_scraper import ProductScraper from studio_navigator import StudioNavigator from .exporter import save_csv 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 "top studios" in lowered: self._state.current_state = "Opening Top Studios" self._state.progress = max(self._state.progress, 20) elif "view all studios" in lowered: self._state.current_state = "Opening View All Studios" self._state.progress = max(self._state.progress, 25) elif raw_message.startswith("Scanning listing page"): self._state.current_state = "Scanning listing" try: self._state.current_page = int(raw_message.split("page")[1].split(":")[0].strip()) except Exception: pass self._state.progress = max(self._state.progress, 40) elif raw_message.startswith(" Found "): try: count = int(raw_message.split("Found")[1].split("product")[0].strip()) self._state.total_items_found = max(self._state.total_items_found, count) except Exception: pass elif "scraping" in lowered: self._state.current_state = "Scraping products" elif "export" in lowered: self._state.current_state = "Exporting" self._state.progress = max(self._state.progress, 90) elif "login successful" in lowered: self._state.current_state = "Authenticated" self._state.progress = max(self._state.progress, 20) def _set_state(self, **updates: Any) -> None: with self._lock: for key, value in updates.items(): setattr(self._state, key, value) def _update_from_scan(self, payload: Dict[str, Any]) -> None: updates = {} if "state" in payload: updates["current_state"] = payload["state"] if "current_page" in payload: updates["current_page"] = int(payload["current_page"] or 0) if "found_items" in payload: updates["total_items_found"] = int(payload["found_items"] or 0) if updates: self._set_state(**updates) 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, username: str, password: str, studio: str, min_price: float) -> Dict[str, Any]: if self.is_running(): return {"ok": False, "error": "A run is already in progress"} if not username or not password or not studio: return {"ok": False, "error": "Username, password, and studio are required"} try: min_price = float(min_price) except Exception: return {"ok": False, "error": "Minimum price must be numeric"} 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 = { "username": username, "password": password, "studio": studio.strip(), "min_price": min_price, } 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: # Use absolute path for logs directory (project root) project_root = Path(__file__).resolve().parent.parent.parent logs_dir = project_root / "logs" self.append_log(f"DEBUG: project_root={project_root}, logs_dir={logs_dir}, logs_dir_exists={logs_dir.exists()}") session = BrowserSession(headless=True, state_dir=".browser_state", timeout=30000, logs_dir=str(logs_dir)) self._session = session records: List[Dict[str, Any]] = [] 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, payload["username"], payload["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 self.append_log("Phase 2 ▶ Studio navigation") self._set_state(progress=25, current_state="Opening Top Studios") navigator = StudioNavigator(session) studio_input = payload["studio"] if self._is_direct_studio_url(studio_input): studio_url = self._normalize_url(navigator, studio_input) logger.info(f"Direct studio URL detected, navigating: {studio_url}") # Try a direct navigation first (user requested opening the exact link) if not session.goto(studio_url): logger.warning("Direct navigation failed, falling back to navigator") if not navigator.navigate_to_studio_url(studio_url): self._set_state(last_error="Could not open studio page", current_state="Studio navigation failed") self.append_log("Could not open studio page") return else: time.sleep(1) logger.info("Direct studio page loaded") else: # If input looks like a URL (entry or relative), navigate directly to it if self._looks_like_url(studio_input): studio_url = self._normalize_url(navigator, studio_input) logger.info(f"Entry URL detected, navigating directly: {studio_url}") # Try direct navigation first if not session.goto(studio_url): logger.warning("Direct navigation failed, falling back to navigator") if not navigator.navigate_to_studio_url(studio_url): self._set_state(last_error="Could not open studio page", current_state="Studio navigation failed") self.append_log("Could not open studio page") return else: time.sleep(1) logger.info("Direct studio page loaded") else: # Treat input as a studio name and search the directory studio_url = navigator.find_studio_by_name(studio_input) if not studio_url: self._set_state(last_error=f"Could not find studio: {studio_input}", current_state="Studio not found") self.append_log(f"Could not find studio: {studio_input}") return if self._stop_event.is_set(): self._set_state(current_state="Stopped by user") return self.append_log("Phase 3 ▶ Listing scan") self._set_state(progress=40, current_state="Scanning listing") listing = ListingScraper( session=session, min_price=payload["min_price"], stop_event=self._stop_event, status_callback=self._update_from_scan, ) self.append_log("Phase 4 ▶ Product detail scraping") self._set_state(progress=75, current_state="Scraping products") scraper = ProductScraper(session=session, retry_count=3, stop_event=self._stop_event) scraped_count = 0 try: for idx, pinfo in enumerate(listing.iter_qualifying_products(studio_url), 1): if self._stop_event.is_set(): self.append_log("Stopped by user") self._set_state(current_state="Stopped by user") break scraped_count += 1 self._set_state( total_items_found=scraped_count, progress=min(89, 75 + min(14, scraped_count)), current_state=f"Scraping product {idx}", ) self.append_log(f"Scraping {idx}: {pinfo.get('title', '')}") record = scraper.scrape_product(pinfo["url"]) if not record.get("title") and pinfo.get("title"): record["title"] = pinfo["title"] if not record.get("price") and pinfo.get("price") is not None: record["price"] = f"${pinfo['price']:.2f}" records.append(record) time.sleep(0.2) except Exception as e: self.append_log(f"⚠ Error during product scraping: {e}") logger.error(f"Product scraping error: {e}") self._set_state(current_state=f"Scraping paused due to error (scraped {scraped_count} items)") if scraped_count == 0 and not self._stop_event.is_set(): self._set_state(current_state="No qualifying products found") self.append_log("No qualifying products found") self.append_log("Phase 5 ▶ Export") self._set_state(progress=95, current_state="Exporting CSV") csv_path = save_csv(records, fallback_studio=studio_input) self._set_state(last_csv_path=csv_path) if self._stop_event.is_set(): self._set_state(current_state="Stopped by user", progress=min(95, self.snapshot()["progress"])) else: self._set_state(current_state="Completed", progress=100) self.append_log(f"Export complete → {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}") if records: try: csv_path = save_csv(records, fallback_studio=payload["studio"]) self._set_state(last_csv_path=csv_path) except Exception: pass 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 @staticmethod def _looks_like_url(value: str) -> bool: return value.startswith("http://") or value.startswith("https://") or value.startswith("/") or "dvd_search.php" in value @staticmethod def _is_direct_studio_url(value: str) -> bool: v = (value or "").lower() if "dvd_search.php" in v and ("type=studioid" in v or "search=" in v): return True if "studios.php" in v and "studio" in v: return True return False @staticmethod def _normalize_url(navigator: StudioNavigator, studio_input: str) -> str: if studio_input.startswith("http://") or studio_input.startswith("https://"): url = studio_input else: url = navigator._make_absolute(studio_input) return navigator._ensure_price_sort(url) 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()