产研效率提升 - 工具篇 - 消息中心

 #!/usr/bin/env python# --coding:utf-8--
from http.server import BaseHTTPRequestHandler, HTTPServerfrom os import pathimport jsonfrom urllib import request, parse

APP_ID = "cli_XXXX"APP_SECRET = "XXXX"APP_VERIFICATION_TOKEN = "XXXX"
class RequestHandler(BaseHTTPRequestHandler):    def do_POST(self):        # 解析请求 body        req_body = self.rfile.read(int(self.headers['content-length']))        obj = json.loads(req_body.decode("utf-8"))        print(req_body)
        # 校验 verification token 是否匹配，token 不匹配说明该回调并非来自开发平台        token = obj.get("token", "")        if token != APP_VERIFICATION_TOKEN:            print("verification token not match, token =", token)            self.response("")            return
        # 根据 type 处理不同类型事件        type = obj.get("type", "")        if "url_verification" == type:  # 验证请求 URL 是否有效            self.handle_request_url_verify(obj)        elif "event_callback" == type:  # 事件回调            # 获取事件内容和类型，并进行相应处理，此处只关注给机器人推送的消息事件            event = obj.get("event")            if event.get("type", "") == "message":                self.handle_message(event)                return        return
    def handle_request_url_verify(self, post_obj):        # 原样返回 challenge 字段内容        challenge = post_obj.get("challenge", "")        rsp = {'challenge': challenge}        self.response(json.dumps(rsp))        return
    def handle_message(self, event):        # 此处只处理 text 类型消息，其他类型消息忽略        msg_type = event.get("msg_type", "")        if msg_type != "text":            print("unknown msg_type =", msg_type)            self.response("")            return
        # 调用发消息 API 之前，先要获取 API 调用凭证：tenant_access_token        access_token = self.get_tenant_access_token()        if access_token == "":            self.response("")            return
        # 机器人 echo 收到的消息        self.send_message(access_token, event.get("open_id"), event.get("text"))        self.response("")        return
    def response(self, body):        self.send_response(200)        self.send_header('Content-Type', 'application/json')        self.end_headers()        self.wfile.write(body.encode())
    def get_tenant_access_token(self):        url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal/"        headers = {            "Content-Type" : "application/json"        }        req_body = {            "app_id": APP_ID,            "app_secret": APP_SECRET        }
        data = bytes(json.dumps(req_body), encoding='utf8')        req = request.Request(url=url, data=data, headers=headers, method='POST')        try:            response = request.urlopen(req)        except Exception as e:            print(e.read().decode())            return ""
        rsp_body = response.read().decode('utf-8')        rsp_dict = json.loads(rsp_body)        code = rsp_dict.get("code", -1)        if code != 0:            print("get tenant_access_token error, code =", code)            return ""        return rsp_dict.get("tenant_access_token", "")
    def send_message(self, token, open_id, text):        url = "https://open.feishu.cn/open-apis/message/v4/send/"
        headers = {            "Content-Type": "application/json",            "Authorization": "Bearer " + token        }        req_body = {            "open_id": open_id,            "msg_type": "text",            "content": {                "text": text            }        }
        data = bytes(json.dumps(req_body), encoding='utf8')        req = request.Request(url=url, data=data, headers=headers, method='POST')        try:            response = request.urlopen(req)        except Exception as e:            print(e.read().decode())            return
        rsp_body = response.read().decode('utf-8')        rsp_dict = json.loads(rsp_body)        code = rsp_dict.get("code", -1)        if code != 0:            print("send message error, code = ", code, ", msg =", rsp_dict.get("msg", ""))
def run():    port = 8000    server_address = ('', port)    httpd = HTTPServer(server_address, RequestHandler)    print("start.....")    httpd.serve_forever()
if __name__ == '__main__':    run()

# -*- coding: utf-8 -*-# @Auth : hanxu# @Time : 2021-06import osimport loggingimport datetimeimport jsonimport requestsfrom flask import Flask, requestimport pymysql
app = Flask(__name__)

@app.before_requestdef before_request():    try:        #可以做参数校验        url = request.full_path        method = request.method        headers = request.headers        body = request.json        print(url)        print(method)        print(headers)        print(body)    except Exception as e:        logging.info(e)

@app.route('/info/notice/create', methods=['POST'])def info_create():    """    消息通知    :return:    """    json_param = request.json    bug_info = json_param['events']    bug_create = json_param['events']['bug::create']    new = bug_create['new']    product_name = new['custom_field_6']    bug_id = bug_create['id']    created_user = bug_create['user']    if exists_key(new, 'current_owner'):        current_owner = new['current_owner']    else:        current_owner = 'hanxu;'    if exists_key(new, 'title'):        title = new['title']    else:        title = '无标题'    if exists_key(new, 'de'):        developer = new['de']    else:        developer = ''
    if exists_key(new, 'te'):        tester = new['te']    else:        tester = ''    if exists_key(new, 'severity'):        severity = new['severity']    else:        severity = ''
    logging.info(        "产品名称{};缺陷ID{};标题:{};严重等级{};处理人{};创建人{};开发人{};测试人{}".format(product_name, bug_id, title, severity,                                                                    current_owner, created_user,                                                                    developer, tester))    chart_id='' #如果不需要在群中发送消息就换成用户open_id即可    params={        "receive_id": "%s" % chart_id,        "msg_type": "post",        "content": "{\"zh_cn\":{\"title\":\"bug单创建通知\",\"content\":[[{\"tag\":\"text\",\"text\":\"缺陷ID:\"},{\"tag\":\"text\",\"text\":\"%s\"}],[{\"tag\":\"text\",\"text\":\"缺陷标题:\"},{\"tag\":\"text\",\"text\":\"%s\"}],[{\"tag\":\"text\",\"text\":\"严重等级:\"},{\"tag\":\"text\",\"text\":\"%s\"}],[{\"tag\":\"text\",\"text\":\"处理人:\"}%s],[{\"tag\":\"text\",\"text\":\"创建人:\"},{\"tag\":\"text\",\"text\":\"%s\"}],[{\"tag\":\"text\",\"text\":\"开发者:\"},{\"tag\":\"text\",\"text\":\"%s\"}],[{\"tag\":\"text\",\"text\":\"测试者:\"},{\"tag\":\"text\",\"text\":\"%s\"}],[{\"tag\":\"text\",\"text\":\"bug单链接 :\"},{\"tag\":\"a\",\"href\":\"https://www.tapd.cn/44232473/bugtrace/bugs/view?bug_id=%s\",\"text\":\"链接\"}]]}}" % (            bug_id[-7:], title, severity, receiver_tag, created_user, developer, tester, bug_id)    }    create_res = send_message_to_user_at_chart(get_app_access_token(),params)    logging.info(f"bug单创建结果:{create_res}")    return "200"

def exists_key(bug, key):    """    判断key值是否存在    :param bug:    :param key:    :return:    """    return key in bug.keys()

def log_ininit():    """    初始化日志    :return:    """    log_path = os.popen('pwd').read().strip()  # 日志文件路径    logging.basicConfig(level=logging.INFO,                        format='%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s',                        filename='%s/app-%s.log' % (log_path, datetime.datetime.now().strftime('%Y-%m-%d')),                        filemode='w')
    logger = logging.getLogger(__name__)    sh = logging.StreamHandler()  # 往屏幕上输出    # sh.setFormatter('%(asctime)s %(filename)s[line:%(lineno)d] %(levelname)s %(message)s')  # 设置屏幕上显示的格式    logger.setLevel(logging.INFO)    logger.addHandler(sh)

def get_app_access_token():    """    获取 app_access_token    :return:    """    url = "https://open.feishu.cn/open-apis/auth/v3/app_access_token/internal"    param = json.dumps({        "app_id": "cli_xxx",        "app_secret": "xxxxxxxx"    })    header = {        "Content-Type": 'application/json; charset=utf-8'    }    r = requests.post(url, data=param, headers=header)    return "Bearer {}".format(r.json()['app_access_token'])

def send_message_to_user_at_chart(access_token, params):    """    在群里给某个人发送消息    :param access_token: token    :param params: 参数    :return:    """    url = "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id" #这里可根据tupen换成open_id    header = {        "Content-Type": 'application/json',        "Authorization": '%s' % access_token    }    r = requests.post(url, data=json.dumps(params), headers=header)    return r.text
if __name__ == "__main__":    log_ininit()    app.run(host="0.0.0.0", port=8000, debug=True)

{    "auto_task_id": "1020355972000001849",      "auto_task_action_id": "1020355972000001749",      "workspace_id": 21000072,    "events": {        "story::status_change": {            "workspace_id": 21000072,            "user": "lincoln",            "object_type": "story",            "id": "1021000072854809635",            "timestamp": 1595388790,            "new": {                "completed": "2020-07-22 11:33:10",                "id": "1021000072854809635",                "status": "resolved"            },            "old": {                "completed": null,                "id": "1021000072854809635",                "status": "developing"            },            "event_key": "story::status_change",            "from": "web"        }    },    "actions": [        "timer_trigger_search":{                "story":[                    1020358496854838221,                    1020358496854835273,                    1020358496854832911                ]            },     ],    "condition": null}