Python使用Selenium模擬瀏覽器自動(dòng)操作功能

class Smoking:
 """定義Smoking類"""

 # 瀏覽器驅(qū)動(dòng)
 __driver: webdriver = None
 # 配置幫助類
 __cfg_info: dict = {}
 # 日志幫助類
 __log_helper: LogHelper = None
 # 主程序目錄
 __work_path: str = ''
 # 是否正在運(yùn)行
 __running: bool = False
 # 無貨
 __no_stock = 'Currently Out of Stock'
 # 線程等待秒數(shù)
 __wait_sec = 2

 def __init__(self, work_path, cfg_info, log_helper: LogHelper):
  """初始化"""
  self.__cfg_info = cfg_info
  self.__log_helper = log_helper
  self.__work_path = work_path
  self.__wait_sec = int(cfg_info['wait_sec'])
  # 如果小于2，則等于2
  self.__wait_sec = (2 if self.__wait_sec < 2 else self.__wait_sec)

 def checkIsExistsById(self, id):
  """通過ID判斷是否存在"""
  try:
   i = 0
   while self.__running and i < 3:
    if len(self.__driver.find_elements_by_id(id)) > 0:
     break
    else:
     time.sleep(self.__wait_sec)
     i = i + 1
   return len(self.__driver.find_elements_by_id(id)) > 0
  except BaseException as e:
   return False

 def checkIsExistsByName(self, name):
  """通過名稱判斷是否存在"""
  try:
   i = 0
   while self.__running and i < 3:
    if len(self.__driver.find_elements_by_name(name)) > 0:
     break
    else:
     time.sleep(self.__wait_sec)
     i = i + 1
   return len(self.__driver.find_elements_by_name(name)) > 0
  except BaseException as e:
   return False

 def checkIsExistsByPath(self, path):
  """通過xpath判斷是否存在"""
  try:
   i = 0
   while self.__running and i < 3:
    if len(self.__driver.find_elements_by_xpath(path)) > 0:
     break
    else:
     time.sleep(self.__wait_sec)
     i = i + 1
   return len(self.__driver.find_elements_by_xpath(path)) > 0
  except BaseException as e:
   return False

 def checkIsExistsByClass(self, cls):
  """通過class名稱判斷是否存在"""
  try:
   i = 0
   while self.__running and i < 3:
    if len(self.__driver.find_elements_by_class_name(cls)) > 0:
     break
    else:
     time.sleep(self.__wait_sec)
     i = i + 1
   return len(self.__driver.find_elements_by_class_name(cls)) > 0
  except BaseException as e:
   return False

 def checkIsExistsByLinkText(self, link_text):
  """判斷LinkText是否存在"""
  try:
   i = 0
   while self.__running and i < 3:
    if len(self.__driver.find_elements_by_link_text(link_text)) > 0:
     break
    else:
     time.sleep(self.__wait_sec)
     i = i + 1
   return len(self.__driver.find_elements_by_link_text(link_text)) > 0
  except BaseException as e:
   return False

 def checkIsExistsByPartialLinkText(self, link_text):
  """判斷包含LinkText是否存在"""
  try:
   i = 0
   while self.__running and i < 3:
    if len(self.__driver.find_elements_by_partial_link_text(link_text)) > 0:
     break
    else:
     time.sleep(self.__wait_sec)
     i = i + 1
   return len(self.__driver.find_elements_by_partial_link_text(link_text)) > 0
  except BaseException as e:
   return False

 # def waiting(self, *locator):
 #  """等待完成"""
 #  # self.__driver.switch_to.window(self.__driver.window_handles[1])
 #  Wait(self.__driver, 60).until(EC.visibility_of_element_located(locator))

 def login(self, username, password):
  """登錄"""
  # 5. 點(diǎn)擊鏈接跳轉(zhuǎn)到登錄頁面
  self.__driver.find_element_by_link_text('賬戶登錄').click()
  # 6. 輸入賬號(hào)密碼
  # 判斷是否加載完成
  # self.waiting((By.ID, "email"))
  if self.checkIsExistsById('email'):
   self.__driver.find_element_by_id('email').send_keys(username)
   self.__driver.find_element_by_id('password').send_keys(password)
   # 7. 點(diǎn)擊登錄按鈕
   self.__driver.find_element_by_id('sign-in').click()

 def working(self, item_id):
  """工作狀態(tài)"""
  while self.__running:
   try:
    # 正常獲取信息
    if self.checkIsExistsById('string'):
     self.__driver.find_element_by_id('string').clear()
     self.__driver.find_element_by_id('string').send_keys(item_id)
     self.__driver.find_element_by_id('string').send_keys(Keys.ENTER)
    # 判斷是否查詢到商品
    xpath = "http://div[@class='specialty-header search']/div[@class='specialty-description']/div[" \
      "@class='gt-450']/span[2] "
    if self.checkIsExistsByPath(xpath):
     count = int(self.__driver.find_element_by_xpath(xpath).text)
     if count < 1:
      time.sleep(self.__wait_sec)
      self.__log_helper.put('沒有查詢到item id =' + item_id + '對(duì)應(yīng)的信息')
      continue
    else:
     time.sleep(self.__wait_sec)
     self.__log_helper.put('沒有查詢到item id2 =' + item_id + '對(duì)應(yīng)的信息')
     continue
    # 判斷當(dāng)前庫存是否有貨

    xpath2 = "http://div[@class='product-list']/div[@class='product']/div[@class='price-and-detail']/div[" \
       "@class='price']/span[@class='noStock'] "
    if self.checkIsExistsByPath(xpath2):
     txt = self.__driver.find_element_by_xpath(xpath2).text
     if txt == self.__no_stock:
      # 當(dāng)前無貨
      time.sleep(self.__wait_sec)
      self.__log_helper.put('查詢一次' + item_id + '，無貨')
      continue

    # 鏈接path2
    xpath3 = "http://div[@class='product-list']/div[@class='product']/div[@class='imgDiv']/a"
    # 判斷是否加載完畢
    # self.waiting((By.CLASS_NAME, "imgDiv"))
    if self.checkIsExistsByPath(xpath3):
     self.__driver.find_element_by_xpath(xpath3).click()
     time.sleep(self.__wait_sec)
     # 加入購物車
     if self.checkIsExistsByClass('add-to-cart'):
      self.__driver.find_element_by_class_name('add-to-cart').click()
      self.__log_helper.put('加入購物車成功，商品item-id:' + item_id)
      break
     else:
      self.__log_helper.put('未找到加入購物車按鈕')
    else:
     self.__log_helper.put('沒有查詢到，可能是商品編碼不對(duì)，或者已下架')
   except BaseException as e:
    self.__log_helper.put(e)

 def startRun(self):
  """運(yùn)行起來"""
  try:
   self.__running = True
   url: str = self.__cfg_info['url']
   username = self.__cfg_info['username']
   password = self.__cfg_info['password']
   item_id = self.__cfg_info['item_id']
   if url is None or len(url) == 0 or username is None or len(username) == 0 or password is None or len(
     password) == 0 or item_id is None or len(item_id) == 0:
    self.__log_helper.put('配置信息不全，請(qǐng)檢查config.cfg文件是否為空，然后再重啟')
    return
   if self.__driver is None:
    options = webdriver.IeOptions()
    options.add_argument('encoding=UTF-8')
    options.add_argument('Accept= text / css, * / *')
    options.add_argument('Accept - Language= zh - Hans - CN, zh - Hans;q = 0.5')
    options.add_argument('Accept - Encoding= gzip, deflate')
    options.add_argument('user-agent=Mozilla/5.0 (Windows NT 10.0; WOW64; Trident/7.0; rv:11.0) like Gecko')
    # 2. 定義瀏覽器驅(qū)動(dòng)對(duì)象
    self.__driver = webdriver.Ie(executable_path=self.__work_path + r'\IEDriverServer.exe', options=options)
   self.run(url, username, password, item_id)
  except BaseException as e:
   self.__log_helper.put('運(yùn)行過程中出錯(cuò)，請(qǐng)重新打開再試')

 def run(self, url, username, password, item_id):
  """運(yùn)行起來"""
  # 3. 訪問網(wǎng)站
  self.__driver.get(url)
  # 4. 最大化窗口
  self.__driver.maximize_window()
  if self.checkIsExistsByLinkText('賬戶登錄'):
   # 判斷是否登錄：未登錄
   self.login(username, password)
  if self.checkIsExistsByPartialLinkText('歡迎回來'):
   # 判斷是否登錄：已登錄
   self.__log_helper.put('登錄成功，下一步開始工作了')
   self.working(item_id)
  else:
   self.__log_helper.put('登錄失敗，請(qǐng)?jiān)O(shè)置賬號(hào)密碼')

 def stop(self):
  """停止"""
  try:
   self.__running = False
   # 如果驅(qū)動(dòng)不為空，則關(guān)閉
   self.close_browser_nicely(self.__driver)
   if self.__driver is not None:
    self.__driver.quit()
    # 關(guān)閉后切要為None，否則啟動(dòng)報(bào)錯(cuò)
    self.__driver = None
  except BaseException as e:
   print('Stop Failure')
  finally:
   self.__driver = None

 def close_browser_nicely(self, browser):
  try:
   browser.execute_script("window.onunload=null; window.onbeforeunload=null")
  except Exception as err:
   print("Fail to execute_script:'window.onunload=null; window.onbeforeunload=null'")

  socket.setdefaulttimeout(10)
  try:
   browser.quit()
   print("Close browser and firefox by calling quit()")
  except Exception as err:
   print("Fail to quit from browser, error-type:%s, reason:%s" % (type(err), str(err)))
  socket.setdefaulttimeout(30)

class LogHelper:
 """日志幫助類"""
 __queue: queue.Queue = None # 隊(duì)列
 __logging: logging.Logger = None # 日志
 __running: bool = False # 是否記錄日志

 def __init__(self, log_path):
  """初始化類"""
  self.__queue = queue.Queue(1000)
  self.init_log(log_path)

 def put(self, value):
  """添加數(shù)據(jù)"""
  # 記錄日志
  self.__logging.info(value)
  # 添加到隊(duì)列
  if self.__queue.qsize() < self.__queue.maxsize:
   self.__queue.put(value)

 def get(self):
  """獲取數(shù)據(jù)"""
  if self.__queue.qsize() > 0:
   try:
    return self.__queue.get(block=False)
   except BaseException as e:
    return None
  else:
   return None

 def init_log(self, log_path):
  """初始化日志"""
  self.__logging = logging.getLogger()
  self.__logging.setLevel(logging.INFO)
  # 日志
  rq = time.strftime('%Y%m%d%H%M', time.localtime(time.time()))
  log_name = log_path + rq + '.log'
  logfile = log_name
  # if not os.path.exists(logfile):
  #  # 創(chuàng)建空文件
  #  open(logfile, mode='r')
  fh = logging.FileHandler(logfile, mode='a', encoding='UTF-8')
  fh.setLevel(logging.DEBUG) # 輸出到file的log等級(jí)的開關(guān)
  # 第三步，定義handler的輸出格式
  formatter = logging.Formatter("%(asctime)s - %(filename)s[line:%(lineno)d] - %(levelname)s: %(message)s")
  fh.setFormatter(formatter)
  # 第四步，將logger添加到handler里面
  self.__logging.addHandler(fh)

 def get_running(self):
  # 獲取當(dāng)前記錄日志的狀態(tài)
  return self.__running

 def set_running(self, v: bool):
  # 設(shè)置當(dāng)前記錄日志的狀態(tài)

  self.__running = v

class ConfigHelper:
 """初始化數(shù)據(jù)類"""

 __config_dir = None
 __dic_cfg = {}

 def __init__(self, config_dir):
  """初始化"""
  self.__config_dir = config_dir

 def ReadConfigInfo(self):
  """得到配置項(xiàng)"""
  parser = ConfigParser()
  parser.read(self.__config_dir + r"\config.cfg")
  section = parser.sections()[0]
  items = parser.items(section)
  self.__dic_cfg.clear()
  for item in items:
   self.__dic_cfg.__setitem__(item[0], item[1])

 def getConfigInfo(self):
  """獲取配置信息"""
  if len(self.__dic_cfg) == 0:
   self.ReadConfigInfo()
  return self.__dic_cfg

class MyThread(threading.Thread):
 """后臺(tái)監(jiān)控線程"""

 def __init__(self, tid, name, smoking: Smoking, log_helper: LogHelper):
  """線程初始化"""
  threading.Thread.__init__(self)
  self.threadID = tid
  self.name = name
  self.smoking = smoking
  self.log_helper = log_helper

 def run(self):
  print("開啟線程： " + self.name)
  self.log_helper.put("開啟線程： " + self.name)
  # 獲取鎖，用于線程同步
  # lock = threading.Lock()
  # lock.acquire()
  self.smoking.startRun()
  # 釋放鎖，開啟下一個(gè)線程
  # lock.release()
  print("結(jié)束線程： " + self.name)
  self.log_helper.put("結(jié)束線程： " + self.name)