226 lines
10 KiB
Python
226 lines
10 KiB
Python
import time
|
|
import openpyxl
|
|
import os
|
|
import csv
|
|
import pexpect
|
|
import sys
|
|
import re
|
|
from PIL import Image
|
|
import pytesseract
|
|
from selenium import webdriver
|
|
from selenium.webdriver.common.by import By
|
|
from selenium.webdriver.common.keys import Keys
|
|
from selenium.webdriver.chrome.service import Service
|
|
from selenium.webdriver.chrome.options import Options
|
|
from selenium.webdriver.support.ui import WebDriverWait
|
|
from selenium.webdriver.support import expected_conditions as EC
|
|
from selenium.common.exceptions import TimeoutException, NoSuchElementException, ElementClickInterceptedException
|
|
from webdriver_manager.chrome import ChromeDriverManager
|
|
|
|
# Function to log processing details
|
|
def log_processing(message, log_file):
|
|
timestamp = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
|
|
with open(log_file, "a") as log:
|
|
log.write(f"{timestamp} - {message}\n")
|
|
|
|
def create_ssh_tunnel(dcu_ip, ssh_user="callysta_icon", ssh_host="10.232.4.113", ssh_port=21112, ssh_password="Callysta_icon2024!", log_file="processing_log.txt"):
|
|
ssh_command = f"ssh -L 8888:{dcu_ip}:80 {ssh_user}@{ssh_host} -p {ssh_port}"
|
|
log_processing(f"Starting SSH tunnel to {dcu_ip}", log_file)
|
|
try:
|
|
child = pexpect.spawn(ssh_command)
|
|
child.expect(f"{ssh_user}@{ssh_host}'s password: ", timeout=60)
|
|
child.sendline(ssh_password)
|
|
child.expect(pexpect.EOF)
|
|
log_processing(f"SSH tunnel to {dcu_ip} established.", log_file)
|
|
return child
|
|
except pexpect.exceptions.EOF:
|
|
log_processing(f"Errir: SSH connection failed to {dcu_ip}.", log_file)
|
|
return None
|
|
except pexpect.exceptions.TIMEOUT:
|
|
log_processing(f"Warning: SSH connection timed out to {dcu_ip}. Reconnecting", log_file)
|
|
return None
|
|
|
|
def wait_for_page_to_load(driver, timeout=30, log_file="processing_log.txt"):
|
|
"""Wait for the page to completely load."""
|
|
log_processing("Waiting for the page to load", log_file)
|
|
try:
|
|
WebDriverWait(driver, timeout).until(
|
|
lambda d: d.execute_script("return document.readyState") == "complete"
|
|
)
|
|
log_processing("Page fully loaded.", log_file)
|
|
except TimeoutException:
|
|
log_processing("Timeout waiting for page load.", log_file)
|
|
raise Exception("Timeout waiting for page load.")
|
|
|
|
def click_running_state_tab(driver, log_file="processing_log.txt"):
|
|
"""Ensure the 'Running State' tab is clickable and click it."""
|
|
log_processing("Attempting to click 'Running State' tab", log_file)
|
|
try:
|
|
running_state_tab = WebDriverWait(driver, 10).until(
|
|
EC.presence_of_element_located((By.ID, "running-state"))
|
|
)
|
|
driver.execute_script("arguments[0].scrollIntoView(true);", running_state_tab)
|
|
WebDriverWait(driver, 10).until(EC.element_to_be_clickable((By.ID, "running-state")))
|
|
|
|
if "sun-navbar_item is-active" not in running_state_tab.get_attribute("class"):
|
|
running_state_tab.click()
|
|
log_processing("Clicked 'Running State' tab.", log_file)
|
|
else:
|
|
log_processing("'Running State' tab already active.", log_file)
|
|
except TimeoutException:
|
|
log_processing("Timeout waiting for 'Running State' tab.", log_file)
|
|
raise Exception("Timeout waiting for 'Running State' tab.")
|
|
except Exception as e:
|
|
log_processing(f"Error clicking 'Running State' tab: {e}", log_file)
|
|
raise
|
|
|
|
def extract_text_and_verify(dcu_ip, expected_ip, log_file="processing_log.txt"):
|
|
screenshot_path = f"/Users/macos/Documents/SINDIGILIVE/CLIENTS/ICON+/Project-HES/Update_DCU/evidence/{dcu_ip}.png"
|
|
status = "Failed"
|
|
dcu_id = "Not found"
|
|
|
|
log_processing(f"Starting text extraction for {dcu_ip}", log_file)
|
|
try:
|
|
image = Image.open(screenshot_path)
|
|
log_processing(f"Screenshot opened for {dcu_ip}", log_file)
|
|
|
|
extracted_text = pytesseract.image_to_string(image)
|
|
log_processing(f"Extracted text: {extracted_text}", log_file)
|
|
|
|
dcu_id_match = re.search(r'\b\d{14}\b', extracted_text)
|
|
if dcu_id_match:
|
|
dcu_id = dcu_id_match.group(0)
|
|
log_processing(f"DCU ID found: {dcu_id}", log_file)
|
|
else:
|
|
log_processing("DCU ID not found in text.", log_file)
|
|
|
|
if expected_ip in extracted_text and "Success" in extracted_text:
|
|
status = "Success"
|
|
log_processing(f"Expected IP and Success found for {dcu_ip}", log_file)
|
|
else:
|
|
log_processing(f"Expected IP or 'Success' not found in text for {dcu_ip}", log_file)
|
|
|
|
except FileNotFoundError:
|
|
log_processing(f"Screenshot not found for {dcu_ip} at {screenshot_path}", log_file)
|
|
status = "Screenshot not found"
|
|
|
|
return dcu_id, status
|
|
|
|
def verify_dcu_update(driver, expected_ip, dcu_ip, retries=5, log_file="processing_log.txt"):
|
|
save_directory = "/Users/macos/Documents/SINDIGILIVE/CLIENTS/ICON+/Project-HES/Update_DCU/evidence"
|
|
os.makedirs(save_directory, exist_ok=True)
|
|
screenshot_path = os.path.join(save_directory, f"{dcu_ip}.png")
|
|
|
|
log_processing(f"Starting verification for {dcu_ip}", log_file)
|
|
|
|
for attempt in range(retries):
|
|
try:
|
|
log_processing(f"Attempt {attempt + 1} for {dcu_ip}", log_file)
|
|
WebDriverWait(driver, 15).until(lambda d: d.execute_script("return document.readyState") == "complete")
|
|
|
|
iframe = driver.find_element(By.ID, 'home_content_iframe')
|
|
driver.switch_to.frame(iframe)
|
|
|
|
element = WebDriverWait(driver, 30).until(
|
|
EC.presence_of_element_located((By.XPATH, '/html/body/div[1]/div/div[2]/div[2]/div[1]/div/div[1]/div[2]'))
|
|
)
|
|
|
|
actual_text = element.text
|
|
log_processing(f"Text found for {dcu_ip}: {actual_text}", log_file)
|
|
|
|
if actual_text == expected_ip:
|
|
log_processing(f"Verification successful for {dcu_ip}", log_file)
|
|
driver.save_screenshot(screenshot_path)
|
|
return
|
|
else:
|
|
log_processing(f"Expected: {expected_ip}, found: {actual_text} for {dcu_ip}", log_file)
|
|
raise Exception("Verification failed: Text does not match expected IP.")
|
|
|
|
except Exception as e:
|
|
log_processing(f"Error during attempt {attempt + 1} for {dcu_ip}: {e}", log_file)
|
|
driver.switch_to.default_content()
|
|
driver.refresh()
|
|
time.sleep(5)
|
|
|
|
raise Exception(f"Verification failed for {dcu_ip} after {retries} attempts.")
|
|
|
|
def update_dcu_ip(updated_log, ip_list, processing_log):
|
|
log_processing("Starting DCU IP update process", processing_log)
|
|
for ip in ip_list:
|
|
log_processing(f"Processing DCU IP: {ip}", processing_log)
|
|
|
|
with open(updated_log, "a", newline="") as log:
|
|
writer = csv.writer(log)
|
|
if log.tell() == 0:
|
|
writer.writerow(["DCU IP", "DCU ID", "Updated IP", "Update Status"])
|
|
|
|
ssh_password = "Callysta_icon2024!"
|
|
ssh_process = create_ssh_tunnel(ip, ssh_password=ssh_password)
|
|
|
|
chrome_options = webdriver.ChromeOptions()
|
|
chrome_options.binary_location = "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome"
|
|
chrome_options.add_argument("--headless")
|
|
|
|
driver_path = "/Users/macos/Documents/SINDIGILIVE/CLIENTS/ICON+/Project-HES/Update_DCU/chromedriver/chromedriver"
|
|
driver = webdriver.Chrome(service=Service(driver_path), options=chrome_options)
|
|
|
|
try:
|
|
driver.get("http://localhost:8888")
|
|
WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "username")))
|
|
driver.find_element(By.ID, "username").send_keys("admin")
|
|
driver.find_element(By.ID, "password").send_keys("jh1296")
|
|
driver.find_element(By.ID, "submit_btn").click()
|
|
|
|
WebDriverWait(driver, 20).until(
|
|
EC.element_to_be_clickable((By.ID, "parameter-setting"))
|
|
).click()
|
|
|
|
iframes = driver.find_elements(By.TAG_NAME, "iframe")
|
|
if iframes:
|
|
driver.switch_to.frame(iframes[0])
|
|
|
|
new_ip = "10.232.107.250:9032"
|
|
WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "main_ip")))
|
|
ip_field = driver.find_element(By.ID, "main_ip")
|
|
dcu_id = "Not found"
|
|
|
|
for _ in range(3):
|
|
ip_field.clear()
|
|
current_value = ip_field.get_attribute("value").strip()
|
|
if current_value == "":
|
|
break
|
|
time.sleep(2)
|
|
else:
|
|
raise Exception("Failed to clear the IP field!")
|
|
|
|
ip_field.send_keys(new_ip)
|
|
driver.find_element(By.ID, "main_host_submit").click()
|
|
|
|
driver.back()
|
|
wait_for_page_to_load(driver, log_file=processing_log)
|
|
click_running_state_tab(driver, log_file=processing_log)
|
|
verify_dcu_update(driver, expected_ip=new_ip, dcu_ip=ip, log_file=processing_log)
|
|
|
|
# Extract text and verify, then update the log immediately after
|
|
dcu_id, status = extract_text_and_verify(ip, new_ip, log_file=processing_log)
|
|
writer.writerow([ip, dcu_id, new_ip, status]) # Write to log before terminating SSH
|
|
|
|
except TimeoutException as e:
|
|
log_processing(f"Timeout error for {ip}: {e}", processing_log)
|
|
writer.writerow([ip, "Not found", "Not found", "Failed - Timeout"])
|
|
except Exception as e:
|
|
log_processing(f"Error processing {ip}: {e}", processing_log)
|
|
writer.writerow([ip, "Not found", "Not found", "Failed - Error"])
|
|
finally:
|
|
if ssh_process and ssh_process.isalive():
|
|
log_processing(f"Terminating SSH tunnel for {ip}", processing_log)
|
|
ssh_process.terminate()
|
|
driver.quit()
|
|
|
|
if __name__ == "__main__":
|
|
updated_log = "/Users/macos/Documents/SINDIGILIVE/CLIENTS/ICON+/Project-HES/Update_DCU/update_log.csv"
|
|
processing_log = "/Users/macos/Documents/SINDIGILIVE/CLIENTS/ICON+/Project-HES/Update_DCU/processing_log.txt"
|
|
|
|
ip_list = sys.argv[1:]
|
|
update_dcu_ip(updated_log, ip_list, processing_log)
|