#!/usr/bin/env python # -*- encoding: utf-8 -*- """ @Version: 1.0 @Python Version:3.6.6 @Author: ludq1 @Email: ludq1@chinaunicom.cn @date: 2023/04/07 11:40:00 @Description: """ import json import re import requests from .globalconst import HttpConst from .my_baseexception import MyBaseException, create_base_exception from .my_stringutils import MyStringUtils from .my_utils import MyUtils class HttpTrans: """ http传输处理 """ # 定义要透传的header_key _header_key = [ 'Cookie' , 'x-request-id' , 'x-b3-traceid' , 'x-b3-spanid' , 'x-b3-parentspanid' , 'x-b3-sampled' , 'x-b3-flags' , 'x-ot-span-context' ] def __init__(self, http_header: dict = None): r""" Args: http_header: dict http的请求头 """ if http_header is not None: try: # 尝试将 http_header 转成 dict 并赋值 self._header = dict(http_header) except: raise ValueError('http_header 不能转换为字典') else: self._header = None def generate_header(self, content_type: str = None, dict_header: dict = None) -> dict: """ 生成http头信息 Args: content_type: string 传输类型 dict_header: dict 额外头信息 Returns: dict """ # 检查参数 if not (dict_header is None or isinstance(dict_header, dict)): raise ValueError('dict_header 不是字典') ret_header = {} # 加入需要透传的 self._header if self._header: # 转换为标准的字符串kv字典 formatted_dict = MyUtils.format_dict_to_str_dict(self._header) # 过滤出指定的key filtered_keys = filter(lambda tmp_k: tmp_k in self._header_key, formatted_dict.keys()) # 生成过滤后的字典 filtered_dict = {tmp_k: formatted_dict.get(tmp_k) for tmp_k in filtered_keys} # 更新 ret_dict ret_header.update(filtered_dict) # 设置参数默认值-- str_content_type if not MyStringUtils.is_empty(content_type): str_content_type = str(content_type) else: str_content_type = HttpConst.CONTENT_TYPE_URLENCODED # 更新 ret_dict ret_header.update({ HttpConst.HEADER_KEY_CONTENT_TYPE: str_content_type }) # 加入额外的HEADER KEY if dict_header: formatted_dict = MyUtils.format_dict_to_str_dict(dict_header) # 更新 ret_dict ret_header.update(formatted_dict) # 移除 Request Method 这个key,因为该key存在会造成requests请求失效 ret_header.pop(HttpConst.HEADER_KEY_REQUEST_METHOD, None) return ret_header def get_data(self, urlstr: str, trans_data: dict = None, dict_header: dict = None) -> requests.Response: """ post 传输数据 Args: urlstr: trans_data: 要附加到 urlstr 上的 key-value 字典 , kv 会自动编码 dict_header: 传输时使用的header Returns: """ # 检查参数 if not (trans_data is None or isinstance(trans_data, dict)): raise ValueError('trans_data 不是字典') if trans_data: trans_data = MyUtils.format_dict_to_str_dict(trans_data) headers = self.generate_header( content_type=HttpConst.CONTENT_TYPE_URLENCODED, dict_header=dict_header ) cookies = None if HttpConst.HEADER_KEY_COOKIE in headers: cookies = headers.get(HttpConst.HEADER_KEY_COOKIE) return requests.get(url=urlstr, headers=headers, params=trans_data, cookies=cookies) def post_encodeddata_standard(self, urlstr: str, trans_data: dict = None, dict_header: dict = None) -> requests.Response: """ post 传输数据,返回 Response对象 Args: urlstr: string httpurl trans_data: 传输的内容 ,字典 , kv会自动编码 dict_header: 传输时使用的header Returns: """ if not (trans_data is None or isinstance(trans_data, dict)): raise ValueError('trans_data 不是字典') trans_data_tmp = None content_type = HttpConst.CONTENT_TYPE_URLENCODED if dict_header is not None and HttpConst.HEADER_KEY_CONTENT_TYPE in dict_header.keys(): content_type = dict_header.get(HttpConst.HEADER_KEY_CONTENT_TYPE) if HttpConst.CONTENT_TYPE_JSON == dict_header.get(HttpConst.HEADER_KEY_CONTENT_TYPE): trans_data_tmp = json.dumps(trans_data) if trans_data_tmp is None: trans_data_tmp = MyUtils.format_dict_to_str_dict(trans_data) headers = self.generate_header( content_type=content_type, dict_header=dict_header ) regexp = (r'^https://') match = re.search(regexp, urlstr.strip(), re.U) if (match is not None): return requests.post(url=urlstr, headers=headers, data=trans_data_tmp, verify=False) else: return requests.post(url=urlstr, headers=headers, data=trans_data_tmp) def post_encodedata(self, urlstr: str, trans_data: dict = None, dict_header: dict = None, **dict_arg) -> str: """ post 传输数据,返回rensponse内容字符串,传输异常时返回 标准错误的json字符串,所以该方法适用于标准的传输 Args: urlstr: string httpurl trans_data: 传输的内容 , 字典,kv会自动编码 dict_header: 传输时使用的header **dict_arg: 额外参数 Returns: 标准格式的 json字符串 """ # 以后额外扩展 # if dict_arg: # pass try: m_response = self.post_encodeddata_standard( urlstr=urlstr, trans_data=trans_data, dict_header=dict_header ) if not m_response.ok: raise create_base_exception( '访问url出错,访问地址:{0},返回码{1}'.format(urlstr, m_response.status_code) ) else: str_ret = m_response.text # 检查返回结果是不是有效的 jsonstr dict_ret = MyUtils.jsonstr2dict(str_ret) if dict_ret is None: raise create_base_exception( '返回的内容不是有效的json字符串:{}'.format(str_ret) ) return str_ret except requests.exceptions.RequestException as exception: return MyUtils.dict2jsonstr(MyBaseException.format_exception_to_standard_dict(exception)) except MyBaseException as exception: return MyUtils.dict2jsonstr(MyBaseException.format_exception_to_standard_dict(exception)) except Exception as exception: return MyUtils.dict2jsonstr(MyBaseException.format_exception_to_standard_dict(exception)) def patch_encodeddata_standard(self, urlstr: str, trans_data: dict = None, dict_header: dict = None) -> requests.Response: """ patch 传输数据,返回 Response对象 Args: urlstr: string httpurl trans_data: 传输的内容 ,字典 , kv会自动编码 dict_header: 传输时使用的header Returns: """ if not (trans_data is None or isinstance(trans_data, dict)): raise ValueError('trans_data 不是字典') trans_data_tmp = None content_type = HttpConst.CONTENT_TYPE_PATH_JSON if dict_header is not None and HttpConst.HEADER_KEY_CONTENT_TYPE in dict_header.keys(): content_type = dict_header.get(HttpConst.HEADER_KEY_CONTENT_TYPE) if HttpConst.CONTENT_TYPE_PATH_JSON == dict_header.get(HttpConst.HEADER_KEY_CONTENT_TYPE): trans_data_tmp = json.dumps(trans_data) if trans_data_tmp is None: trans_data_tmp = MyUtils.format_dict_to_str_dict(trans_data) headers = self.generate_header( content_type=content_type, dict_header=dict_header ) return requests.patch(url=urlstr, headers=headers, data=trans_data_tmp)