bt-source/panel/mod/project/push/taskMod.py

# coding: utf-8
# -------------------------------------------------------------------
# 宝塔Linux面板
# -------------------------------------------------------------------
# Copyright (c) 2015-2017 宝塔软件(http:#bt.cn) All rights reserved.
# -------------------------------------------------------------------
# Author: baozi <baozi@bt.cn>
# -------------------------------------------------------------------
# 新告警通道管理模块
# ------------------------------
import json
import os
from typing import List
from mod.base import json_response

from mod.base.push_mod import PushManager, TaskConfig, TaskRecordConfig, TaskTemplateConfig, PushSystem, SenderConfig
from mod.base.push_mod import update_mod_push_system, UPDATE_MOD_PUSH_FILE, load_task_template_by_file, \
    UPDATE_VERSION_FILE, update_mod_push_system2
from mod.base.msg import update_mod_push_msg
from mod.base.push_mod.rsync_push import load_rsync_template
from mod.base.push_mod.task_manager_push import load_task_manager_template
from mod.base.push_mod.load_push import load_load_template
from mod.base.push_mod.util import ExecShell, debug_log
from mod.base.push_mod.web_log_push import load_web_log_template


def update_mod():
    try:
        with open(UPDATE_VERSION_FILE, 'r') as f:
            if f.read() == "11.5":
                pl = False
            else:
                pl = True
    except:
        pl = True

    if pl:
        load_task_template_by_file("/www/server/panel/mod/base/push_mod/site_push_template.json")
        load_task_template_by_file("/www/server/panel/mod/base/push_mod/system_push_template.json")
        load_task_template_by_file("/www/server/panel/mod/base/push_mod/database_push_template.json")
        load_task_template_by_file("/www/server/panel/mod/base/push_mod/ssl_push_template.json")
        load_task_template_by_file("/www/server/panel/mod/base/push_mod/ftp_push_template.json")
        load_task_template_by_file("/www/server/panel/mod/base/push_mod/safe_mod_push_template.json")
        load_task_template_by_file("/www/server/panel/mod/base/push_mod/monitor_push_template.json")
        load_task_template_by_file("/www/server/panel/mod/base/push_mod/mod_node_push_template.json")
        load_rsync_template()
        load_task_manager_template()
        load_load_template()
        load_web_log_template()
        update_mod_push_system2()
        with open(UPDATE_VERSION_FILE, "w") as f:
            f.write("11.5")
            # debug_log(">>>>>>> update_mod_push_system <<<<<<<<")

    if not os.path.exists(UPDATE_MOD_PUSH_FILE):
        update_mod_push_msg()
        update_mod_push_system()

    check_mod_push_file = "/www/server/panel/data/mod_push_data/check_mod_push_file.pl"
    if not os.path.exists(check_mod_push_file):
        ExecShell('nohup btpython /www/server/panel/script/migrate_push_tasks.py > /dev/null 2>&1 &')

try:
    update_mod()
except:
    pass
del update_mod


class main(PushManager):

    def get_task_list(self, get=None):
        sf = kf = None
        template_ids = None
        if get:
            sf = get.get("status/s", "").lower()
            if sf and sf in ("true", "1"):
                sf = True
            elif sf and sf in ("false", "0"):
                sf = False
            else:
                sf = None

            kf = get.get("keyword/s", "").lower()
            try:
                template_ids = [str(i) for i in json.loads(get.get("template_ids/s", "[]"))]
            except:
                template_ids = None

        res = self._get_task_list(status=sf, keyword=kf, template_id=template_ids)
        return json_response(status=True, data=res)

    def _get_task_list(self, status: bool=None, keyword: str=None, template_id: List[str] = None):
        res = TaskConfig().config
        # 按创建时间排序
        res.sort(key=lambda x: x["create_time"], reverse=True)
        can_call_template_ids = TaskTemplateConfig.can_call_template_ids()
        res = [task for task in res if task["template_id"] in can_call_template_ids]

        # 根据状态过滤任务
        if status is not None:
            res = [task for task in res if bool(task["status"]) == status]

        if template_id is not None and isinstance(template_id, list) and len(template_id) > 0:
            res = [task for task in res if task["template_id"] in template_id]

        for i in res:
            i['view_msg'] = self.get_view_msg_format(i)

        # 根据关键词过滤任务
        if keyword is not None and keyword:
            filtered_res = []
            for task in res:
                task_match = False
                if keyword in task["view_msg"].lower() or keyword in task["title"].lower():
                    task_match = True
                else:
                    sc = SenderConfig()
                    # 通道类型映射，包含模糊匹配规则
                    channel_map = {
                        "wx_account": "微信公众号",
                        "mail": "邮箱",
                        "webhook": "自定义通道",
                        "feishu": "飞书",
                        "dingding": "钉钉",
                        "weixin": "企业微信",
                        "sms": "短信"
                    }

                    for sender_id in task["sender"]:
                        sender = sc.get_by_id(sender_id)
                        if not sender:
                            continue
                        sender_title = sender.get("data", {}).get("title", "").lower()
                        sender_type = sender.get("sender_type", "")
                        sender_type_name = channel_map.get(sender_type, "")
                        if keyword in sender_title or keyword in sender_type or keyword in sender_type_name:
                            task_match = True
                            break

                if task_match:
                    filtered_res.append(task)

            res = filtered_res

        return res

    @staticmethod
    def get_task_record(get):
        page = 1
        size = 10
        try:
            if hasattr(get, "page"):
                page = int(get.page.strip())
            if hasattr(get, "size"):
                size = int(get.size.strip())
            task_id = get.task_id.strip()
        except (AttributeError, ValueError, TypeError):
            return json_response(status=False, msg="参数错误")

        t = TaskRecordConfig(task_id)
        t.config.sort(key=lambda x: x["create_time"])
        page = max(page, 1)
        size = max(size, 1)
        count = len(t.config)
        data = t.config[(page - 1) * size: page * size]
        return json_response(status=True, data={
            "count": count,
            "list": data,
        })

    def clear_task_record(self, get):
        try:
            task_id = get.task_id.strip()
        except (AttributeError, ValueError, TypeError):
            return json_response(status=False, msg="参数错误")
        self.clear_task_record_by_task_id(task_id)

        return json_response(status=True, msg="清除成功")

    @staticmethod
    def remove_task_records(get):
        try:
            task_id = get.task_id.strip()
            record_ids = set(json.loads(get.record_ids.strip()))
        except (AttributeError, ValueError, TypeError):
            return json_response(status=False, msg="参数错误")
        task_records = TaskRecordConfig(task_id)
        for i in range(len(task_records.config) - 1, -1, -1):
            if task_records.config[i]["id"] in record_ids:
                del task_records.config[i]

        task_records.save_config()
        return json_response(status=True, msg="清除成功")

    @staticmethod
    def get_task_template_list(get=None):
        res = []
        p_sys = PushSystem()
        tags = set()
        for i in TaskTemplateConfig().config:
            if not i['used']:
                continue
            to = p_sys.get_task_object(i["id"], i["load_cls"])
            if not to:
                continue
            t = to.filter_template(i["template"])
            if not t:
                continue
            i["template"] = t
            tags.update(set(t.get("tags", [])))
            res.append(i)

        return json_response(status=True, data=res)

    @classmethod
    def get_view_msg_format(cls, task: dict) -> str:
        return TaskTemplateConfig.task_view_msg_format(task)

    def monitor_push_list(self, get):
        res = self._get_task_list(template_id=["130", "131", "132"])
        site_name = get.get("site_name/s", "")
        if site_name:
            res = [i for i in res if site_name == i["task_data"]["site"]]
        return json_response(status=True, data=res)