96 lines
3.0 KiB
Python
96 lines
3.0 KiB
Python
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
|
||
from utils.Ntfy import ntfy
|
||
|
||
class htmlUtils:
|
||
headers = {'User-Agent': UserAgent().random}
|
||
url_data = {}
|
||
|
||
@classmethod
|
||
def getHTML(cls, curl,type=None):
|
||
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 <= 10:
|
||
try:
|
||
print(f"请求地址：{curl}")
|
||
res = s.get(curl,stream=True, headers=cls.headers, timeout=180)
|
||
time.sleep(3)
|
||
if type == "bytes":
|
||
url_text = res
|
||
else:
|
||
url_text = html.fromstring(res.text)
|
||
except:
|
||
print(f'Retry! 第{count}次')
|
||
time.sleep(3)
|
||
traceback.print_exc()
|
||
count += 1
|
||
continue
|
||
else:
|
||
ntfy.sendMsg(f"fail 请求失败：{curl}")
|
||
exit()
|
||
break
|
||
if type == "None":
|
||
data = { curl : url_text}
|
||
cls.url_data.update(data)
|
||
return url_text
|
||
|
||
@classmethod
|
||
def getBytes(cls, url):
|
||
return cls.getHTML(url,type="bytes")
|
||
|
||
@classmethod
|
||
def getJSON(cls,curl):
|
||
count = 0
|
||
if count < 5:
|
||
try:
|
||
res = requests.get(curl, headers=cls.headers, timeout=180)
|
||
time.sleep(1.5)
|
||
data_json = res.json()
|
||
return data_json
|
||
except:
|
||
print(f"请求失败，重试中... {curl}")
|
||
count += 1
|
||
time.sleep(1)
|
||
cls.getJSON(curl)
|
||
else:
|
||
ntfy.sendMsg(f"重试请求失败... {curl}, 已退出")
|
||
exit()
|
||
|
||
@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 |