selenium fetchpdf checkpoint

2025-06-09 19:59:58 +05:30
parent d3cf01b65b
commit f585111b23
8 changed files with 309 additions and 66 deletions
--- a/apps/SeleniumService/agent.py
+++ b/apps/SeleniumService/agent.py
@@ -1,7 +1,7 @@
 from fastapi import FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
 import uvicorn
-from selenium_worker import AutomationMassDHP
+from selenium_worker import AutomationMassHealth

 app = FastAPI()

@@ -12,15 +12,38 @@ app.add_middleware(
    allow_headers=["*"],
 )

-@app.post("/run")
-async def run_bot(request: Request):
+
+# Endpoint: Step 1 — Start the automation
+@app.post("/start-workflow")
+async def start_workflow(request: Request):
    data = await request.json()
    try:
-        bot = AutomationMassDHP(data)
-        result = bot.main_workflow("https://providers.massdhp.com/providers_login.asp")
+        bot = AutomationMassHealth(data)
+        result = bot.main_workflow_upto_step2("https://providers.massdhp.com/providers_login.asp")
        return result
    except Exception as e:
        return {"status": "error", "message": str(e)}

+# Endpoint: Step 2 — Extract the PDF content after manual submission
+@app.post("/fetch-pdf")
+async def fetch_pdf():
+    try:
+        bot = AutomationMassHealth().get_last_instance()
+        if not bot:
+            return {"status": "error", "message": "No running automation session"}
+        
+        pdf_data = bot.reach_to_pdf()
+        if pdf_data.get("status") != "success":
+            return {"status": "error", "message": pdf_data.get("message")}
+
+        return {
+            "status": "success",
+            "pdf_url": pdf_data["pdf_url"],
+            "pdf_base64": pdf_data["pdf_bytes"]
+        }
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+    
+
 if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=5002)
--- a/apps/SeleniumService/agent_test_1.py
+++ b/apps/SeleniumService/agent_test_1.py
@@ -0,0 +1,26 @@
+from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
+import uvicorn
+from selenium_worker import AutomationMassDHP
+
+app = FastAPI()
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # Replace with your frontend domain for security
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+@app.post("/run")
+async def run_bot(request: Request):
+    data = await request.json()
+    try:
+        bot = AutomationMassDHP(data)
+        result = bot.main_workflow("https://providers.massdhp.com/providers_login.asp")
+        return result
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=5002)
--- a/apps/SeleniumService/agent_test_2.py
+++ b/apps/SeleniumService/agent_test_2.py
@@ -0,0 +1,42 @@
+from fastapi import FastAPI, Request
+from fastapi.middleware.cors import CORSMiddleware
+import uvicorn
+from selenium_worker import AutomationMassHealth
+
+app = FastAPI()
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],  # Replace with your frontend domain for security
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+# Endpoint: Step 1 — Start the automation
+@app.post("/start-workflow")
+async def start_workflow(request: Request):
+    data = await request.json()
+    try:
+        bot = AutomationMassHealth(data)
+        result = bot.main_workflow_upto_step2("https://abc.com/providers_login.asp")
+        return result
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+
+# Endpoint: Step 2 — Extract the PDF content after manual submission
+@app.post("/fetch-pdf")
+async def fetch_pdf():
+    try:
+        bot = AutomationMassHealth().get_last_instance()
+        pdf_data = bot.reach_to_pdf()
+
+        if not pdf_data:
+            return {"status": "error", "message": "Failed to fetch PDF"}
+        return {"status": "success", "pdf_data": pdf_data}
+    except Exception as e:
+        return {"status": "error", "message": str(e)}
+    
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=5002)
--- a/apps/SeleniumService/selenium_worker.py
+++ b/apps/SeleniumService/selenium_worker.py
@@ -11,12 +11,16 @@ from datetime import datetime
 import tempfile
 import base64
 import os
+import requests
+import json

+class AutomationMassHealth:    
+    last_instance = None  

-class AutomationMassDHP:      
    def __init__(self, data):
        self.headless = False
        self.driver = None
+        AutomationMassHealth.last_instance = self

        self.data = data
        self.claim = data.get("claim", {})
@@ -31,6 +35,10 @@ class AutomationMassDHP:
        self.serviceLines = self.claim.get("serviceLines", [])
        self.missingTeethStatus = self.claim.get("missingTeethStatus", "")
        self.missingTeeth = self.claim.get("missingTeeth", {})
+    
+    @staticmethod
+    def get_last_instance():
+        return AutomationMassHealth.last_instance

    def config_driver(self):
        options = webdriver.ChromeOptions()
@@ -279,35 +287,98 @@ class AutomationMassDHP:
            return "ERROR:REMARKS FAILED"

        return "Success"
+    
+
+    def reach_to_pdf(self):
+        wait = WebDriverWait(self.driver, 30)
+
+        try:
+            print("Waiting for PDF link to appear on success page...")
+            pdf_link_element = wait.until(
+                EC.element_to_be_clickable((By.XPATH, "//a[contains(@href, '.pdf')]"))
+            )
+            print("PDF link found. Clicking it...")
+
+            # Click the PDF link
+            pdf_link_element.click()
+            time.sleep(5)
+
+            existing_windows = self.driver.window_handles
+
+            # Wait for the new tab
+            WebDriverWait(self.driver, 90).until(
+                lambda d: len(d.window_handles) > len(existing_windows)
+            )
+
+            print("Switching to PDF tab...")
+            self.driver.switch_to.window(self.driver.window_handles[1])


-    def main_workflow(self, url):
+            time.sleep(2)
+            current_url = self.driver.current_url
+            print(f"Switched to PDF tab. Current URL: {current_url}")
+
+
+             # Get full PDF URL in case it's a relative path
+            pdf_url = pdf_link_element.get_attribute("href")
+            if not pdf_url.startswith("http"):
+                base_url = self.driver.current_url.split("/providers")[0]
+                pdf_url = f"{base_url}/{pdf_url}"
+
+            # Get cookies from Selenium session, saving just for my referece while testing. in prod just use below one line
+            # cookies = {c['name']: c['value'] for c in self.driver.get_cookies()} 
+            # 1. Get raw Selenium cookies (list of dicts)
+            raw_cookies = self.driver.get_cookies()
+            with open("raw_cookies.txt", "w") as f:
+                json.dump(raw_cookies, f, indent=2)
+
+            formatted_cookies = {c['name']: c['value'] for c in raw_cookies}
+            with open("formatted_cookies.txt", "w") as f:
+                for k, v in formatted_cookies.items():
+                    f.write(f"{k}={v}\n")
+
+            # Use requests to download the file using session cookies
+            print("Downloading PDF content via requests...")
+            pdf_response = requests.get(pdf_url, cookies=formatted_cookies)
+
+            if pdf_response.status_code == 200:
+                print("PDF successfully fetched (bytes length):")
+                return {
+                "status": "success",
+                "pdf_bytes": base64.b64encode(pdf_response.content).decode(),
+            }
+            else:
+                print("Failed to fetch PDF. Status:", pdf_response.status_code, pdf_response)
+                return {
+                "status": "error",
+                "message": pdf_response,
+            }
+
+        except Exception as e:
+            print(f"ERROR: {str(e)}")
+            return {
+                "status": "error",
+                "message": str(e),
+            }
+        
+    def main_workflow_upto_step2(self, url):
        self.config_driver()
        print("Reaching Site :", url)
        self.driver.maximize_window()
        self.driver.get(url)
        time.sleep(3)
-        value = self.login()
-        if value.startswith("ERROR"):
-            # self.driver.close()
-            return value
-        

-        time.sleep(5)
-        value2 = self.step1()
-        if value2.startswith("ERROR"):
-            # self.driver.close()
-            return value2
+        if self.login().startswith("ERROR"):
+            return {"status": "error", "message": "Login failed"}

-        time.sleep(5)
-        value3 = self.step2()
-        if value3.startswith("ERROR"):
-            # self.driver.close()
-            return value3
+        if self.step1().startswith("ERROR"):
+            return {"status": "error", "message": "Step1 failed"}

+        if self.step2().startswith("ERROR"):
+            return {"status": "error", "message": "Step2 failed"}

-        input("should Close?") # here it sholud get confirmation from the frontend, 
-        
-        self.driver.close()
-
-        return {"status": "success", "message": "Successfully submitted the form."}
+        print("Waiting for user to manually submit form in browser...")
+        return {
+            "status": "waiting_for_user",
+            "message": "Automation paused. Please submit the form manually in browser."
+        }