使用多个线程/进程并行从 Excel 读取/写入数据



>我正在做一个项目,该项目需要在点击提交按钮时填写表单(从 excel 表中读取输入数据(并获得输出值(并将其保存到 excel 工作表(。我已经设法用Python+Selenium自动化了这个过程,但是浏览xls文件的所有行需要相当长的时间。

注意:XLUtils是另一个带有帮助程序函数的 python 文件。我还使用openpyxlxlrd从 excel 文件读取/写入

下面是高级别的代码(无线程(:

#Initial set-up
url = "https://url.com"
driver = webdriver.Chrome(options=options, executable_path=r'C:/chromedriver.exe')
driver.get(url)
driver.maximize_window()
#Location of the input sheet
path = "C:/Book1.xlsx"
#Get number of rows from the excel input sheet
rows = XLUtils.getRowCount()
#Log into website
XLUtils.login(driver)
#Loop through all records from the input sheet
for row_num in range(4, rows + 1):
try:
XLUtils.fill_out_form(driver, path, row_num)
except:
print("Element not found and test failed")
continue
#Log out and close Browser
XLUtils.logout(driver)

下面是一个线程示例,但它非常简单,如下所示:

def test_logic():
url = "https://www.google.com"
driver = webdriver.Chrome(options=options, executable_path=r'C:/chromedriver.exe')
driver.get(url)
driver.maximize_window()
search_bar = driver.find_element_by_xpath("//input[@name='q']").send_keys("python threading")
time.sleep(1)
# driver.quit()
num_browsers = 5   # Number of browsers to spawn
thread_list = list()
# Start test
for i in range(num_browsers):
t = threading.Thread(name='Test {}'.format(i), target=test_logic)
t.start()
time.sleep(1)
print(t.name + ' started!')
thread_list.append(t)
# Wait for all threads to complete
for thread in thread_list:
thread.join()
print('Test completed!')

我的问题是:如何合并两个代码以实现我的目标?这仅仅是将我想复制的过程包装在一个函数中,然后将这个函数传递给线程代码段的问题吗?如果是这样,我将如何设法以 5 个为批次循环整个 xls 文件,并从/向生成的 excel 文件读取/写入每个文件。

这是一个棘手的问题,老实说,我没有找到类似的问题,所以任何帮助将不胜感激。



import XL_utility # Importing from my second file
from selenium import webdriver
import unittest
import time


class Test_Excel(unittest.TestCase):

def setUp(self):

self.driver = webdriver.Chrome("C:Program Files (x86)chromedriver.exe")
self.driver.implicitly_wait(10)
self.driver.get("https://eu-uat2-int.weissr-cloud.com/auth")
self.driver.maximize_window()

return self.driver

def test_read_excel_file(self):

path = "...location..."
print(path.title())

rows = XL_utility.getRowCount(path, "Sheet1")

for r in range(2, rows + 1):
username = XL_utility.readData(path, "Sheet1", r, 1)
password = XL_utility.readData(path, "Sheet1", r, 2)

self.driver.find_element_by_name("username").send_keys(username)
self.driver.find_element_by_name("password").send_keys(password)

self.driver.find_element_by_class_name("auth- 
form__submit").click()

time.sleep(10)


if self.driver.title == "Weissr":
print("Test is passed!")
XL_utility.writeData(path, "Sheet1", r, 3, "Test passed")
else:
print("Test failed")
XL_utility.writeData(path, "Sheet1", r, 3, "Test failed")

self.driver.back()
return self.driver



if __name__ == "__main__":
unittest.main()

# The following code is from the second file...


import openpyxl


def getRowCount(file, sheetName):
workbook = openpyxl.load_workbook(file)
sheet = workbook.get_sheet_by_name(sheetName)
return sheet.max_row


def getColumnCount(file, sheetName):
workbook = openpyxl.load_workbook(file)
sheet = workbook.get_sheet_by_name(sheetName)
return sheet.max_column


def readData(file, sheetName, rownum, columnno):
workbook = openpyxl.load_workbook(file)
sheet = workbook.get_sheet_by_name(sheetName)
return sheet.cell(row=rownum, column=columnno).value


def writeData(file, sheetName, rownum, columnno, data):
workbook = openpyxl.load_workbook(file)
sheet = workbook.get_sheet_by_name(sheetName)
sheet.cell(row=rownum, column=columnno).value = data
workbook.save(file)

最新更新