赞
踩
示例代码
import time
import traceback
import logging
from typing import Callable
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.common.exceptions import NoSuchElementException, WebDriverException, TimeoutException
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
# 设置日志配置
logging.basicConfig(level=logging.ERROR)
logger = logging.getLogger(__name__)
# 假设 DOME 是一个配置字典,用于控制 DEBUG 模式
DOME = {
'DEBUG': False,
'debug': False
}
def exr_func(*args, **kwargs):
# 在每次重试前执行的操作,可以在此添加刷新页面等逻辑
driver = kwargs.get('driver')
if driver:
driver.refresh()
time.sleep(2) # 等待页面刷新完成
def dec_retry(num=1, exr: Callable = lambda *_, **__: None, catch=Exception, exclude=Exception, out_error=True):
"""装饰器,重复执行func并忽略异常,超过次数抛出异常或返回空,此函数默认输出异常信息"""
def decorator(func):
def dec(*args, **kwargs):
for i in range(1, num + 1):
try:
return func(*args, **kwargs)
except exclude as err:
raise err
except catch as err:
logger.error(f'第 {i} 次执行失败')
if out_error:
logger.error(err, traceback.format_exc(), sep='\n')
if DOME.get('DEBUG') or DOME.get('debug'):
logger.error('已开启 DEBUG 模式,直接抛出异常')
raise err
if i >= num:
logger.error(f'已执行 {num} 次,抛出异常')
raise err
logger.error('开始重试')
exr(*args, **kwargs)
return dec
return decorator
@dec_retry(num=3, exr=exr_func, catch=(NoSuchElementException, WebDriverException, TimeoutException))
def login(driver, username, password):
# 访问登录页面
driver.get("https://example.com/login")
# 等待页面加载完成
WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "username")))
# 输入用户名
username_field = driver.find_element(By.ID, "username")
username_field.send_keys(username)
# 输入密码
password_field = driver.find_element(By.ID, "password")
password_field.send_keys(password)
# 点击登录按钮
login_button = driver.find_element(By.ID, "loginButton")
login_button.click()
# 等待页面加载完成并检查是否登录成功
WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "logoutButton")))
print("登录成功")
return True
def main():
# 设置WebDriver
driver = webdriver.Chrome()
username = "your_username"
password = "your_password"
try:
login(driver, username, password, driver=driver)
except Exception as e:
logger.error(f"最终登录失败,异常: {e}")
driver.quit()
if __name__ == "__main__":
main()
代码解释
exr_func
函数: 定义了在每次重试前执行的操作,例如刷新页面。dec_retry
装饰器: 实现了重试机制,捕获指定异常,在重试次数用尽前会执行 exr_func
函数。login
函数: 实现具体的登录操作,并被 dec_retry
装饰器装饰。main
函数: 设置 WebDriver,尝试登录,并在失败时输出日志信息。类装饰器可以更灵活地管理状态,并且可以更好地组织复杂的逻辑。
import time
import traceback
import logging
from typing import Callable
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.common.exceptions import NoSuchElementException, WebDriverException, TimeoutException
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
# 设置日志配置
logging.basicConfig(level=logging.ERROR)
logger = logging.getLogger(__name__)
# 假设 DOME 是一个配置字典,用于控制 DEBUG 模式
DOME = {
'DEBUG': False,
'debug': False
}
def exr_func(*args, **kwargs):
driver = kwargs.get('driver')
if driver:
driver.refresh()
time.sleep(2) # 等待页面刷新完成
class RetryDecorator:
def __init__(self, num=1, exr: Callable = lambda *_, **__: None, catch=Exception, exclude=Exception, out_error=True):
self.num = num
self.exr = exr
self.catch = catch
self.exclude = exclude
self.out_error = out_error
def __call__(self, func):
def wrapped_func(*args, **kwargs):
for i in range(1, self.num + 1):
try:
return func(*args, **kwargs)
except self.exclude as err:
raise err
except self.catch as err:
logger.error(f'第 {i} 次执行失败')
if self.out_error:
logger.error(err, traceback.format_exc(), sep='\n')
if DOME.get('DEBUG') or DOME.get('debug'):
logger.error('已开启 DEBUG 模式,直接抛出异常')
raise err
if i >= self.num:
logger.error(f'已执行 {self.num} 次,抛出异常')
raise err
logger.error('开始重试')
self.exr(*args, **kwargs)
return wrapped_func
@RetryDecorator(num=3, exr=exr_func, catch=(NoSuchElementException, WebDriverException, TimeoutException))
def login(driver, username, password):
driver.get("https://example.com/login")
WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "username")))
username_field = driver.find_element(By.ID, "username")
username_field.send_keys(username)
password_field = driver.find_element(By.ID, "password")
password_field.send_keys(password)
login_button = driver.find_element(By.ID, "loginButton")
login_button.click()
WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "logoutButton")))
print("登录成功")
return True
def main():
driver = webdriver.Chrome()
username = "your_username"
password = "your_password"
try:
login(driver, username, password, driver=driver)
except Exception as e:
logger.error(f"最终登录失败,异常: {e}")
driver.quit()
if __name__ == "__main__":
main()
使用闭包使代码更简洁。
import time
import traceback
import logging
from typing import Callable
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.common.exceptions import NoSuchElementException, WebDriverException, TimeoutException
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
logging.basicConfig(level=logging.ERROR)
logger = logging.getLogger(__name__)
DOME = {
'DEBUG': False,
'debug': False
}
def exr_func(*args, **kwargs):
driver = kwargs.get('driver')
if driver:
driver.refresh()
time.sleep(2)
def dec_retry(num=1, exr: Callable = lambda *_, **__: None, catch=Exception, exclude=Exception, out_error=True):
def decorator(func):
def wrapped_func(*args, **kwargs):
for i in range(1, num + 1):
try:
return func(*args, **kwargs)
except exclude as err:
raise err
except catch as err:
logger.error(f'第 {i} 次执行失败')
if out_error:
logger.error(err, traceback.format_exc(), sep='\n')
if DOME.get('DEBUG') or DOME.get('debug'):
logger.error('已开启 DEBUG 模式,直接抛出异常')
raise err
if i >= num:
logger.error(f'已执行 {num} 次,抛出异常')
raise err
logger.error('开始重试')
exr(*args, **kwargs)
return wrapped_func
return decorator
@dec_retry(num=3, exr=exr_func, catch=(NoSuchElementException, WebDriverException, TimeoutException))
def login(driver, username, password):
driver.get("https://example.com/login")
WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "username")))
username_field = driver.find_element(By.ID, "username")
username_field.send_keys(username)
password_field = driver.find_element(By.ID, "password")
password_field.send_keys(password)
login_button = driver.find_element(By.ID, "loginButton")
login_button.click()
WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "logoutButton")))
print("登录成功")
return True
def main():
driver = webdriver.Chrome()
username = "your_username"
password = "your_password"
try:
login(driver, username, password, driver=driver)
except Exception as e:
logger.error(f"最终登录失败,异常: {e}")
driver.quit()
if __name__ == "__main__":
main()
wrapt
模块wrapt
是一个强大的装饰器库,可以更轻松地编写装饰器,特别是处理复杂逻辑时。
import time
import traceback
import logging
import wrapt
from typing import Callable
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.common.exceptions import NoSuchElementException, WebDriverException, TimeoutException
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
logging.basicConfig(level=logging.ERROR)
logger = logging.getLogger(__name__)
DOME = {
'DEBUG': False,
'debug': False
}
def exr_func(*args, **kwargs):
driver = kwargs.get('driver')
if driver:
driver.refresh()
time.sleep(2)
def dec_retry(num=1, exr: Callable = lambda *_, **__: None, catch=Exception, exclude=Exception, out_error=True):
@wrapt.decorator
def wrapper(wrapped, instance, args, kwargs):
for i in range(1, num + 1):
try:
return wrapped(*args, **kwargs)
except exclude as err:
raise err
except catch as err:
logger.error(f'第 {i} 次执行失败')
if out_error:
logger.error(err, traceback.format_exc(), sep='\n')
if DOME.get('DEBUG') or DOME.get('debug'):
logger.error('已开启 DEBUG 模式,直接抛出异常')
raise err
if i >= num:
logger.error(f'已执行 {num} 次,抛出异常')
raise err
logger.error('开始重试')
exr(*args, **kwargs)
return wrapper
@dec_retry(num=3, exr=exr_func, catch=(NoSuchElementException, WebDriverException, TimeoutException))
def login(driver, username, password):
driver.get("https://example.com/login")
WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "username")))
username_field = driver.find_element(By.ID, "username")
username_field.send_keys(username)
password_field = driver.find_element(By.ID, "password")
password_field.send_keys(password)
login_button = driver.find_element(By.ID, "loginButton")
login_button.click()
WebDriverWait(driver, 10).until(EC.presence_of_element_located((By.ID, "logoutButton")))
print("登录成功")
return True
def main():
driver = webdriver.Chrome()
username = "your_username"
password = "your_password"
try:
login(driver, username, password, driver=driver)
except Exception as e:
logger.error(f"最终登录失败,异常: {e}")
driver.quit()
if __name__ == "__main__":
main()
赞
踩
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。