Model/backend/pashub_fetcher/token_getter.py

78 lines
2.4 KiB
Python

import os
from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeoutError
from utils.logger import setup_logger
logger = setup_logger()
def get_token_from_local_storage(email: str, password: str) -> str:
logger.info("Starting Playwright flow")
# For local testing / debugging, we save videos of the flow
video_dir = os.path.join(os.path.dirname(__file__), "videos")
os.makedirs(video_dir, exist_ok=True)
with sync_playwright() as p:
browser = p.chromium.launch(
headless=True,
args=["--no-sandbox", "--disable-dev-shm-usage"],
)
context = browser.new_context(
record_video_dir=video_dir,
record_video_size={"width": 1280, "height": 720},
)
page = context.new_page()
try:
logger.info("Navigating to site...")
page.goto("https://pashub.net/", timeout=30000)
logger.info("Filling login form...")
email_input = page.locator("#email")
email_input.wait_for(state="visible", timeout=10000)
email_input.fill(email)
password_input = page.locator("#password")
password_input.wait_for(state="visible", timeout=10000)
password_input.focus()
password_input.fill(password)
logger.info("Submitting login...")
page.wait_for_selector("#btn-login", state="visible", timeout=10000)
with page.expect_navigation(timeout=15000):
page.click("#btn-login")
page.wait_for_timeout(3000)
if "login" in page.url.lower():
raise Exception("Login failed (still on login page)")
logger.info(f"Login likely successful. URL: {page.url}")
token = page.evaluate(
"""() => {
return localStorage.getItem('token');
}"""
)
if not token:
raise Exception("Login succeeded but no token found")
return token
except PlaywrightTimeoutError as e:
raise Exception(f"Timeout during login flow: {str(e)}")
except Exception as e:
raise Exception(f"Unexpected error: {str(e)}")
finally:
logger.info("Closing browser context (saving video)...")
context.close()
browser.close()
logger.info(f"Video(s) saved in: {video_dir}")