211 lines
7.0 KiB
Python
211 lines
7.0 KiB
Python
# -*- 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 = {}
|
|
# <class 'dict'>: {'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) |