forked from Trustie-Study-Group/PRWelBot4SECourse
72 lines
2.2 KiB
Python
72 lines
2.2 KiB
Python
#!/usr/bin/python
|
||
import sys
|
||
sys.path.append('..')
|
||
from flask import Flask, jsonify, request
|
||
from flask_cors import CORS, cross_origin
|
||
import json
|
||
import services.pullService as pullService
|
||
import datetime
|
||
from commons.logUtil import logger
|
||
|
||
app = Flask(__name__)
|
||
|
||
# 测试接口
|
||
@app.route('/hello', methods=['get'])
|
||
@cross_origin()
|
||
def hello():
|
||
# 返回JSON格式的问候语
|
||
return jsonify({
|
||
"code": 0,
|
||
"msg": "hello"
|
||
})
|
||
|
||
# 功能：处理新建或关闭的PR（Pull Request）相关的Webhook请求
|
||
@app.route('/prwelcome', methods=['POST'])
|
||
@cross_origin()
|
||
def pr_welcome():
|
||
try:
|
||
# 获取请求的JSON数据
|
||
payload = request.json
|
||
print(payload)
|
||
# 记录日志
|
||
logger.info("获取webhook信息成功:" + str(payload))
|
||
|
||
# 如果是新建的PR
|
||
if payload["action"] == 'opened':
|
||
# 提取PR的相关信息
|
||
pull_id = payload["pull_request"]["id"]
|
||
index = payload["pull_request"]["number"]
|
||
time = payload["pull_request"]["created_at"][:19]
|
||
owner = payload["pull_request"]["user"]["login"]
|
||
repo = payload["repository"]["full_name"]
|
||
created = datetime.datetime.strptime(time, "%Y-%m-%dT%H:%M:%S")
|
||
check_time = created + datetime.timedelta(hours=2)
|
||
values = (pull_id, index, owner, repo, created, check_time, 0)
|
||
# 将提取的信息存储到数据库
|
||
pullService.insert_pull(values)
|
||
|
||
# 如果PR被关闭
|
||
elif payload["action"] == 'closed':
|
||
# 提取PR的相关信息
|
||
index = payload["pull_request"]["number"]
|
||
owner = payload["pull_request"]["user"]["login"]
|
||
repo = payload["repository"]["full_name"]
|
||
# 更新数据库中的PR状态
|
||
pullService.update_pull(index, owner, repo)
|
||
|
||
# 返回成功的响应
|
||
return jsonify({
|
||
"code": 0,
|
||
"msg": "success"
|
||
})
|
||
except Exception as e:
|
||
# 返回异常响应
|
||
return jsonify({
|
||
"code": -1,
|
||
"msg": "exception:" + str(e)
|
||
})
|
||
|
||
if __name__ == '__main__':
|
||
# 启动Flask应用
|
||
app.run(port=3090)
|