需要用的的包：haralyzer 主要目的：解析har文件
下载包命令：pip3 install haralyzer
from haralyzer import HarParser, HarPage
from urllib.parse import urlparse, urlunparse
import json
import logging
import requests
import re
import os
logger = logging.getLogger('django')
class CharlesAnalysis:
@staticmethod
def charlesAnalysis(harFile, rjgc, token):
"""
:param harFile: charles文件地址
:param rjgc: 环境区分
:param token: token值
:return:
"""
"""
解析har文件
"""
with open(harFile, 'r') as f:
har_parser = HarParser(json.loads(f.read()))
charles = []
for entry in har_parser.har_data['entries']:
"""
过滤har文件 筛选出有效请求
"""
if entry['request']['method'] == "OPTIONS":
pass
else:
# 正则提取域名
pattern = r"(https?://[^/]+)/.*"
match = re.match(pattern, entry['request']['url'])
if match:
domain_name = str(match.group(1)) + "/"
logger.info(domain_name)
# 获取请求url
url = entry['request']['url'].split(domain_name)[1].split("?")[0]
logger.info("请求的url：" + str(url))
# 获取 YaPiURL
yaPiUrl = url.split("/", 1)[1]
logger.info("yaPiUrl：" + str(yaPiUrl))
# 获取请求方式
method = entry['request']['method']
logger.info("请求方式：" + str(method))
# 请求头解析
new_headers = {}
headers = entry['request']['headers']
for j in headers:
if ":" in j['name']:
pass
else:
new_headers[j['name']] = j['value']
# 获取请求头
logger.info("请求头：" + str(new_headers))
# 变更请求头中 cookie的值
new_headers.update({"cookie": rjgc + "=" + token})
json_headers = json.dumps(new_headers)
logger.info("转换cookie：" + str(new_headers))
# 解析har文件 整体请求数据
logger.info("整体请求数据：" + str(entry['request']))
newQueryString = {}
postData = "{}"
if 'postData' in entry['request']:
postData = entry['request']['postData']['text']
if "queryString" in entry['request']:
if entry['request']['queryString'] == []:
queryString = entry['request']['queryString']
logger.info("请求参数queryString：" + str(queryString))
else:
newQueryString = entry['request']['queryString']
logger.info("请求参数queryString：" + str(newQueryString))
else:
pass
logger.info("请求参数postData：" + str(postData))
else:
newQueryString = entry['request']['queryString']
logger.info("请求参数queryString：" + str(newQueryString))
# 根据平台定义的规则 转换请求方式：1:get 2.post 3:put 4:delete
api_method = ""
if method == "GET":
api_method = 1
elif method == "POST":
api_method = 2
elif method == "PUT":
api_method = 3
elif method == "DELETE":
api_method = 4
# 质量平台json传参必填，需要判断 如空则 赋值 {}
if postData == "":
postData = "{}"
else:
pass
# 解析所有相关数据后，进行组合拼接 获取需要的数据
charles_dict = {
"yaPiUrl": yaPiUrl,
"url": url,
"method": api_method,
"header": json_headers,
"postData": postData,
"queryString": newQueryString,
"domain_name": domain_name
}
charles.append(charles_dict)
logger.info("解析后的参数：" + str(charles))
return charles
@staticmethod
def yaPiDirectory(yaPi_url):
"""
yaPi_url-搜索yapi的url
"""
# 判断运行环境，生产环境：SERVER ，对应的URL值是YAPI请求的域名
ENV_PROFILE = os.getenv("ENV")
if ENV_PROFILE == "SERVER":
url = 'xxxx'
else:
url = 'xxxx'
yaPiBody = {"email": "xxxxxx", "password": "xxxxx"}
header = {
"Content-Type": "application/json"
}
# YaPi登录接口
yaPi = requests.post(url="http://" + url + ":3000/api/user/login", json=yaPiBody)
cookie = yaPi.cookies
# yapi搜索接口
# yaPi_url = "/api/operate/brand/list"
# YaPi接口数据-搜索
yapi_api_search = requests.get(url="http://" + url + ":3000/api/project/search?q=" + yaPi_url, headers=header, cookies=cookie).json()['data']['interface'][0]
# yapi 获取接口详细信息
yapi_interface = requests.get(url="http://" + url + ":3000/api/interface/get?id=" + str(yapi_api_search['_id']), headers=header, cookies=cookie).json()
# 二级目录名称
yapi_directory_two = requests.get(url="http://" + url + ":3000/api/interface/list_menu?project_id=" + str(yapi_interface['data']['project_id']), headers=header, cookies=cookie).json()
# 获取二级目录名称
for directory_two in yapi_directory_two['data']:
if directory_two['_id'] == yapi_interface['data']['catid']:
directory_two_yapi = directory_two['name']
break
else:
pass
# 一级目录名称
yapi_directory_one = ""
# 获取组id
yapi_group_id = requests.get(url="http://" + url + ":3000/api/project/get?id=" + str(yapi_api_search['projectId']), headers=header, cookies=cookie).json()['data']['group_id']
# yapi 项目信息-获取一级目录名称
yapi_project = requests.get(url="http://" + url + ":3000/api/project/list?group_id=" + str(yapi_group_id) + "&page=1&limit=100", headers=header, cookies=cookie).json()
# 获取 一级目录名称
for project in yapi_project['data']['list']:
if project['_id'] == yapi_api_search['projectId']:
yapi_directory_one = project['name']
break
else:
pass
yaPiUrl = "http://" + url + ":3000/project/" + str(yapi_api_search['projectId']) + "/interface/api/" + str(yapi_api_search['_id'])
# 获取的一、二、三级目录名称以及 该接口在yapi中的url访问地址，数据组合
api_directory = {
"yapi_directory_one": yapi_directory_one,
"yapi_directory_two": directory_two_yapi,
"yapi_directory_three": yapi_interface['data']['title'],
"yaPiApiUrl": yaPiUrl,
}
return api_directory
if __name__ == '__main__':
domain = "https://xxxxxx"
harfile = "/Users/xxxxxx"
a = CharlesAnalysis()
print(a.charlesAnalysis(harfile, "${token}", "${jgc}"))
yaPi_url = "/api/xxxxx"
print(a.yaPiDirectory(yaPi_url))