| class Smoking:
?"""定義Smoking類"""
?# 瀏覽器驅(qū)動(dòng)
?__driver: webdriver = None
?# 配置幫助類
?__cfg_info: dict = {}
?# 日志幫助類
?__log_helper: LogHelper = None
?# 主程序目錄
?__work_path: str = ''
?# 是否正在運(yùn)行
?__running: bool = False
?# 無(wú)貨
?__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):
??"""通過(guò)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):
??"""通過(guò)名稱判斷是否存在"""
??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):
??"""通過(guò)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):
??"""通過(guò)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)到登錄頁(yè)面
??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)前庫(kù)存是否有貨
????xpath1 = "http://div[@class='product-list']/div[@class='product']/div[@class='price-and-detail']/div[" \
???????"@class='price']/span[@class='noStock'] "
????if self.checkIsExistsByPath(xpath1):
?????txt = self.__driver.find_element_by_xpath(xpath1).text
?????if txt == self.__no_stock:
??????# 當(dāng)前無(wú)貨
??????time.sleep(self.__wait_sec)
??????self.__log_helper.put('查詢一次' + item_id + '，無(wú)貨')
??????continue
????# 鏈接path1
????xpath2 = "http://div[@class='product-list']/div[@class='product']/div[@class='imgDiv']/a"
????# 判斷是否加載完畢
????# self.waiting((By.CLASS_NAME, "imgDiv"))
????if self.checkIsExistsByPath(xpath2):
?????self.__driver.find_element_by_xpath(xpath2).click()
?????time.sleep(self.__wait_sec)
?????# 加入購(gòu)物車
?????if self.checkIsExistsByClass('add-to-cart'):
??????self.__driver.find_element_by_class_name('add-to-cart').click()
??????self.__log_helper.put('加入購(gòu)物車成功，商品item-id:' + item_id)
??????break
?????else:
??????self.__log_helper.put('未找到加入購(gòu)物車按鈕')
????else:
?????self.__log_helper.put('沒有查詢到，可能是商品編碼不對(duì)，或者已下架')
???except BaseException as e:
????self.__log_helper.put(e)
?def startRun(self):
??"""運(yùn)行起來(lái)"""
??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)行過(guò)程中出錯(cuò)，請(qǐng)重新打開再試')
?def run(self, url, username, password, item_id):
??"""運(yùn)行起來(lái)"""
??# 3. 訪問(wèn)網(wǎng)站
??self.__driver.get(url)
??# 4. 最大化窗口
??self.__driver.maximize_window()
??if self.checkIsExistsByLinkText('賬戶登錄'):
???# 判斷是否登錄：未登錄
???self.login(username, password)
??if self.checkIsExistsByPartialLinkText('歡迎回來(lái)'):
???# 判斷是否登錄：已登錄
???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)
|
文章來(lái)源：http://www.zghlxwxcb.cn/news/detail-797508.html
文章來(lái)源地址http://www.zghlxwxcb.cn/news/detail-797508.html
![[爬蟲]2.2.1 使用Selenium庫(kù)模擬瀏覽器操作](https://imgs.yssmx.com/Uploads/2024/02/594545-1.jpg)
