Model/backend/ecmk_fetcher/handler/handler.py
2026-04-01 08:18:06 +00:00

88 lines
2.6 KiB
Python

import os
from enum import Enum
from typing import Any, Mapping
from playwright.sync_api import sync_playwright, TimeoutError as PlaywrightTimeoutError
class file_download_button_types(Enum):
ASSESSOR_HUB_SITENOTE_REPORT = 11
CERTIFICATE = 9
SITENOTE_REPORT = 8
RAW_XML = 7
SAP_WORK_SHEET = 15
def download_report():
username = ""
password = ""
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
context = browser.new_context()
page = context.new_page()
try:
# 1. Go to site
page.goto("https://assessorhub.net/", timeout=30000)
# 2. Login (UPDATE selectors if needed)
username_input = page.locator("#Username")
password_input = page.locator("#Password")
username_input.wait_for(state="visible", timeout=10000)
username_input.fill(username)
password_input.wait_for(state="visible", timeout=10000)
password_input.fill(password)
# 3. Submit login
with page.expect_navigation(timeout=15000):
page.click("button[type='submit']")
# 4. Verify login succeeded
if "login" in page.url.lower():
raise Exception("Login failed")
print("Login successful:", page.url)
# 5. Navigate to the assessment detail page
page.goto(
"https://assessorhub.net/Assessments/Assessments/Detail/1bd9fd74-08f6-4fc1-b2f7-3a13a8f9084d?returnUrl=/Companies/Assessments",
timeout=30000,
)
# 6. Locate the correct download button
button = page.locator("a.download-report-btn[data-report-type='11']")
button.wait_for(state="visible", timeout=10000)
# 7. Click and capture the download
with page.expect_download(timeout=30000) as download_info:
button.click()
download = download_info.value
# 8. Save file locally
filename = download.suggested_filename
save_path = os.path.join(os.getcwd(), filename)
download.save_as(save_path)
print(f"Downloaded file saved to: {save_path}")
except PlaywrightTimeoutError as e:
raise Exception(f"Timeout occurred: {str(e)}")
finally:
context.close()
browser.close()
def handler(event: Mapping[str, Any], context: Any) -> None:
download_report()
if __name__ == "__main__":
event = {"Records": [{"body": "{}"}]}
handler(event, None)