main.py
#!/usr/bin/env python#-*- coding:UTF-8 -*-（添加）from sre_constants import SUCCESSfrom unittest import resultfrom flask import Flask, requestfrom service.servers import Serversfrom flask import Flask, request, jsonifyapp = Flask(__name__)config={"yuanyuzhou":{"secret":"12345678","wx_webhook":"https://qyapi./cgi-bin/webhook/send?key=123456","branch":"dev,master",#允许的分支"dir":"/www/wwwroot/dev/,/www/wwwroot/master/"#分支所对应的目录}}#项目一@app.route('/yuanyuzhou', methods=['POST'])def post_data():server= Servers()result=server.git(config['yuanyuzhou'])if result==200:return jsonify({"status": result})if result==401:return "token认证无效", resultif __name__ == '__main__':app.run(port=5000)
servers.py
#!/usr/bin/env python#-*- coding:UTF-8 -*-（添加）import requestsimport osimport jsonimport timeimport subprocessfrom flask import Flask, requestclass Servers(object):def mkdir(self,path):path=path.strip()path=path.rstrip("\\")isExists=os.path.exists(path)if not isExists:os.makedirs(path) return Trueelse:return Falsedef mklog(self,body):t=time.strftime('%Y%m%d')dirName="log/"+tself.mkdir(dirName)name=dirName+"/"+time.strftime('%Y%m%d%H:%M:%S')+"_request_log.txt"f=open(name,'a')f.write(body)f.close()def git(self,config):data = request.datadata = json.loads(data)# 认证签名是否有效signature = request.headers.get('X-Gitee-Token', '')if signature != config['secret']:return 401# 运行shell脚本，更新代码checkBranch=self.checkBranch(data['ref'])branch=str.split(config['branch'],",")path=str.split(config['dir'],",")realBranch=""realPath=""gitBranch=Falsefor i in range(len(branch)) :if(checkBranch==branch[i]):realBranch=branch[i]realPath=path[i]gitBranch=Truebreak# if gitBranch:#res = subprocess.Popen('cd '+realPath+' && git pull origin '+realBranch+'',stdout=subprocess.PIPE,shell=True)#success=False#result=""#for line in res.stdout.readlines():# f="@"+line.decode("utf-8").strip()# if f.find("Updating")>0:# success=True# result+=line.decode("utf-8").strip()+"\n"#server= Servers()#server.mklog(result)#self.wxWebHook(data,success,config)if gitBranch:command='cd '+realPath+' && git pull origin '+realBranch+''#output = os.popen('cd '+realPath+' && git pull origin '+realBranch+'','r',1)cmdstring=subprocess.getstatusoutput(command)[1]outresult="cmdstring{0}".format(cmdstring)success=Trueif outresult.find("error")>0:success=Falseserver= Servers()server.mklog(outresult)self.wxWebHook(data,success,config,mark)return 200#企业微信 webhookdef wxWebHook(self,data,status,config):if(status):status="<font color=\"#26f443\">成功</font>"else:status="<font color=\"#f51832\">失败</font>"gjData={"msgtype": "markdown","markdown": {"content": "实例<font color=\"#1f97eb\">"+data['project']['name']+"</font>\n"+">构建人: <font color=\"#000000\">"+data['user_name']+"</font>\n"+">构建时间: <font color=\"#000000\">"+data['project']['created_at']+"</font>\n"+">构建状态: "+status+"\n"+">构建分支: <font color=\"#000000\">"+data['ref']+"</font>"#refs/heads/dev}}data = json.dumps(gjData)headers = {'Content-Type':'application/json'}rep = requests.post(url=config['wx_webhook'], data=data, headers=headers)return rep.text#检测分支 def checkBranch(self,ref):arr=str.split(ref,"/")return arr[2]
源码包：gitee+企业微信自动构建git/download/zhang804633234/85177501
