from fake_useragent import UserAgent import requests from lxml import html import traceback import time from urllib3.util.retry import Retry from requests.adapters import HTTPAdapter class htmlUtils: headers = {'User-Agent': UserAgent().random} url_data = {} @classmethod def getHTML(cls, curl): rstr = r"[\/\\\:\*\?\"\<\>\|\.]" #  '/ \ : * ? " < > |' #file_url = re.sub(rstr, "_", curl) keys = cls.url_data.keys() url_text = cls.url_data.get(curl) retries = Retry(total=3, backoff_factor=0.1, status_forcelist=[ 500, 502, 503, 504 ]) s = requests.Session() s.mount('http://', HTTPAdapter(max_retries=retries)) s.mount('https://', HTTPAdapter(max_retries=retries)) count = 1 #数据为空则获取数据 while url_text == None: if count < 5: try: print("请求地址:",curl) res = s.get(curl,stream=True, headers=cls.headers, timeout=20) url_text = html.fromstring(res.text) except: print(f'Retry! 第{count}次') time.sleep(3) traceback.print_exc() count += 1 continue else: print(f"fail 请求失败:{curl}") break data = { curl : url_text} cls.url_data.update(data) return url_text @classmethod def getBytes(cls, url): return requests.get(url, headers=cls.headers,stream=True,timeout=30) @classmethod def getJSON(cls,curl): res = requests.get(curl, headers=cls.headers,timeout=30) data_json = res.json() return data_json @classmethod def xpathData(cls,c_xpath,url=None,num=None,not_eq=None): if url == None: url = cls.temp_url else: cls.temp_url = url result = [] #获取html实体数据 et = cls.getHTML(url) #比对数据 count = 1 xpaths = et.xpath(c_xpath) for x in xpaths: if not x == not_eq: result.append(x) count +=1 if not num == None: try: result = result[num] except: result = None return result