Gitead/DentalManagementE
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ddma eligibiltiy - v5 done

Browse Source
This commit is contained in:
Potenz
2025-12-12 00:30:35 +05:30
parent ef02d6454e
commit a4877e4ced
2 changed files with 0 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
apps/SeleniumService/helpers_ddma_eligibility.py
View File

@@ -6,7 +6,6 @@ from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.support import expected_conditions as EC
from selenium.common.exceptions import WebDriverException from selenium.common.exceptions import WebDriverException
import pickle
from selenium_DDMA_eligibilityCheckWorker import AutomationDeltaDentalMAEligibilityCheck from selenium_DDMA_eligibilityCheckWorker import AutomationDeltaDentalMAEligibilityCheck
@@ -175,37 +174,6 @@ async def start_ddma_run(sid: str, data: dict, url: str):
s["last_activity"] = time.time() s["last_activity"] = time.time()
await asyncio.sleep(0.5) await asyncio.sleep(0.5)
# Wait for post-OTP login to complete and then save cookies
try:
driver = s["driver"]
wait = WebDriverWait(driver, 30)
# Wait for dashboard element or URL change indicating success
logged_in_el = wait.until(
EC.presence_of_element_located(
(By.XPATH, "//a[text()='Member Eligibility' or contains(., 'Member Eligibility')]")
)
)
# If found, save cookies
if logged_in_el:
try:
# Prefer direct save to avoid subtle create_if_missing behavior
cookies = driver.get_cookies()
pickle.dump(cookies, open(bot.cookies_path, "wb"))
print(f"[start_ddma_run] Saved {len(cookies)} cookies after OTP to {bot.cookies_path}")
except Exception as e:
print("[start_ddma_run] Warning saving cookies after OTP:", e)
except Exception as e:
# If waiting times out, still attempt a heuristic check by URL
cur = s["driver"].current_url if s.get("driver") else ""
print("[start_ddma_run] Post-OTP dashboard detection timed out. Current URL:", cur)
if "dashboard" in cur or "providers" in cur:
try:
cookies = s["driver"].get_cookies()
pickle.dump(cookies, open(bot.cookies_path, "wb"))
print(f"[start_ddma_run] Saved {len(cookies)} cookies after OTP (URL heuristic).")
except Exception as e2:
print("[start_ddma_run] Warning saving cookies after OTP (heuristic):", e2)
except Exception as e: except Exception as e:
s["status"] = "error" s["status"] = "error"
s["message"] = f"Failed to submit OTP into page: {e}" s["message"] = f"Failed to submit OTP into page: {e}"

168
apps/SeleniumService/selenium_DDMA_eligibilityCheckWorker.py
View File

@@ -9,10 +9,6 @@ from webdriver_manager.chrome import ChromeDriverManager
import time import time
import os import os
import base64 import base64
import stat
import pickle
import traceback
import urllib
class AutomationDeltaDentalMAEligibilityCheck: class AutomationDeltaDentalMAEligibilityCheck:
def __init__(self, data): def __init__(self, data):
@@ -31,9 +27,6 @@ class AutomationDeltaDentalMAEligibilityCheck:
self.download_dir = os.path.abspath("seleniumDownloads") self.download_dir = os.path.abspath("seleniumDownloads")
os.makedirs(self.download_dir, exist_ok=True) os.makedirs(self.download_dir, exist_ok=True)
self.cookies_path = os.path.abspath("cookies.pkl")
def config_driver(self): def config_driver(self):
options = webdriver.ChromeOptions() options = webdriver.ChromeOptions()
if self.headless: if self.headless:
@@ -52,145 +45,10 @@ class AutomationDeltaDentalMAEligibilityCheck:
driver = webdriver.Chrome(service=s, options=options) driver = webdriver.Chrome(service=s, options=options)
self.driver = driver self.driver = driver
def _origin_from_url(self, url):
p = urllib.parse.urlparse(url)
origin = f"{p.scheme}://{p.netloc}"
return origin
def try_cookie_login(self, url):
try:
if not os.path.exists(self.cookies_path):
print(f"[try_cookie_login] cookies file not found at: {self.cookies_path}")
return "NO_COOKIES"
origin = self._origin_from_url(url)
print(f"origin {origin}")
print(f"[try_cookie_login] Attempting cookie-login. origin={origin}, cookies_path={self.cookies_path}")
# load origin first
self.driver.get(origin)
time.sleep(1)
try:
cookies = pickle.load(open(self.cookies_path, "rb"))
print(f"[try_cookie_login] Loaded {len(cookies)} cookies from file.")
except Exception as e:
print("[try_cookie_login] Failed to load cookies.pkl:", e)
traceback.print_exc()
return "FAILED"
added = 0
for c in cookies:
try:
self.driver.add_cookie(c)
added += 1
except Exception as ex:
print(f"[try_cookie_login] warning adding cookie {c.get('name')}: {ex}")
print(f"[try_cookie_login] Added {added}/{len(cookies)} cookies to browser.")
# now load target url
time.sleep(2)
self.driver.get(origin)
time.sleep(2)
# detect logged-in state
try:
WebDriverWait(self.driver, 8).until(
EC.presence_of_element_located((By.XPATH, "//a[text()='Member Eligibility' or contains(., 'Member Eligibility')]"))
)
# refresh cookie file with any updates
try:
current = self.driver.get_cookies()
pickle.dump(current, open(self.cookies_path, "wb"))
print("[try_cookie_login] Cookie-login success; cookies.pkl refreshed.")
except Exception as e:
print("[try_cookie_login] Warning saving refreshed cookies:", e)
return "Success"
except TimeoutException:
print("[try_cookie_login] Cookie-login did not find dashboard element.")
return "FAILED"
except Exception as e:
print("[try_cookie_login] Unexpected exception:", e)
traceback.print_exc()
return "FAILED"
def handle_cookies(self, url, create_if_missing=True):
try:
origin = self._origin_from_url(url)
self.driver.get(origin)
time.sleep(1)
if not os.path.exists(self.cookies_path):
if not create_if_missing:
print("[handle_cookies] cookies file missing and create_if_missing=False")
return False
try:
input("add ?")
cookies = self.driver.get_cookies()
pickle.dump(cookies, open(self.cookies_path, "wb"))
print(f"[handle_cookies] Saved {len(cookies)} cookies to {self.cookies_path}")
return True
except Exception as e:
print("[handle_cookies] failed to save cookies:", e)
traceback.print_exc()
return False
else:
try:
cookies = pickle.load(open(self.cookies_path, "rb"))
print(f"[handle_cookies] Loaded {len(cookies)} cookies from {self.cookies_path}")
except Exception as e:
print("[handle_cookies] failed to load cookies.pkl:", e)
traceback.print_exc()
return False
# ensure on origin to add cookies
self.driver.get(origin)
time.sleep(1)
added = 0
for c in cookies:
try:
self.driver.add_cookie(c)
added += 1
except Exception:
pass
print(f"[handle_cookies] Re-applied {added}/{len(cookies)} cookies.")
time.sleep(1)
self.driver.refresh()
time.sleep(2)
try:
new_cookies = self.driver.get_cookies()
pickle.dump(new_cookies, open(self.cookies_path, "wb"))
print(f"[handle_cookies] Refreshed cookies.pkl with {len(new_cookies)} cookies.")
except Exception as e:
print("[handle_cookies] failed to refresh cookies.pkl:", e)
traceback.print_exc()
return True
except Exception as e:
print("[handle_cookies] Unexpected exception:", e)
traceback.print_exc()
return False
def login(self, url): def login(self, url):
wait = WebDriverWait(self.driver, 30) wait = WebDriverWait(self.driver, 30)
try: try:
print(f"[login] Starting login for url: {url}")
cookie_attempt = self.try_cookie_login(url)
print(f"[login] cookie_attempt result: {cookie_attempt}")
if cookie_attempt == "Success":
print("[login] Logged in via cookies.")
return "Success"
elif cookie_attempt == "NO_COOKIES":
print("[login] No cookies present; will do credential login.")
else:
print("[login] Cookie-login failed; falling back to username/password login.")
# credential login flow
self.driver.get(url) self.driver.get(url)
print("[login] On login page, attempting to fill credentials.")
email_field = wait.until(EC.presence_of_element_located((By.XPATH, "//input[@name='username' and @type='text']"))) email_field = wait.until(EC.presence_of_element_located((By.XPATH, "//input[@name='username' and @type='text']")))
email_field.clear() email_field.clear()
email_field.send_keys(self.massddma_username) email_field.send_keys(self.massddma_username)
@@ -223,32 +81,6 @@ class AutomationDeltaDentalMAEligibilityCheck:
return "OTP_REQUIRED" return "OTP_REQUIRED"
except TimeoutException: except TimeoutException:
print("[login] No OTP input detected in allowed time.") print("[login] No OTP input detected in allowed time.")
# check for dashboard element
try:
logged_in_el = WebDriverWait(self.driver, 8).until(
EC.presence_of_element_located((By.XPATH, "//a[text()='Member Eligibility' or contains(., 'Member Eligibility')]"))
)
if logged_in_el:
print("[login] Credential login succeeded; calling handle_cookies to save/refresh cookies.")
try:
self.handle_cookies(self.driver.current_url, create_if_missing=True)
except Exception as e:
print("[login] handle_cookies raised:", e)
traceback.print_exc()
return "Success"
except TimeoutException:
cur = self.driver.current_url or ""
print(f"[login] Post-login element not found. Current URL: {cur}")
if "dashboard" in cur or "providers" in cur:
print("[login] URL looks like dashboard; treating as success and saving cookies.")
try:
self.handle_cookies(self.driver.current_url, create_if_missing=True)
except Exception:
pass
return "Success"
return "ERROR:LOGIN FAILED - unable to detect success or OTP"
except Exception as e: except Exception as e:
print("[login] Exception during login:", e) print("[login] Exception during login:", e)
return f"ERROR:LOGIN FAILED: {e}" return f"ERROR:LOGIN FAILED: {e}"