class SeleniumUtil def __init__self driver selfdriver = driver # selfdisplay = settingsCHROME_DISPLAY def set_driverself driver selfdriver = driver def safe_get_elements_
可以将driver的定义放在BaseSpider类中,然后在BaseSpider的构造方法中初始化driver。这样可以保证每次运行程序时都只会创建一个driver实例。在程序运行完成后,可以在DFMProcessor的start方法中调用driver.quit()来退出driver。
修改后的代码如下:
class BaseSpider(ABC):
def __init__(self):
self.driver = None
self.browser_ = None
self.records = list()
self.now_time = datetime.datetime.now()
def set_driver(self, driver):
self.driver = driver
self.browser_ = SeleniumUtil(self.driver)
def add_record(self, urn, class_name, operate, exception, ele_message):
"""错误记录
urn:urn class_name:类名 operate:操作 exception:错误信息 ele_message:元素信息
"""
self.records.extend([self.now_time, urn, class_name, operate, exception, ele_message])
QueryResult.insert_data(INSERT_OPERATION_LOG, self.records)
self.records.clear()
def get_search_url(self, url):
self.browser_.get_page(url)
def check_login(self):
"""登录验证逻辑实现"""
self.browser_.get_page(LOGIN_URL) # 登录
try:
self.browser_.input_with_xpath(By.NAME, TEXTBOX_LOGIN_ELEMENT, settings['dfm_username'],
ErrorMsg.TEXT_BOX_NOT_DISPLAYED)
self.browser_.input_with_xpath(By.NAME, TEXTBOX_PWD_ELEMENT,
generate_utils.transform_str(settings['dfm_pwd']),
ErrorMsg.TEXT_BOX_NOT_DISPLAYED)
self.browser_.click_with_xpath(By.NAME, BUTTON_LOGIN_ELEMENT,
LogErrorMsg.LOGIN_CLICK_BUTTON_NOT_DISPLAYED)
self.browser_.check_loging_with(By.XPATH, CUSTOM_LOGIN_ELEMENT)
return True
except SpiderException as e:
log_error(e)
self.add_record(False, self.__class__.__name__, Operate.LOGIN.value, e.message, e.xpath)
return False
class EmailSpider(BaseSpider):
def __int__(self):
BaseSpider.__init__(self) # 调用父类构造方法
def get_advanced_and_ingestion(self):
"""advanced 和ingestion 按钮"""
self.get_search_url(SEARCH_PAGE) # search页面
self.browser_.click_with_xpath(By.LINK_TEXT, ADVANCED_SEARCH_ELEMENT,
EmailErrorMsg.ADVANCED_SEARCH_NOR_DISPLAYED) # Advanced Search 按钮
self.browser_.click_with_xpath(By.LINK_TEXT, EMAIL_INGESTION_DATA_ELEMENT,
EmailErrorMsg.INGESTION_DATA_NOR_DISPLAYED) # Email Ingestion data 按钮
class DFMProcessor(BaseProcessor):
def __init__(self):
BaseProcessor.__init__(self)
self.name = 'DFMProcessor'
self.lock = ''
self.display = settings['CHROME_DISPLAY']
self.baseSpider = BaseSpider()
def process(self):
pass
def get_download_path(self):
return ''
def start(self):
try:
self.lock.acquire()
logger.info(r" >>> 开始执行 {}".format(self.name))
driver = get_chrome_driver(self.get_download_path(), display=self.display)
self.baseSpider.set_driver(driver) # 设置driver给BaseSpider
status = self.baseSpider.check_login()
if status:
self.process()
logger.info(r" >>> 完成执行 {}".format(self.name))
except Exception as e:
log_error(e)
finally:
self.baseSpider.driver.quit()
self.lock.release()
@singleton
class EmailIngestionProcessor(DFMProcessor):
def __init__(self):
DFMProcessor.__init__(self)
self.lock = threading.Lock()
self.name = "EmailIngestionProcessor"
self.emailSpider = EmailSpider()
def get_download_path(self):
cur_date = get_strtime('%Y-%m-%d')
download_path = os.path.join(settings['email_ingestion_path'], cur_date)
if not os.path.exists(download_path):
os.makedirs(download_path)
return download_path
def process(self):
self.emailSpider.get_advanced_and_ingestion()
def run():
EmailIngestionProcessor().start()
if __name__ == '__main__':
run()
``
原文地址: https://www.cveoy.top/t/topic/ixux 著作权归作者所有。请勿转载和采集!