# -*- coding:utf-8 -*- """ Author: qiaoxinjiu Email: xinjiu.qiao@allschool.com Create Date: 2022/04/26 5:58 下午 """ import os import sys input_team_name = sys.argv BASIC_PATH = os.path.dirname(os.path.abspath(__file__)) TEAM_PATH = os.path.abspath(os.path.join(BASIC_PATH, '../../../{}'.format("base_framework"))) sys.path.append(TEAM_PATH) PROJECT_PATH = os.path.abspath(os.path.join(BASIC_PATH, '../../..')) sys.path.append(PROJECT_PATH) from base_framework.public_tools.sqlhelper import MySqLHelper import json import requests obj_mysql_helper = MySqLHelper() class Handle_tools: def __init__(self): pass def query_interface_sql(self): ''' 查询qa的余量接口信息 :return: ''' query_sql = ''' SELECT in_url,controller_name FROM sparkatp.interface_info WHERE id IN ( SELECT a.id FROM (SELECT * from sparkatp.interface_info) a WHERE ( a.swagger_id in (SELECT id FROM sparkatp.swagger_info WHERE team="UBRD") ) AND a.created_time > "2022-01-01 00:00:50" AND is_used = 1 and at_numbers = 0 and offline=0 and jira_id is null) ORDER BY created_time''' query_sql_r = """SELECT si.id AS interface_id, rpgm.jira_id, si.in_url FROM ( SELECT ii.id, ii.in_url, ii.jira_id FROM sparkatp.interface_info ii WHERE ii.swagger_id IN ( SELECT id FROM sparkatp.swagger_info WHERE team = "UBRD" ) AND ii.created_time > "2022-01-01 00:00:50" AND ii.is_used = 1 AND ii.at_numbers = 0 AND ii.offline = 0 AND ii.jira_id IS NULL ) si INNER JOIN ( SELECT rp.interface_id, rp.jira_id, MAX( modified_time ) FROM request_parameters rp WHERE rp.jira_id IS NOT NULL AND rp.jira_id <> 'None' AND rp.jira_id <> '' GROUP BY rp.interface_id ) rpgm ON rpgm.interface_id = si.id;""" # query_sql = " SELECT * FROM sparkatp.interface_info WHERE id IN (SELECT a.id FROM(SELECT * from sparkatp.interface_info) a WHERE( a.swagger_id in (SELECT id FROM sparkatp.swagger_info WHERE team='UBRD') ) AND a.created_time > '2022-01-01 00:00:50' AND is_used = 1 and at_numbers = 0 and offline=0 and jira_id is null) ORDER BY created_time" query_sql_r_result = obj_mysql_helper.select_all(query_sql_r) query_sql_dict = {} query_sql_dict_temp = {} # : {'interface_id': 924646, 'jira_id': 'HHC-48875', 'in_url': 'http://peppa-parent-api.qa.huohua.cn/classes/audition/enter'} for qsrr in query_sql_r_result: in_url = "{}_{}".format(str(qsrr.get('interface_id')).strip(' '), qsrr.get('in_url').strip(' ')) if qsrr.get('jira_id') in query_sql_dict.keys(): if in_url in query_sql_dict.get(qsrr.get('jira_id')): pass else: query_sql_dict[qsrr.get('jira_id')].append(in_url) else: query_sql_dict[qsrr.get('jira_id')] = [in_url] query_sql_s = """SELECT si.id AS interface_id, rpgm.jira_id, si.in_url FROM ( SELECT ii.id, ii.in_url, ii.jira_id FROM sparkatp.interface_info ii WHERE ii.swagger_id IN ( SELECT id FROM sparkatp.swagger_info WHERE team = "UBRD" ) AND ii.created_time > "2022-01-01 00:00:50" AND ii.is_used = 1 AND ii.at_numbers = 0 AND ii.offline = 0 AND ii.jira_id IS NULL ) si INNER JOIN ( SELECT rp.interface_id, rp.jira_id, MAX( modified_time ) FROM response_parameters rp WHERE rp.jira_id IS NOT NULL AND rp.jira_id <> 'None' AND rp.jira_id <> '' GROUP BY rp.interface_id ) rpgm ON rpgm.interface_id = si.id;""" query_sql_s_result = obj_mysql_helper.select_all(query_sql_s) for qsrs in query_sql_s_result: in_url = "{}_{}".format(str(qsrs.get('interface_id')).strip(' '), qsrs.get('in_url').strip(' ')) if qsrs.get('jira_id') in query_sql_dict.keys(): if in_url in query_sql_dict[qsrs.get('jira_id')]: pass else: query_sql_dict[qsrs.get('jira_id')].append(in_url) else: query_sql_dict[qsrs.get('jira_id')] = [in_url] return query_sql_dict def send_log_info(self): ''' 每天进行日志提醒 :return: ''' feishu_name_id = {"罗志鹏": "7020366259502153730", "谯新久": "7020366258071715842", "陈洁": "7020370251997069314", "蒲思宇": "7076270364313108481", "张楠": "7076270369128349697", "刘涛婷": "7020366262240854017"} order_list = ["谯新久", "陈洁", "刘涛婷", "罗志鹏", "蒲思宇", "张楠"] # for name in order_list: def send_feishu(self, inferfaces): ''' 发送飞书 :return: ''' at_user_list = [{"tag": "at", "user_id": "{}".format("7020366259502153730")}] message_data = {"msg_type": "post", "content": { "post": {"zh_cn": {"title": "有存在的接口未实现自动化哦", "content": [[{"tag": "text", "text": "问题数据："}], [{"tag": "text", "text": "{}".format(inferfaces)}], at_user_list]}}}} web_hook = "https://open.feishu.cn/open-apis/bot/v2/hook/40696c86-264a-4222-a40c-cfd64a05dffd" headers = {"Content-Type": "application/json"} json_data = json.dumps(message_data) print(json_data) rsp = requests.post(url=web_hook, data=json_data, headers=headers) return rsp def send_message_by_feishu(self, web_hook, data): headers = {"Content-Type": "application/json"} json_data = json.dumps(data) rsp = requests.post(url=web_hook, data=json_data, headers=headers) return rsp def compare_txt(self, su_t, st_t): ''' 对比两个文档，返回不同的信息 :param su_t: :param st_t: :return: ''' file_object1 = open(su_t, 'r', encoding='utf-8') file_object2 = open(st_t, 'r', encoding='utf-8') try: while True: line = file_object1.readline() lines = file_object2.readline() if line.strip() != lines.strip(): print("pc :", line) print("right:", lines) else: continue finally: file_object1.close() file_object2.close() if __name__ == '__main__': test = Handle_tools() # get_interfaces = test.query_interface_sql() # if get_interfaces: # test.send_feishu(json.dumps(get_interfaces)) # else: # print("都已经完成") su_t = r"D:\UiVispark\spark1.txt" st_t = r"D:\UiVispark\spark2.txt" test.compare_txt(st_t=st_t,su_t=su_t)