feat(backend): 单据终止提醒 TencentBlueKing#9133

dbm-ui/backend/core/notify/handlers.py

@@ -9,7 +9,9 @@
 specific language governing permissions and limitations under the License.
 """
 import logging
+import re
 import textwrap
+from datetime import datetime, timedelta, timezone
 
 from celery import shared_task
 from django.utils.translation import ugettext as _
@@ -77,7 +79,7 @@ def get_msg_type(cls):
     def get_actions(msg_type, ticket):
         """获取bkchat操作按钮"""
         # TODO: 暂时去掉[待确认]按钮
-        if ticket.status not in [TicketStatus.APPROVE]:
+        if not ticket or ticket.status not in [TicketStatus.APPROVE]:
             return []
 
         todo = ticket.todo_of_ticket.filter(status=TodoStatus.TODO).first()
@@ -113,15 +115,14 @@ def get_title_color(phase):
         else:
             return "warning"
 
-    def render_title_content(self, msg_type, title, content, ticket, phase, receivers):
+    def render_title_content(self, msg_type, title, content, phase, receivers):
         """重新渲染标题和内容样式，bkchat有特定要求"""
         # title 要加上样式
-        title = _("「DBM」：您有{ticket_type}单据 「{ticket_id}」<font color='{color}'>{status}</font>").format(
-            ticket_type=TicketType.get_choice_label(ticket.ticket_type),
-            ticket_id=ticket.id,
-            status=TicketStatus.get_choice_label(phase),
-            color=self.get_title_color(phase),
-        )
+        pattern = r"(?P<title>「DBM」：.+「[0-9]+?」)(?P<msg>.+)"
+        replace = r"\g<title><font color='{}'>\g<msg></font>".format(self.get_title_color(phase))
+        title = re.sub(pattern, replace, title)
+        # 终止提醒(如果有)也需要加上样式
+        title = re.sub(r"(?P<msg>长期未处理.+)", r"<font color='red'>\g<msg></font>", title)
 
         # content要去掉点击详情，即最后一行，并且加上@通知人
         content = "\n".join(content.split("\n")[:-1])
@@ -133,7 +134,7 @@ def render_title_content(self, msg_type, title, content, ticket, phase, receiver
 
     def send_msg(self, msg_type, context):
         ticket, phase, receivers = context["ticket"], context["phase"], context["receivers"]
-        title, content = self.render_title_content(msg_type, self.title, self.content, ticket, phase, receivers)
+        title, content = self.render_title_content(msg_type, self.title, self.content, phase, receivers)
         ticket_operators = ticket.get_current_operators()
         approvers = list(dict.fromkeys(ticket_operators["operators"] + ticket_operators["helpers"]))
         msg_info = {
@@ -226,17 +227,15 @@ class NotifyAdapter:
 
     register_notify_class = [CmsiHandler, BkChatHandler]
 
-    def __init__(self, ticket_id: int, flow_id: int = None):
+    def __init__(self, ticket_id: int, deadline: int = None):
         """
         @param ticket_id: 单据ID
-        @param flow_id: 流程ID
         """
         # 初始化单据，流程信息
         try:
             self.ticket = Ticket.objects.get(id=ticket_id)
-            self.flow = Flow.objects.get(id=flow_id) if flow_id else self.ticket.current_flow()
         except (Ticket.DoesNotExist, Flow.DoesNotExist):
-            raise NotifyBaseException(_("无法初始化通知适配器，无法找到此单据{}或流程{}").format(ticket_id, flow_id))
+            raise NotifyBaseException(_("无法初始化通知适配器，无法找到此单据{}").format(ticket_id))
 
         # 当前阶段，对于运行中发通知的单据，实际上是【待继续】，这里做一次转换
         self.phase = TicketStatus.INNER_TODO if self.ticket.status == TicketStatus.RUNNING else self.ticket.status
@@ -245,6 +244,8 @@ def __init__(self, ticket_id: int, flow_id: int = None):
         self.bk_biz_id = self.ticket.bk_biz_id
         self.receivers = self.get_receivers()
         self.clusters = [cluster["immute_domain"] for cluster in self.ticket.details.pop("clusters", {}).values()]
+        # 单据终止时间，用于终止提醒
+        self.deadline = deadline
 
     @classmethod
     def get_support_msg_types(cls):
@@ -316,6 +317,13 @@ def render_msg_template(self, msg_type: str):
             "detail_address": self.ticket.url,
             "terminate_reason": self.ticket.get_terminate_reason(),
         }
+
+        # 如果有终止时间，说明是一个终止提醒
+        if self.deadline:
+            timeout = datetime.now(timezone.utc) + timedelta(hours=self.deadline)
+            timeout = timeout.astimezone().strftime("%Y-%m-%d %H:%M:%S%z")
+            title += _("\n长期未处理，{}小时后({})将超时终止").format(self.deadline, timeout)
+
         content = textwrap.dedent(template.render(payload))
         return title, content
 
@@ -352,6 +360,6 @@ def send_msg(self):
 
 
 @shared_task
-def send_msg(ticket_id: int, flow_id: int = None):
+def send_msg(ticket_id: int, deadline: int = None):
     # 可异步发送消息，非阻塞路径默认不抛出异常
-    NotifyAdapter(ticket_id, flow_id).send_msg()
+    NotifyAdapter(ticket_id, deadline).send_msg()

dbm-ui/backend/db_services/dbpermission/db_account/serializers.py

@@ -172,7 +172,7 @@ class RuleTypeSerializer(serializers.Serializer):
             required=False,
         )
         ddl = serializers.ListField(
-            help_text=_("dml"),
+            help_text=_("ddl"),
             child=serializers.ChoiceField(choices=PrivilegeType.MySQL.DDL.get_choices()),
             required=False,
         )

dbm-ui/backend/ticket/tasks/ticket_tasks.py

@@ -26,6 +26,7 @@
 from backend.components import BKLogApi
 from backend.configuration.constants import PLAT_BIZ_ID, DBType
 from backend.constants import DEFAULT_SYSTEM_USER
+from backend.core import notify
 from backend.db_meta.enums import ClusterType, InstanceInnerRole
 from backend.db_meta.models import Cluster, StorageInstance
 from backend.ticket.builders.common.constants import MYSQL_CHECKSUM_TABLE, MySQLDataRepairTriggerMode
@@ -238,24 +239,12 @@ def auto_clear_expire_flow(cls):
 
         # 一次批量只操作100个单据
         batch = 100
-        now = datetime.now()
+        now = datetime.now(timezone.utc)
         # 只考虑平台级别的过期配置，暂不考虑业务和集群粒度
         ticket_configs = TicketFlowsConfig.objects.filter(bk_biz_id=PLAT_BIZ_ID)
 
-        def get_expire_flow_tickets(expire_type):
-            """获取超时过期的过滤条件"""
-            qs, ticket_ids = [], []
-            for cnf in ticket_configs:
-                expire_days = cnf.configs.get(FlowTypeConfig.EXPIRE_CONFIG, TICKET_EXPIRE_DEFAULT_CONFIG)[expire_type]
-                if expire_days < 0:
-                    continue
-                qs.append(Q(update_at__lt=now - timedelta(days=expire_days), ticket__ticket_type=cnf.ticket_type))
-
-            # 如果设置为无限制过期，则不进行过滤
-            if not qs:
-                return ticket_ids
-
-            filters = reduce(operator.or_, qs)
+        def filter_tickets(filters, expire_type):
+            ticket_ids = []
             # itsm: 审批中的流程
             if expire_type == TicketExpireType.ITSM:
                 filters &= Q(flow_type=FlowType.BK_ITSM, status=TicketFlowStatus.RUNNING)
@@ -273,14 +262,59 @@ def get_expire_flow_tickets(expire_type):
 
             return ticket_ids
 
+        def get_expire_flow_tickets(expire_type):
+            """获取超时过期的单据"""
+            qs = []
+            for cnf in ticket_configs:
+                expire = cnf.configs.get(FlowTypeConfig.EXPIRE_CONFIG, TICKET_EXPIRE_DEFAULT_CONFIG)[expire_type]
+                # -1表示无限制，不参与终止
+                if expire < 0:
+                    continue
+                qs.append(Q(update_at__lt=now - timedelta(days=expire), ticket__ticket_type=cnf.ticket_type))
+
+            # 如果设置为无限制过期，则不进行过滤
+            if not qs:
+                return []
+
+            ticket_ids = filter_tickets(reduce(operator.or_, qs), expire_type)
+            return ticket_ids
+
+        def remind_expire_tickets(expire_type):
+            """获取即将超时需要提醒的单据"""
+            deadline_hours = [3, 72]
+            for hour in deadline_hours:
+                qs = []
+                for cnf in ticket_configs:
+                    expire = cnf.configs.get(FlowTypeConfig.EXPIRE_CONFIG, TICKET_EXPIRE_DEFAULT_CONFIG)[expire_type]
+                    # -1表示无限制，不参与提醒
+                    if expire < 0:
+                        continue
+                    # 超时提醒的区间是1小时，左闭右开，
+                    # 即 terminate - hour - 1 <= now < terminate - hour; terminate = update_at + expire_days
+                    st = now - timedelta(days=expire) + timedelta(hours=hour)
+                    ed = now - timedelta(days=expire) + timedelta(hours=hour + 1)
+                    qs.append(Q(update_at__gte=st, update_at__lt=ed, ticket__ticket_type=cnf.ticket_type))
+
+                if not qs:
+                    continue
+
+                ticket_ids = filter_tickets(reduce(operator.or_, qs), expire_type)
+                for ticket_id in ticket_ids:
+                    notify.send_msg.apply_async(
+                        args=(
+                            ticket_id,
+                            hour,
+                        )
+                    )
+
         # 根据超时保护类型，获取需要过期处理的单据
         expire_ticket_ids = []
         for expire_type in TicketExpireType.get_values():
             expire_ticket_ids.extend(get_expire_flow_tickets(expire_type))
+            remind_expire_tickets(expire_type)
 
         # 终止单据
         TicketHandler.revoke_ticket(ticket_ids=expire_ticket_ids[:batch], operator=DEFAULT_SYSTEM_USER)
-        # print(expire_ticket_ids)
 
 
 # ----------------------------- 异步执行任务函数 ----------------------------------------