selenium_elm_fengshen/run_window_v2.py

import configparser
import logging
import os
import re
from time import sleep
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager
from webdriver_manager.microsoft import EdgeChromiumDriverManager
from webdriver_manager.firefox import GeckoDriverManager
import ssl
os.environ['WDM_SSL_VERIFY'] = "false"

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# 读取配置文件
config = configparser.ConfigParser()
config.read('config.ini', encoding="utf-8")

# 从配置文件中读取参数
url_base = config.get('base', 'url_base')
url_login = r"https://mozi-login.alibaba-inc.com/?APP_NAME=LPD_TEAM_AEOLUS&BACK_URL=https%3A%2F%2Faeolus.ele.me"
url_home = config.get('base', 'url_home')
url_work = config.get('base', 'url_work')
account = config.get('base', 'account')
password = config.get('base', 'password')
interval = int(config.get('base', 'interval'))
driver_type = config.get('base', 'driver')
driver_path_edge = config.get('base', 'driver_path_edge')
driver_path_chrome = config.get('base', 'driver_path_chrome')

# WebDriver初始化
def init_webdriver(driver_type):
    options = Options()
    manager = {
        "firefox": GeckoDriverManager,
        "edge": EdgeChromiumDriverManager,
        "chrome": ChromeDriverManager
    }[driver_type.lower()]

    driver_path = None  # 初始化为None，以防安装失败
    try:
        driver_path = manager().install()
        logging.info("找到驱动" + driver_path)
        service = Service(executable_path=driver_path)

        if driver_type.lower() == "firefox":
            oprofile = webdriver.FirefoxOptions()
            oprofile.accept_insecure_certs = True
            driver = webdriver.Firefox(service=service, options=oprofile)
        elif driver_type.lower() == "edge":
            driver = webdriver.Edge(service=service, options=options)
        else:  # chrome
            options.add_argument('--ignore-certificate-errors')
            driver = webdriver.Chrome(service=service, options=options)

        logging.info(f"成功 - 使用{driver_type}")
        return driver

    except Exception as e:
        logging.error(f"无法创建WebDriver实例: {e}")
        if driver_type.lower() == "edge" and driver_path_edge:  # 只有在Edge类型且路径已知的情况下才尝试使用已存在的驱动
            edge_options = Options()  # 创建Edge的Options实例
            service = Service(executable_path=driver_path_edge)  # 使用已知的驱动路径
            driver = webdriver.Edge(service=service, options=edge_options)  # 传入Options实例
            logging.info("使用自行安装的Edge")
            return driver
        else:
            edge_options = Options()  # 创建Edge的Options实例
            service = Service(executable_path=driver_path_chrome)  # 使用已知的驱动路径
            driver = webdriver.Edge(service=service, options=edge_options)  # 传入Options实例
            logging.info("使用自行安装的Chrome")
            return driver

# 登录流程
def login(driver, url_login, account, password):
    driver.get(url_login)
    WebDriverWait(driver, 10).until(EC.url_to_be(url_login))

    username_field = driver.find_element(By.NAME, 'domainAccount')
    username_field.clear()
    username_field.send_keys(account)

    password_field = driver.find_element(By.NAME, 'password')
    password_field.clear()
    password_field.send_keys(password)

    login_button = WebDriverWait(driver, 10).until(
        EC.element_to_be_clickable((By.CLASS_NAME, 'sso-btn-submit'))
    )
    login_button.click()

    WebDriverWait(driver, 10).until_not(EC.url_to_be(url_login))
    logging.info("登录成功！")
# 将在线时长转换为分钟
def parse_online_time(online_time_str):
    match = re.match(r'(\d*)小时(\d*)分钟', online_time_str)
    if match:
        hours, minutes = match.groups()
        hours = int(hours) if hours else 0
        minutes = int(minutes) if minutes else 0
        return hours * 60 + minutes
    else:
        # 如果字符串不匹配任何时间格式，假设它是以分钟表示的
        return int(online_time_str) if online_time_str.isdigit() else 0
# 主要逻辑
def main():
    driver = init_webdriver(driver_type)
    if driver is None:
        logging.error("驱动初始化失败")
        return "驱动异常"
    else:
        logging.info("驱动初始化完毕")

    driver.get(url_base) # 开始进入网页

    while True:
        if driver.current_url == url_login:
            sleep(3)
            # 填写手机号码
            username_field = driver.find_element(By.NAME, 'domainAccount')
            username_field.clear()
            username_field.send_keys(account)
            # 填写密码
            password_field = driver.find_element(By.NAME, 'password')
            password_field.clear()
            password_field.send_keys(password)

            # 使用WebDriverWait等待登录按钮变为可点击状态
            login_button = WebDriverWait(driver, 10).until(
                EC.element_to_be_clickable((By.CLASS_NAME, 'sso-btn-submit'))
            )
            login_button.click()
            logging.info("登录完成")
            # 等待一段时间，确保登录过程完成
            sleep(3)
        if driver.current_url == url_home:
            logging.info("已经进入后台主页")
            sleep(1) # 等待加载，跳转到工作目录
            driver.get(url_work)
            sleep(1)  # 等待加载，跳转到工作目录
            logging.info("已经跳转到工作页面；"+driver.current_url)
        if driver.current_url == url_work:

            sleep(3)
            logging.info("开始提取数据")
            # 任务1: 提取目标在线时长和目标完单量
            logging.info("任务1：提取目标在线时长和目标完单量")
            target_info_element = driver.find_element(By.XPATH,
                                                      '//*[@id="root"]/div/div/div/div/div[2]/div/div/div/div/div[2]/div[2]/div/div[1]/div/div[1]/div/div/div[2]/span')
            target_info_text = target_info_element.text
            logging.info(target_info_text)
            target_online_time = int(re.search(r'(\d+)小时(\d+)分钟', target_info_text).group(1)) * 60 + int(
                re.search(r'(\d+)小时(\d+)分钟', target_info_text).group(2))
            logging.info("目标时间（min）："+str(target_online_time))
            target_order_completion = int(re.search(r'目标完单量：(\d+)', target_info_text).group(1))
            logging.info("目标单量："+str(target_order_completion))
            # 任务2: 提取表格数据并确定目标列的索引
            logging.info("任务2：提取表格数据并确定目标列的索引")
            headers =[]
            for i in range(1, 13):  # 假设有12列
                logging.info("开始查找表头:"+str(i))
                # 使用字符串格式化构建XPath表达式
                xpath_expression = f'//*[@id="root"]/div/div/div/div/div[2]/div/div/div/div/div[2]/div[2]/div/div[2]/div/div/div/div/div[1]/table/thead/tr/th[{i}]/div'
                # 使用XPath表达式查找元素
                header_element = driver.find_element(By.XPATH, xpath_expression)

                if header_element.text=="" or header_element.text==None:
                    logging.error("方法1 失败")
                    xpath_expression = f'//*[@id="root"]/div/div/div/div/div[2]/div/div/div/div/div[2]/div[2]/div/div[2]/div/div/div/div/div[1]/table/thead/tr/th[{i}]/div/span[1]'
                    header_element = driver.find_element(By.XPATH, xpath_expression)
                # if header_element.text=="" or header_element.text==None:
                #     logging.error("方法2 失败")
                #     xpath_expression = f'//*[@id="root"]/div/div/div/div/div[2]/div/div/div/div/div[2]/div[2]/div/div[2]/div/div/div/div/div[1]/table/thead/tr/th[{i}]/div/span[1]/div/div'
                #     header_element = driver.find_element(By.XPATH, xpath_expression)
                if header_element.text=="" or header_element.text==None:
                    logging.error("方法3 失败")
                    xpath_expression = f'/html/body/div[1]/div/div/div/div/div[2]/div/div/div/div/div[2]/div[2]/div/div[2]/div/div/div/div/div[1]/table/thead/tr/th[{i}]/div/span[1]'
                    header_element = driver.find_element(By.XPATH, xpath_expression)
                if header_element.text == "" or header_element.text == None:
                    logging.error("方法4 失败")
                    logging.error("依旧没有找到内容"+str(i))
                    logging.error("开始手动补齐")
                    if i ==8:
                        headers.append("全天完单量")
                    elif i==9:
                        headers.append("时段内在线时长")
                    elif i == 10:
                        headers.append("时段内背单时长")
                    elif i == 11:
                        headers.append("时段内完单量")
                    elif i == 12:
                        headers.append("配送中单量")
                else:
                    logging.error(str(i)+" : "+str(header_element.text))
                    headers.append(header_element.text.replace(" ",""))
            header_to_index = {header: index for index, header in enumerate(headers)}
            online_time_header = "全天在线时长"
            order_completion_header = "全天完单量"
            logging.info("获取表头如下：")
            logging.info(headers)

            # 任务3: 根据表格数据和目标值，筛选出不合格的人
            logging.info("任务3：根据表格数据和目标值，筛选出不合格的人")
            unqualified_persons = []
            rows = driver.find_elements(By.XPATH, '//tbody/tr')
            for row in rows:
                cells = row.find_elements(By.TAG_NAME, 'td')
                row_data = [cell.text for cell in cells]
                name = row_data[header_to_index["姓名"]]
                online_time = row_data[header_to_index[online_time_header]]
                order_completion = row_data[header_to_index[order_completion_header]]

                # 将在线时长转换为分钟
                total_online_time = parse_online_time(online_time)

                # 比较在线时长和完单量
                if total_online_time < target_online_time or int(order_completion) < target_order_completion:
                    unqualified_persons.append(name)

            logging.info(
                f"目标在线时长: {target_online_time // 60}小时{target_online_time % 60}分钟, 目标完单量: {target_order_completion}")
            logging.info("不合格人员名单:", unqualified_persons)
            logging.info("准备工作……休息中……请等待休息完毕……")
            sleep(interval)




if __name__ == '__main__':
    main()