需求背景：
目前業(yè)務(wù)線上問(wèn)題登記在tapd的項(xiàng)目中般渡，客服提單后，各產(chǎn)研無(wú)法感知到有新工單進(jìn)來(lái)。為了提高產(chǎn)研對(duì)線上問(wèn)題的感知度馏艾、提高工單收斂速度。開發(fā)了實(shí)時(shí)機(jī)器人每5分鐘拉取數(shù)據(jù)推送到產(chǎn)研群里奴愉，開發(fā)了超時(shí)機(jī)器人每日下班前推送還未解決的工單琅摩。
技術(shù)棧：
python +mysql+tapd+企微機(jī)器人+jenkins
實(shí)現(xiàn)思路：
一、讀取tapd數(shù)據(jù)
-
1：調(diào)用tapd官方api,直接讀取項(xiàng)目下的各類信息
https://www.tapd.cn/help/show#1120003271001000086
tapd-api示例 2：利用方法1讀取后定時(shí)寫入mysql數(shù)據(jù)庫(kù)-本次使用方法2
二锭硼、組裝企微機(jī)器人消息體
參考企微官方-群機(jī)器人配置api
https://developer.work.weixin.qq.com/document/path/91770
群機(jī)器人
三房资、企微群中新建機(jī)器人，拿到webhookurl
新建群機(jī)器人
四檀头、python腳本調(diào)用企微機(jī)器人發(fā)送接口推送消息
超時(shí)機(jī)器人效果圖
實(shí)時(shí)機(jī)器人效果圖
五轰异、持續(xù)集成到j(luò)enkins定時(shí)執(zhí)行推送
調(diào)用click模塊封裝好岖沛，編寫shell腳本調(diào)用，jenkins中構(gòu)建添加shell腳本就行搭独。
實(shí)時(shí)機(jī)器人示例：python xx/tapd_remind_robot.py -rtu https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=x -rtn 請(qǐng)?zhí)幚韝x人抓緊處理新工單婴削！ -t xx產(chǎn)品中心 -g xx
完整代碼：
-初學(xué)python寫的很臭很長(zhǎng)勿噴
#!/usr/bin/python
# -*- coding: UTF-8 -*-
# @Time : 2021/5/14 17:06
# @Author : yanfa
# @File : tapd_remind_robot.py
# TAPD工單統(tǒng)計(jì)機(jī)器人-超時(shí)、實(shí)時(shí)
import logging
import click
import pymysql
from sshtunnel import SSHTunnelForwarder
import requests
import json
# ssh
SSH_HOST = "xxx"
SSH_PORT = 22
SSH_USERNAME = "xxxx"
SSH_PASSWORD = "xxxx"
REMOTE_BIND_HOST = "xxx"
REMOTE_BIND_PORT = xxxx
# mysql
MYSQL_HOST = "127.0.0.1"
MYSQL_PORT = ""
MYSQL_USER = "xxx"
MYSQL_PASSWD = "xxx"
MYSQL_DB = "xxx" # tadp主庫(kù)
MYSQL_DB_01 = "xxx" # 獲取用戶userid
MYSQL_CHARSET = "utf8"
# 定義工單鏈接前半部分
project_path = "https://www.tapd.cn/xxx/bugtrace/bugs/view?bug_id="
class Tapd_Remind_Robot():
# 初始化必傳的url
def __init__(self, url):
self.url = url
# SSH跳板機(jī)鏈接MySQL數(shù)據(jù)庫(kù)-tadp主庫(kù)
def connect_mysql(self):
server = SSHTunnelForwarder(ssh_address_or_host=(SSH_HOST, SSH_PORT),
ssh_username=SSH_USERNAME,
ssh_password=SSH_PASSWORD,
remote_bind_address=(REMOTE_BIND_HOST, REMOTE_BIND_PORT))
server.start()
conn = pymysql.connect(host='127.0.0.1',
port=server.local_bind_port,
user=MYSQL_USER,
passwd=MYSQL_PASSWD,
db=MYSQL_DB,
charset=MYSQL_CHARSET)
return conn
# SSH跳板機(jī)鏈接MySQL數(shù)據(jù)庫(kù)-獲取用戶userid
def connect_mysql01(self):
server = SSHTunnelForwarder(ssh_address_or_host=(SSH_HOST, SSH_PORT),
ssh_username=SSH_USERNAME,
ssh_password=SSH_PASSWORD,
remote_bind_address=(REMOTE_BIND_HOST, REMOTE_BIND_PORT))
server.start()
conn = pymysql.connect(host='127.0.0.1',
port=server.local_bind_port,
user=MYSQL_USER,
passwd=MYSQL_PASSWD,
db=MYSQL_DB_01,
charset=MYSQL_CHARSET)
return conn
# 執(zhí)行SQL命令-tapd庫(kù)
def execute_sql(self, sql):
db = self.connect_mysql()
cursor = db.cursor()
try:
cursor.execute(sql)
results = cursor.fetchall()
return results
except:
print("Error")
db.close()
# 執(zhí)行SQL命令-db_user_center庫(kù)
def execute_sql01(self, sql):
db = self.connect_mysql01()
cursor = db.cursor()
try:
cursor.execute(sql)
results = cursor.fetchall()
return results
except:
print("Error")
db.close()
# 推送消息
def post_bot(self, url, data):
headers = {'Content-Type': 'application/json'}
response = requests.post(url=url, headers=headers, data=json.dumps(data))
# 后續(xù)可能有用
return response
# 定義tapd_name轉(zhuǎn)化為userid方法
def find_userid(self, pending_list_01):
# 不以;結(jié)尾特殊的元素末尾添加;
list0 = []
for j in pending_list_01:
if j.endswith(';'):
list0.append(j)
else:
j = j + ';'
list0.append(j)
# print("處理不以；結(jié)尾的元素后：", list0)
# 第一次去重
pending_list = list(set(list0))
# print("第一次去重后的列表為：", pending_list)
# 定義一個(gè)字符串存儲(chǔ)拼接的用戶信息
list1 = ''
for i in pending_list:
list1 += i
# print("拼接用戶信息后的列表：", list1)
# 去除字符最后的字符"；"
list2 = list1.strip(';')
# print("字符串去除尾部標(biāo)點(diǎn)為：", list2)
# 切割
list3 = list2.split(";")
# print("字符串分割后：", list3)
# 第二次去重
list4 = list(set(list3))
# print("第二次去重后的列表為：", list4)
# 強(qiáng)轉(zhuǎn)成元祖后再?gòu)?qiáng)轉(zhuǎn)為str拼接在sql
list5 = str(tuple(list4))
# print("列表轉(zhuǎn)為元祖再轉(zhuǎn)為str：",list5)
# 特殊處理元祖只有單個(gè)元素的','
if len(list4) == 1:
list6 = []
for i in list5:
list6.append(i)
# print("遍歷字符串：", list6)
# 刪除倒數(shù)第二個(gè)元素','
del list6[-2]
# print('處理嗓蘑，后：', list6)
# 重新合并
list7 = "".join(list6)
# print('重新合并：', str(list7))
# 拼接sql
userid_list = "SELECT a.userid FROM t_admin_user a where a.tapd_name in" + list7 + ";"
print("查詢userid的sql為：", userid_list)
else:
# 拼接sql
userid_list = "SELECT a.userid FROM t_admin_user a where a.tapd_name in" + list5 + ";"
print("查詢userid的sql為：", userid_list)
userid_list01 = self.execute_sql01(userid_list)
# print("處理人userid元祖為：", userid_list01)
# 定義一個(gè)列表存儲(chǔ)遍歷元祖的值
list8 = []
for e in userid_list01:
list8.append(e[0]) # 只查詢子元祖第一個(gè)值userid
print('最終處理人列表為：', list8)
return list8
# 定義發(fā)送消息主方法
def send_msg_time_out(self, timeout_notice, title, group_new):
# 定義查詢語(yǔ)句-有截止時(shí)間，取【當(dāng)前時(shí)間 - 截止時(shí)間】>=0 算超時(shí)
# 定義被刪除的部分
deleted = """('xxx')"""
sql01 = "SELECT count(0) FROM(SELECT *, TIMESTAMPDIFF( DAY, DATE_FORMAT( deadline, '%Y-%m-%d'), CURDATE() ) AS date FROM t_tapd_bug_order ) AS a WHERE a.module = '技術(shù)問(wèn)題' AND SUBSTR(a.id,1,12)=114249027600 AND a.`status` IN ( 'in_progress', 'unconfirmed') AND SUBSTR(a.custom_field_two,1,6) = " + group_new + " AND a.deadline !='0000-00-00 00:00:00' AND a.id NOT IN " + deleted + " AND a.date >= 0;"
sql02 = "SELECT count(0) FROM(SELECT *, TIMESTAMPDIFF( DAY, DATE_FORMAT( deadline, '%Y-%m-%d'), CURDATE() ) AS date FROM t_tapd_bug_order ) AS a WHERE a.module = '技術(shù)問(wèn)題' AND SUBSTR(a.id,1,12)=xxxAND a.`status` IN ( 'in_progress', 'unconfirmed') AND SUBSTR(a.custom_field_two,1,6) = " + group_new + " AND a.deadline !='0000-00-00 00:00:00' AND a.id NOT IN " + deleted + " AND a.date >=3 AND a.date <5;"
sql03 = "SELECT count(0) FROM(SELECT *, TIMESTAMPDIFF( DAY, DATE_FORMAT( deadline, '%Y-%m-%d'), CURDATE() ) AS date FROM t_tapd_bug_order ) AS a WHERE a.module = '技術(shù)問(wèn)題' AND SUBSTR(a.id,1,12)=xxxAND a.`status` IN ( 'in_progress', 'unconfirmed') AND SUBSTR(a.custom_field_two,1,6) = " + group_new + " AND a.deadline !='0000-00-00 00:00:00' AND a.id NOT IN " + deleted + " AND a.date >= 5 AND a.date <7;"
sql04 = "SELECT count(0) FROM(SELECT *, TIMESTAMPDIFF( DAY, DATE_FORMAT( deadline, '%Y-%m-%d'), CURDATE() ) AS date FROM t_tapd_bug_order ) AS a WHERE a.module = '技術(shù)問(wèn)題' AND SUBSTR(a.id,1,12)=xxxAND a.`status` IN ( 'in_progress', 'unconfirmed') AND SUBSTR(a.custom_field_two,1,6) = " + group_new + " AND a.deadline !='0000-00-00 00:00:00' AND a.id NOT IN " + deleted + " AND a.date >= 7;"
# 定義查詢語(yǔ)句-無(wú)截止時(shí)間虫溜，取【當(dāng)前時(shí)間 - 創(chuàng)建時(shí)間】>=3 算超時(shí)
sql05 = "SELECT count(0) FROM(SELECT *, TIMESTAMPDIFF( DAY, DATE_FORMAT( created, '%Y-%m-%d'), CURDATE() ) AS date FROM t_tapd_bug_order ) AS a WHERE a.module = '技術(shù)問(wèn)題' AND SUBSTR(a.id,1,12)=xxx AND a.`status` IN ( 'in_progress', 'unconfirmed') AND SUBSTR(a.custom_field_two,1,6) = " + group_new + " AND a.deadline ='0000-00-00 00:00:00' AND a.id NOT IN " + deleted + " AND a.date >= 3 AND a.date <5;"
sql06 = "SELECT count(0) FROM(SELECT *, TIMESTAMPDIFF( DAY, DATE_FORMAT( created, '%Y-%m-%d'), CURDATE() ) AS date FROM t_tapd_bug_order ) AS a WHERE a.module = '技術(shù)問(wèn)題' AND SUBSTR(a.id,1,12)=114249027600 AND a.`status` IN ( 'in_progress', 'unconfirmed') AND SUBSTR(a.custom_field_two,1,6) = " + group_new + " AND a.deadline ='0000-00-00 00:00:00' AND a.id NOT IN " + deleted + " AND a.date >= 5 AND a.date <7;"
sql07 = "SELECT count(0) FROM(SELECT *, TIMESTAMPDIFF( DAY, DATE_FORMAT( created, '%Y-%m-%d'), CURDATE() ) AS date FROM t_tapd_bug_order ) AS a WHERE a.module = '技術(shù)問(wèn)題' AND SUBSTR(a.id,1,12)=114249027600 AND a.`status` IN ( 'in_progress', 'unconfirmed') AND SUBSTR(a.custom_field_two,1,6) = " + group_new + " AND a.deadline ='0000-00-00 00:00:00' AND a.id NOT IN " + deleted + " AND a.date >= 7;"
sql08 = "SELECT count(0) FROM(SELECT *, TIMESTAMPDIFF( DAY, DATE_FORMAT( created, '%Y-%m-%d'), CURDATE() ) AS date FROM t_tapd_bug_order ) AS a WHERE a.module = '技術(shù)問(wèn)題' AND SUBSTR(a.id,1,12)=114249027600 AND a.`status` IN ( 'in_progress', 'unconfirmed') AND SUBSTR(a.custom_field_two,1,6) = " + group_new + " AND a.deadline ='0000-00-00 00:00:00' AND a.id NOT IN " + deleted + " AND a.date >= 3;"
sql09 = "SELECT * FROM((SELECT a.id, a.title AS '工單標(biāo)題', a.created AS '創(chuàng)建時(shí)間', a.current_owner AS '處理人', a.custom_field_two AS '所屬小組', a.date AS '超時(shí)天數(shù)' FROM (SELECT *, TIMESTAMPDIFF( DAY, DATE_FORMAT( deadline, '%Y-%m-%d'), CURDATE()) AS date FROM t_tapd_bug_order ) AS a WHERE a.module = '技術(shù)問(wèn)題' AND SUBSTR(a.id,1,12)=114249027600 AND a.`status` IN ( 'in_progress', 'unconfirmed') AND SUBSTR(a.custom_field_two,1,6) = " + group_new + "AND a.deadline != '0000-00-00 00:00:00' AND a.id NOT IN('xxx','xx','xx') AND a.date >= 0 ) UNION (SELECT a.id, a.title AS '工單標(biāo)題', a.created AS '創(chuàng)建時(shí)間', a.current_owner AS '處理人', a.custom_field_two AS '所屬小組', a.date AS '超時(shí)天數(shù)' FROM (SELECT *, TIMESTAMPDIFF( DAY, DATE_FORMAT( created, '%Y-%m-%d' ), CURDATE( ) ) AS date FROM t_tapd_bug_order ) AS a WHERE a.module = '技術(shù)問(wèn)題' AND SUBSTR(a.id,1,12)=114249027600 AND a.`status` IN ( 'in_progress', 'unconfirmed') AND SUBSTR(a.custom_field_two,1,6) = " + group_new + " AND a.deadline = '0000-00-00 00:00:00' AND a.id NOT IN " + deleted + " AND a.date >= 3 )) AS e ORDER BY e.`超時(shí)天數(shù)` desc;"
# print("sql01為：%s\nsql02為：%s\nsql03為：%s\nsql04為：%s\nsql05為：%s\nsql06為：%s\nsql06為：%s\nsql08為：%s\nsql09為：%s"%(sql01,sql02,sql03,sql04,sql05,sql06,sql07,sql08,sql09))
# 超時(shí)定義：
# 如果填寫了截止時(shí)間或者截止時(shí)間不等于00: 00:00 - ---則取當(dāng)前時(shí)間 - 截止時(shí)間 >= 0算超時(shí)
# 否則沒(méi)填寫截止時(shí)間，則用當(dāng)前時(shí)間 - 創(chuàng)建時(shí)間 >= 3才算超時(shí)
# 獲取超時(shí)工單總數(shù)/各天數(shù)股缸，數(shù)據(jù)庫(kù)返回一個(gè)元祖且元素也是元祖衡楞，通過(guò)下標(biāo)取2層值
# 執(zhí)行后相加處理
bug_count = self.execute_sql(sql01)[0][0] + self.execute_sql(sql08)[0][0]
bug_03 = self.execute_sql(sql02)[0][0] + self.execute_sql(sql05)[0][0]
bug_05 = self.execute_sql(sql03)[0][0] + self.execute_sql(sql06)[0][0]
bug_07 = self.execute_sql(sql04)[0][0] + self.execute_sql(sql07)[0][0]
bug_all = self.execute_sql(sql09)
print("全部超時(shí)個(gè)數(shù)為：", bug_count)
# print("超時(shí)3天個(gè)數(shù)為：", bug_03)
# print("超時(shí)5天個(gè)數(shù)為：", bug_05)
# print("超時(shí)7天個(gè)數(shù)為：", bug_07)
# print("sql09為：",sql09)
# print("bug_all為：", bug_all)
# 推送的報(bào)文 %d格式化數(shù)值，%s格式化字符敦姻，多個(gè)值用%(a,b,c)
# 定義匯總部分
head = """>【%s】\n超時(shí)工單數(shù)量匯總：<font color = "warning">%d單</font>\n>超時(shí)三天： <font color = "comment">%d單</font>\n>超時(shí)五天： <font color = "comment">%d單</font>\n>超時(shí)七天： <font color = "comment">%d單</font>""" % (
title, bug_count, bug_03, bug_05, bug_07)
# 定義詳細(xì)清單-壓縮前
body = ''
pending_list_01 = []
for i in bug_all:
# 將處理人遍歷存儲(chǔ)至新列表pending_list_01
pending_list_01.append(i[3])
body01 = """\n\n>工單標(biāo)題：<font color="comment">%s</font>\n>工單鏈接：[%s%s](%s%s)\n>創(chuàng)建時(shí)間：<font color="comment">%s</font>\n>處理人：<font color="comment">%s</font>\n>所屬小組：<font color="comment">%s</font>\n>超時(shí)天數(shù)：<font color="warning">%s天</font>""" % (
i[1], project_path, i[0], project_path, i[0], i[2], i[3], i[4], i[5])
# 前后加空格換行瘾境，+=遍歷疊加
body += '\n\n' + body01 + '\n\n'
# print("處理人列表為：", pending_list_01)
# 定義請(qǐng)求報(bào)文，將2部分字符串拼接
markdown = {
"msgtype": "markdown",
"markdown": {
"content": head + body
}
}
# 定義詳細(xì)清單-壓縮后(因?yàn)榻涌赾ontent字節(jié)數(shù)不能大于4096镰惦，所以需要壓縮)
body_short = ''
pending_list_01 = []
for i in bug_all:
# 將處理人遍歷存儲(chǔ)至新列表pending_list_01
pending_list_01.append(i[3])
body_short_01 = """\n\n>工單標(biāo)題：<font color="comment">%s</font>\n>處理人：<font color="comment">%s</font>\n>超時(shí)天數(shù)：<font color="warning">%s天</font>""" % (
i[1], i[3], i[5])
# 前后加空格換行寄雀，+=遍歷疊加
body_short += '\n\n' + body_short_01 + '\n\n'
# print("處理人列表為：", pending_list_01)
# 定義請(qǐng)求報(bào)文，將2部分字符串拼接
markdown_short = {
"msgtype": "markdown",
"markdown": {
"content": head + body_short
}
}
# 獲取content字節(jié)數(shù),注意采用 UTF-8 編碼方式陨献，先
# 使用 encode() 方法盒犹，將字符串進(jìn)行編碼后再獲取它的字節(jié)數(shù)
count_byte = len((head + body).encode())
short_count_byte = len((head + body_short).encode())
print("壓縮前content字節(jié)數(shù)為：", count_byte)
print("壓縮后content字節(jié)數(shù)為：", short_count_byte)
# 添加判斷，超時(shí)工單數(shù)為0且大于4096字節(jié)則不執(zhí)行
if bug_count != 0:
if count_byte <= 4096:
# print("超時(shí)工單匯總明細(xì)為：",markdown)
resp01 = self.post_bot(url=self.url, data=markdown)
# print("超時(shí)工單匯總明細(xì)-響應(yīng)為：", resp01.text)
assert resp01.text.__contains__("""errmsg":"ok""")
print("超時(shí)工單匯總明細(xì)-發(fā)送成功")
# 定義@處理人的消息
notice = {
"msgtype": "text",
"text": {
"content": timeout_notice,
"mentioned_list": self.find_userid(pending_list_01) # 調(diào)用用戶userid轉(zhuǎn)化方法
}
}
resp02 = self.post_bot(url=self.url, data=notice)
print("通知處理人消息為：", notice)
# print("通知處理人消息-響應(yīng)為", resp02.text)
assert resp02.text.__contains__("""errmsg":"ok""")
print("通知處理人-發(fā)送成功")
elif short_count_byte <= 4096:
# print("超時(shí)工單匯總明細(xì)為：", markdown_short)
resp01 = self.post_bot(self.url, markdown_short)
# print("超時(shí)工單匯總明細(xì)-響應(yīng)為：", resp01.text)
assert resp01.text.__contains__("""errmsg":"ok""")
print("超時(shí)工單匯總明細(xì)-發(fā)送成功")
# 定義@處理人的消息
notice = {
"msgtype": "text",
"text": {
"content": timeout_notice,
"mentioned_list": self.find_userid(pending_list_01) # 調(diào)用用戶userid轉(zhuǎn)化方法
}
}
print("通知處理人消息為：", notice)
resp02 = self.post_bot(self.url, notice)
# print("通知處理人消息-響應(yīng)為：", resp02.text)
assert resp02.text.__contains__("""errmsg":"ok""")
print("通知處理人-發(fā)送成功")
else:
# print("壓縮前-超時(shí)工單匯總明細(xì)為：",markdown)
# print("壓縮后-超時(shí)工單匯總明細(xì)為：", markdown_short)
head = """>\n【%s】\n超時(shí)工單數(shù)量匯總：<font color = "warning">%d單</font>\n>超時(shí)三天： <font color = "comment">%d單</font>\n>超時(shí)五天： <font color = "comment">%d單</font>\n>超時(shí)七天： <font color = "comment">%d單</font>\n\n""" % (
title, bug_count, bug_03, bug_05, bug_07)
body_short = ''
top = '\n【超時(shí)工單Top10】\n'
pending_list_01 = []
# 壓縮后還是大于4098則只取top10
for i in bug_all[:10]:
# 將處理人遍歷存儲(chǔ)至新列表pending_list_01
pending_list_01.append(i[3])
body_short_01 = """>工單標(biāo)題：<font color="comment">%s</font>\n>處理人：<font color="comment">%s</font>\n>超時(shí)天數(shù)：<font color="warning">%s天</font>""" % (
i[1], i[3], i[5])
# 前后加空格換行眨业，+=遍歷疊加
body_short += '\n\n' + body_short_01 + '\n\n'
print("前10處理人列表為：", pending_list_01)
short_count_byte = len((head + top + body_short).encode())
print('壓縮后前10條長(zhǎng)度為：', short_count_byte)
# 定義請(qǐng)求報(bào)文急膀，將2部分字符串拼接
markdown_short = {
"msgtype": "markdown",
"markdown": {
"content": head + top + body_short
}
}
# 定義@處理人的消息
notice = {
"msgtype": "text",
"text": {
"content": timeout_notice,
"mentioned_list": self.find_userid(pending_list_01) # 調(diào)用用戶userid轉(zhuǎn)化方法
}
}
resp01 = self.post_bot(self.url, markdown_short)
# print("超時(shí)工單匯總明細(xì)-響應(yīng)為：", resp01.text)
assert resp01.text.__contains__("""errmsg":"ok""")
print("超時(shí)工單匯總明細(xì)-發(fā)送成功")
print("通知處理人消息為：", notice)
resp02 = self.post_bot(self.url, notice)
# print("通知處理人消息-響應(yīng)為：", resp02.text)
assert resp02.text.__contains__("""errmsg":"ok""")
print("通知處理人-發(fā)送成功")
# logging.error("字節(jié)數(shù)為%s，markdown內(nèi)容最長(zhǎng)不超過(guò)4096個(gè)字節(jié)" % short_count_byte)
else:
logging.error("暫無(wú)超時(shí)記錄")
def send_msg_real_time(self, realtime_notice, title, group_new):
# 定義查詢語(yǔ)句
realtime_count_sql = "SELECT COUNT(0) from(SELECT min(b.created),b.bug_id FROM t_tapd_bug_order a, t_tapd_bug_order_change b WHERE a.id = b.bug_id AND b.bug_id IN(SELECT b.bug_id FROM t_tapd_bug_order_change b WHERE b.created >= DATE_SUB( now(), INTERVAL 15 MINUTE ) AND SUBSTR( b.bug_id, 1, 12 ) = xxx AND b.old_value = 'feedback' AND b.new_value = 'unconfirmed') AND DATE_FORMAT( a.created, '%Y-%m-%d' ) = date_sub( curdate( ), INTERVAL 0 DAY ) AND a.module = '技術(shù)問(wèn)題' AND SUBSTR( a.id, 1, 12 ) = xxx AND a.`status` = 'unconfirmed' AND SUBSTR(a.custom_field_two,1,6) = " + group_new + " GROUP BY b.bug_id ORDER BY b.bug_id) AS c;"
realtime_list_sql = "SELECT a.id, a.title AS '工單標(biāo)題', a.custom_field_four AS '店鋪', a.bugtype AS '類型', a.`status` AS '英文狀態(tài)',( CASE WHEN a.`status` = 'feedback' THEN '待核實(shí)' WHEN a.`status` = 'acknowledged' THEN '已核實(shí)' WHEN a.`status` = 'unconfirmed' THEN '待評(píng)估' END) AS '狀態(tài)', a.current_owner AS '處理人', a.custom_field_two AS '所屬小組' FROM t_tapd_bug_order a, t_tapd_bug_order_change b WHERE a.id = b.bug_id AND b.created IN ( SELECT min( b.created ) FROM t_tapd_bug_order a, t_tapd_bug_order_change b WHERE a.id = b.bug_id AND b.bug_id IN ( SELECT DISTINCT b.bug_id FROM t_tapd_bug_order_change b WHERE b.created >= DATE_SUB( now( ), INTERVAL 15 MINUTE ) AND SUBSTR( b.bug_id, 1, 12 ) = xxx AND b.old_value = 'feedback' AND b.new_value = 'unconfirmed' ) AND DATE_FORMAT( a.created, '%Y-%m-%d' ) = date_sub( curdate( ), INTERVAL 0 DAY ) AND a.module = '技術(shù)問(wèn)題' AND SUBSTR( a.id, 1, 12 ) = xxx AND a.`status` = 'unconfirmed' AND SUBSTR(a.custom_field_two,1,6) = " + group_new + " GROUP BY b.bug_id ORDER BY b.bug_id );"
print("第一個(gè)sql為：%s\n第二個(gè)sql為：%s" % (realtime_count_sql, realtime_list_sql))
# 執(zhí)行查詢語(yǔ)句
realtime_bug_count = self.execute_sql(realtime_count_sql)[0][0]
realtime_bug_detail = self.execute_sql(realtime_list_sql)
print("新增工單數(shù)為：", realtime_bug_count)
# 推送的報(bào)文 %d格式化數(shù)值龄捡，%s格式化字符卓嫂，多個(gè)值用%(a,b,c)
# 定義匯總部分
head = """>【%s】\n當(dāng)前新增技術(shù)工單數(shù)：<font color = "warning">%d單</font>""" % (title, realtime_bug_count)
# 定義詳細(xì)清單-壓縮前
body = ''
pending_list_01 = []
for i in realtime_bug_detail:
pending_list_01.append(i[6])
# 鏈接需要[鏈接名稱](http://www....)
body01 = """\n\n>工單標(biāo)題：<font color="comment">%s</font>\n>工單鏈接：[%s%s](%s%s)\n>店鋪：<font color="comment">%s</font>\n>類型：<font color="comment">%s</font>\n>狀態(tài)：<font color="comment">%s</font>\n>處理人：<font color="comment">%s</font>\n>所屬小組：<font color="comment">%s</font>""" % (
i[1], project_path, i[0], project_path, i[0], i[2], i[3], i[5], i[6], i[7])
# 前后加空格換行，+=遍歷疊加
body += '\n\n' + body01 + '\n\n'
# print("處理人列表為：", pending_list_01)
# 定義請(qǐng)求報(bào)文聘殖，將2部分字符串拼接
markdown = {
"msgtype": "markdown",
"markdown": {
"content": head + body
}
}
# 定義詳細(xì)清單-壓縮后
body_short = ''
pending_list_01 = []
for i in realtime_bug_detail:
pending_list_01.append(i[6])
# 鏈接需要[鏈接名稱](http://www....)
body_short_01 = """\n\n>工單標(biāo)題：<font color="comment">%s</font>\n>工單鏈接：[%s%s](%s%s)\n>處理人：<font color="comment">%s</font>""" % (
i[1], project_path, i[0], project_path, i[0], i[6])
# 前后加空格換行晨雳，+=遍歷疊加
body_short += '\n\n' + body_short_01 + '\n\n'
markdown_short = {
"msgtype": "markdown",
"markdown": {
"content": head + body_short
}
}
# 獲取content字節(jié)數(shù),注意采用 UTF-8 編碼方式，先
# 使用 encode() 方法奸腺，將字符串進(jìn)行編碼后再獲取它的字節(jié)數(shù)
count_byte = len((head + body).encode())
short_count_byte = len((head + body_short).encode())
# print("content字節(jié)數(shù)為：", count_byte)
# 添加判斷餐禁，實(shí)時(shí)工單數(shù)為0且大于4096字節(jié)則不執(zhí)行
if realtime_bug_count != 0:
# 定義@處理人的消息
notice = {
"msgtype": "text",
"text": {
"content": realtime_notice,
"mentioned_list": self.find_userid(pending_list_01) # 調(diào)用用戶userid轉(zhuǎn)化方法
}
}
if count_byte <= 4096:
# print("新工單匯總明細(xì)為：",markdown)
resp01 = self.post_bot(self.url, markdown)
# print("新工單匯總明細(xì)-響應(yīng)為：", resp01.text)
assert resp01.text.__contains__("""errmsg":"ok""")
print("新工單匯總明細(xì)-發(fā)送成功")
print("通知處理人消息為：", notice)
resp02 = self.post_bot(self.url, notice)
# print("通知處理人消息-響應(yīng)為：", resp02.text)
assert resp02.text.__contains__("""errmsg":"ok""")
print("通知處理人-發(fā)送成功")
elif short_count_byte <= 4096:
# print("新工單匯總明細(xì)為：", markdown_short)
resp01 = self.post_bot(self.url, markdown_short)
# print("新工單匯總明細(xì)-響應(yīng)為：", resp01.text)
assert resp01.text.__contains__("""errmsg":"ok""")
print("新工單匯總明細(xì)-發(fā)送成功")
print("通知處理人消息為：", notice)
resp02 = self.post_bot(self.url, notice)
# print("通知處理人消息-響應(yīng)為：", resp02.text)
assert resp02.text.__contains__("""errmsg":"ok""")
print("通知處理人-發(fā)送成功")
else:
logging.error("字節(jié)數(shù)為%s，markdown內(nèi)容最長(zhǎng)不超過(guò)4096個(gè)字節(jié)" % short_count_byte)
else:
logging.error("暫無(wú)新記錄")
# 定義產(chǎn)研自動(dòng)轉(zhuǎn)化分組方法
def group_transform(self, group):
group_list=['xx','xx','xx']
if group in group_list:
group = "'" + group + "產(chǎn)品中心'"
else:
logging.error("請(qǐng)?zhí)顚懻_的產(chǎn)研名稱")
return group
@click.command()
@click.option('-rtu', '--realtime_url', required=False, help='定義實(shí)時(shí)工單機(jī)器人webhook,為url格式突照，如 https://qyapi.weixin.qq.com')
@click.option('-rtn', '--realtime_notice', required=False, default='請(qǐng)?zhí)幚砣俗ゾo處理新工單帮非！', type=str, help='定義實(shí)時(shí)工單處理人的提醒文案')
@click.option('-tou', '--timeout_url', required=False, help='定義超時(shí)工單機(jī)器人webhook,為url格式，如 https://qyapi.weixin.qq.com')
@click.option('-ton', '--timeout_notice', required=False, default='請(qǐng)?zhí)幚砣俗ゾo處理超時(shí)工單！', type=str, help='定義超時(shí)工單處理人的提醒文案')
@click.option('-t', '--title', required=False, default='數(shù)據(jù)產(chǎn)品中心', type=str, help='定義標(biāo)題末盔，如【數(shù)據(jù)產(chǎn)品中心】')
@click.option('-g', '--group', required=False, default='數(shù)據(jù)', type=str,
help="定義處理人所在產(chǎn)研,如基礎(chǔ)或者9筑舅，映射關(guān)系為：1-直播 2-營(yíng)銷 3-xxxx 4-xxx ")
# 定義機(jī)器人運(yùn)行方法
# 實(shí)時(shí)機(jī)器人示例：python test/py_interface_automation/tapd_remind_robot.py -rtu https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=x -rtn 請(qǐng)?zhí)幚韝x人抓緊處理新工單！ -t xx產(chǎn)品中心 -g xx
# 超時(shí)機(jī)器人示例：python test/py_interface_automation/tapd_remind_robot.py -tou https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx -ton 請(qǐng)?zhí)幚砣俗ゾo處理超時(shí)工單陨舱！-t xx產(chǎn)品中心 -g xx
def run_remind(timeout_url, realtime_url, timeout_notice='', realtime_notice='', title='', group=''):
# 添加判斷超時(shí)機(jī)器人url對(duì)象存在且實(shí)時(shí)機(jī)器人url對(duì)象不存在
if timeout_url is not None and realtime_url is None:
print("url為：%s\n提醒信息為：%s\n標(biāo)題為：%s\n處理分組前:%s" % (timeout_url, timeout_notice, title, group))
# 實(shí)例化
trr = Tapd_Remind_Robot(timeout_url)
# 調(diào)用發(fā)送超時(shí)工單提醒方法
group_new = trr.group_transform(group)
print("處理分組后：", group_new)
trr.send_msg_time_out(timeout_notice, title, group_new)
elif realtime_url is not None and timeout_url is None:
print("url為：%s\n提醒信息為：%s\n標(biāo)題為：%s\n處理分組前:%s" % (realtime_url, realtime_notice, title, group))
# 實(shí)例化
trr = Tapd_Remind_Robot(realtime_url)
# 調(diào)用發(fā)送實(shí)時(shí)工單提醒方法
group_new = trr.group_transform(group)
print("處理分組后：", group_new)
trr.send_msg_real_time(realtime_notice, title, group_new)
else:
logging.error("請(qǐng)錄入正確的url")
if __name__ == '__main__':
# 調(diào)用主運(yùn)行方法
run_remind()```
