mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-27 23:35:01 +00:00
ae1da26558
The service was the last one living wholly under backend/. It now follows the same layering as abri and the other newer services: domain/pashub_fetcher/ core file classification, subfolders infrastructure/pashub_fetcher/ PasHub client, token getter, wire DTOs orchestration/ PashubFetcherOrchestrator (was PashubService) applications/pashub_fetcher/ lambda handler, trigger request, dev tooling core_files.py is split along the layer boundary: the domain module keeps the filename/evidence-category classification rules and no longer imports infrastructure.postgres, while the CoreFiles -> FileTypeEnum translation moves to infrastructure/pashub_fetcher/core_file_types.py. Tests move into the tests/ tree by layer. Note this puts them in the only suite CI currently runs (unit_tests.yml is disabled), so these 73 tests now execute on PRs for the first time; they were previously reachable only via the legacy pytest.ini testpaths. sharepoint_renamer's image now copies just domain/pashub_fetcher/ rather than the whole service, since SharepointSubfolders is all it needed. Behaviour is unchanged. tests/ goes 9927 -> 10000 passed (+73, exactly the tests that moved in); the legacy suite keeps its same 17 pre-existing failures and 11 errors. tests/test_lambda_packaging.py confirms both changed Dockerfiles still copy their handler's full import closure. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
103 lines
3.1 KiB
Python
103 lines
3.1 KiB
Python
import glob
|
|
import os
|
|
import shutil
|
|
|
|
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, record_video: bool = False
|
|
) -> str:
|
|
logger.info("Starting Playwright flow")
|
|
|
|
with sync_playwright() as p:
|
|
logger.info("Playwright server started")
|
|
|
|
browser = p.chromium.launch(
|
|
headless=True,
|
|
args=[
|
|
"--no-sandbox",
|
|
"--disable-dev-shm-usage",
|
|
"--disable-gpu",
|
|
"--single-process",
|
|
"--no-zygote",
|
|
],
|
|
)
|
|
logger.info("Chromium launched successfully")
|
|
|
|
video_dir = None
|
|
if record_video:
|
|
video_dir = os.path.join(os.path.dirname(__file__), "videos")
|
|
os.makedirs(video_dir, exist_ok=True)
|
|
|
|
if record_video:
|
|
context = browser.new_context(
|
|
record_video_dir=video_dir,
|
|
record_video_size={"width": 1280, "height": 720},
|
|
)
|
|
else:
|
|
context = browser.new_context()
|
|
page = context.new_page()
|
|
logger.info("Page created")
|
|
|
|
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:
|
|
context.close()
|
|
browser.close()
|
|
|
|
for pattern in (
|
|
"/tmp/playwright-artifacts-*",
|
|
"/tmp/playwright_chromiumdev_profile-*",
|
|
):
|
|
for path in glob.glob(pattern):
|
|
shutil.rmtree(path, ignore_errors=True)
|
|
|
|
if record_video and video_dir:
|
|
logger.info(f"Video(s) saved in: {video_dir}")
|