def get_sonar(): try: global prod_name url="http://xXXX:9000/api/measures/component?component={}&metricKeys=new_bugs,bugs,new_coverage,coverage," "new_vulnerabilities,vulnerabilities,new_duplicated_lines_density,duplicated_lines_density".format(prod_name) res=requests.get(url) message = json.loads(res.text) message_tmp = message['component']['measures'] messages = [{i['metric']: i.get('value',i.get('periods',[{'value':None}])[0]['value'])} for i in message_tmp] results={} for i in messages: results.update(i) print results message = 'bug:{},新增bug:{},漏洞:{},新增漏洞:{},覆盖率:{:.2f}, 重复率:{:.2f},新代码重复率:{:.2f},状态:OK'.format(results['bugs'],results['new_bugs'], results['vulnerabilities'],results['new_vulnerabilities'], float(results['coverage']), float(results['duplicated_lines_density']), float(results['new_duplicated_lines_density'])) print message return message except Exception as e: return None
获取信息并发钉钉
#coding=utf8 import requests import json import os import time import sys reload(sys) sys.setdefaultencoding('utf8') from requests.auth import HTTPBasicAuth import ssl ssl._create_default_https_context = ssl._create_unverified_context requests.packages.urllib3.disable_warnings() time.sleep( 100 ) prod_name=os.environ.get('prod_name','') server_ip='XXXX:9000' def send_msg(): global prod_name token='' url = 'http://oapi.dingtalk.com/robot/send?access_token='+token headers = { 'Content-Type': 'application/json;charset=utf-8' } while 1: try: # 6.5 if 'XXXX' in server_ip: msg=requests.get(url="http://XXXX:8082/?projectname="+prod_name) # 7.9 if 'XXXX' in server_ip: msg = requests.get(url="http://{}/api/measures/component?component={}&metricKeys=" "new_bugs,bugs,new_coverage,coverage,new_vulnerabilities," "vulnerabilities,new_duplicated_lines_density,duplicated_lines_density".format(server_ip,prod_name) ) message=msg.text print message msg_url="http://{}/dashboard?id={}".format(server_ip,prod_name) except Exception as e: print e pass if '10.1.63.97' in server_ip and '漏洞' in str(message): break if '10.1.63.96' in server_ip : try: message=json.loads(message) if message['component']['measures']: message_tmp= message['component']['measures'] messages=["i['metric']:i['value'] " for i in message_tmp] message=','.join(messages) break except Exception as e: print e pass print message title=prod_name+"代码质量报告" formdata = { "msgtype": "markdown", "markdown": { "title":prod_name+"代码质量报告", "text": '## {} '.format(title)+str(message)+' [查看报告]({})'.format(msg_url) } } #钉钉发报告 #res = requests.post(url=url, data=json.dumps(formdata), headers=headers, proxies={"http": "http://XXXX:8005"},verify=False) #print(res.text) print '代码扫描完毕' send_msg()