834 lines
35 KiB
Python
834 lines
35 KiB
Python
"""
|
|
scraper.py — Selenium automation to log in to Factors.AI and download the
|
|
Account Profiles CSV.
|
|
"""
|
|
import glob
|
|
import logging
|
|
import os
|
|
import random
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from selenium import webdriver
|
|
from selenium.common.exceptions import NoSuchElementException, TimeoutException, StaleElementReferenceException
|
|
from selenium.webdriver.chrome.options import Options
|
|
from selenium.webdriver.chrome.service import Service
|
|
from selenium.webdriver.common.by import By
|
|
from selenium.webdriver.common.keys import Keys
|
|
from selenium.webdriver.support import expected_conditions as EC
|
|
from selenium.webdriver.support.ui import WebDriverWait
|
|
from webdriver_manager.chrome import ChromeDriverManager
|
|
from selenium.common.exceptions import TimeoutException
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_PAGE_LOAD_TIMEOUT = 30 # seconds to wait for page transitions
|
|
_ELEMENT_TIMEOUT = 20 # seconds to wait for elements
|
|
_DOWNLOAD_TIMEOUT = 90 # seconds to wait for file download
|
|
|
|
|
|
class FactorsAIScraper:
|
|
"""Downloads the Account Profiles CSV export from app.factors.ai."""
|
|
|
|
def __init__(self, download_dir: str, headless: bool = True):
|
|
self.download_dir = str(Path(download_dir).resolve())
|
|
self.headless = headless
|
|
self.driver: webdriver.Chrome | None = None
|
|
|
|
# ── Driver setup ──────────────────────────────────────────────────────────
|
|
|
|
def _setup_driver(self) -> None:
|
|
options = Options()
|
|
|
|
if self.headless:
|
|
options.add_argument("--headless=new")
|
|
|
|
# Use a persistent Chrome profile to keep cookies and session active
|
|
profile_dir = os.path.join(self.download_dir, "chrome_profile")
|
|
os.makedirs(profile_dir, exist_ok=True)
|
|
# Clean up stale chrome locks
|
|
lock_file = os.path.join(profile_dir, "SingletonLock")
|
|
if os.path.islink(lock_file) or os.path.exists(lock_file):
|
|
try:
|
|
os.remove(lock_file)
|
|
logger.info("Cleaned up stale SingletonLock from Chrome profile.")
|
|
except Exception as e:
|
|
logger.warning(f"Could not remove SingletonLock: {e}")
|
|
options.add_argument(f"--user-data-dir={profile_dir}")
|
|
|
|
options.add_argument("--no-sandbox")
|
|
options.add_argument("--disable-dev-shm-usage")
|
|
options.add_argument("--disable-gpu")
|
|
options.add_argument("--window-size=1920,1080")
|
|
options.add_argument("--disable-extensions")
|
|
options.add_argument("--disable-blink-features=AutomationControlled")
|
|
options.add_experimental_option("excludeSwitches", ["enable-automation"])
|
|
options.add_experimental_option("useAutomationExtension", False)
|
|
|
|
prefs = {
|
|
"download.default_directory": self.download_dir,
|
|
"download.prompt_for_download": False,
|
|
"download.directory_upgrade": True,
|
|
"safebrowsing.enabled": False,
|
|
"safebrowsing.disable_download_protection": True,
|
|
}
|
|
options.add_experimental_option("prefs", prefs)
|
|
|
|
service = Service(ChromeDriverManager().install())
|
|
self.driver = webdriver.Chrome(service=service, options=options)
|
|
self.driver.set_page_load_timeout(_PAGE_LOAD_TIMEOUT)
|
|
|
|
# Enable downloads explicitly when headless (CDP call)
|
|
if self.headless:
|
|
try:
|
|
self.driver.execute_cdp_cmd(
|
|
"Page.setDownloadBehavior",
|
|
{"behavior": "allow", "downloadPath": self.download_dir},
|
|
)
|
|
except Exception:
|
|
pass # Older ChromeDriver builds may not support this
|
|
|
|
# ── Helpers ───────────────────────────────────────────────────────────────
|
|
|
|
def _wait(
|
|
self,
|
|
by: str,
|
|
value: str,
|
|
timeout: int = _ELEMENT_TIMEOUT,
|
|
condition: str = "presence",
|
|
):
|
|
ec_map = {
|
|
"presence": EC.presence_of_element_located,
|
|
"clickable": EC.element_to_be_clickable,
|
|
"visible": EC.visibility_of_element_located,
|
|
}
|
|
return WebDriverWait(self.driver, timeout).until(
|
|
ec_map[condition]((by, value))
|
|
)
|
|
|
|
def _try_selectors(self, selectors: list[tuple], timeout: int = 8):
|
|
"""Return the first element found from a list of (By.*, selector) pairs, waiting at most timeout seconds total."""
|
|
deadline = time.time() + timeout
|
|
while True:
|
|
for by, sel in selectors:
|
|
try:
|
|
elements = self.driver.find_elements(by, sel)
|
|
for el in elements:
|
|
if el.is_displayed() and el.is_enabled():
|
|
return el
|
|
except Exception:
|
|
pass
|
|
if time.time() > deadline:
|
|
break
|
|
time.sleep(0.2)
|
|
return None
|
|
|
|
def _click(self, element) -> None:
|
|
"""Click an element, falling back to JS click if intercepted."""
|
|
try:
|
|
element.click()
|
|
except Exception as e:
|
|
logger.warning(f"Standard click failed ({e}), attempting JS click...")
|
|
self.driver.execute_script("arguments[0].click();", element)
|
|
|
|
def _type_humanlike(self, element, text: str) -> None:
|
|
"""Type text into an element with a short random delay between keystrokes."""
|
|
element.click()
|
|
from selenium.webdriver.common.keys import Keys
|
|
try:
|
|
element.send_keys(Keys.CONTROL + "a")
|
|
element.send_keys(Keys.BACKSPACE)
|
|
except Exception:
|
|
element.clear()
|
|
|
|
for char in text:
|
|
element.send_keys(char)
|
|
time.sleep(random.uniform(0.04, 0.12))
|
|
|
|
try:
|
|
self.driver.execute_script("arguments[0].dispatchEvent(new Event('input', { bubbles: true }));", element)
|
|
self.driver.execute_script("arguments[0].dispatchEvent(new Event('change', { bubbles: true }));", element)
|
|
except Exception:
|
|
pass
|
|
|
|
# ── Login ─────────────────────────────────────────────────────────────────
|
|
|
|
def login(self, email: str, password: str) -> None:
|
|
logger.info("Opening https://app.factors.ai/ …")
|
|
self.driver.get("https://app.factors.ai/")
|
|
time.sleep(3)
|
|
|
|
# Check if we are already logged in
|
|
current_url = self.driver.current_url.lower()
|
|
if all(kw not in current_url for kw in ("login", "signin", "sign-in")):
|
|
logger.info("Already logged in (session restored from profile).")
|
|
return
|
|
|
|
max_attempts = 3
|
|
for attempt in range(1, max_attempts + 1):
|
|
logger.info(f"Login attempt {attempt}/{max_attempts} …")
|
|
|
|
# Find input fields
|
|
email_input = self._try_selectors([
|
|
(By.CSS_SELECTOR, "input[name='email']"),
|
|
(By.CSS_SELECTOR, "input[type='email']"),
|
|
(By.XPATH, "//input[@placeholder[contains(translate(.,'EMAIL','email'),'email')]]"),
|
|
(By.XPATH, "//input[@autocomplete='username']"),
|
|
(By.XPATH, "//input[@autocomplete='email']"),
|
|
], timeout=5)
|
|
|
|
if email_input is None:
|
|
self._save_screenshot("login_page_not_found.png")
|
|
raise RuntimeError("Could not find email input on the login page.")
|
|
|
|
self._type_humanlike(email_input, email)
|
|
logger.info("Email entered.")
|
|
|
|
pwd_input = self._wait(
|
|
By.CSS_SELECTOR, "input[type='password']", timeout=10
|
|
)
|
|
self._type_humanlike(pwd_input, password)
|
|
logger.info("Password entered.")
|
|
|
|
submit = self._try_selectors([
|
|
(By.CSS_SELECTOR, "button[type='submit']"),
|
|
(By.XPATH, "//button[contains(translate(.,'LOGIN','login'),'login') or contains(translate(.,'SIGN IN','sign in'),'sign in')]"),
|
|
(By.CSS_SELECTOR, "button.ant-btn-primary"),
|
|
], timeout=10)
|
|
|
|
if submit is None:
|
|
raise RuntimeError("Could not find login submit button.")
|
|
|
|
self._click(submit)
|
|
logger.info("Login form submitted.")
|
|
time.sleep(5)
|
|
|
|
# Check if login failed (e.g. error message displayed)
|
|
try:
|
|
err = WebDriverWait(self.driver, 5).until(
|
|
EC.visibility_of_element_located((
|
|
By.XPATH,
|
|
"//*[self::div or self::span or self::p or contains(@class,'error') or contains(@class,'message')][contains(text(), 'Login failed') or contains(text(), 'invalid') or contains(text(), 'incorrect')]"
|
|
))
|
|
)
|
|
logger.warning(f"Login failed message detected: {err.text}")
|
|
if attempt < max_attempts:
|
|
sleep_time = 45 * attempt
|
|
logger.info(f"Waiting {sleep_time} seconds before retrying login...")
|
|
time.sleep(sleep_time)
|
|
# Refresh the page to start clean
|
|
self.driver.get("https://app.factors.ai/")
|
|
time.sleep(3)
|
|
continue
|
|
else:
|
|
raise RuntimeError(f"Login rejected by Factors.AI after {max_attempts} attempts: {err.text}")
|
|
except TimeoutException:
|
|
# No failure message found, verify if we are logged in
|
|
current_url = self.driver.current_url.lower()
|
|
if all(kw not in current_url for kw in ("login", "signin", "sign-in")):
|
|
logger.info(f"Logged in successfully. URL: {self.driver.current_url}")
|
|
return
|
|
else:
|
|
# No error message, but still on login page. Wait and check again.
|
|
time.sleep(3)
|
|
if all(kw not in self.driver.current_url.lower() for kw in ("login", "signin", "sign-in")):
|
|
logger.info(f"Logged in successfully. URL: {self.driver.current_url}")
|
|
return
|
|
|
|
raise RuntimeError("Still on login page after submission.")
|
|
|
|
# ── Navigate to Account Profiles ──────────────────────────────────────────
|
|
|
|
def _navigate_to_accounts(self) -> None:
|
|
"""Find and open the Account Profiles page."""
|
|
logger.info("Navigating to Account Profiles page …")
|
|
|
|
# Derive base URL (e.g. https://app.factors.ai) from current URL
|
|
parts = self.driver.current_url.split("/")
|
|
base_url = "/".join(parts[:3]) # ['https:', '', 'app.factors.ai']
|
|
|
|
# 1. If we are already on the base URL, just wait for the download button to be present
|
|
from urllib.parse import urlparse
|
|
parsed = urlparse(self.driver.current_url)
|
|
path = parsed.path.rstrip("/")
|
|
if path in ("", "/", "/v2/accounts"):
|
|
logger.info("Already on the home/accounts URL. Waiting for the Accounts page to load...")
|
|
self._dismiss_upgrade_modal()
|
|
if self._download_button_present(timeout=25):
|
|
logger.info("Accounts page loaded successfully on the home page. Waiting 5s for data loading/hydration...")
|
|
time.sleep(5)
|
|
return
|
|
|
|
# 2. Check if the download button is already present (e.g. from a previous navigation)
|
|
self._dismiss_upgrade_modal()
|
|
if self._download_button_present(timeout=3):
|
|
logger.info("Already on the Account Profiles page. Waiting 5s for data loading/hydration...")
|
|
time.sleep(5)
|
|
return
|
|
|
|
# 3. Try clicking a sidebar / nav link first (fastest)
|
|
nav_candidates = [
|
|
(By.ID, "fa-at-link--accounts"),
|
|
(By.XPATH, "//a[contains(translate(@href,'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz'),'account')]"),
|
|
(By.XPATH, "//span[contains(translate(text(),'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz'),'account')]/ancestor::a"),
|
|
(By.XPATH, "//li[contains(translate(.,'ABCDEFGHIJKLMNOPQRSTUVWXYZ','abcdefghijklmnopqrstuvwxyz'),'account')]//a"),
|
|
]
|
|
link = self._try_selectors(nav_candidates, timeout=5)
|
|
if link:
|
|
try:
|
|
self._click(link)
|
|
time.sleep(1)
|
|
if self._download_button_present(timeout=5):
|
|
logger.info(f"Accounts page found via nav. URL: {self.driver.current_url}. Waiting 5s for data loading/hydration...")
|
|
time.sleep(5)
|
|
return
|
|
except Exception:
|
|
pass
|
|
|
|
# 4. Try known URL patterns
|
|
for slug in [
|
|
"/v2/accounts",
|
|
"/accounts",
|
|
"/",
|
|
]:
|
|
url = base_url + slug
|
|
logger.info(f" Trying {url} …")
|
|
try:
|
|
self.driver.get(url)
|
|
if self._download_button_present(timeout=3):
|
|
logger.info(f"Accounts page found at: {url}. Waiting 5s for data loading/hydration...")
|
|
time.sleep(5)
|
|
return
|
|
except Exception:
|
|
pass
|
|
|
|
# Last resort: dump screenshot and raise
|
|
self._save_screenshot("accounts_not_found.png")
|
|
raise RuntimeError(
|
|
"Could not navigate to the Account Profiles page. "
|
|
"Screenshot saved → tmp_download/accounts_not_found.png"
|
|
)
|
|
|
|
def _download_button_present(self, timeout: int = 6) -> bool:
|
|
"""Return True if the toolbar download button is visible on the page."""
|
|
combined_selector = (
|
|
"button#fa-at-btn--export-csv, "
|
|
"button[id*='export-csv'], "
|
|
"[class*='profileToolbarIconBtn'], "
|
|
"button[title*='Download'], "
|
|
"button[title*='Export']"
|
|
)
|
|
try:
|
|
WebDriverWait(self.driver, timeout).until(
|
|
EC.presence_of_element_located((By.CSS_SELECTOR, combined_selector))
|
|
)
|
|
return True
|
|
except TimeoutException:
|
|
# Fallback to xpath check
|
|
try:
|
|
WebDriverWait(self.driver, 1).until(
|
|
EC.presence_of_element_located((
|
|
By.XPATH,
|
|
"//button[.//svg[@data-icon='download'] or contains(@class,'profileToolbarIconBtn')]"
|
|
))
|
|
)
|
|
return True
|
|
except TimeoutException:
|
|
return False
|
|
|
|
def _click_download_button(self) -> None:
|
|
logger.info("Clicking the download toolbar button …")
|
|
|
|
for attempt in range(3):
|
|
self._dismiss_upgrade_modal()
|
|
|
|
# Find all profile toolbar buttons
|
|
candidates = self.driver.find_elements(By.CSS_SELECTOR, "[class*='profileToolbarIconBtn']")
|
|
btn = None
|
|
for c in candidates:
|
|
try:
|
|
html = c.get_attribute("outerHTML") or ""
|
|
if "download" in html or "fa-download" in html:
|
|
if c.is_displayed() and c.is_enabled():
|
|
btn = c
|
|
break
|
|
except Exception:
|
|
pass
|
|
|
|
if btn is None:
|
|
logger.warning(f"Download button not found on attempt {attempt + 1}")
|
|
time.sleep(2)
|
|
continue
|
|
|
|
try:
|
|
self.driver.execute_script("arguments[0].scrollIntoView(true);", btn)
|
|
time.sleep(0.5)
|
|
self.driver.execute_script("arguments[0].click();", btn)
|
|
logger.info(f"Download button clicked via JS (attempt {attempt + 1}).")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to click button on attempt {attempt + 1}: {e}")
|
|
time.sleep(2)
|
|
continue
|
|
|
|
time.sleep(2.0)
|
|
|
|
modals = self.driver.find_elements(By.CSS_SELECTOR, ".ant-modal-content")
|
|
if any(m.is_displayed() for m in modals):
|
|
logger.info("Modal/dropdown detected after click.")
|
|
return
|
|
logger.warning("Modal/dropdown did not open. Retrying click...")
|
|
|
|
self._save_screenshot("download_btn_click_failed.png")
|
|
raise RuntimeError("Failed to open export modal after multiple clicks.")
|
|
|
|
# ── Handle the export modal ───────────────────────────────────────────────
|
|
|
|
def _handle_modal(self, fields_to_select: list[str] | None) -> None:
|
|
logger.info("Waiting for CSV export modal …")
|
|
self._dismiss_upgrade_modal()
|
|
|
|
# Find the visible modal that is actually the export modal
|
|
time.sleep(2) # Give time for the modal to open
|
|
modals = self.driver.find_elements(By.CSS_SELECTOR, ".ant-modal-content")
|
|
modal = None
|
|
for m in modals:
|
|
if m.is_displayed():
|
|
text = m.text.lower()
|
|
if any(kw in text for kw in ("export", "field", "select", "column", "include")):
|
|
modal = m
|
|
break
|
|
|
|
if modal is None:
|
|
# Fallback to waiting for the first visible modal
|
|
modal = WebDriverWait(self.driver, _ELEMENT_TIMEOUT).until(
|
|
EC.visibility_of_element_located((By.CSS_SELECTOR, ".ant-modal-content"))
|
|
)
|
|
|
|
time.sleep(1.0) # let the virtual list fully render
|
|
|
|
# --- Optionally override field selection ---
|
|
if fields_to_select is not None:
|
|
# 1. Clear all
|
|
try:
|
|
clear_btn = modal.find_element(
|
|
By.XPATH,
|
|
".//button[.//*[contains(text(),'Clear All')]]",
|
|
)
|
|
self._click(clear_btn)
|
|
time.sleep(0.6)
|
|
logger.info("Cleared all field selections.")
|
|
except NoSuchElementException:
|
|
logger.warning("'Clear All' button not found — skipping.")
|
|
|
|
# 2. Scroll through the virtual list and check each desired field
|
|
for field_val in fields_to_select:
|
|
self._select_field_in_modal(modal, field_val)
|
|
|
|
# --- Click Export CSV ---
|
|
export_btn = None
|
|
|
|
# 1. Look for any button with "export" or "csv" in text anywhere inside the modal
|
|
try:
|
|
buttons = modal.find_elements(By.TAG_NAME, "button")
|
|
for btn in buttons:
|
|
text = (btn.text or "").lower()
|
|
if "export" in text or "csv" in text:
|
|
export_btn = btn
|
|
break
|
|
except Exception:
|
|
pass
|
|
|
|
# 2. Fallback to primary button inside the modal
|
|
if export_btn is None:
|
|
try:
|
|
export_btn = modal.find_element(By.CSS_SELECTOR, ".ant-btn-primary")
|
|
except NoSuchElementException:
|
|
pass
|
|
|
|
if export_btn is None:
|
|
self._save_screenshot("export_btn_not_found.png")
|
|
raise RuntimeError("Could not find the 'Export CSV' button in modal.")
|
|
|
|
self._click(export_btn)
|
|
logger.info("'Export CSV' button clicked.")
|
|
time.sleep(1.5)
|
|
self._dismiss_upgrade_modal()
|
|
|
|
# def _select_field_in_modal(self, modal, field_val: str) -> None:
|
|
# """Scroll inside the modal virtual list until the checkbox is found, then check it."""
|
|
# scrollable = modal.find_element(By.CSS_SELECTOR, ".rc-virtual-list-holder")
|
|
# max_scrolls = 20
|
|
|
|
# for _ in range(max_scrolls):
|
|
# try:
|
|
# # Find the label or span containing the exact field name
|
|
# label_el = modal.find_element(
|
|
# By.XPATH,
|
|
# f".//span[text()='{field_val}'] | .//label[contains(.,'{field_val}')] | .//*[text()='{field_val}']"
|
|
# )
|
|
|
|
# # Scroll it into view inside the virtual list container
|
|
# self.driver.execute_script("arguments[0].scrollIntoView({block: 'center'});", label_el)
|
|
# time.sleep(0.1)
|
|
|
|
# # Locate the parent/wrapper label and the checkbox input
|
|
# parent_label = label_el.find_element(By.XPATH, "./ancestor-or-self::label[contains(@class,'ant-checkbox-wrapper')]")
|
|
# checkbox = parent_label.find_element(By.CSS_SELECTOR, "input[type='checkbox']")
|
|
|
|
# if not checkbox.is_selected():
|
|
# self.driver.execute_script("arguments[0].click();", checkbox)
|
|
# time.sleep(0.2)
|
|
# logger.info(f"Selected field: {field_val}")
|
|
# return
|
|
# except (NoSuchElementException, StaleElementReferenceException):
|
|
# # Scroll down inside the modal list to reveal more items
|
|
# self.driver.execute_script(
|
|
# "arguments[0].scrollTop += 100;", scrollable
|
|
# )
|
|
# time.sleep(1.5)
|
|
|
|
# logger.warning(f"Field not found in modal: {field_val}")
|
|
|
|
def _select_field_in_modal(self, modal, field_val: str) -> None:
|
|
"""
|
|
Select a field inside the Export CSV modal.
|
|
|
|
Uses the modal's built-in search box instead of scrolling the
|
|
virtual list. This is much more reliable with Ant Design's
|
|
rc-virtual-list because only matching rows are rendered.
|
|
"""
|
|
|
|
wait = WebDriverWait(self.driver, 15)
|
|
|
|
try:
|
|
# Locate search input
|
|
search_box = wait.until(
|
|
lambda d: modal.find_element(
|
|
By.CSS_SELECTOR,
|
|
"input[placeholder='Search properties']"
|
|
)
|
|
)
|
|
|
|
# Clear any previous search
|
|
search_box.click()
|
|
search_box.send_keys(Keys.CONTROL, "a")
|
|
search_box.send_keys(Keys.DELETE)
|
|
|
|
# Type desired field
|
|
search_box.send_keys(field_val)
|
|
|
|
# Wait until matching checkbox appears
|
|
checkbox = wait.until(
|
|
lambda d: modal.find_element(
|
|
By.XPATH,
|
|
f"""
|
|
.//label[
|
|
.//*[normalize-space(text())="{field_val}"]
|
|
]//input[@type='checkbox']
|
|
"""
|
|
)
|
|
)
|
|
|
|
# Tick if needed
|
|
if not checkbox.is_selected():
|
|
self.driver.execute_script(
|
|
"arguments[0].click();",
|
|
checkbox
|
|
)
|
|
|
|
wait.until(lambda d: checkbox.is_selected())
|
|
|
|
logger.info(f"Selected field: {field_val}")
|
|
|
|
except TimeoutException:
|
|
logger.warning(f"Field not found in modal: {field_val}")
|
|
|
|
finally:
|
|
# Clear search so next iteration starts fresh
|
|
try:
|
|
search_box = modal.find_element(
|
|
By.CSS_SELECTOR,
|
|
"input[placeholder='Search properties']"
|
|
)
|
|
|
|
search_box.send_keys(Keys.CONTROL, "a")
|
|
search_box.send_keys(Keys.DELETE)
|
|
|
|
wait.until(
|
|
lambda d: search_box.get_attribute("value") == ""
|
|
)
|
|
|
|
except Exception:
|
|
pass
|
|
|
|
# ── Wait for the CSV download to finish ───────────────────────────────────
|
|
|
|
def _wait_for_download(self, timeout: int = _DOWNLOAD_TIMEOUT) -> str:
|
|
logger.info(f"Waiting up to {timeout}s for CSV download …")
|
|
deadline = time.time() + timeout
|
|
|
|
while time.time() < deadline:
|
|
self._dismiss_upgrade_modal()
|
|
|
|
all_csvs = glob.glob(os.path.join(self.download_dir, "*.csv"))
|
|
in_flight = glob.glob(os.path.join(self.download_dir, "*.crdownload"))
|
|
|
|
logger.info(f"Checking downloads. CSVs: {len(all_csvs)}, in-flight: {len(in_flight)}")
|
|
|
|
complete = [f for f in all_csvs if not f.endswith(".crdownload")]
|
|
if complete and not in_flight:
|
|
latest = max(complete, key=os.path.getmtime)
|
|
logger.info(f"Download complete → {latest}")
|
|
return latest
|
|
|
|
time.sleep(1)
|
|
|
|
self._save_screenshot("download_timeout.png")
|
|
raise TimeoutError(
|
|
f"CSV download did not complete within {timeout} seconds."
|
|
)
|
|
|
|
# ── Utilities ─────────────────────────────────────────────────────────────
|
|
|
|
def _save_screenshot(self, filename: str) -> None:
|
|
if self.driver:
|
|
path = os.path.join(self.download_dir, filename)
|
|
try:
|
|
self.driver.save_screenshot(path)
|
|
logger.info(f"Screenshot saved → {path}")
|
|
except Exception:
|
|
pass
|
|
|
|
def _cleanup_download_dir(self) -> None:
|
|
"""Remove any leftover CSVs from previous runs."""
|
|
for f in glob.glob(os.path.join(self.download_dir, "*.csv")):
|
|
try:
|
|
os.remove(f)
|
|
except OSError:
|
|
pass
|
|
|
|
# ── Project switching and Modals ──────────────────────────────────────────
|
|
|
|
def _dismiss_upgrade_modal(self) -> None:
|
|
"""Find and dismiss the 'Upgrade to experience the full power of Factors' / 'Go beyond signals' modal if present."""
|
|
try:
|
|
modals = self.driver.find_elements(
|
|
By.XPATH,
|
|
"//div[contains(@class, 'ant-modal-content')][.//*[contains(text(), 'Upgrade to experience') or contains(text(), 'Go beyond signals') or contains(text(), 'Book a Demo')]]"
|
|
)
|
|
for m in modals:
|
|
if m.is_displayed():
|
|
logger.info("Upgrade/Paywall modal detected! Dismissing...")
|
|
try:
|
|
close_btn = m.find_element(
|
|
By.CSS_SELECTOR,
|
|
"button.ant-modal-close, button[aria-label='Close'], .ant-modal-close-x, button[class*='close']"
|
|
)
|
|
self._click(close_btn)
|
|
logger.info("Successfully dismissed upgrade modal.")
|
|
time.sleep(1)
|
|
except Exception as click_err:
|
|
logger.warning(f"Could not click close button on upgrade modal: {click_err}")
|
|
except Exception as e:
|
|
logger.debug(f"Error checking/dismissing upgrade modal: {e}")
|
|
|
|
def _dismiss_popups(self) -> None:
|
|
"""Attempt to dismiss any promotional or feedback modals that might block interaction."""
|
|
self._dismiss_upgrade_modal()
|
|
logger.info("Checking for any blocking popups/modals to dismiss...")
|
|
|
|
# 1. Look for G2 review popup / "Don't show this again" buttons
|
|
try:
|
|
g2_buttons = self.driver.find_elements(
|
|
By.XPATH,
|
|
"//button[contains(translate(text(), 'DONT SHOW', 'dont show'), 'dont show') or contains(., \"Don't show this again\")]"
|
|
)
|
|
for btn in g2_buttons:
|
|
if btn.is_displayed():
|
|
logger.info("Dismissing G2 / 'Don't show this again' popup.")
|
|
self._click(btn)
|
|
time.sleep(1)
|
|
except Exception as e:
|
|
logger.debug(f"Error checking/dismissing G2 popup: {e}")
|
|
|
|
# 2. Look for general modal close buttons (e.g. ant-modal-close, ant-modal-close-x)
|
|
try:
|
|
close_buttons = self.driver.find_elements(
|
|
By.CSS_SELECTOR,
|
|
".ant-modal-close, button.ant-modal-close, .ant-modal-close-x, [class*='ant-modal-close']"
|
|
)
|
|
for btn in close_buttons:
|
|
if btn.is_displayed():
|
|
# We do NOT want to close the project switch modal or fields modal!
|
|
parent_modal_text = ""
|
|
try:
|
|
parent_modal = btn.find_element(By.XPATH, "./ancestor::div[contains(@class, 'ant-modal-content')]")
|
|
parent_modal_text = parent_modal.text.lower()
|
|
except Exception:
|
|
pass
|
|
|
|
if "switch" in parent_modal_text or "export" in parent_modal_text or "csv" in parent_modal_text:
|
|
logger.info("Skipping modal close for functional modal (switch project/export).")
|
|
continue
|
|
|
|
logger.info("Clicking modal close button to dismiss popup.")
|
|
self._click(btn)
|
|
time.sleep(1)
|
|
except Exception as e:
|
|
logger.debug(f"Error dismissing modal popups: {e}")
|
|
|
|
def _switch_project(self, target_project: str) -> None:
|
|
"""Switch the current project to the target_project if not already active."""
|
|
if not target_project:
|
|
logger.info("No target project specified; skipping switch.")
|
|
return
|
|
|
|
# Dismiss any initially shown modals/popups
|
|
self._dismiss_popups()
|
|
|
|
def clean_name(name: str) -> str:
|
|
return name.replace(" ", "").replace("-", "").replace("_", "").lower().strip()
|
|
|
|
clean_target = clean_name(target_project)
|
|
|
|
logger.info(f"Ensuring target project is set to: '{target_project}'")
|
|
|
|
# 1. Locate the main project dropdown button
|
|
try:
|
|
dropdown_btn = self._wait(
|
|
By.ID, "fa-at-dropdown--projects", timeout=20, condition="clickable"
|
|
)
|
|
except TimeoutException:
|
|
self._save_screenshot("project_dropdown_not_found.png")
|
|
raise RuntimeError("Could not find the project dropdown button (#fa-at-dropdown--projects).")
|
|
|
|
current_project_text = dropdown_btn.text
|
|
logger.info(f"Main project button text: {current_project_text!r}")
|
|
|
|
# If it's already active on the top nav, we're done
|
|
if clean_target in clean_name(current_project_text):
|
|
logger.info(f"Project '{target_project}' is already active.")
|
|
return
|
|
|
|
# 2. Click the dropdown button to open the project selector
|
|
logger.info("Clicking project dropdown button...")
|
|
self._click(dropdown_btn)
|
|
time.sleep(2)
|
|
|
|
# 3. Check the active project inside the popover/modal
|
|
try:
|
|
active_proj_div = self._wait(
|
|
By.CSS_SELECTOR,
|
|
"div[class*='ProjectModal__active_project_div']",
|
|
timeout=10,
|
|
condition="clickable"
|
|
)
|
|
except TimeoutException:
|
|
active_proj_div = None
|
|
|
|
if active_proj_div:
|
|
active_proj_text = active_proj_div.text
|
|
logger.info(f"Active project in modal/dropdown menu: {active_proj_text!r}")
|
|
|
|
if clean_target in clean_name(active_proj_text):
|
|
logger.info(f"Project '{target_project}' is already selected in the active div.")
|
|
# Close the popover by pressing escape
|
|
self.driver.find_element(By.TAG_NAME, "body").send_keys(Keys.ESCAPE)
|
|
time.sleep(1)
|
|
return
|
|
|
|
# Click active project div to reveal the project list
|
|
logger.info("Clicking active project div to open the list of all projects...")
|
|
self._click(active_proj_div)
|
|
time.sleep(2)
|
|
|
|
# 4. Search/filter using the search input
|
|
try:
|
|
search_input = self.driver.find_element(By.CSS_SELECTOR, "input[placeholder='Search Project']")
|
|
search_input.clear()
|
|
search_input.send_keys(target_project)
|
|
logger.info(f"Typed '{target_project}' into project search input.")
|
|
time.sleep(1.5)
|
|
except NoSuchElementException:
|
|
logger.info("Project list search input not found; scanning list directly.")
|
|
|
|
# Find the project item that matches the target project
|
|
project_items = self.driver.find_elements(
|
|
By.CSS_SELECTOR, "div[class*='ProjectModal__project_item']"
|
|
)
|
|
target_item = None
|
|
for item in project_items:
|
|
item_text = item.text
|
|
if clean_target in clean_name(item_text):
|
|
target_item = item
|
|
break
|
|
|
|
if not target_item:
|
|
self._save_screenshot("project_item_not_found.png")
|
|
raise RuntimeError(
|
|
f"Could not find project '{target_project}' in the project list. "
|
|
"Screenshot saved → tmp_download/project_item_not_found.png"
|
|
)
|
|
|
|
logger.info(f"Selecting project item: {target_item.text!r}")
|
|
self._click(target_item)
|
|
time.sleep(2)
|
|
|
|
# 5. Handle the project switch confirmation modal if it appears
|
|
try:
|
|
switch_modal = WebDriverWait(self.driver, 5).until(
|
|
EC.visibility_of_element_located((
|
|
By.XPATH,
|
|
"//div[contains(@class, 'ant-modal-content')][.//h4[contains(translate(text(), 'SWITCH', 'switch'), 'switch')]]"
|
|
))
|
|
)
|
|
logger.info("Switch confirmation modal appeared.")
|
|
switch_btn = switch_modal.find_element(
|
|
By.XPATH,
|
|
".//button[contains(@class, 'ant-btn-primary') or .//span[text()='Switch']]"
|
|
)
|
|
self._click(switch_btn)
|
|
logger.info("Clicked 'Switch' on confirmation modal.")
|
|
time.sleep(5) # Wait for page to reload/redirect
|
|
except TimeoutException:
|
|
logger.info("No switch confirmation modal appeared; switch may have been instant.")
|
|
|
|
# Dismiss any popups after switching
|
|
self._dismiss_popups()
|
|
|
|
# ── Public entry point ────────────────────────────────────────────────────
|
|
|
|
def run(
|
|
self,
|
|
email: str,
|
|
password: str,
|
|
target_project: str = "Inbound-Tech4Biz",
|
|
fields_to_select: list[str] | None = None,
|
|
) -> str:
|
|
"""
|
|
Full end-to-end automation run.
|
|
|
|
Returns:
|
|
Path to the freshly downloaded CSV file.
|
|
"""
|
|
self._cleanup_download_dir()
|
|
|
|
try:
|
|
self._setup_driver()
|
|
self.login(email, password)
|
|
self._switch_project(target_project)
|
|
self._navigate_to_accounts()
|
|
self._click_download_button()
|
|
self._handle_modal(fields_to_select)
|
|
return self._wait_for_download()
|
|
|
|
except Exception as exc:
|
|
self._save_screenshot("scraper_failure.png")
|
|
logger.error(f"Scraper failed: {exc}", exc_info=True)
|
|
raise
|
|
|
|
finally:
|
|
if self.driver:
|
|
self.driver.quit()
|
|
logger.info("Browser closed.")
|