161 lines
6.2 KiB
Python
161 lines
6.2 KiB
Python
import configparser
|
||
import logging
|
||
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
|
||
|
||
|
||
# 读取配置文件
|
||
config = configparser.ConfigParser()
|
||
config.read('config.ini', encoding="utf-8")
|
||
|
||
# 从配置文件中读取参数
|
||
url_base= config.get('base', 'url_base')
|
||
# url_login= config.get('base', 'url_login')
|
||
url_login = "https://mozi-login.alibaba-inc.com/?APP_NAME=LPD_TEAM_AEOLUS&BACK_URL=https%3A%2F%2Faeolus.ele.me"
|
||
print(url_login)
|
||
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 = config.get('base', 'interval')
|
||
driver_type = config.get('base', 'driver')
|
||
def open_url_with_selenium(url):
|
||
# 创建ChromeOptions对象
|
||
options = Options()
|
||
# 确保无头模式没有被启用
|
||
# options.add_argument('--headless') # 注释掉这行
|
||
if driver_type.lower()=="firefox":
|
||
driver_path = GeckoDriverManager().install()
|
||
try:
|
||
service = Service(executable_path=driver_path)
|
||
driver = webdriver.Firefox(service=service, options=options)
|
||
logging.info("成功 - 使用firefox")
|
||
except Exception as e:
|
||
logging.error("无法创建WebDriver实例: %s", str(e))
|
||
return None
|
||
elif driver_type.lower()=="edge":
|
||
driver_path = EdgeChromiumDriverManager().install()
|
||
# 创建WebDriver实例
|
||
try:
|
||
service = Service(executable_path=driver_path)
|
||
driver = webdriver.Edge(service=service, options=options)
|
||
logging.info("成功 - 使用edge")
|
||
except Exception as e:
|
||
logging.error("无法创建WebDriver实例: %s", str(e))
|
||
return None
|
||
else: # 默认谷歌驱动
|
||
# 确保chromedriver已经安装并获取其路径
|
||
driver_path = ChromeDriverManager().install()
|
||
# 创建Service对象
|
||
# 创建WebDriver实例
|
||
try:
|
||
service = Service(executable_path=driver_path)
|
||
driver = webdriver.Chrome(service=service, options=options)
|
||
logging.info("成功 - 使用chrome")
|
||
except Exception as e:
|
||
logging.error("无法创建WebDriver实例: %s", str(e))
|
||
return None
|
||
try:
|
||
# 打开指定的URL
|
||
logging.info("开始访问"+url_base)
|
||
driver.get(url_base)
|
||
sleep(3) # 等待3秒
|
||
while True:
|
||
if driver.current_url==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)
|
||
|
||
# 使用WebDriverWait等待登录按钮变为可点击状态
|
||
login_button = WebDriverWait(driver, 10).until(
|
||
EC.element_to_be_clickable((By.CLASS_NAME, 'sso-btn-submit'))
|
||
)
|
||
login_button.click()
|
||
# 等待一段时间，确保登录过程完成
|
||
sleep(3)
|
||
break
|
||
print("当前URL：", driver.current_url)
|
||
print("当前URL：", driver.current_url)
|
||
while True:
|
||
if driver.current_url==url_home:
|
||
# 登录成功，已经进入了主页
|
||
print("cookie",driver.get_cookies())
|
||
# 跳转到 https://aeolus.ele.me/?targetId=3818#/group/schedule/view/manage
|
||
driver.get(url_work)
|
||
if driver.current_url == url_work:
|
||
break
|
||
|
||
# 开始查找数据
|
||
while driver.current_url == url_work:
|
||
sleep(interval)
|
||
# 查找目标值
|
||
# 定位到包含目标完单量的div元素
|
||
print("开始查找 - 定位到包含目标完单量的div元素")
|
||
div_element = driver.find_element(By.XPATH, "//div[@class='ant-space-item'][2]/span")
|
||
print(div_element.text)
|
||
print("开始查找 - 定位到包含目标完单量的div元素")
|
||
# 定位表格
|
||
table = WebDriverWait(driver, 10).until(
|
||
EC.presence_of_element_located((By.CSS_SELECTOR, ".ant-table-thead")))
|
||
|
||
# 获取表头
|
||
headers = []
|
||
header_elements = table.find_elements(By.TAG_NAME, "th")
|
||
for header in header_elements:
|
||
headers.append(header.text)
|
||
|
||
print("Headers:", headers)
|
||
|
||
# 定位表体
|
||
tbody = WebDriverWait(driver, 10).until(
|
||
EC.presence_of_element_located((By.CSS_SELECTOR, ".ant-table-tbody")))
|
||
|
||
# 获取表体数据
|
||
rows = tbody.find_elements(By.TAG_NAME, "tr")
|
||
data = []
|
||
for row in rows:
|
||
cols = row.find_elements(By.TAG_NAME, "td")
|
||
row_data = [col.text for col in cols]
|
||
data.append(row_data)
|
||
|
||
print("Data:")
|
||
for row in data:
|
||
print(row)
|
||
|
||
|
||
|
||
|
||
|
||
# 等待一段时间，确保登录过程完成
|
||
sleep(5)
|
||
|
||
# 检查是否登录成功
|
||
if 'login' not in driver.current_url:
|
||
print("登录成功！")
|
||
else:
|
||
print("登录失败，请检查用户名和密码。")
|
||
|
||
# 返回WebDriver实例，而不是关闭它
|
||
return driver
|
||
except Exception as e:
|
||
print(e)
|
||
return None
|
||
|
||
if __name__ == '__main__':
|
||
driver = open_url_with_selenium(url_base)
|
||
# 如果你想要在后续操作中使用driver，可以在这里添加代码
|
||
# 例如：driver.get('another_url')
|
||
# 在你完成所有操作后，记得调用driver.quit()来关闭WebDriver实例 |