280 lines
12 KiB
Python
280 lines
12 KiB
Python
import time
|
||
import openpyxl
|
||
import os
|
||
import csv
|
||
import pexpect
|
||
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.webdriver.chrome.service import Service
|
||
from selenium.webdriver.chrome.options import Options
|
||
from selenium.common.exceptions import TimeoutException, NoSuchElementException, ElementClickInterceptedException
|
||
from selenium.webdriver.common.action_chains import ActionChains
|
||
from webdriver_manager.chrome import ChromeDriverManager
|
||
|
||
def create_ssh_tunnel(dcu_ip, ssh_user="callysta_icon", ssh_host="10.232.4.113", ssh_port=21112, ssh_password="Callysta_icon2024!"):
|
||
ssh_command = f"ssh -L 8888:{dcu_ip}:80 {ssh_user}@{ssh_host} -p {ssh_port}"
|
||
child = pexpect.spawn(ssh_command)
|
||
try:
|
||
# Expect the password prompt and send the password
|
||
child.expect("callysta_icon@10.232.4.113's password: ", timeout=60)
|
||
child.sendline(ssh_password)
|
||
child.expect(pexpect.EOF)
|
||
print("SSH tunnel established.")
|
||
except pexpect.exceptions.EOF:
|
||
print("Error: SSH connection failed.")
|
||
except pexpect.exceptions.TIMEOUT:
|
||
print("Error: SSH connection timed out.")
|
||
|
||
if child.isalive():
|
||
print("SSH tunnel is active.")
|
||
else:
|
||
print("Warning: SSH tunnel may not be active.")
|
||
|
||
return child # Return the child process to keep the tunnel open
|
||
|
||
def wait_for_page_to_load(driver, timeout=30):
|
||
"""Wait for the page to completely load."""
|
||
try:
|
||
WebDriverWait(driver, timeout).until(
|
||
lambda d: d.execute_script("return document.readyState") == "complete"
|
||
)
|
||
print("Halaman berhasil dimuat sepenuhnya.")
|
||
except TimeoutException:
|
||
raise Exception("Timeout menunggu halaman selesai dimuat.")
|
||
|
||
def click_running_state_tab(driver):
|
||
"""Ensure the 'Running State' tab is clickable and click it."""
|
||
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"))
|
||
)
|
||
|
||
# Periksa apakah tab sudah aktif
|
||
if "sun-navbar_item is-active" not in running_state_tab.get_attribute("class"):
|
||
running_state_tab.click()
|
||
print("Tab 'Running State' diklik.")
|
||
else:
|
||
print("Tab 'Running State' sudah aktif.")
|
||
except TimeoutException:
|
||
print("Timeout menunggu tab 'Running State' untuk dapat diklik.")
|
||
raise Exception("Tab 'Running State' tidak dapat diklik.")
|
||
except Exception as e:
|
||
print(f"Error clicking the 'Running State' tab: {e}")
|
||
raise
|
||
|
||
def extract_text_and_verify(dcu_ip, expected_ip):
|
||
# Path to the directory and file name based on expected IP
|
||
screenshot_path = f"/Users/macos/Documents/SINDIGILIVE/CLIENTS/ICON+/Project-HES/Update_DCU/evidence/{dcu_ip}.png"
|
||
csv_path = "/Users/macos/Documents/SINDIGILIVE/CLIENTS/ICON+/Project-HES/Update_DCU/extract_log.csv"
|
||
status = "Failed"
|
||
dcu_id = "Not found"
|
||
|
||
try:
|
||
# Load the image
|
||
image = Image.open(screenshot_path)
|
||
print(f"Tangkapan layar dibuka: {screenshot_path}")
|
||
|
||
# Extract text from the image using Tesseract
|
||
extracted_text = pytesseract.image_to_string(image)
|
||
print("Extracted Text:")
|
||
print(extracted_text)
|
||
|
||
# Search for the DCU ID using regex for a 14-digit sequence
|
||
dcu_id_match = re.search(r'\b\d{14}\b', extracted_text)
|
||
if dcu_id_match:
|
||
dcu_id = dcu_id_match.group(0)
|
||
print(f"DCU ID found: {dcu_id}")
|
||
else:
|
||
print("DCU ID not found in extracted text.")
|
||
|
||
# Verify if expected IP and "Success" are in the extracted text
|
||
if expected_ip in extracted_text and "Success" in extracted_text:
|
||
status = "Success"
|
||
print(f"Found: {expected_ip} and Success")
|
||
else:
|
||
print(f"{expected_ip} atau status 'Success' tidak ditemukan dalam teks yang diekstrak.")
|
||
|
||
except FileNotFoundError:
|
||
print(f"Tangkapan layar tidak ditemukan di {screenshot_path}")
|
||
status = "Screenshot not found"
|
||
|
||
return dcu_id, status
|
||
|
||
def verify_dcu_update(driver, expected_ip, dcu_ip, retries=3):
|
||
"""Verify if the DCU update is successful by checking the IP and connection state."""
|
||
save_directory = "/Users/macos/Documents/SINDIGILIVE/CLIENTS/ICON+/Project-HES/Update_DCU/evidence"
|
||
os.makedirs(save_directory, exist_ok=True) # Create the folder if it doesn’t exist
|
||
screenshot_path = os.path.join(save_directory, f"{dcu_ip}.png")
|
||
|
||
# Full XPath dari elemen target
|
||
full_xpath = '/html/body/div[1]/div/div[2]/div[2]/div[1]/div/div[1]/div[2]'
|
||
|
||
for attempt in range(retries):
|
||
try:
|
||
print(f"Attempt {attempt + 1} to verify DCU update...")
|
||
|
||
# Wait until the page is fully loaded
|
||
WebDriverWait(driver, 15).until(lambda d: d.execute_script("return document.readyState") == "complete")
|
||
print("Page fully loaded.")
|
||
|
||
# Switch to iframe
|
||
iframe = driver.find_element(By.ID, 'home_content_iframe')
|
||
driver.switch_to.frame(iframe)
|
||
|
||
# Wait for the presence of the element
|
||
element = WebDriverWait(driver, 30).until(
|
||
EC.presence_of_element_located((By.XPATH, full_xpath))
|
||
)
|
||
print("Target element is present.")
|
||
|
||
# Get the text from the element
|
||
actual_text = element.text
|
||
print(f"Found text: {actual_text}")
|
||
|
||
# Check if the text matches the desired IP
|
||
if actual_text == expected_ip:
|
||
print(f"DCU update verification successful: {actual_text}")
|
||
|
||
# Take a screenshot
|
||
screenshot_taken = driver.save_screenshot(screenshot_path)
|
||
if screenshot_taken:
|
||
print(f"Screenshot saved at: {screenshot_path}")
|
||
else:
|
||
print("Failed to take screenshot. Check WebDriver setup.")
|
||
|
||
# Exit the loop if successful
|
||
return
|
||
|
||
else:
|
||
print(f"Text does not match expected value. Expected: {expected_ip}, Found: {actual_text}")
|
||
raise Exception("Verification failed due to mismatched text.")
|
||
|
||
except Exception as e:
|
||
print(f"Error during attempt {attempt + 1}: {e}")
|
||
|
||
# Refresh the page on error
|
||
print("Page or data not found, attempting to refresh the page...")
|
||
driver.switch_to.default_content() # Return to the main context before refreshing
|
||
driver.refresh()
|
||
time.sleep(5) # Wait a bit before retrying
|
||
|
||
# If all attempts fail, raise an exception
|
||
raise Exception(f"Failed to verify DCU update after {retries} attempts.")
|
||
|
||
def update_dcu_ip(excel_file, log_file):
|
||
# Load the Excel file
|
||
wb = openpyxl.load_workbook(excel_file)
|
||
sheet = wb.active
|
||
|
||
# Open log file for writing header only
|
||
with open(log_file, "w", newline="") as log:
|
||
writer = csv.writer(log)
|
||
writer.writerow(["DCU IP", "DCU ID", "Updated IP", "Update Status"]) # Write CSV header
|
||
|
||
for row in range(2, sheet.max_row + 1):
|
||
# Get the DCU IP from Excel and handle None values
|
||
cell_value = sheet.cell(row=row, column=5).value
|
||
if cell_value is None:
|
||
print(f"Row {row}: DCU IP is empty. Skipping...")
|
||
writer.writerow(["Not found", "Not found", "Not found", "Failed - Empty IP"])
|
||
continue # Skip this row
|
||
|
||
dcu_ip = str(cell_value).strip('="')
|
||
print(f"Processing DCU IP: {dcu_ip}")
|
||
|
||
# Set up SSH tunnel for each DCU IP
|
||
ssh_password = "Callysta_icon2024!"
|
||
ssh_process = create_ssh_tunnel(dcu_ip, ssh_password=ssh_password)
|
||
|
||
# Set up Chrome WebDriver for each DCU IP
|
||
chrome_options = Options()
|
||
chrome_options.binary_location = "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome"
|
||
chrome_options.add_argument("--headless")
|
||
chrome_options.add_argument("--disable-gpu")
|
||
chrome_options.add_argument("--no-sandbox")
|
||
chrome_options.add_argument("--disable-dev-shm-usage")
|
||
|
||
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:
|
||
# Access the DCU web application and login
|
||
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()
|
||
|
||
# Wait for "parameter-setting" element and click
|
||
WebDriverWait(driver, 20).until(
|
||
EC.element_to_be_clickable((By.ID, "parameter-setting"))
|
||
).click()
|
||
|
||
# Handle iframe if needed
|
||
iframes = driver.find_elements(By.TAG_NAME, "iframe")
|
||
if iframes:
|
||
driver.switch_to.frame(iframes[0])
|
||
|
||
# Update the MDC IP
|
||
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"
|
||
|
||
# Clear the field and set the new IP
|
||
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!")
|
||
|
||
# Enter the new IP and submit
|
||
ip_field.send_keys(new_ip)
|
||
time.sleep(2)
|
||
driver.find_element(By.ID, "main_host_submit").click()
|
||
time.sleep(2)
|
||
|
||
# Verification step
|
||
driver.back()
|
||
wait_for_page_to_load(driver)
|
||
click_running_state_tab(driver)
|
||
verify_dcu_update(driver, expected_ip=new_ip, dcu_ip=dcu_ip)
|
||
|
||
# Log success
|
||
dcu_id, status = extract_text_and_verify(dcu_ip, new_ip)
|
||
writer.writerow([dcu_ip, dcu_id, new_ip, status])
|
||
|
||
except TimeoutException as e:
|
||
print(f"Timeout error for DCU IP {dcu_ip}: {e}")
|
||
writer.writerow([dcu_ip, "Not found", "Not found", "Failed - Timeout"])
|
||
except Exception as e:
|
||
print(f"Error processing DCU IP {dcu_ip}: {e}")
|
||
writer.writerow([dcu_ip, "Not found", "Not found", "Failed - Error"])
|
||
finally:
|
||
# Ensure each SSH session and driver instance is closed after processing
|
||
if ssh_process.isalive():
|
||
ssh_process.terminate()
|
||
driver.quit() # Close the WebDriver instance
|
||
|
||
# Main function
|
||
if __name__ == "__main__":
|
||
excel_file = "/Users/macos/Documents/SINDIGILIVE/CLIENTS/ICON+/Project-HES/Update_DCU/DCU-LIST.xlsx"
|
||
log_file = "/Users/macos/Documents/SINDIGILIVE/CLIENTS/ICON+/Project-HES/Update_DCU/update_log.csv"
|
||
update_dcu_ip(excel_file, log_file)
|