Charles导入功能实现 代码说明

需要用的的包：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))