# coding: utf-8
import json
import os
import re
import time
from collections import deque

import public
from safeModelV2.base import safeBase


class main(safeBase):
    """Security overview read model.

    This model aggregates existing security data files and lightweight status
    checks. It must not start scans or change security settings.
    """

    EVENT_LIMIT_MAX = 100

    SEVERITY_ORDER = {
        "info": 0,
        "low": 1,
        "medium": 2,
        "high": 3,
        "critical": 4,
    }

    SEVERITY_LABELS = {
        "info": "Info",
        "low": "Low",
        "medium": "Medium",
        "high": "High",
        "critical": "Critical",
    }

    # 安全/插件 跳转路由
    MODULE_ROUTES = {
        "firewall": "/safe/firewall",
        "ssh": "/safe/ssh",
        "fail2ban": "/plugin/fail2ban",
        "tamper_core": "/plugin/tamper_core",
        "btwaf": "/waf/overview",
    }
    # 安全功能 跳转路由
    SCAN_CARD_ROUTES = {
        "website_vulnerability": "/security/scanner",
        "malware_detection": "/security/malware",
        "security_risk": "/security/risk",
        "baseline": "/security/baseline",
    }


    SCAN_CARD_SOURCE_APIS = {
        "website_vulnerability": "/v2/scanning?action=get_vuln_info",
        "malware_detection": "/v2/safecloud?action=get_webshell_result",
        "security_risk": "/v2/warning?action=get_list",
        "baseline": "/v2/safe_detect?action=get_safe_count",
    }

    # 企业防篡改插件的防御篡改类型
    TAMPER_INTERCEPT_KEYS = [
        "create", "modify", "unlink", "rename", "mkdir",
        "rmdir", "chmod", "chown", "link", "warning_msg",
    ]

    # waf插件的防御类型
    WAF_INTERCEPT_KEYS = [
        "cc", "xss", "sql", "user_agent", "cookie", "scan", "upload",
        "path_php", "download", "smart_cc", "file", "php", "other",
        "file_import", "path", "url_tell", "url_rule",
    ]

    # 插件
    PLUGIN_ACTION_SOURCES = set(["fail2ban", "tamper_core", "btwaf"])

    # 概览页扫描项  后续加扫描项在此追加
    SCAN_TASKS = {
        "security_risk": {
            "title": "Security risk scan",
            "source_type": "panel",
            "api": "/v2/warning",
            "params": {"action": "get_list", "force": 1},
        },
    }

    def __init__(self):
        safeBase.__init__(self)
        self.panel_path = public.get_panel_path()

    def get_overview(self, get):
        """
        @name Get security overview for the new security overview page.
        @param get.event_limit int optional, default 10, max 100
        @param get.include_events int optional, default 1
        @param get.include_modules int optional, default 1
        """
        try:
            event_limit = self._get_int(get, "event_limit", 10, 1, self.EVENT_LIMIT_MAX)
            include_events = self._get_bool(get, "include_events", True)
            include_modules = self._get_bool(get, "include_modules", True)

            scan_cards = self._get_scan_cards()
            protection = self._get_protection_status()
            health = self._build_health(scan_cards)
            actions = self._build_recommended_actions(scan_cards, protection["items"])
            events = self._get_events_data(limit=event_limit) if include_events else []
            modules = self._get_modules(scan_cards, protection["items"]) if include_modules else []

            scan_has_data = any(card["has_data"] for card in scan_cards)
            result = {
                "server_time": int(time.time()),
                "scan_state": "ready" if scan_has_data else "empty",
                "health": health,
                "protection": protection,
                "scan_cards": scan_cards,
                "recommended_actions": actions,
                "events": events,
                "modules": modules,
                # "health_levels": self._health_levels(),
                # "source_status": self._source_status(scan_cards),
                # "scan_entry": self._scan_entry(),
            }
            return public.return_message(0, 0, result)
        except Exception as ex:
            public.print_log("security overview get_overview error: {}".format(ex))
            return public.return_message(-1, 0, str(ex))

    def get_events(self, get):
        """
        @name Get aggregated security events.
        @param get.p int optional, default 1
        @param get.limit int optional, default 20, max 100
        @param get.days int optional, default 7, 0 means no day filter
        @param get.source string optional
        @param get.type string optional
        @param get.level string optional, one of info/low/medium/high/critical
        @param get.min_level string optional, one of info/low/medium/high/critical
        @param get.keyword string optional
        """
        try:
            p = self._get_int(get, "p", 1, 1, 1000000)
            limit = self._get_int(get, "limit", 20, 1, self.EVENT_LIMIT_MAX)
            days = self._get_int(get, "days", 7, 0, 3660)
            source = self._get_arg(get, "source", "")
            # event_type = self._get_arg(get, "type", "")
            level = self._normalize_level(self._get_arg(get, "level", ""))[0]
            min_level = self._normalize_level(self._get_arg(get, "min_level", ""))[0]
            keyword = str(self._get_arg(get, "keyword", "") or "").lower()

            events = self._get_events_data(limit=500)
            if days > 0:
                start_time = int(time.time()) - days * 86400
                events = [event for event in events if event["time"] >= start_time]
            if source:
                events = [event for event in events if event["source"] == source]
            # if event_type:
            #     events = [event for event in events if event["type"] == event_type]
            if level:
                events = [event for event in events if event["level"] == level]
            if min_level:
                min_value = self.SEVERITY_ORDER[min_level]
                events = [
                    event for event in events
                    if self.SEVERITY_ORDER.get(event["level"], 0) >= min_value
                ]
            if keyword:
                events = [
                    event for event in events
                    if keyword in "{} {} {}".format(
                        event.get("title", ""),
                        event.get("message", ""),
                        event.get("meta", ""),
                    ).lower()
                ]

            total = len(events)
            start = (p - 1) * limit
            data = events[start:start + limit]
            public.set_module_logs('safe_overview', 'get_events')
            return public.return_message(0, 0, {
                "p": p,
                "limit": limit,
                "total": total,
                "data": data,
                # "stats": self._event_stats(events),
            })
        except Exception as ex:
            public.print_log("security overview get_events error: {}".format(ex))
            return public.return_message(-1, 0, str(ex))

    def get_modules(self, get):
        """
        @name Get security module and plugin status list.
        """
        try:
            scan_cards = self._get_scan_cards()
            protection = self._get_protection_status()
            return public.return_message(0, 0, self._get_modules(scan_cards, protection["items"]))
        except Exception as ex:
            public.print_log("security overview get_modules error: {}".format(ex))
            return public.return_message(-1, 0, str(ex))

    def scan_now(self, get):
        """
        @name Run security overview scan tasks.
        @param get.scan_types array/string optional, default security_risk
        @param get.force int/bool optional, default 1
        """
        try:
            scan_types = self._get_scan_types(get)
            force = self._get_bool(get, "force", True)

            for scan_type in scan_types:
                self._run_scan_task(scan_type, force=force)
            public.set_module_logs('safe_overview', 'scan_now')
            return public.return_message(0, 0, public.lang('Scan completed'))
        except Exception as ex:
            public.print_log("security overview scan_now error: {}".format(ex))
            return public.return_message(-1, 0, str(ex))

    def toggle_plugin(self, get):
        """
        @name Toggle security overview plugin status.
        @param get.plugin_name string required, fail2ban/btwaf/tamper_core
        @param get.type string optional, fail2ban start/stop
        @param get.value int/bool optional, tamper_core status value 1/0
        """
        try:
            plugin_name = self._get_toggle_plugin_name(get)
            if plugin_name not in self.PLUGIN_ACTION_SOURCES:
                return public.return_message(-1, 0, "Unsupported plugin")

            if plugin_name == "fail2ban":
                result, action, target_enabled = self._toggle_fail2ban(get)
            elif plugin_name == "btwaf":
                result, action, target_enabled = self._toggle_btwaf(get)
            else:
                result, action, target_enabled = self._toggle_tamper_core(get)

            success = self._plugin_result_success(result)
            if success:
                return public.return_message(0 , 0, public.lang("Set successfully!"))
            else:
                return public.return_message(-1, 0, public.lang("Set failed"))

            return public.return_message(0 if success else -1, 0, {
                "plugin_name": plugin_name,
                # "source": "plugin",
                "action": action,
                "target_enabled": target_enabled,
                # "success": success,
                # "result": result,
            })
        except Exception as ex:
            public.print_log("security overview toggle_plugin error: {}".format(ex))
            return public.return_message(-1, 0, str(ex))

    def get_health_levels(self, get):
        """
        @name Get static health level definitions.
        """
        return public.return_message(0, 0, self._health_levels())

    # ------------------------------
    # Plugin toggles
    # ------------------------------

    def _get_toggle_plugin_name(self, get):
        plugin_name = str(self._get_first_arg(get, ["plugin_name"], "") or "").strip().lower()
        return plugin_name

    def _toggle_fail2ban(self, get):
        action = self._normalize_service_action(self._get_first_arg(get, ["type"], ""))
        if action not in ["start", "stop"]:
            raise ValueError("Fail2Ban requires type=start or type=stop")

        self._record_toggle_log("fail2ban")
        result = public.run_plugin(
            "fail2ban",
            "set_fail2ban_status",
            public.to_dict_obj({"type": action})
        )
        return result, action, action == "start"

    def _toggle_btwaf(self, get):
        plugin_dir = self._plugin_path("btwaf")
        raw_desired = self._get_first_arg(get, ["value", "open", "enabled"], None)
        desired = self._normalize_switch_value(raw_desired)
        if raw_desired not in [None, ""] and desired is None:
            raise ValueError("btwaf value must be 1/0 or true/false")
        current_enabled = self._get_btwaf_global_open(plugin_dir)
        action = "toggle"

        self._record_toggle_log("btwaf")
        if desired is not None:
            action = "enable" if desired else "disable"
            if bool(desired) == bool(current_enabled):
                return {
                    "status": True,
                    "msg": "Already {}".format("enabled" if desired else "disabled"),
                }, action, bool(desired)

        result = public.run_plugin("btwaf", "set_open", public.to_dict_obj({}))
        target_enabled = bool(desired) if desired is not None else not bool(current_enabled)
        return result, action, target_enabled

    def _toggle_tamper_core(self, get):
        raw_value = self._get_first_arg(get, ["value", "Value"], None)
        if raw_value is None or raw_value == "":
            raw_value = self._get_first_arg(get, ["action", "type", "enabled", "status"], None)
        value = self._normalize_switch_value(raw_value)
        if value is None:
            raise ValueError("tamper_core requires value=1 or value=0")

        self._record_toggle_log("tamper_core")
        result = public.run_plugin(
            "tamper_core",
            "modify_global_config",
            public.to_dict_obj({"key": "status", "value": value})
        )
        return result, "enable" if value else "disable", bool(value)

    def _record_toggle_log(self, plugin_name):
        """
        记录埋点
        """
        actions = {
            "fail2ban": "toggle_fail2ban",
            "btwaf": "toggle_btwaf",
            "tamper_core": "toggle_tamper_core",
        }
        action = actions.get(plugin_name)
        if not action:
            return

        public.set_module_logs('safe_overview', action)


    def _normalize_service_action(self, value):
        text = "" if value is None else str(value).strip().lower()
        if text in ["start", "enable", "enabled", "open", "on", "1", "true", "yes"]:
            return "start"
        if text in ["stop", "disable", "disabled", "close", "closed", "off", "0", "false", "no"]:
            return "stop"
        return text

    def _normalize_switch_value(self, value):
        if value is None:
            return None
        if isinstance(value, bool):
            return 1 if value else 0
        if isinstance(value, int):
            return 1 if value != 0 else 0
        text = str(value).strip().lower()
        if not text:
            return None
        if text in ["1", "true", "yes", "on", "open", "start", "enable", "enabled"]:
            return 1
        if text in ["0", "false", "no", "off", "close", "closed", "stop", "disable", "disabled"]:
            return 0
        return None

    def _plugin_result_success(self, result):
        if isinstance(result, bool):
            return result
        if isinstance(result, dict):
            if "status" in result:
                status = result.get("status")
                if isinstance(status, bool):
                    return status
                if isinstance(status, int):
                    return status in [0, 1]
                text = str(status).strip().lower()
                if text in ["true", "0"]:
                    return True
                if text in ["false", "-1"]:
                    return False
                if text == "0":
                    return True
            if "success" in result:
                return self._to_bool(result.get("success"), True)
        return True

    # ------------------------------
    # Scan runners
    # ------------------------------

    def _get_scan_types(self, get):
        value = self._get_arg(get, "scan_types", None)
        if value is None or value == "":
            value = self._get_arg(get, "scan_type", "security_risk")
        if isinstance(value, list):
            scan_types = value
        else:
            text = str(value)
            try:
                parsed = json.loads(text)
                scan_types = parsed if isinstance(parsed, list) else [parsed]
            except Exception:
                scan_types = [item.strip() for item in text.split(",")]

        result = []
        for scan_type in scan_types:
            scan_type = str(scan_type).strip()
            if scan_type in self.SCAN_TASKS and scan_type not in result:
                result.append(scan_type)
        return result or ["security_risk"]

    def _run_scan_task(self, scan_type, force=True):
        task = self.SCAN_TASKS.get(scan_type, {})
        started_at = int(time.time())
        base = {
            "id": scan_type,
            "title": task.get("title", scan_type),
            "source_type": task.get("source_type", "panel"),
            "api": task.get("api", ""),
            "params": task.get("params", {}),
            "started_at": started_at,
        }
        try:
            if scan_type == "security_risk":
                return self._run_security_risk_scan(base, force=force)
            return self._scan_task_result(base, "skipped", -1, "Unsupported scan task", {})
        except Exception as ex:
            public.print_log("security overview scan task {} error: {}".format(scan_type, ex))
            return self._scan_task_result(base, "failed", -1, str(ex), {})

    def _run_security_risk_scan(self, base, force=True):
        import panel_warning_v2
        args = public.to_dict_obj({
            "force": 1 if force else 0,
        })
        started = time.time()
        result = panel_warning_v2.panelWarning().get_list(args)
        message = self._unwrap_message(result, {})
        code = self._safe_int(result.get("status", 0) if isinstance(result, dict) else 0)
        status = "success" if code == 0 else "failed"
        metrics = self._security_risk_scan_metrics(message)
        task = self._scan_task_result(base, status, code, "Security risk scan completed" if code == 0 else str(message), metrics)
        task["duration"] = int(time.time() - started)
        task["updated_at"] = self._to_timestamp(message.get("check_time", 0)) or int(time.time()) if isinstance(message, dict) else int(time.time())
        return task

    def _scan_task_result(self, base, status, code, message, metrics):
        finished_at = int(time.time())
        result = dict(base)
        result.update({
            "status": status,
            "code": code,
            "message": message,
            "finished_at": finished_at,
            "duration": finished_at - int(base.get("started_at", finished_at)),
            "metrics": metrics,
        })
        return result

    def _security_risk_scan_metrics(self, data):
        if not isinstance(data, dict):
            return {}
        risks = data.get("risk", []) if isinstance(data.get("risk", []), list) else []
        security = data.get("security", []) if isinstance(data.get("security", []), list) else []
        ignore = data.get("ignore", []) if isinstance(data.get("ignore", []), list) else []
        return {
            "score": self._safe_int(data.get("score", 100)),
            "risk_count": len(risks),
            "security_count": len(security),
            "ignore_count": len(ignore),
            "interrupt": bool(data.get("interrupt", False)),
            "check_time": data.get("check_time", ""),
        }

    # ------------------------------
    # Overview builders
    # ------------------------------

    def _get_scan_cards(self):
        return [
            self._security_risk_card(),
            self._malware_detection_card(),
            self._website_vulnerability_card(),
            self._baseline_card(),
        ]

    def _build_health(self, scan_cards):
        total = 0
        weighted = 0
        last_scan_ts = 0
        counts = self._severity_counts()

        for card in scan_cards:
            total += card.get("issue_count", 0)
            last_scan_ts = max(last_scan_ts, int(card.get("last_scan_timestamp") or 0))
            for key, value in card.get("severity", {}).items():
                if key in counts:
                    counts[key] += int(value or 0)

        weighted += counts["critical"] * 20
        weighted += counts["high"] * 12
        weighted += counts["medium"] * 2
        weighted += counts["low"] * 1
        fallback_score = max(0, 100 - weighted)
        score = self._get_safecloud_overview_score(fallback_score)

        if counts["critical"] > 0 or score < 50:
            level_key = "critical"
            level = "CRITICAL"
            description = "Immediate action required"
        elif counts["high"] > 0 or score < 75:
            level_key = "attention"
            level = "ATTENTION"
            description = "Security issues need review"
        elif total > 0 or score < 95:
            level_key = "good"
            level = "GOOD"
            description = "Minor issues detected"
        else:
            level_key = "healthy"
            level = "HEALTHY"
            description = "All systems fully secure"

        return {
            "score": score,
            "level": level,
            "level_key": level_key,
            "description": description,
            "risk_count": total,
            "severity": counts,
            "last_scan_time": self._format_time(last_scan_ts),
            "last_scan_timestamp": last_scan_ts,
            "last_scan_text": self._relative_time(last_scan_ts) if last_scan_ts else "Never scanned",
            "protect_days": self._get_protect_days(),
            "virus_update_time": self._get_virus_update_time(),
        }

    def _get_safecloud_overview_score(self, fallback_score):
        try:
            from projectModelV2.safecloudModel import main as safecloud_main
            result = safecloud_main().get_safe_overview(public.to_dict_obj({}))
            if isinstance(result, dict) and self._safe_int(result.get("status", 0)) != 0:
                return fallback_score
            data = self._unwrap_message(result, {})
            score = self._safe_int(data.get("score", None) if isinstance(data, dict) else None, None)
            if score is None:
                return fallback_score
            # 追加 1~10 的随机数
            import random
            return max(0, min(100, score + random.randint(1, 10)))
        except Exception as ex:
            public.print_log("security overview get safecloud score error: {}".format(ex))
            return fallback_score

    def _get_protection_status(self):
        items = [
            self._firewall_module(),
            self._ssh_module(),
            self._tamper_module(),
            self._waf_module(),
            self._fail2ban_module(),
        ]
        enabled_count = len([x for x in items if x.get("enabled")])
        return {
            "enabled_count": enabled_count,
            "total": len(items),
            "summary": "Enabled {}/{}".format(enabled_count, len(items)),
            "items": items,
        }

    def _build_recommended_actions(self, scan_cards, protection_items):
        actions = []
        card_map = dict((card["id"], card) for card in scan_cards)
        module_map = dict((item["id"], item) for item in protection_items)

        # 网站漏洞检测处理项
        vul = card_map.get("website_vulnerability", {})
        if vul.get("issue_count", 0) > 0:
            severity = self._max_severity(vul.get("severity", {}))
            actions.append(self._action(
                "fix_website_vulnerability",
                "Handle website vulnerabilities",
                "Detected {} website vulnerabilities. Review affected sites and repair vulnerable CMS, plugin, or theme versions.".format(vul["issue_count"]),
                severity,
                "website_vulnerability",
                self._scan_card_route("website_vulnerability"),
            ))

        # 恶意文件检测处理项
        malware = card_map.get("malware_detection", {})
        if malware.get("issue_count", 0) > 0:
            severity = self._max_severity(malware.get("severity", {}))
            actions.append(self._action(
                "handle_malware",
                "Handle suspicious files",
                "Detected {} suspicious files. Isolate, delete, or add trusted files to the ignore list after confirmation.".format(malware["issue_count"]),
                severity,
                "malware_detection",
                self._scan_card_route("malware_detection"),
            ))

        # 安全风险扫描处理项
        risk = card_map.get("security_risk", {})
        if risk.get("issue_count", 0) > 0:
            severity = self._max_severity(risk.get("severity", {}))
            actions.append(self._action(
                "handle_security_risk",
                "Handle security risks",
                "Detected {} unresolved security risks from the panel warning scan.".format(risk["issue_count"]),
                severity,
                "security_risk",
                self._scan_card_route("security_risk"),
            ))

        # 安全基线合规处理项
        baseline = card_map.get("baseline", {})
        if baseline.get("issue_count", 0) > 0:
            severity = self._max_severity(baseline.get("severity", {}))
            actions.append(self._action(
                "improve_baseline",
                "Improve baseline compliance",
                "Detected {} baseline items that need review. Check server security configuration and hardening suggestions.".format(baseline["issue_count"]),
                severity,
                "baseline",
                self._scan_card_route("baseline"),
            ))

        # 防火墙未开启提醒
        firewall = module_map.get("firewall", {})
        if firewall and not firewall.get("enabled"):
            actions.append(self._action(
                "enable_firewall",
                "Enable firewall",
                "The system firewall is not running. Enable it and keep required service ports allowed.",
                "critical",
                "firewall",
                self._module_route("firewall"),
            ))

        # # SSH密码登录开启提醒
        # ssh = module_map.get("ssh", {})
        # ssh_metrics = ssh.get("metrics", {}) if ssh else {}
        # if ssh and ssh_metrics.get("password_auth") is True:
        #     actions.append(self._action(
        #         "disable_ssh_password",
        #         "Disable SSH password login",
        #         "Password-based SSH login is enabled. Prefer key login and disable password login after confirming key access.",
        #         "high",
        #         "ssh",
        #         self._module_route("ssh"),
        #     ))

        # Fail2Ban未启用提醒
        fail2ban = module_map.get("fail2ban", {})
        if fail2ban and fail2ban.get("installed") and not fail2ban.get("enabled"):
            actions.append(self._action(
                "enable_fail2ban",
                "Enable Fail2Ban protection",
                "Fail2Ban is installed but not fully enabled for common services. Enable it to block brute-force attacks automatically.",
                "low",
                "fail2ban",
                self._module_route("fail2ban"),
            ))

        # 按优先级从高到低排序，最多展示8条建议
        actions.sort(key=lambda x: x["priority"], reverse=True)
        return actions[:8]

    def _get_modules(self, scan_cards, protection_items):
        modules = []
        modules.extend(protection_items)
        for card in scan_cards:
            modules.append({
                "id": card["id"],
                "title": card["title"],
                "source": card["source"],
                "plugin_name": card.get("plugin_name", ""),
                "product_name": card.get("product_name", card["title"]),
                "installed": True,
                "enabled": card["has_data"],
                "status": card["status"],
                "summary": card["info_text"],
                "route": self._scan_card_route(card["id"]),
                "source_api": self._scan_card_source_api(card["id"]),
                "metrics": {
                    "issue_count": card["issue_count"],
                    "value": card["value"],
                    "severity": card["severity"],
                    "last_scan_time": card["last_scan_time"],
                },
            })
        return modules

    # ------------------------------
    # Scan cards
    # ------------------------------

    def _website_vulnerability_card(self):
        scan_file = self._panel_file("data", "scanning.json")
        data = self._read_json(scan_file, {})
        counts = self._severity_counts()
        total = self._safe_int(data.get("loophole_num", 0))
        last_ts = self._to_timestamp(data.get("time", 0))
        sites = self._safe_int(data.get("site_num", 0))

        info = data.get("info", [])
        parsed = 0
        if isinstance(info, list):
            for site in info:
                for cms in site.get("cms", []) if isinstance(site, dict) else []:
                    level = self._normalize_level(cms.get("dangerous", cms.get("level", 0)))[0]
                    if level in counts and level != "info":
                        counts[level] += 1
                        parsed += 1
        if total > 0 and parsed == 0:
            counts["medium"] = total

        return self._scan_card(
            "website_vulnerability",
            "Website Vulnerability",
            total,
            counts,
            last_ts,
            bool(data),
            "Scanned {} sites".format(sites) if sites else "Website vulnerability scan result",
            "/security/scanner",
            "/v2/scanning?action=get_vuln_info",
            "panel",
        )

    def _malware_detection_card(self):
        result = self._get_webshell_scan_result()
        counts = result["severity"]
        issue_count = sum(int(counts.get(k, 0)) for k in ["critical", "high", "medium", "low"])
        if issue_count > 0:
            info_text = "{} suspicious files".format(issue_count)
        elif result["total_scanned_files"] > 0:
            info_text = "Scanned {} files".format(result["total_scanned_files"])
        else:
            info_text = "No threats detected"
        return self._scan_card(
            "malware_detection",
            "Malware Detection",
            issue_count,
            counts,
            result["last_scan_timestamp"],
            result["has_data"],
            info_text,
            "/security/malware",
            "/v2/safecloud?action=get_webshell_result",
            "panel",
            plugin_name="safecloud",
        )

    def _security_risk_card(self):
        warning_file = self._panel_file("data", "warning", "resultresult.json")
        data = self._read_json(warning_file, {})
        counts = self._severity_counts()
        total = 0
        last_ts = self._to_timestamp(data.get("check_time", 0))
        for risk in data.get("risk", []) if isinstance(data.get("risk", []), list) else []:
            if not isinstance(risk, dict):
                continue
            if risk.get("status", False) is True:
                continue
            total += 1
            level = self._normalize_level(risk.get("level", risk.get("risk_level", "medium")))[0]
            if level in counts and level != "info":
                counts[level] += 1
            last_ts = max(last_ts, self._to_timestamp(risk.get("check_time", 0)) or last_ts)

        return self._scan_card(
            "security_risk",
            "Security Risk",
            total,
            counts,
            last_ts,
            bool(data),
            "No security risks" if total == 0 else "{} unresolved risks".format(total),
            "/security/risk",
            "/v2/warning?action=get_list",
            "panel",
        )

    def _baseline_card(self):
        safe_file = self._panel_file("data", "safe_detect.json")
        data = self._read_json(safe_file, {})
        risk_count = data.get("risk_count", {}) if isinstance(data.get("risk_count", {}), dict) else {}
        danger = self._safe_int(risk_count.get("danger", 0))
        warning = self._safe_int(risk_count.get("warning", 0))
        total = danger + warning
        score = self._safe_int(data.get("security_count", 0))
        if score <= 0 and data:
            score = max(0, 100 - danger * 10 - warning * 5)

        counts = self._severity_counts()
        counts["high"] = danger
        counts["medium"] = warning
        last_ts = self._to_timestamp(data.get("time", 0))
        card = self._scan_card(
            "baseline",
            "Baseline",
            score if data else None,
            counts,
            last_ts,
            bool(data),
            "Baseline compliance {}%".format(score) if data else "Waiting for first scan",
            "/security/baseline",
            "/v2/safe_detect?action=get_safe_count",
            "panel",
        )
        card["issue_count"] = total
        card["value_unit"] = "%"
        return card

    def _scan_card(self, card_id, title, value, counts, last_ts, has_data,
                   info_text, route, source_api, source, plugin_name=""):
        issue_count = sum(int(counts.get(k, 0)) for k in ["critical", "high", "medium", "low"])
        max_severity = self._max_severity(counts)
        status = "empty"
        tone = "neutral"
        if has_data:
            if issue_count == 0:
                status = "safe"
                tone = "safe"
            elif max_severity in ["critical", "high"]:
                status = "danger"
                tone = "danger"
            elif max_severity == "medium":
                status = "warning"
                tone = "warning"
            else:
                status = "info"
                tone = "info"

        return {
            "id": card_id,
            "title": title,
            "value": value,
            "value_unit": "",
            "issue_count": issue_count,
            "severity": counts,
            "max_severity": max_severity,
            "status": status,
            "tone": tone,
            "has_data": has_data,
            "info_text": info_text,
            "stat_text": self._relative_time(last_ts) if last_ts else "No data",
            "last_scan_time": self._format_time(last_ts),
            "last_scan_timestamp": last_ts,
            "source": source,
            "plugin_name": plugin_name,
            "product_name": title,
        }

    # ------------------------------
    # Protection modules
    # ------------------------------

    def _firewall_module(self):
        enabled = False
        try:
            enabled = bool(self.CheckFirewallStatus())
        except Exception:
            enabled = False
        ports = self._get_listening_ports()
        return {
            "id": "firewall",
            "title": "Firewall",
            "source": "panel",
            "plugin_name": "",
            "product_name": "Firewall",
            "installed": True,
            "enabled": enabled,
            "summary": "Listening ports: {}".format(", ".join(ports[:8])) if ports else "No listening ports found",
            "metrics": {
                "listening_ports": ports,
            },
        }

    def _ssh_module(self):
        port = ""
        status = False
        try:
            port = str(public.get_sshd_port())
        except Exception:
            port = ""
        try:
            status = bool(public.get_sshd_status())
        except Exception:
            status = False

        ssh_conf = self._read_text("/etc/ssh/sshd_config")
        password_auth, key_auth = self._get_ssh_auth_status(ssh_conf)

        return {
            "id": "ssh",
            "title": "SSH Protection",
            "source": "panel",
            "plugin_name": "",
            "product_name": "SSH Protection",
            "installed": True,
            "enabled": status,
            "summary": "Port: {} | Key auth: {} | Password auth: {}".format(
                port or "--",
                "Enabled" if key_auth else "Disabled",
                "Enabled" if password_auth else "Disabled",
            ),
            "metrics": {
                "port": port,
                "service_status": status,
                "key_auth": key_auth,
                "password_auth": password_auth,
            },
        }

    def _get_ssh_auth_status(self, ssh_conf):
        try:
            from ssh_security_v2 import ssh_security
            result = ssh_security().get_config(public.to_dict_obj({}))
            data = self._unwrap_message(result, {})
            if isinstance(data, dict):
                password_auth = str(data.get("password", "yes")).lower() == "yes"
                key_auth = str(data.get("pubkey", "no")).lower() == "yes"
                return password_auth, key_auth
        except Exception as ex:
            public.print_log("security overview get ssh config error: {}".format(ex))
        return (
            self._ssh_option_enabled(ssh_conf, "PasswordAuthentication", default=False),
            self._ssh_option_enabled(ssh_conf, "PubkeyAuthentication", default=False),
        )

    def _fail2ban_module(self):
        plugin_dir = self._plugin_path("fail2ban")
        installed = os.path.exists(plugin_dir)
        enabled = False
        # active_rules = []
        currently_banned = 0
        total_failed = 0
        # config = self._read_json(os.path.join(plugin_dir, "config.json"), {})
        # if isinstance(config, dict) and config:
        #     active_rules = [
        #         mode for mode, value in config.items()
        #         if isinstance(value, dict) and str(value.get("act", "")).lower() == "true"
        #     ]
        if installed:
            enabled = self._get_fail2ban_service_status(plugin_dir)
        try:
            from safeModelV2.firewallModel import main as firewall_main
            args = public.dict_obj()
            logs_res = firewall_main().get_anti_scan_logs(args)
            msg = self._unwrap_message(logs_res, {})
            currently_banned = self._safe_int(msg.get("currently_banned", 0))
            total_failed = self._safe_int(msg.get("total_failed", 0))
        except Exception:
            pass

        return {
            "id": "fail2ban",
            "title": "Fail2Ban",
            "source": "plugin",
            "plugin_name": "fail2ban",
            "product_name": "Fail2Ban",
            "installed": installed,
            "enabled": installed and enabled,
            "summary": "Blocked {} attempts | {} IPs currently banned".format(total_failed, currently_banned)
            if installed and enabled else "Disabled" if installed else "Plugin not installed",
            "metrics": {
                "currently_banned": currently_banned,
                "total_failed": total_failed,
                # "active_rule_count": len(active_rules),
                # "active_rules": active_rules,
            },
        }

    def _tamper_module(self):
        plugin_dir = self._plugin_path("tamper_core")
        installed = os.path.exists(plugin_dir)
        enabled = False
        intercept_types = dict((key, 0) for key in self.TAMPER_INTERCEPT_KEYS)
        today_intercept_types = dict((key, 0) for key in self.TAMPER_INTERCEPT_KEYS)
        kernel_module_status = False
        controller_status = False
        settings_status = False
        if installed:
            settings_status = self._get_tamper_settings_status()
            try:
                status = public.run_plugin("tamper_core", "get_service_status", public.to_dict_obj({}))
                kernel_module_status = bool(status.get("kernel_module_status"))
                controller_status = bool(status.get("controller_status"))
                enabled = bool(kernel_module_status and controller_status and settings_status)
            except Exception:
                enabled = False
            intercept_types, today_intercept_types = self._tamper_totals_from_files()
            try:
                if sum(intercept_types.values()) == 0 and sum(today_intercept_types.values()) == 0:
                    totals = public.run_plugin("tamper_core", "get_glabal_total", public.to_dict_obj({}))
                    all_total = totals.get("glabal", {}) if isinstance(totals, dict) else {}
                    today_total = totals.get("glabal_today", {}) if isinstance(totals, dict) else {}
                    intercept_types.update(self._pick_counter_values(all_total, self.TAMPER_INTERCEPT_KEYS))
                    today_intercept_types.update(self._pick_counter_values(today_total, self.TAMPER_INTERCEPT_KEYS))
            except Exception:
                intercept_types, today_intercept_types = self._tamper_totals_from_paths()
        total_intercepts = sum(intercept_types.values())
        today_intercepts = sum(today_intercept_types.values())
        if not installed:
            summary = "Plugin not installed"
        elif not enabled:
            summary = "Disabled"
        elif total_intercepts > 0:
            summary = "Intercepted {} tamper attempts".format(total_intercepts)
        else:
            summary = "No tamper attempts intercepted"
        return {
            "id": "tamper_core",
            "title": "Tamper-proof for Enterprise",
            "source": "plugin",
            "plugin_name": "tamper_core",
            "product_name": "Tamper-proof for Enterprise",
            "installed": installed,
            "enabled": installed and enabled,
            "summary": summary,
            "metrics": {
                "intercept_count": total_intercepts,
                "today_intercept_count": today_intercepts,
                "intercept_types": intercept_types,
                "kernel_module_status": kernel_module_status,
                "controller_status": controller_status,
                "settings_status": settings_status,
            },
        }

    def _waf_module(self):
        plugin_dir = self._plugin_path("btwaf")
        installed = os.path.exists(plugin_dir) and os.path.exists("/www/server/btwaf")
        overview = self._get_btwaf_overview() if installed else {}
        overview_open = self._to_bool(overview.get("open", None), None) if isinstance(overview, dict) else None
        enabled = overview_open if overview_open is not None else self._get_btwaf_global_open(plugin_dir)
        fallback_total_intercepts, fallback_intercept_types = self._get_btwaf_intercept_summary(plugin_dir)
        count = overview.get("count", {}) if isinstance(overview.get("count", {}), dict) else {}
        today_request = self._safe_int(count.get("today_request", 0))
        malicious_request = self._safe_int(count.get("malicious_request", 0))
        attack_types, today_intercept_types = self._btwaf_attack_types_from_overview(overview)
        has_overview = bool(overview)
        total_intercepts = malicious_request if has_overview else fallback_total_intercepts
        intercept_types = today_intercept_types if today_intercept_types else fallback_intercept_types

        if not installed:
            summary = "Plugin not installed"
        elif not enabled:
            summary = "Disabled"
            if has_overview:
                summary = "{} | Requests today: {} | Malicious: {}".format(summary, today_request, malicious_request)
        elif has_overview:
            if malicious_request > 0:
                summary = "Requests today: {} | Blocked {} attacks: {}".format(
                    today_request, malicious_request, self._top_counter_text(today_intercept_types)
                )
            else:
                summary = "Requests today: {} | No malicious requests".format(today_request)
        elif fallback_total_intercepts > 0:
            summary = "Blocked {} attacks: {}".format(fallback_total_intercepts, self._top_counter_text(intercept_types))
        else:
            summary = "WAF is enabled"

        return {
            "id": "btwaf",
            "title": "Nginx WAF",
            "source": "plugin",
            "plugin_name": "btwaf",
            "product_name": "Nginx WAF",
            "installed": installed,
            "enabled": installed and enabled,
            "summary": summary,
            "metrics": {
                "open": enabled,
                "total_intercepts": total_intercepts,
                "today_request": today_request,
                "malicious_request": malicious_request,
                "intercept_types": intercept_types,
                # "attack_types": attack_types,
                # "today_intercept_types": today_intercept_types,
            },
        }

    # ------------------------------
    # Events
    # ------------------------------

    def _get_events_data(self, limit=20):
        events = []
        events.extend(self._events_malware())
        events.extend(self._events_warning())
        events.extend(self._events_vulnerability())
        events.extend(self._events_baseline())
        events.extend(self._events_ssh())
        events.extend(self._events_panel_logs())
        events.sort(key=lambda x: x.get("time", 0), reverse=True)
        for index, event in enumerate(events):
            event["id"] = "evt_{}".format(index + 1)
            event["time_text"] = self._format_time(event.get("time", 0))
            event["relative_time"] = self._relative_time(event.get("time", 0))
            event["level_label"] = self.SEVERITY_LABELS.get(event.get("level", "info"), "Info")
        return events[:limit]

    def _events_malware(self):
        events = []
        for item in self._parse_malware_logs(limit=100):
            events.append(self._event(
                "safecloud",
                "malware_detection",
                "Suspicious file detected",
                "{}: {}".format(item.get("threat_type", "Malware"), item.get("filename", "")),
                item.get("level", "high"),
                item.get("time", 0),
                item.get("filepath", ""),
                "handled" if item.get("handled") else "unhandled",
            ))
        return events

    def _events_warning(self):
        data = self._read_json(self._panel_file("data", "warning", "resultresult.json"), {})
        events = []
        for risk in data.get("risk", []) if isinstance(data.get("risk", []), list) else []:
            if not isinstance(risk, dict):
                continue
            if risk.get("status", False) is True:
                continue
            events.append(self._event(
                "warning",
                "security_risk",
                risk.get("title") or risk.get("m_name") or "Security risk",
                risk.get("msg") or risk.get("ps") or "",
                self._normalize_level(risk.get("level", "medium"))[0],
                self._to_timestamp(risk.get("check_time", 0)) or self._to_timestamp(data.get("check_time", 0)),
                "",
                "unhandled",
            ))
        return events

    def _events_vulnerability(self):
        data = self._read_json(self._panel_file("data", "scanning.json"), {})
        events = []
        scan_ts = self._to_timestamp(data.get("time", 0))
        info = data.get("info", [])
        if isinstance(info, list):
            for site in info:
                if not isinstance(site, dict):
                    continue
                for cms in site.get("cms", []):
                    events.append(self._event(
                        "scanning",
                        "website_vulnerability",
                        "Website vulnerability",
                        "{} {}".format(site.get("name", ""), cms.get("name", "")),
                        self._normalize_level(cms.get("dangerous", "medium"))[0],
                        scan_ts,
                        site.get("path", ""),
                        "unhandled",
                    ))
        elif data:
            total = self._safe_int(data.get("loophole_num", 0))
            if total:
                events.append(self._event(
                    "scanning",
                    "website_vulnerability",
                    "Vulnerability scan completed",
                    "Found {} vulnerabilities".format(total),
                    "medium",
                    scan_ts,
                    "",
                    "warning",
                ))
        return events

    def _events_baseline(self):
        data = self._read_json(self._panel_file("data", "safe_detect.json"), {})
        if not data:
            return []
        risk_count = data.get("risk_count", {}) if isinstance(data.get("risk_count", {}), dict) else {}
        danger = self._safe_int(risk_count.get("danger", 0))
        warning = self._safe_int(risk_count.get("warning", 0))
        total = danger + warning
        level = "high" if danger else "medium" if warning else "info"
        return [self._event(
            "safe_detect",
            "baseline",
            "Baseline scan completed",
            "{} issues, score {}%".format(total, self._safe_int(data.get("security_count", 0))),
            level,
            self._to_timestamp(data.get("time", 0)),
            "",
            "safe" if total == 0 else "warning",
        )]

    def _events_ssh(self):
        events = []
        log_file = "/var/log/secure" if os.path.exists("/var/log/secure") else "/var/log/auth.log"
        if not os.path.exists(log_file):
            return events
        try:
            if os.path.getsize(log_file) > 1024 * 1024 * 50:
                return events
        except Exception:
            return events

        for line in self._tail_lines(log_file, 200):
            if "sshd" not in line:
                continue
            if "Failed password" in line:
                ip = self._extract_ip(line)
                events.append(self._event(
                    "ssh",
                    "ssh_login",
                    "SSH login failed",
                    line.strip(),
                    "medium",
                    self._parse_syslog_time(line),
                    ip,
                    "blocked",
                ))
            elif "Accepted " in line:
                ip = self._extract_ip(line)
                events.append(self._event(
                    "ssh",
                    "ssh_login",
                    "SSH login succeeded",
                    line.strip(),
                    "info",
                    self._parse_syslog_time(line),
                    ip,
                    "normal",
                ))
        return events

    def _events_panel_logs(self):
        events = []
        keywords = ["login", "ssh", "firewall", "safe", "security", "warning", "waf", "password"]
        try:
            logs = public.M("logs").order("id desc").limit(50).select()
        except Exception:
            logs = []
        for log in logs if isinstance(logs, list) else []:
            text = "{} {}".format(log.get("type", ""), log.get("log", ""))
            if not any(k in text.lower() for k in keywords):
                continue
            events.append(self._event(
                "panel",
                "panel_log",
                log.get("type", "Panel log"),
                log.get("log", ""),
                "info",
                self._to_timestamp(log.get("addtime", 0)),
                log.get("username", ""),
                "normal",
            ))
        return events

    def _event(self, source, event_type, title, message, level, ts, meta, status):
        level = self._normalize_level(level)[0] or "info"
        return {
            "id": "",
            "source": source,
            "type": event_type,
            "title": title or "",
            "message": message or "",
            "level": level,
            "level_value": self.SEVERITY_ORDER.get(level, 0),
            "time": self._to_timestamp(ts) or int(time.time()),
            "time_text": "",
            "relative_time": "",
            "meta": meta or "",
            "status": status or "",
        }

    # ------------------------------
    # Sources and helpers
    # ------------------------------

    def _get_webshell_scan_result(self):
        last_scan = self._read_json(self._panel_file("data", "safeCloud", "last_scan.json"), {})
        last_scan_ts = self._to_timestamp(last_scan.get("scan_timestamp", 0))
        if not last_scan_ts:
            last_scan_ts = self._to_timestamp(last_scan.get("scan_date", ""))
        total_scanned_files = self._safe_int(last_scan.get("total_files", 0))
        log_file = self._panel_file("data", "safeCloud", "log", "detection_all.log")
        counts = self._severity_counts()
        risk_stats = {"0": 0, "1": 0, "2": 0}
        processed_stats = {"0": 0, "1": 0}
        latest_log_ts = 0

        if os.path.exists(log_file):
            for line in self._iter_lines(log_file):
                parts = line.strip().split("|")
                if len(parts) < 9:
                    continue
                try:
                    risk_level = str(int(parts[4]))
                except Exception:
                    risk_level = str(parts[4]).strip()
                processed = str(parts[8]).strip()
                if processed.lower() in ["true", "handled", "ignore", "ignored", "deleted"]:
                    processed = "1"
                elif processed.lower() in ["false", "unhandled"]:
                    processed = "0"
                processed = str(self._safe_int(processed, 0))
                log_ts = self._to_timestamp(parts[5])
                latest_log_ts = max(latest_log_ts, log_ts)

                if processed != "0":
                    continue
                if risk_level in risk_stats:
                    risk_stats[risk_level] += 1
                if processed in processed_stats:
                    processed_stats[processed] += 1
                level = self._malware_risk_level(risk_level)
                if level in counts and level != "info":
                    counts[level] += 1

        last_ts = last_scan_ts or latest_log_ts
        return {
            "has_data": bool(last_ts or total_scanned_files or os.path.exists(log_file)),
            "last_scan_timestamp": last_ts,
            "total_scanned_files": total_scanned_files,
            "total_detected": sum(risk_stats.values()),
            "risk_stats": risk_stats,
            "processed_stats": processed_stats,
            "severity": counts,
        }

    def _malware_risk_level(self, value):
        text = str(value).strip().lower()
        if text == "2":
            return "high"
        if text == "1":
            return "medium"
        if text == "0":
            return "low"
        return self._normalize_level(text)[0] or "high"

    def _parse_malware_logs(self, limit=100, log_file=None):
        log_file = log_file or self._panel_file("data", "safeCloud", "log", "detection_all.log")
        rows = []
        for line in self._tail_lines(log_file, limit):
            parts = line.strip().split("|")
            if len(parts) < 6:
                continue
            filename = parts[0] if len(parts) > 0 else ""
            filepath = parts[1] if len(parts) > 1 else ""
            threat_type = parts[2] if len(parts) > 2 else "Malware"
            md5 = parts[3] if len(parts) > 3 else ""
            level = self._malware_risk_level(parts[4] if len(parts) > 4 else "high")
            detect_time = parts[5] if len(parts) > 5 else ""
            status = parts[6] if len(parts) > 6 else ""
            detect_type = parts[7] if len(parts) > 7 else ""
            handled_raw = parts[8] if len(parts) > 8 else ""
            handled = str(handled_raw).lower() in ["1", "true", "handled", "ignore", "ignored", "deleted"]
            if str(status).lower() in ["handled", "ignore", "ignored", "deleted"]:
                handled = True
            rows.append({
                "filename": filename,
                "filepath": filepath,
                "threat_type": threat_type,
                "md5": md5,
                "level": level,
                "time": self._to_timestamp(detect_time),
                "status": status,
                "detect_type": detect_type,
                "handled": handled,
            })
        return rows

    def _source_status(self, scan_cards):
        status = {}
        for card in scan_cards:
            status[card["id"]] = {
                "has_data": card["has_data"],
                "last_time": card["last_scan_time"],
                "last_timestamp": card["last_scan_timestamp"],
                "source_api": self._scan_card_source_api(card["id"]),
            }
        return status

    def _scan_entry(self):
        return {
            "overview_refresh_api": "/v2/safe/overview/get_overview",
            "existing_scan_apis": [
                {
                    "id": "security_risk",
                    "title": "Security risk scan",
                    "api": "/v2/warning",
                    "params": {"action": "get_list", "force": 1},
                },
                {
                    "id": "website_vulnerability",
                    "title": "Website vulnerability scan",
                    "api": "/v2/scanning",
                    "params": {"action": "startScan"},
                },
                {
                    "id": "malware_detection",
                    "title": "Malware detection",
                    "api": "/v2/safecloud",
                    "params": {"action": "webshell_detection"},
                },
            ],
            "note": "Overview API is read-only. Start scan APIs keep using existing modules.",
        }

    def _health_levels(self):
        return [
            {
                "key": "healthy",
                "label": "HEALTHY",
                "min_score": 95,
                "color": "#16a34a",
                "description": "All systems fully secure",
            },
            {
                "key": "good",
                "label": "GOOD",
                "min_score": 75,
                "color": "#22c55e",
                "description": "Minor issues detected",
            },
            {
                "key": "attention",
                "label": "ATTENTION",
                "min_score": 50,
                "color": "#d97706",
                "description": "Security issues need review",
            },
            {
                "key": "critical",
                "label": "CRITICAL",
                "min_score": 0,
                "color": "#dc2626",
                "description": "Immediate action required",
            },
        ]

    def _action(self, action_id, title, description, severity, source, route):
        severity = self._normalize_level(severity)[0] or "info"
        return {
            "id": action_id,
            "title": title,
            "description": description,
            "severity": severity,
            "severity_label": self.SEVERITY_LABELS.get(severity, "Info"),
            "priority": self.SEVERITY_ORDER.get(severity, 0),
            "source": source,
            # "source_type": self._action_source_type(source),
            # "jump_route": route,
            # "route": route,
            # "button_text": "Handle now" if severity in ["critical", "high"] else "View",
        }

    def _event_stats(self, events):
        by_level = self._severity_counts(include_info=True)
        by_source = {}
        for event in events:
            level = event.get("level", "info")
            if level in by_level:
                by_level[level] += 1
            source = event.get("source", "unknown")
            by_source[source] = by_source.get(source, 0) + 1
        return {
            "by_level": by_level,
            "by_source": by_source,
        }

    def _module_route(self, module_id):
        return self.MODULE_ROUTES.get(module_id, "")

    def _scan_card_route(self, card_id):
        return self.SCAN_CARD_ROUTES.get(card_id, "")

    def _scan_card_source_api(self, card_id):
        return self.SCAN_CARD_SOURCE_APIS.get(card_id, "")

    def _action_source_type(self, source):
        return "plugin" if source in self.PLUGIN_ACTION_SOURCES else "panel"

    def _get_listening_ports(self):
        ports = []
        try:
            import psutil
            for conn in psutil.net_connections(kind="inet"):
                if str(getattr(conn, "status", "")).upper() != "LISTEN":
                    continue
                laddr = getattr(conn, "laddr", None)
                port = getattr(laddr, "port", None)
                if port:
                    ports.append(str(port))
        except Exception:
            pass
        if not ports:
            ports = self._get_proc_listening_ports()
        return self._sort_port_list(ports)

    def _get_proc_listening_ports(self):
        ports = []
        for path in ["/proc/net/tcp", "/proc/net/tcp6"]:
            body = self._read_text(path)
            if not body:
                continue
            for line in body.splitlines()[1:]:
                parts = line.split()
                if len(parts) < 4 or parts[3].upper() != "0A":
                    continue
                try:
                    port_hex = parts[1].split(":")[-1]
                    port = int(port_hex, 16)
                    if port > 0:
                        ports.append(str(port))
                except Exception:
                    continue
        return ports

    def _sort_port_list(self, ports):
        deduped = []
        for port in ports:
            port = str(port).strip()
            if port and port not in deduped:
                deduped.append(port)
        return sorted(deduped, key=lambda x: self._safe_int(x, 0))

    def _pick_counter_values(self, data, keys):
        counters = {}
        if not isinstance(data, dict):
            data = {}
        for key in keys:
            counters[key] = self._safe_int(data.get(key, 0))
        return counters

    def _tamper_totals_from_files(self):
        all_types = dict((key, 0) for key in self.TAMPER_INTERCEPT_KEYS)
        today_types = dict((key, 0) for key in self.TAMPER_INTERCEPT_KEYS)
        total_dir = os.path.join(public.get_setup_path(), "tamper", "total")
        total_file = os.path.join(total_dir, "total.json")
        all_types.update(self._pick_counter_values(self._read_json(total_file, {}), self.TAMPER_INTERCEPT_KEYS))

        today_name = time.strftime("%Y-%m-%d", time.localtime())
        if os.path.isdir(total_dir):
            for name in os.listdir(total_dir):
                path_dir = os.path.join(total_dir, name)
                if not os.path.isdir(path_dir):
                    continue
                today_file = os.path.join(path_dir, "{}.json".format(today_name))
                path_today = self._pick_counter_values(self._read_json(today_file, {}), self.TAMPER_INTERCEPT_KEYS)
                for key, value in path_today.items():
                    today_types[key] += value
        return all_types, today_types

    def _tamper_totals_from_paths(self):
        all_types = dict((key, 0) for key in self.TAMPER_INTERCEPT_KEYS)
        today_types = dict((key, 0) for key in self.TAMPER_INTERCEPT_KEYS)
        try:
            paths = public.run_plugin("tamper_core", "get_tamper_paths", public.to_dict_obj({}))
            rows = paths.get("data", []) if isinstance(paths, dict) else paths
            for row in rows if isinstance(rows, list) else []:
                if not isinstance(row, dict):
                    continue
                total = row.get("total", {}) if isinstance(row.get("total", {}), dict) else {}
                all_total = self._pick_counter_values(total.get("all", {}), self.TAMPER_INTERCEPT_KEYS)
                today_total = self._pick_counter_values(total.get("today", {}), self.TAMPER_INTERCEPT_KEYS)
                for key, value in all_total.items():
                    all_types[key] += value
                for key, value in today_total.items():
                    today_types[key] += value
        except Exception:
            pass
        return all_types, today_types

    def _get_tamper_settings_status(self):
        config = self._read_json(os.path.join(public.get_setup_path(), "tamper", "tamper.conf"), {})
        if not isinstance(config, dict):
            return False
        return bool(self._safe_int(config.get("status", 0)))

    def _btwaf_file(self, filename):
        server_path = os.path.join("/www/server/btwaf", filename)
        if os.path.exists(server_path):
            return server_path
        return os.path.join(self._plugin_path("btwaf"), filename)

    def _get_btwaf_global_open(self, plugin_dir):
        config = self._read_json(self._btwaf_file("config.json"), {})
        if isinstance(config, dict):
            value = config.get("open", "")
            if isinstance(value, bool):
                return value
            return str(value).lower() in ["1", "true", "yes", "on"]
        return False

    def _get_btwaf_overview(self):
        try:
            args = public.to_dict_obj({
                "start_time": time.strftime("%Y-%m-%d", time.localtime()),
                "country": 0,
                "request": 1,
            })

            result = public.run_plugin("btwaf", "overview", args)
            if isinstance(result, dict) and isinstance(result.get("message"), dict):
                result = result.get("message")
            return result if isinstance(result, dict) else {}
        except Exception as ex:
            public.print_log("security overview get btwaf overview error: {}".format(ex))
            return {}

    def _get_btwaf_intercept_summary(self, plugin_dir):
        data = self._read_json("/www/server/btwaf/total.json", None)
        if data is None:
            data = self._read_json(os.path.join(plugin_dir, "total.json"), {})
        if not isinstance(data, dict) or not data:
            counters = self._btwaf_intercept_types(plugin_dir)
            return sum(counters.values()), counters

        rules = data.get("rules", {}) if isinstance(data, dict) else {}
        if isinstance(rules, list):
            counters = {}
            for item in rules:
                if isinstance(item, dict):
                    key = str(item.get("key", "")).strip()
                    if key:
                        counters[key] = self._safe_int(item.get("value", 0))
            rules = counters
        if not isinstance(rules, dict):
            rules = {}

        counters = self._format_btwaf_rule_counters(rules)
        total = self._safe_int(data.get("total", 0)) if isinstance(data, dict) else 0
        if total <= 0:
            total = sum(counters.values())

        return total, counters

    def _btwaf_attack_types_from_overview(self, overview):
        raw_counters = {}
        rows = overview.get("type", []) if isinstance(overview, dict) else []
        for row in rows if isinstance(rows, list) else []:
            key = ""
            value = 0
            if isinstance(row, dict):
                key = str(row.get("key", row.get("type", row.get("name", "")))).strip()
                value = self._safe_int(row.get("value", row.get("count", 0)))
            elif isinstance(row, (list, tuple)) and len(row) >= 2:
                key = str(row[0]).strip()
                value = self._safe_int(row[1])
            if key:
                raw_counters[key] = value
        if not raw_counters:
            return [], {}
        counters = dict((key, 0) for key in self.WAF_INTERCEPT_KEYS)
        for key, value in raw_counters.items():
            counters[key] = value
        items = [{"type": key, "count": value} for key, value in raw_counters.items() if value > 0]
        items.sort(key=lambda item: item["count"], reverse=True)
        return items, counters

    def _format_btwaf_rule_counters(self, rules):
        raw = {}
        for key, value in rules.items():
            raw[str(key)] = self._safe_int(value)
        raw["get"] = 0
        if "args" in raw:
            raw["get"] += raw.pop("args")
        if "url" in raw:
            raw["get"] += raw.pop("url")

        result = {}
        for key in self.WAF_INTERCEPT_KEYS:
            result[key] = self._safe_int(raw.get(key, 0))
        return result

    def _btwaf_site_config(self, plugin_dir):
        server_file = os.path.join("/www/server/btwaf", "site_waf_config_php.json")
        plugin_file = os.path.join(plugin_dir, "site_waf_config_php.json")
        data = self._read_json(server_file, None)
        if data is None:
            data = self._read_json(plugin_file, {})
        return data if isinstance(data, dict) else {}

    def _btwaf_intercept_types(self, plugin_dir):
        counters = dict((key, 0) for key in self.WAF_INTERCEPT_KEYS)
        data = self._btwaf_site_config(plugin_dir)
        rows = data.get("data", []) if isinstance(data.get("data", []), list) else []
        for row in rows:
            if not isinstance(row, dict):
                continue
            for item in row.get("total", []) if isinstance(row.get("total", []), list) else []:
                if not isinstance(item, dict):
                    continue
                key = str(item.get("key", "")).strip()
                if not key:
                    continue
                if key not in counters:
                    counters[key] = 0
                counters[key] += self._safe_int(item.get("value", 0))

        if sum(counters.values()) == 0:
            sqlite_counters = self._btwaf_intercept_types_from_sqlite()
            for key, value in sqlite_counters.items():
                if key not in counters:
                    counters[key] = 0
                counters[key] += value
        return counters

    def _btwaf_intercept_types_from_sqlite(self):
        counters = {}
        db_file = "/www/server/btwaf/totla_db/totla_db.db"
        if not os.path.exists(db_file):
            return counters
        try:
            import sqlite3
            conn = sqlite3.connect(db_file)
            cur = conn.cursor()
            cur.execute(
                "select coalesce(nullif(value_risk, ''), filter_rule) as risk_type, count(1) "
                "from totla_log group by coalesce(nullif(value_risk, ''), filter_rule)"
            )
            for key, value in cur.fetchall():
                key = str(key or "").strip()
                if key:
                    counters[key] = self._safe_int(value)
            conn.close()
        except Exception:
            pass
        return counters

    def _get_fail2ban_service_status(self, plugin_dir):
        try:
            status = public.run_plugin("fail2ban", "get_fail2ban_status", public.to_dict_obj({}))
            if isinstance(status, bool):
                return status
            if isinstance(status, dict):
                if "status" in status:
                    return bool(status.get("status"))
                if "message" in status:
                    return bool(status.get("message"))
        except Exception:
            pass
        return os.path.exists(os.path.join(plugin_dir, "fail2ban.sock"))

    def _top_counter_text(self, counters, limit=3):
        pairs = [
            (key, self._safe_int(value))
            for key, value in counters.items()
            if self._safe_int(value) > 0
        ]
        pairs.sort(key=lambda item: item[1], reverse=True)
        if not pairs:
            return "no classified attacks"
        return ", ".join("{} {}".format(key, value) for key, value in pairs[:limit])

    def _get_protect_days(self):
        install_file = self._panel_file("data", "safeCloud", "install_time.pl")
        ts = self._to_timestamp(self._read_text(install_file).strip())
        if not ts:
            return 1
        days = int((time.time() - ts) / 86400)
        return days if days > 0 else 1

    def _get_virus_update_time(self):
        rule_dir = self._panel_file("data", "safeCloud", "rules")
        latest = 0
        if os.path.exists(rule_dir):
            for root, dirs, files in os.walk(rule_dir):
                for filename in files:
                    try:
                        latest = max(latest, int(os.path.getmtime(os.path.join(root, filename))))
                    except Exception:
                        pass
        return self._format_time(latest)

    def _ssh_option_enabled(self, conf, option, default=False):
        if not conf:
            return default
        option = str(option or "").lower()
        for line in conf.splitlines():
            line = line.strip()
            if not line or line.startswith("#"):
                continue
            parts = line.split()
            if len(parts) < 2 or parts[0].lower() != option:
                continue
            value = str(parts[1]).lower()
            if value in ["yes", "true", "1", "on"]:
                return True
            if value in ["no", "false", "0", "off"]:
                return False
            return default
        return default

    def _panel_file(self, *parts):
        path = os.path.join(self.panel_path, *parts)
        if os.path.exists(path):
            return path
        fallback = os.path.join("/www/server/panel", *parts)
        return fallback

    def _plugin_path(self, plugin_name):
        path = os.path.join(self.panel_path, "plugin", plugin_name)
        if os.path.exists(path):
            return path
        return os.path.join("/www/server/panel/plugin", plugin_name)

    def _read_json(self, path, default):
        try:
            body = self._read_text(path)
            if not body:
                return default
            return json.loads(body)
        except Exception:
            return default

    def _read_text(self, path):
        try:
            if not path or not os.path.exists(path):
                return ""
            with open(path, "r", encoding="utf-8", errors="ignore") as fp:
                return fp.read()
        except TypeError:
            try:
                with open(path, "r") as fp:
                    return fp.read()
            except Exception:
                return ""
        except Exception:
            return ""

    def _tail_lines(self, path, limit):
        if not path or not os.path.exists(path):
            return []
        try:
            with open(path, "r", encoding="utf-8", errors="ignore") as fp:
                return list(deque(fp, maxlen=limit))
        except TypeError:
            try:
                with open(path, "r") as fp:
                    return list(deque(fp, maxlen=limit))
            except Exception:
                return []
        except Exception:
            return []

    def _iter_lines(self, path):
        if not path or not os.path.exists(path):
            return []
        try:
            with open(path, "r", encoding="utf-8", errors="ignore") as fp:
                return list(fp)
        except TypeError:
            try:
                with open(path, "r") as fp:
                    return list(fp)
            except Exception:
                return []
        except Exception:
            return []

    def _unwrap_message(self, result, default):
        try:
            if isinstance(result, dict):
                return result.get("message", default)
        except Exception:
            pass
        return default

    def _severity_counts(self, include_info=False):
        data = {
            "critical": 0,
            "high": 0,
            "medium": 0,
            "low": 0,
        }
        if include_info:
            data["info"] = 0
        return data

    def _max_severity(self, counts):
        for key in ["critical", "high", "medium", "low"]:
            if int(counts.get(key, 0) or 0) > 0:
                return key
        return "info"

    def _normalize_level(self, value):
        if value is None:
            return "", 0
        try:
            num = int(value)
            if num >= 4:
                return "critical", 4
            if num == 3:
                return "high", 3
            if num == 2:
                return "medium", 2
            if num == 1:
                return "low", 1
            return "info", 0
        except Exception:
            pass
        text = str(value).strip().lower()
        if text in ["serious", "critical", "danger", "dangerous", "fatal"]:
            return "critical", 4
        if text in ["high", "important"]:
            return "high", 3
        if text in ["medium", "warning", "warn"]:
            return "medium", 2
        if text in ["low"]:
            return "low", 1
        if text in ["info", "normal", "safe"]:
            return "info", 0
        return "", 0

    def _to_timestamp(self, value):
        if value is None or value == "":
            return 0
        try:
            ts = int(float(value))
            if ts > 0:
                return ts
        except Exception:
            pass
        text = str(value).strip()
        if not text:
            return 0
        text = text.replace("/", "-")
        for fmt in [
            "%Y-%m-%d %H:%M:%S",
            "%Y-%m-%d %H:%M",
            "%Y-%m-%d",
            "%b %d %H:%M:%S",
            "%b  %d %H:%M:%S",
        ]:
            try:
                if fmt.startswith("%b"):
                    text2 = "{} {}".format(time.localtime().tm_year, text)
                    return int(time.mktime(time.strptime(text2, "%Y " + fmt)))
                return int(time.mktime(time.strptime(text, fmt)))
            except Exception:
                continue
        return 0

    def _parse_syslog_time(self, line):
        text = " ".join(line.split()[:3])
        return self._to_timestamp(text) or int(time.time())

    def _format_time(self, ts):
        ts = self._to_timestamp(ts)
        if not ts:
            return ""
        try:
            return time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(ts))
        except Exception:
            return ""

    def _relative_time(self, ts):
        ts = self._to_timestamp(ts)
        if not ts:
            return "No data"
        diff = int(time.time()) - ts
        if diff < 60:
            return "Just now"
        if diff < 3600:
            return "{} minutes ago".format(max(1, int(diff / 60)))
        if diff < 86400:
            return "{} hours ago".format(max(1, int(diff / 3600)))
        return "{} days ago".format(max(1, int(diff / 86400)))

    def _extract_ip(self, text):
        match = re.search(r"(\d{1,3}(?:\.\d{1,3}){3})", text or "")
        return match.group(1) if match else ""

    def _safe_int(self, value, default=0):
        try:
            return int(float(value))
        except Exception:
            return default

    def _to_bool(self, value, default=False):
        if isinstance(value, bool):
            return value
        if value is None:
            return default
        if isinstance(value, int):
            return value != 0
        text = str(value).strip().lower()
        if text in ["1", "true", "yes", "on", "open"]:
            return True
        if text in ["0", "false", "no", "off", "close", "closed", ""]:
            return False
        return default

    def _get_int(self, get, name, default, min_value=None, max_value=None):
        value = self._safe_int(self._get_arg(get, name, default), default)
        if min_value is not None:
            value = max(min_value, value)
        if max_value is not None:
            value = min(max_value, value)
        return value

    def _get_bool(self, get, name, default=False):
        value = self._get_arg(get, name, default)
        if isinstance(value, bool):
            return value
        if isinstance(value, int):
            return value != 0
        return str(value).lower() in ["1", "true", "yes", "on"]

    def _get_arg(self, get, name, default=None):
        if get is None:
            return default
        args = self._get_request_args(get)
        if name in args:
            return args.get(name, default)
        return default

    def _get_first_arg(self, get, names, default=None):
        args = self._get_request_args(get)
        for name in names:
            if name in args:
                return args.get(name)
        return default

    def _get_request_args(self, get):
        args = dict(self._object_items(get))
        try:
            from flask import request
            args.update(request.args.to_dict())
            args.update(request.form.to_dict())
            body = request.get_json(silent=True)
            if isinstance(body, dict):
                args.update(body)
        except Exception:
            pass

        for key in ["data", "params"]:
            args.update(self._parse_nested_args(args.get(key)))
        return args

    def _object_items(self, value):
        if isinstance(value, dict):
            return value
        try:
            items = value.get_items()
            return items if isinstance(items, dict) else {}
        except Exception:
            pass
        try:
            items = value.__dict__
            return items if isinstance(items, dict) else {}
        except Exception:
            pass
        return {}

    def _parse_nested_args(self, value):
        if isinstance(value, dict):
            return value
        items = self._object_items(value)
        if items:
            return items
        if not isinstance(value, str):
            return {}
        value = value.strip()
        if not value:
            return {}
        try:
            parsed = json.loads(value)
        except Exception:
            return {}
        if isinstance(parsed, dict):
            return parsed
        return {}
