feat(notify-expire-priv):增加查询权限到期提醒

common/utils/sendmsg.py

@@ -93,7 +93,8 @@ def send_email(self, subject, body, to, **kwargs):
             main_msg = email.mime.multipart.MIMEMultipart()
 
             # 添加文本内容
-            text_msg = email.mime.text.MIMEText(body, "plain", "utf-8")
+            content_type = kwargs.get("content_type", "plain")
+            text_msg = email.mime.text.MIMEText(body, content_type, "utf-8")
             main_msg.attach(text_msg)
 
             # 添加附件

sql/notify.py

@@ -144,7 +144,7 @@ def render_audit(self):
         # 获取审核信息
         audit_id = self.audit.audit_id
         base_url = self.sys_config.get(
-            "archery_base_url", "http://127.0.0.1:8000"
+            "archery_base_url", "http://127.0.0.1:9123"
         ).rstrip("/")
         workflow_url = "{base_url}/workflow/{audit_id}".format(
             base_url=base_url, audit_id=self.audit.audit_id
@@ -281,7 +281,7 @@ def render_audit(self):
 
     def render_execute(self):
         base_url = self.sys_config.get(
-            "archery_base_url", "http://127.0.0.1:8000"
+            "archery_base_url", "http://127.0.0.1:9123"
         ).rstrip("/")
         audit_handler = AuditV2(workflow=self.workflow, audit=self.audit)
         review_info = audit_handler.get_review_info()

sql/query_privileges.py

@@ -23,7 +23,15 @@
 from common.config import SysConfig
 from common.utils.const import WorkflowStatus, WorkflowType, WorkflowAction
 from common.utils.extend_json_encoder import ExtendJSONEncoder
-from sql.models import QueryPrivilegesApply, QueryPrivileges, Instance, ResourceGroup
+from sql.engines.goinception import GoInceptionEngine
+from sql.models import (
+    QueryPrivilegesApply,
+    QueryPrivileges,
+    Instance,
+    ResourceGroup,
+    Users,
+)
+from common.utils.sendmsg import MsgSender
 from sql.notify import notify_for_audit
 from sql.utils.resource_group import user_groups, user_instances
 from sql.utils.workflow_audit import Audit, AuditException, get_auditor
@@ -625,3 +633,94 @@ def _query_apply_audit_call_back(apply_id, workflow_status):
                 for table_name in apply_queryset.table_list.split(",")
             ]
         QueryPrivileges.objects.bulk_create(insert_list)
+
+
+def query_priv_expire_reminder():
+    """
+    查询权限到期提醒任务
+    扫描 valid_date 在 3 天内到期的权限，向权限持有人和管理员发送提醒邮件
+    """
+    now = datetime.datetime.now()
+    today = now.date()
+    expire_threshold = today + datetime.timedelta(days=3)
+
+    # 查询即将到期的权限（未删除）
+    expire_privs = QueryPrivileges.objects.filter(
+        valid_date__gte=today,
+        valid_date__lte=expire_threshold,
+        is_deleted=0,
+    ).order_by("user_name", "valid_date")
+
+    if not expire_privs.exists():
+        logger.info("暂无即将到期的查询权限，跳过提醒")
+        return
+
+    # 按用户聚合权限
+    user_priv_map = {}
+    for priv in expire_privs:
+        user_priv_map.setdefault(priv.user_name, []).append(priv)
+
+    # 获取管理员邮件列表
+    admin_emails = list(
+        Users.objects.filter(is_superuser=True, email__isnull=False)
+        .exclude(email="")
+        .values_list("email", flat=True)
+    )
+
+    msg_sender = MsgSender()
+    base_url = SysConfig().get("archery_base_url", "http://127.0.0.1:9123").rstrip("/")
+
+    for user_name, privs in user_priv_map.items():
+        try:
+            user = Users.objects.get(username=user_name)
+            user_email = user.email
+        except Users.DoesNotExist:
+            logger.warning(f"用户 {user_name} 不存在，跳过权限到期提醒")
+            continue
+
+        if not user_email:
+            logger.warning(f"用户 {user_name} 未配置邮箱，跳过权限到期提醒")
+            continue
+
+        # 构造 HTML 邮件内容
+        rows_html = ""
+        for priv in privs:
+            priv_type_display = "库权限" if priv.priv_type == 1 else "表权限"
+            tb_name = priv.table_name if priv.priv_type == 2 else "-"
+            rows_html += (
+                f"<tr><td>{priv.instance.instance_name}</td>"
+                f"<td>{priv.db_name}</td>"
+                f"<td>{tb_name}</td>"
+                f"<td>{priv_type_display}</td>"
+                f"<td>{priv.valid_date}</td>"
+                f"<td>{priv.limit_num}</td></tr>"
+            )
+
+        msg_content = f"""<html>
+<body>
+<p>您好，</p>
+<p>您有以下查询权限将在 3 天内到期，请及时申请续期：</p>
+<table border="1" cellpadding="6" cellspacing="0" style="border-collapse: collapse; border: 1px solid #ddd;">
+  <thead>
+    <tr><th>实例</th><th>数据库</th><th>表</th><th>权限类型</th><th>到期日</th><th>行数限制</th></tr>
+  </thead>
+  <tbody>
+    {rows_html}
+  </tbody>
+</table>
+<p>请及时到查询权限申请页面进行续期申请，以免影响正常使用。<br>
+访问地址：<a href="{base_url}/queryapplylist/">{base_url}/queryapplylist/</a></p>
+<p style="color: gray;">本邮件由 Archery 系统自动发送</p>
+</body>
+</html>"""
+        msg_title = "[Archery]查询权限即将到期提醒"
+
+        # 发送给用户，抄送管理员
+        to = [user_email]
+        list_cc = admin_emails if admin_emails else []
+        msg_sender.send_email(
+            msg_title, msg_content, to, list_cc_addr=list_cc, content_type="html"
+        )
+        logger.info(f"已发送查询权限到期提醒给用户 {user_name}，权限数量：{len(privs)}")
+
+    logger.info(f"查询权限到期提醒任务完成，共提醒 {len(user_priv_map)} 位用户")

sql/test_query_privileges.py

@@ -935,3 +935,99 @@ def test_query_privilege_audit(
     )
     sql_query_apply.refresh_from_db()
     assert sql_query_apply.status == WorkflowStatus.PASSED
+
+
+class TestQueryPrivExpireReminder(TestCase):
+    """测试查询权限到期提醒"""
+
+    def setUp(self):
+        self.superuser = User.objects.create(
+            username="super", is_superuser=True, email="admin@test.com"
+        )
+        self.user = User.objects.create(
+            username="user", display="普通用户", email="user@test.com"
+        )
+        self.user_no_email = User.objects.create(
+            username="user_no_email", display="无邮箱用户"
+        )
+        self.slave = Instance.objects.create(
+            instance_name="test_instance",
+            type="slave",
+            db_type="mysql",
+            host=settings.DATABASES["default"]["HOST"],
+            port=settings.DATABASES["default"]["PORT"],
+            user=settings.DATABASES["default"]["USER"],
+            password=settings.DATABASES["default"]["PASSWORD"],
+        )
+        self.sys_config = SysConfig()
+
+    def tearDown(self):
+        self.superuser.delete()
+        self.user.delete()
+        self.user_no_email.delete()
+        Instance.objects.all().delete()
+        QueryPrivileges.objects.all().delete()
+
+    @patch("sql.query_privileges.MsgSender.send_email")
+    def test_no_expire_privs(self, mock_send_email):
+        """没有即将到期的权限，不发送邮件"""
+        sql.query_privileges.query_priv_expire_reminder()
+        mock_send_email.assert_not_called()
+
+    @patch("sql.query_privileges.MsgSender.send_email")
+    def test_send_reminder(self, mock_send_email):
+        """有即将到期的权限，发送 HTML 邮件给用户并抄送管理员"""
+        expire_date = date.today() + timedelta(days=2)
+        QueryPrivileges.objects.create(
+            user_name=self.user.username,
+            user_display=self.user.display,
+            instance=self.slave,
+            db_name="test_db",
+            table_name="test_table",
+            valid_date=expire_date,
+            limit_num=100,
+            priv_type=2,
+        )
+        sql.query_privileges.query_priv_expire_reminder()
+        mock_send_email.assert_called_once()
+        args = mock_send_email.call_args
+        self.assertIn("查询权限即将到期提醒", args[0][0])
+        self.assertEqual(args[0][2], ["user@test.com"])
+        self.assertEqual(args[1]["list_cc_addr"], ["admin@test.com"])
+        self.assertEqual(args[1]["content_type"], "html")
+        self.assertIn("<table", args[0][1])
+        self.assertIn("</table>", args[0][1])
+
+    @patch("sql.query_privileges.MsgSender.send_email")
+    def test_user_no_email(self, mock_send_email):
+        """用户未配置邮箱，跳过发送"""
+        expire_date = date.today() + timedelta(days=1)
+        QueryPrivileges.objects.create(
+            user_name=self.user_no_email.username,
+            user_display=self.user_no_email.display,
+            instance=self.slave,
+            db_name="test_db",
+            table_name="test_table",
+            valid_date=expire_date,
+            limit_num=100,
+            priv_type=2,
+        )
+        sql.query_privileges.query_priv_expire_reminder()
+        mock_send_email.assert_not_called()
+
+    @patch("sql.query_privileges.MsgSender.send_email")
+    def test_user_not_exist(self, mock_send_email):
+        """用户不存在，跳过发送"""
+        expire_date = date.today() + timedelta(days=1)
+        QueryPrivileges.objects.create(
+            user_name="not_exist_user",
+            user_display="不存在用户",
+            instance=self.slave,
+            db_name="test_db",
+            table_name="test_table",
+            valid_date=expire_date,
+            limit_num=100,
+            priv_type=2,
+        )
+        sql.query_privileges.query_priv_expire_reminder()
+        mock_send_email.assert_not_called()

sql/utils/tasks.py

@@ -53,6 +53,18 @@ def add_sync_ding_user_schedule():
     )
 
 
+def add_query_priv_expire_reminder_schedule():
+    """添加查询权限到期提醒定时任务（每天执行一次）"""
+    del_schedule(name="查询权限到期提醒")
+    schedule(
+        "sql.query_privileges.query_priv_expire_reminder",
+        name="查询权限到期提醒",
+        schedule_type="D",
+        repeats=-1,
+        timeout=-1,
+    )
+
+
 def del_schedule(name):
     """删除schedule"""
     try:

sql_api/apps.py

@@ -1,5 +1,20 @@
 from django.apps import AppConfig
+from django.db import ProgrammingError
 
 
 class SqlApi2Config(AppConfig):
     name = "sql_api"
+
+    def ready(self):
+        # 延迟导入，避免循环引用
+        from sql.utils.tasks import add_query_priv_expire_reminder_schedule
+        from django_q.models import Schedule
+
+        try:
+            # 检查是否已存在定时任务
+            if not Schedule.objects.filter(name="查询权限到期提醒").exists():
+                add_query_priv_expire_reminder_schedule()
+        except ProgrammingError:
+            # 表（django_q_schedule）还不存在，说明正在 makemigrations 或首次 migrate
+            # 忽略即可，等表创建后下次启动时会自动创建任务
+            pass