209 lines
8.2 KiB
Python
209 lines
8.2 KiB
Python
import json
|
||
import re
|
||
import requests
|
||
import logging
|
||
import time
|
||
from lxml import etree
|
||
from types import SimpleNamespace
|
||
from http.cookies import SimpleCookie
|
||
|
||
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
|
||
|
||
|
||
class ExtendedResponse(requests.Response):
|
||
def xpath(self):
|
||
try:
|
||
tree = etree.HTML(self.text)
|
||
return tree
|
||
except Exception as e:
|
||
raise ValueError("XPath解析错误: " + str(e))
|
||
|
||
def to_Dict(self):
|
||
try:
|
||
data = self.json()
|
||
return self.dict_to_obj(data)
|
||
except Exception as e:
|
||
raise ValueError("JSON转换错误: " + str(e))
|
||
|
||
def to_Re_findall(self, regex):
|
||
try:
|
||
data = self.text
|
||
return re.findall(regex, data)
|
||
except Exception as e:
|
||
raise ValueError("Re搜索错误: " + str(e))
|
||
|
||
def cookies_dict(self):
|
||
try:
|
||
# 获取原有的 cookies 字典
|
||
cookie_dict = self.cookies.get_dict()
|
||
# 如果响应头中有 Set-Cookie，则解析并补充 cookies
|
||
if 'Set-Cookie' in self.headers:
|
||
from http.cookies import SimpleCookie
|
||
sc = SimpleCookie()
|
||
sc.load(self.headers['Set-Cookie'])
|
||
for key, morsel in sc.items():
|
||
cookie_dict[key] = morsel.value
|
||
return cookie_dict
|
||
except Exception as e:
|
||
raise ValueError("Cookies转换错误: " + str(e))
|
||
|
||
def save_cookies(self, filepath, format='json'):
|
||
"""
|
||
将当前响应中的cookie信息保存到指定文件中。
|
||
|
||
参数：
|
||
filepath (str): 保存文件的路径
|
||
format (str): 保存格式，支持 'json'、'pickle' 和 'txt' 三种格式，默认为 'json'
|
||
"""
|
||
try:
|
||
cookie_dict = self.cookies_dict()
|
||
if format.lower() == 'json':
|
||
with open(filepath, 'w', encoding='utf-8') as f:
|
||
json.dump(cookie_dict, f, ensure_ascii=False, indent=4)
|
||
elif format.lower() == 'pickle':
|
||
import pickle
|
||
with open(filepath, 'wb') as f:
|
||
pickle.dump(cookie_dict, f)
|
||
elif format.lower() == 'txt':
|
||
with open(filepath, 'w', encoding='utf-8') as f:
|
||
for key, value in cookie_dict.items():
|
||
f.write(f"{key}: {value}\n")
|
||
else:
|
||
raise ValueError("不支持的格式，请选择 'json'、'pickle' 或 'txt'")
|
||
except Exception as e:
|
||
raise ValueError("保存cookies出错: " + str(e))
|
||
|
||
@staticmethod
|
||
def dict_to_obj(d):
|
||
if isinstance(d, dict):
|
||
return SimpleNamespace(**{k: ExtendedResponse.dict_to_obj(v) for k, v in d.items()})
|
||
elif isinstance(d, list):
|
||
return [ExtendedResponse.dict_to_obj(item) for item in d]
|
||
else:
|
||
return d
|
||
|
||
|
||
class MyRequests:
|
||
def __init__(self, base_url, protocol='http', retries=3, proxy_options=True, default_timeout=10,
|
||
default_cookies=None):
|
||
"""
|
||
初始化 MyRequests 对象，自动加载本地 cookies 文件（根据 base_url 生成文件名，如 "www_zhrczp_com_cookies.json"）中的 cookies，
|
||
如果文件存在，则将其加载到 session 中；否则使用 default_cookies（如果提供）更新 session。
|
||
|
||
参数：
|
||
base_url (str): 基础 URL
|
||
protocol (str): 协议（默认为 'http'）
|
||
retries (int): 请求重试次数
|
||
proxy_options (bool): 是否使用代理
|
||
default_timeout (int): 默认超时时间
|
||
default_cookies (dict): 默认的 cookies 字典
|
||
"""
|
||
self.base_url = base_url.rstrip('/')
|
||
self.protocol = protocol
|
||
self.retries = retries
|
||
self.default_timeout = default_timeout
|
||
self.session = requests.Session()
|
||
|
||
if proxy_options:
|
||
self.session.proxies = {"http": "http://127.0.0.1:7890", "https": "http://127.0.0.1:7890"}
|
||
|
||
# 优先使用传入的 default_cookies 更新 session
|
||
if default_cookies:
|
||
self.session.cookies.update(default_cookies)
|
||
|
||
# 根据 base_url 生成 cookies 文件名，将 '.' 替换为 '_'
|
||
self.cookie_file = f"{self.base_url.replace('.', '_')}_cookies.json"
|
||
# 尝试加载本地已保存的 cookies 文件
|
||
try:
|
||
with open(self.cookie_file, 'r', encoding='utf-8') as f:
|
||
loaded_cookies = json.load(f)
|
||
self.session.cookies.update(loaded_cookies)
|
||
logging.info("成功加载本地 cookies")
|
||
except FileNotFoundError:
|
||
logging.info("本地 cookies 文件不存在，将在请求后自动保存")
|
||
except Exception as e:
|
||
logging.error("加载本地 cookies 失败：" + str(e))
|
||
|
||
def _save_cookies(self):
|
||
"""
|
||
将当前 session 中的 cookies 保存到本地文件（基于 base_url 的文件名），以 JSON 格式存储。
|
||
"""
|
||
try:
|
||
with open(self.cookie_file, 'w', encoding='utf-8') as f:
|
||
json.dump(self.session.cookies.get_dict(), f, ensure_ascii=False, indent=4)
|
||
logging.info("cookies 已保存到本地文件：" + self.cookie_file)
|
||
except Exception as e:
|
||
logging.error("保存 cookies 文件失败：" + str(e))
|
||
|
||
def _build_url(self, url):
|
||
if url.startswith("http://") or url.startswith("https://"):
|
||
return url
|
||
return f"{self.protocol}://{self.base_url}/{url.lstrip('/')}"
|
||
|
||
def set_default_headers(self, headers):
|
||
self.session.headers.update(headers)
|
||
|
||
def set_default_cookies(self, cookies):
|
||
self.session.cookies.update(cookies)
|
||
self._save_cookies()
|
||
|
||
def get(self, url, params=None, headers=None, cookies=None, **kwargs):
|
||
full_url = self._build_url(url)
|
||
return self._request("GET", full_url, params=params, headers=headers, cookies=cookies, **kwargs)
|
||
|
||
def post(self, url, data=None, json=None, headers=None, cookies=None, **kwargs):
|
||
full_url = self._build_url(url)
|
||
return self._request("POST", full_url, data=data, json=json, headers=headers, cookies=cookies, **kwargs)
|
||
|
||
def update(self, url, data=None, json=None, headers=None, cookies=None, **kwargs):
|
||
full_url = self._build_url(url)
|
||
return self._request("PUT", full_url, data=data, json=json, headers=headers, cookies=cookies, **kwargs)
|
||
|
||
def delete(self, url, params=None, headers=None, cookies=None, **kwargs):
|
||
full_url = self._build_url(url)
|
||
return self._request("DELETE", full_url, params=params, headers=headers, cookies=cookies, **kwargs)
|
||
|
||
def _request(self, method, url, retries=None, autosave=False, **kwargs):
|
||
if retries is None:
|
||
retries = self.retries
|
||
if 'timeout' not in kwargs:
|
||
kwargs['timeout'] = self.default_timeout
|
||
|
||
try:
|
||
response = self.session.request(method, url, **kwargs)
|
||
response.raise_for_status()
|
||
self.session.cookies.update(response.cookies)
|
||
|
||
if 'Set-Cookie' in response.headers:
|
||
from http.cookies import SimpleCookie
|
||
sc = SimpleCookie()
|
||
sc.load(response.headers['Set-Cookie'])
|
||
for key, morsel in sc.items():
|
||
if morsel.value.lower() != 'deleted':
|
||
self.session.cookies.set(key, morsel.value)
|
||
|
||
if autosave:
|
||
self._save_cookies()
|
||
|
||
response.__class__ = ExtendedResponse
|
||
return response
|
||
|
||
except Exception as e:
|
||
if retries > 0:
|
||
logging.warning(f"请求 {method} {url} 失败，剩余重试次数 {retries}，错误: {e}")
|
||
time.sleep(2 ** (self.retries - retries))
|
||
return self._request(method, url, retries=retries - 1, autosave=autosave, **kwargs)
|
||
else:
|
||
logging.error(f"请求 {method} {url} 重试次数用尽")
|
||
raise e
|
||
|
||
def get_cookies(self):
|
||
try:
|
||
return self.session.cookies.get_dict()
|
||
except Exception as e:
|
||
raise ValueError("获取 cookies 失败：" + str(e))
|
||
|
||
|
||
class MR(MyRequests):
|
||
pass
|