相思资源网 Design By www.200059.com
概述
在进行网站爬取数据的时候，会发现很多网站都进行了反爬虫的处理，如JS加密，Ajax加密，反Debug等方法，通过请求获取数据和页面展示的内容完全不同，这时候就用到Selenium技术，来模拟浏览器的操作，然后获取数据。本文以一个简单的小例子，简述Python搭配Tkinter和Selenium进行浏览器的模拟操作，仅供学习分享使用，如有不足之处，还请指正。
什么是Selenium"color: #ff0000">安装Selenium
通过pip install selenium 进行安装即可，如果速度慢，则可以使用国内的镜像进行安装。
涉及知识点
程序虽小，除了需要掌握的Html ，JavaScript，CSS等基础知识外，本例涉及的Python相关知识点还是蛮多的，具体如下：
- Selenium相关：
Selenium进行元素定位，主要有ID，Name，ClassName，Css Selector，Partial LinkText，LinkText，XPath，TagName等8种方式。
Selenium获取单一元素（如：find_element_by_xpath）和获取元素数组（如：find_elements_by_xpath）两种方式。
Selenium元素定位后，可以给元素进行赋值和取值，或者进行相应的事件操作（如：click）。
- 线程（Thread）相关：
为了防止前台页面卡主，本文用到了线程进行后台操作，如果要定义一个新的线程，只需要定义一个类并继承threading.Thread，然后重写run方法即可。
在使用线程的过程中，为了保证线程的同步，本例用到了线程锁，如：threading.Lock()。
- 队列（queue）相关：
本例将Selenium执行的过程信息，保存到对列中，并通过线程输出到页面显示。queue默认先进先出方式。
对列通过put进行压栈，通过get进行出栈。通过qsize()用于获取当前对列元素个数。
- 日志（logging.Logger）相关：
为了保存Selenium执行过程中的日志，本例用到了日志模块，为Pyhton自带的模块，不需要额外安装。
Python的日志共六种级别，分别是：NOTSET，DEBUG，INFO，WARN，ERROR，FATAL，CRITICAL。
示例效果图
本例主要针对某一配置好的商品ID进行轮询，监控是否有货，有货则加入购物车，无货则继续轮询，如下图所示：
核心代码
本例最核心的代码，就是利用Selenium进行网站的模拟操作，如下所示：
class Smoking:
"""定义Smoking类"""
# 浏览器驱动
__driver: webdriver = None
# 配置帮助类
__cfg_info: dict = {}
# 日志帮助类
__log_helper: LogHelper = None
# 主程序目录
__work_path: str = ''
# 是否正在运行
__running: bool = False
# 无货
__no_stock = 'Currently Out of Stock'
# 线程等待秒数
__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. 点击链接跳转到登录页面
self.__driver.find_element_by_link_text('账户登录').click()
# 6. 输入账号密码
# 判断是否加载完成
# 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. 点击登录按钮
self.__driver.find_element_by_id('sign-in').click()
def working(self, item_id):
"""工作状态"""
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 = "//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 + '对应的信息')
continue
else:
time.sleep(self.__wait_sec)
self.__log_helper.put('没有查询到item id2 =' + item_id + '对应的信息')
continue
# 判断当前库存是否有货
xpath1 = "//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:
# 当前无货
time.sleep(self.__wait_sec)
self.__log_helper.put('查询一次' + item_id + '，无货')
continue
# 链接path1
xpath2 = "//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)
# 加入购物车
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('没有查询到，可能是商品编码不对，或者已下架')
except BaseException as e:
self.__log_helper.put(e)
def startRun(self):
"""运行起来"""
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('配置信息不全，请检查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. 定义浏览器驱动对象
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('运行过程中出错，请重新打开再试')
def run(self, url, username, password, item_id):
"""运行起来"""
# 3. 访问网站
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('登录失败，请设置账号密码')
def stop(self):
"""停止"""
try:
self.__running = False
# 如果驱动不为空，则关闭
self.close_browser_nicely(self.__driver)
if self.__driver is not None:
self.__driver.quit()
# 关闭后切要为None，否则启动报错
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)
其他辅助类
日志类（LogHelper），代码如下：
class LogHelper:
"""日志帮助类"""
__queue: queue.Queue = None # 队列
__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):
"""添加数据"""
# 记录日志
self.__logging.info(value)
# 添加到队列
if self.__queue.qsize() < self.__queue.maxsize:
self.__queue.put(value)
def get(self):
"""获取数据"""
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):
# # 创建空文件
# open(logfile, mode='r')
fh = logging.FileHandler(logfile, mode='a', encoding='UTF-8')
fh.setLevel(logging.DEBUG) # 输出到file的log等级的开关
# 第三步，定义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):
# 获取当前记录日志的状态
return self.__running
def set_running(self, v: bool):
# 设置当前记录日志的状态
self.__running = v
配置类（ConfigHelper）
class ConfigHelper:
"""初始化数据类"""
__config_dir = None
__dic_cfg = {}
def __init__(self, config_dir):
"""初始化"""
self.__config_dir = config_dir
def ReadConfigInfo(self):
"""得到配置项"""
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
线程类（MyThread）
class MyThread(threading.Thread):
"""后台监控线程"""
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()
# 释放锁，开启下一个线程
# lock.release()
print("结束线程： " + self.name)
self.log_helper.put("结束线程： " + self.name)
备注
侠客行 [唐：李白]赵客缦胡缨，吴钩霜雪明。银鞍照白马，飒沓如流星。
十步杀一人，千里不留行。事了拂衣去，深藏身与名。
闲过信陵饮，脱剑膝前横。将炙啖朱亥，持觞劝侯嬴。
三杯吐然诺，五岳倒为轻。眼花耳热后，意气素霓生。
救赵挥金槌，邯郸先震惊。千秋二壮士，烜赫大梁城。
纵死侠骨香，不惭世上英。谁能书阁下，白首太玄经。
相思资源网 Design By www.200059.com
广告合作：本站广告合作请联系QQ：858582 申请时备注：广告合作（否则不回）
免责声明：本站文章均来自网站采集或用户投稿，网站不提供任何软件下载或自行开发的软件！ 如有用户或公司发现本站内容信息存在侵权行为，请邮件告知！ 858582#qq.com
免责声明：本站文章均来自网站采集或用户投稿，网站不提供任何软件下载或自行开发的软件！ 如有用户或公司发现本站内容信息存在侵权行为，请邮件告知！ 858582#qq.com
相思资源网 Design By www.200059.com
暂无Python使用Selenium模拟浏览器自动操作功能的评论...
一口气升级7个大模型SaaS应用，百度智能云：突出一个“开箱即用” - 2024/11/19
这一波大模型产业落地浪潮里，不少企业其实处在 “干瞪眼“的状态。
一种情况是，很多大模型产品看得见却摸不着，在台上一个个遥遥领先——今天Sora技精四座，明天英伟达的机器人又赢得满堂彩，可是到了台下一问：啥时候能用上啊？答曰：遥遥无期。
另一种情况是，企业想用上大模型，却又难免瞻前顾后——既要考虑场景融合，又得兼顾安全性，还要考虑打通现有系统，再加上各种部署成本和繁琐的采购流程……最后只能拂袖：罢了，再等等吧。
这一波大模型产业落地浪潮里，不少企业其实处在 “干瞪眼“的状态。
一种情况是，很多大模型产品看得见却摸不着，在台上一个个遥遥领先——今天Sora技精四座，明天英伟达的机器人又赢得满堂彩，可是到了台下一问：啥时候能用上啊？答曰：遥遥无期。
另一种情况是，企业想用上大模型，却又难免瞻前顾后——既要考虑场景融合，又得兼顾安全性，还要考虑打通现有系统，再加上各种部署成本和繁琐的采购流程……最后只能拂袖：罢了，再等等吧。
稳了！魔兽国服回归的3条重磅消息！官宣时间再确认！
昨天有一位朋友在大神群里分享，自己亚服账号被封号之后居然弹出了国服的封号信息对话框。
这里面让他访问的是一个国服的战网网址，com.cn和后面的zh都非常明白地表明这就是国服战网。
而他在复制这个网址并且进行登录之后，确实是网易的网址，也就是我们熟悉的停服之后国服发布的暴雪游戏产品运营到期开放退款的说明。这是一件比较奇怪的事情，因为以前都没有出现这样的情况，现在突然提示跳转到国服战网的网址，是不是说明了简体中文客户端已经开始进行更新了呢？