Spaces:
Sleeping
Sleeping
| import logging | |
| import re | |
| import time | |
| from typing import Optional | |
| from urllib.parse import urlparse, parse_qs, urlencode, urlunparse | |
| from browser_session import BrowserSession | |
| logger = logging.getLogger("bot") | |
| ALL_STUDIOS_URL = "https://www.adultdvdmarketplace.com/xcart/adult_dvd/studios.php" | |
| class StudioNavigator: | |
| def __init__(self, session: BrowserSession): | |
| self.session = session | |
| def page(self): | |
| return self.session.page | |
| # ------------------------------------------------------------------ | |
| # Navigate by direct URL | |
| # ------------------------------------------------------------------ | |
| def navigate_to_studio_url(self, url: str) -> bool: | |
| """Go to a studio listing URL, ensuring price ordering is set.""" | |
| url = self._ensure_price_sort(url) | |
| logger.info(f"Opening studio URL: {url}") | |
| if not self.session.goto(url): | |
| return False | |
| time.sleep(1) | |
| logger.info("Studio page loaded") | |
| return True | |
| # ------------------------------------------------------------------ | |
| # Navigate by studio name | |
| # ------------------------------------------------------------------ | |
| def _open_view_all_studios(self) -> bool: | |
| page = self.page | |
| top_studios_selectors = [ | |
| "a:has-text('TOP STUDIOS')", | |
| "a:has-text('Top Studios')", | |
| "button:has-text('TOP STUDIOS')", | |
| "button:has-text('Top Studios')", | |
| ] | |
| for sel in top_studios_selectors: | |
| try: | |
| menu = page.locator(sel).first | |
| if menu.is_visible(timeout=1000): | |
| try: | |
| menu.hover(timeout=1000) | |
| except Exception: | |
| pass | |
| menu.click() | |
| time.sleep(1.5) | |
| break | |
| except Exception: | |
| continue | |
| view_all_selectors = [ | |
| "a:has-text('VIEW ALL STUDIOS...')", | |
| "a:has-text('VIEW ALL STUDIOS')", | |
| "a:has-text('View All Studios')", | |
| "button:has-text('VIEW ALL STUDIOS')", | |
| "button:has-text('View All Studios')", | |
| ] | |
| for sel in view_all_selectors: | |
| try: | |
| link = page.locator(sel).first | |
| if link.is_visible(timeout=2000): | |
| logger.info(f"Found View All Studios link with selector: {sel}") | |
| link.click() | |
| page.wait_for_load_state("domcontentloaded", timeout=10000) | |
| time.sleep(1) | |
| logger.info("Opened all studios directory from menu") | |
| return True | |
| except Exception as e: | |
| logger.debug(f"Failed to find/click VIEW ALL STUDIOS with {sel}: {e}") | |
| continue | |
| logger.warning("Could not open View All Studios from menu; using directory URL") | |
| return self.session.goto(ALL_STUDIOS_URL) | |
| def _click_studio_link(self, studio_name: str) -> bool: | |
| page = self.page | |
| escaped = re.escape(studio_name.strip()) | |
| patterns = [ | |
| re.compile(rf"^{escaped}\s+DVD Movies$", re.I), | |
| re.compile(rf"^{escaped}$", re.I), | |
| re.compile(escaped, re.I), | |
| ] | |
| for attempt in range(1, 4): | |
| try: | |
| # Wait for studio directory content to render before searching links. | |
| page.locator("a:has-text('DVD Movies')").first.wait_for(timeout=4000) | |
| except Exception: | |
| pass | |
| for pattern in patterns: | |
| try: | |
| link = page.get_by_role("link", name=pattern).first | |
| if link.is_visible(timeout=2000): | |
| link.click() | |
| page.wait_for_load_state("domcontentloaded", timeout=10000) | |
| time.sleep(1) | |
| return True | |
| except Exception: | |
| continue | |
| try: | |
| anchors = page.locator("a").all() | |
| target = studio_name.lower().strip() | |
| for anchor in anchors: | |
| try: | |
| text = anchor.inner_text(timeout=500).strip().lower() | |
| if target in text and "dvd movies" in text: | |
| anchor.click() | |
| page.wait_for_load_state("domcontentloaded", timeout=10000) | |
| time.sleep(1) | |
| return True | |
| except Exception: | |
| continue | |
| except Exception: | |
| pass | |
| if attempt < 3: | |
| logger.warning(f"Studio link not found on attempt {attempt}; reloading all studios directory") | |
| self.session.goto(ALL_STUDIOS_URL) | |
| time.sleep(1) | |
| return False | |
| def find_studio_by_name(self, studio_name: str) -> Optional[str]: | |
| """Open the all-studios directory from the menu and click the matching studio.""" | |
| logger.info(f"Searching for studio: '{studio_name}'") | |
| if not self._open_view_all_studios(): | |
| return None | |
| if self._click_studio_link(studio_name): | |
| current_url = self.page.url | |
| logger.info(f"Opened studio page: {current_url}") | |
| return self._ensure_price_sort(current_url) | |
| logger.warning(f"Could not click studio link for: {studio_name}") | |
| return None | |
| # ------------------------------------------------------------------ | |
| # Helpers | |
| # ------------------------------------------------------------------ | |
| def _ensure_price_sort(url: str) -> str: | |
| if "order_by=" in url: | |
| return url | |
| sep = "&" if "?" in url else "?" | |
| return f"{url}{sep}order_by=price" | |
| def _make_absolute(href: str, base: str = "https://www.adultdvdmarketplace.com/xcart/") -> str: | |
| if href.startswith("http"): | |
| return href | |
| from urllib.parse import urljoin | |
| return urljoin(base, href) | |