教程介紹:

該教程主要校驗(yàn)思路是根據(jù)news_id獲取新聞和新聞主題數(shù)據(jù),調(diào)用接口獲取企業(yè)數(shù)據(jù)(實(shí)際結(jié)果)
根據(jù)news_id獲取csv中符合條件的數(shù)據(jù),獲取企業(yè)數(shù)據(jù)(預(yù)期結(jié)果),兩者比較,輸出不相等的數(shù)據(jù)(excel輸出,日志輸出)

注意! 公司項(xiàng)目不同,僅供參考,重在思路理解

執(zhí)行腳本 test_match_company.py

# -*- coding: utf-8 -*-

import os, pytest, json
import pandas as pd
from jsonpath import jsonpath
import requests
from util.data.xlwt_tool import write_excel
from util.report.logger_tool import Logger


class TestMatchCompany():
    real_path = os.path.split(os.path.realpath(__file__))[0]
    news_path = real_path + os.sep + "../data/news_content.txt"
    news = pd.read_csv(news_path, sep='^')
    news1 = news.values.tolist()

    def setup_class(self):
        algo_name = "match_company"
        # host = "IP地址"
        host = "ip地址"
        port = "端口號(hào)"
        # match_company_url
        self.url = "http://" + host + ':' + str(port) + '/algorithm/process/' + algo_name
        real_path = os.path.split(os.path.realpath(__file__))[0]
        labels_path = real_path + os.sep + "../data/news_company_label.csv"
        self.labels_expected = pd.read_csv(labels_path, sep=';')
        # self.baseline_cols = ['chinesename','emotionindicator','stockcode']
        self.baseline_cols = ['stockcode', 'companyid']
        self.logging = Logger('../logs/{}.log'.format(algo_name), level='error')
        # self.logging = Logger('../logs/{}.log'.format(algo_name), level='info')

    def test_company_labels(self):
        excel_data_list = []
        for i in self.news1:
            # news_id = str(i[0])
            news_id = i[0]
            title = i[1]
            content = i[2]
            # 獲取match_company算法接口的結(jié)果數(shù)據(jù)
            parameter1 = {'title': title, 'content': content, 'extr_method': 1,
                          'use_skip': 0}  # extr_method默認(rèn)值為0，需要設(shè)為1調(diào)用，且沒有計(jì)劃修改默認(rèn)值
            payload1 = {'parameter': json.dumps(parameter1, ensure_ascii=False)}
            response1 = requests.post(self.url, data=payload1).json()
            json_data = jsonpath(response1,"$..com")
            shijijieguo = []
            for i in json_data[0]:
                code = jsonpath(i,"$..code")
                comcode = jsonpath(i,"$..comcode")
                shijijieguo.append([code[0],comcode[0]])
            shijijieguo.sort()

            # 根據(jù)news_id獲取csv文件中符合條件的數(shù)據(jù)
            expected = self.labels_expected
            baseline_result = expected[expected['newsid'] == news_id][self.baseline_cols]
            data_expect = baseline_result.values.tolist()
            for i in range(len(data_expect)):
                if data_expect[i][0] == 'csf':
                    data_expect[i][0] = ''
            data_expect.sort()

            # 判斷各種異常情況
            if data_expect == [] and shijijieguo == []:
                # excel_data_list.append(["csv和接口都沒查到:"+str(news_id), str(data_expect), str(shijijieguo)])
                self.logging.debug(
                    "接口和csv文件都沒有查到企業(yè)新聞數(shù)據(jù),不做對(duì)比 news_id:{} 預(yù)期結(jié)果為空:{} 接口實(shí)際結(jié)果為空:{}".format(news_id, data_expect, shijijieguo))

            elif data_expect == [] and shijijieguo != []:
                excel_data_list.append([str(news_id), str(data_expect), str(shijijieguo)])
                self.logging.error("對(duì)比不一致 news_id:{} 預(yù)期結(jié)果為空:{} 接口實(shí)際結(jié)果不為空:{}".format(news_id, data_expect, shijijieguo))

            elif data_expect != [] and shijijieguo == []:
                excel_data_list.append([str(news_id), str(data_expect), str(shijijieguo)])
                self.logging.error("對(duì)比不一致 news_id:{} 預(yù)期結(jié)果為空:{} 接口實(shí)際結(jié)果不為空:{}".format(news_id, data_expect, shijijieguo))

            elif data_expect != [] and shijijieguo != []:
                if data_expect != shijijieguo:
                    excel_data_list.append([str(news_id), str(data_expect), str(shijijieguo)])
                    self.logging.error("對(duì)比不一致 news_id:{} 預(yù)期結(jié)果:{} 實(shí)際結(jié)果:{}".format(news_id, data_expect, shijijieguo))
                else:
                    # excel_data_list.append(["一致:"+str(news_id), str(data_expect), str(shijijieguo)])
                    self.logging.info("對(duì)比一致 news_id:{} 預(yù)期結(jié)果:{} 實(shí)際結(jié)果:{}".format(news_id, data_expect, shijijieguo))
        # 將日志輸出到excel
        write_excel("../logs/match_company_error_{}.xls".format(len(excel_data_list)), ['news_id', '預(yù)期結(jié)果', '實(shí)際結(jié)果'], excel_data_list)

結(jié)果輸出工具 write_excel.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Author : Leiyh
# @File : pandas_tool.py
import xlwt


def write_excel(file_name,data_title,data_list,encoding='utf-8'):
    '''

    :param file_name: 文件路徑地址
    :param data_title: excel第一行的標(biāo)題欄
    :param data_list: 二維數(shù)據(jù)列表
    :param encoding:
    :return:
    '''
    # 創(chuàng)建workbook和sheet對(duì)象 注意Workbook的開頭W要大寫
    workbook = xlwt.Workbook(encoding=encoding)
    # 添加一個(gè)名為sheet1的表
    sheet1 = workbook.add_sheet('sheet1', cell_overwrite_ok=True)

    # 向表頭寫入數(shù)據(jù)
    for i in range(len(data_title)):
        sheet1.write(0, i, str(data_title[i]))

    # 向sheet寫入數(shù)據(jù)
    for i in range(len(data_list)):
        for j in range(len(data_title)):
            sheet1.write(i + 1, j, str(data_list[i][j]))

    # 保存數(shù)據(jù)到‘Workbook2.xls’文件中
    workbook.save(file_name)
    print('創(chuàng)建execel完成！')

if __name__ == '__main__':
    # data = get_test_case("C:/softwareData/PycharmProjects/s00-wuling/documents/user/注冊(cè)接口sign_up.xlsx")
    # print(data[0])
    # print(data[1])
    # write_excel("match_company.xls", ['news_id', '預(yù)期結(jié)果', '實(shí)際結(jié)果'],
    #             [['35942860', str(['', 'ICN5025197980', 1]), str(['', 'ICN5025197980', 1])]])
    write_excel("match_company.xls", ['news_id', '預(yù)期結(jié)果', '實(shí)際結(jié)果'],
                [['35942860', "['', 'ICN5025197980', 1]", "['', 'ICN5025197980', 1]"]])

日志輸出工具 logger_tool.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Author : Leiyh
# @File : logger_tool.py
import logging


class Logger(object):
    level_relations = {
        'notset':logging.NOTSET,
        'debug': logging.DEBUG,
        'info': logging.INFO,
        'warning': logging.WARNING,
        'error': logging.ERROR,
        'crit': logging.CRITICAL
    }

    def __init__(self, filename, filemode='w', level='notset',
                 format='%(asctime)s - %(levelname)s: %(message)s'):
        logger = logging.getLogger()
        logger.setLevel(level=self.level_relations.get(level))
        filehandle = logging.FileHandler(filename,filemode)
        formatter = logging.Formatter(format)
        filehandle.setFormatter(formatter)
        logger.addHandler(filehandle)


    def debug(self, msg):
        logging.debug(msg)

    def info(self, msg):
        logging.info(msg)

    def warning(self, msg):
        logging.warning(msg)

    def error(self, msg):
        logging.error(msg)

    def critical(self, msg):
        logging.critical(msg)


if __name__ == '__main__':
    logger = Logger('all.log', level='info')
    logger.info("Start print log")
    logger.debug("Do something")
    logger.warning("Something maybe fail")
    logger.error("error print log")