wecom_it_smart_desk/backend/app/services/it_health_service.py

# =============================================================================
# 企微IT智能服务台 — IT 健康聚合服务
# =============================================================================
# 说明：整合联软(设备信息/CPU/内存/硬盘)、火绒(安全状态/病毒/漏洞)、
#       资产服务(资产编号/启用时间)数据源，返回前端 BasicInfoCard.vue 期望的数据结构。
#
# 数据流：
#   1. 用 employee_id (企微UserID) 作为联软 strusername 查询终端
#   2. 联软 get_dev_all_info() 获取详细硬件/磁盘/网卡信息
#   3. 火绒 list_terminals() 按计算机名匹配，获取安全状态
#   4. 火绒 list_terminal_leaks() 检查漏洞，get_virus_events() 检查病毒
#   5. 资产服务 find_asset() 查资产编号和启用时间
#
# 降级策略：
#   - 联软/火绒未配置 → 返回 Mock 数据（标记 data_source: "mock"）
#   - 联软配置但火绒未配置 → 设备信息真实，安全状态为 pending
#   - 任一API调用失败 → 该部分数据返回 None，不影响其他部分
# =============================================================================

import logging
import time
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional

from sqlalchemy.ext.asyncio import AsyncSession

logger = logging.getLogger(__name__)


class ITHealthService:
    """IT 健康聚合服务。

    从联软、火绒、资产服务获取数据，聚合为前端期望的格式。
    所有外部 API 调用均做了异常隔离——任一数据源失败不影响整体。
    """

    def __init__(self, db: AsyncSession):
        """初始化服务。

        Args:
            db: 数据库会话（用于读取 system_configs 表中的集成配置）
        """
        self.db = db

    async def get_it_health(self, employee_id: str) -> Dict[str, Any]:
        """获取员工终端的 IT 健康信息。

        这是主入口方法，聚合所有数据源，返回前端期望的 JSON 结构。

        Args:
            employee_id: 员工企微 UserID（对应联软的 strusername）

        Returns:
            Dict: 包含 current_device / other_devices / data_source / health_score
        """
        # 尝试从联软获取真实设备信息
        device_info = await self._get_device_from_lianruan(employee_id)

        if device_info is None:
            # 联软不可用 → 返回 Mock 数据
            return self._get_mock_data(employee_id)

        # 联软数据可用，尝试从火绒获取安全状态
        security_info = await self._get_security_from_huorong(
            device_info.get("device_name", "")
        )

        # 尝试从资产服务获取资产编号
        asset_info = await self._get_asset_info(device_info.get("device_name", ""))

        # 聚合数据
        current_device = self._build_current_device(device_info, security_info, asset_info)

        # 获取其他设备（联软中该用户的其他终端）
        other_devices = await self._get_other_devices(employee_id, device_info.get("device_name", ""))

        return {
            "current_device": current_device,
            "other_devices": other_devices,
            "data_source": "real",
            "generated_at": datetime.now(timezone.utc).isoformat(),
        }

    # ==========================================================================
    # 联软数据获取
    # ==========================================================================

    async def _get_device_from_lianruan(self, employee_id: str) -> Optional[Dict[str, Any]]:
        """从联软查终端设备信息。

        流程：
        1. 用 employee_id 作为 strusername 查终端列表
        2. 取第一个（最近活跃的）终端
        3. 调 get_dev_all_info() 获取详细硬件信息

        Args:
            employee_id: 员工账号

        Returns:
            Dict: 设备信息字典，联软不可用时返回 None
        """
        try:
            from app.integrations.lianruan.config import get_lianruan_client

            client = await get_lianruan_client(self.db)

            # 按员工账号查终端列表
            result = await client.query_dev_by_params(strusername=employee_id, per_page=10)
            terminals = result.get("items", [])

            if not terminals:
                logger.info(f"联软未找到 employee_id={employee_id} 的终端")
                return None

            # 取第一个终端（联软默认按最近活跃排序）
            terminal = terminals[0]
            device_name = terminal.strdevname

            if not device_name:
                logger.warning(f"联软返回的终端无计算机名: {terminal}")
                return None

            # 获取详细信息
            detail = await client.get_dev_all_info(strdevname=device_name)

            # 构建设备信息字典
            device = {
                "device_name": device_name,
                "is_online": terminal.istatus == "1",
                "ip_address": terminal.strdevip or detail.strip1,
                "mac": terminal.strmac or detail.strmac,
                "os": detail.stros or "",
                "location": terminal.strdeptname or "",
                "department": terminal.strdeptname or "",
                "switch_name": terminal.strswitchname or "",
                "uptime": self._format_uptime(detail.dtdevuptime),
                "last_online_time": detail.dtdevuptime or "",
                "last_offline_time": detail.dtdevdowntime or "",
                "device_type": detail.strdevtype or "台式机",
                "serial_number": detail.strserialnumber or "",
                "mainboard": detail.strmainboardtype or "",
                # 硬件详情
                "cpu_list": [
                    {"name": c.name, "model": c.model, "vendor": c.vendor}
                    for c in detail.cpu
                ] if detail.cpu else [],
                "memory_list": [
                    {"name": m.name, "capacity": m.capacity, "vendor": m.vendor}
                    for m in detail.memory
                ] if detail.memory else [],
                "logical_disks": [
                    {
                        "label": d.name,
                        "total": d.total_size,
                        "free": d.free_space,
                        "usage_percent": d.usage_percent,
                    }
                    for d in detail.logical_disk
                ] if detail.logical_disk else [],
                "network_cards": [
                    {"name": n.name, "mac": n.mac, "is_wireless": n.is_wireless}
                    for n in detail.network_card
                ] if detail.network_card else [],
            }

            logger.info(f"联软获取设备成功: {device_name} (employee={employee_id})")
            return device

        except Exception as e:
            logger.warning(f"联软获取设备信息失败: {e}")
            return None

    async def _get_other_devices(
        self, employee_id: str, exclude_device: str
    ) -> List[Dict[str, Any]]:
        """获取员工的其他设备（联软中该用户的其他终端）。

        Args:
            employee_id: 员工账号
            exclude_device: 要排除的当前设备名

        Returns:
            List: 其他设备列表
        """
        try:
            from app.integrations.lianruan.config import get_lianruan_client

            client = await get_lianruan_client(self.db)
            result = await client.query_dev_by_params(strusername=employee_id, per_page=10)
            terminals = result.get("items", [])

            other = []
            for t in terminals:
                if t.strdevname and t.strdevname != exclude_device:
                    other.append({
                        "device_type": t.strdevtype or "设备",
                        "device_name": t.strdevname,
                        "last_login_time": t.istatus == "1" and "在线" or "离线",
                        "last_login_location": t.strdeptname or "",
                    })

            return other

        except Exception as e:
            logger.warning(f"获取其他设备失败: {e}")
            return []

    # ==========================================================================
    # 火绒安全数据获取
    # ==========================================================================

    async def _get_security_from_huorong(
        self, computer_name: str
    ) -> Optional[Dict[str, Any]]:
        """从火绒查终端安全状态。

        流程：
        1. list_terminals() 全量分页搜索，按 computer_name 匹配
        2. 找到后 get_terminal_detail() 获取硬件/资产/网络配置
        3. list_terminal_leaks() 检查是否在漏洞清单中
        4. get_virus_events() 查病毒事件统计

        Args:
            computer_name: 计算机名（联软的 strdevname）

        Returns:
            Dict: 安全状态字典，火绒不可用时返回 None
        """
        try:
            from app.integrations.huorong.config import get_huorong_client

            client = await get_huorong_client(self.db)

            # 分页搜索终端，按计算机名匹配
            target_client_id = None
            page = 1
            while page <= 10:  # 最多查10页（2000台）
                result = await client.list_terminals(page=page, per_page=200)
                items = result.get("items", [])

                for item in items:
                    if item.computer_name and item.computer_name.upper() == computer_name.upper():
                        target_client_id = item.client_id
                        break

                if target_client_id:
                    break

                if len(items) < 200:
                    break  # 没有更多数据
                page += 1

            if not target_client_id:
                # 火绒中未找到该终端 → 可能未安装火绒
                return {
                    "huorong_installed": False,
                    "is_online": False,
                    "version": "",
                    "definitions": "",
                    "high_risk_leaks": 0,
                    "virus_count": 0,
                    "virus_uncleaned": 0,
                }

            # 获取终端详情
            detail = await client.get_terminal_detail(
                client_id=target_client_id,
                optional_fields=["hardware", "assets", "netconf"],
            )

            # 检查漏洞清单
            leak_count = 0
            try:
                leaks_result = await client.list_terminal_leaks()
                for leak_item in leaks_result.get("items", []):
                    if leak_item.hostname and leak_item.hostname.upper() == computer_name.upper():
                        leak_count = 1  # 在漏洞清单中说明有高危漏洞
                        break
            except Exception as e:
                logger.warning(f"火绒漏洞查询失败: {e}")

            # 查病毒事件
            virus_count = 0
            virus_uncleaned = 0
            try:
                virus_result = await client.get_virus_events(
                    client_id=target_client_id, type=0
                )
                for stat in virus_result.get("items", []):
                    virus_count += stat.count
                    if stat.result:
                        virus_uncleaned += stat.result.fail + stat.result.ignored
            except Exception as e:
                logger.warning(f"火绒病毒事件查询失败: {e}")

            return {
                "huorong_installed": True,
                "is_online": True,  # 从 list_terminals 已确认存在
                "version": detail.computer_name and "" or "",  # 火绒版本从 list 获取
                "definitions": "",
                "high_risk_leaks": leak_count,
                "virus_count": virus_count,
                "virus_uncleaned": virus_uncleaned,
            }

        except Exception as e:
            logger.warning(f"火绒获取安全状态失败: {e}")
            return None

    # ==========================================================================
    # 资产服务
    # ==========================================================================

    async def _get_asset_info(self, device_name: str) -> Optional[Dict[str, Any]]:
        """从资产服务查设备资产编号和启用时间。

        Args:
            device_name: 计算机名（用于日志，资产查询通过资产编号）

        Returns:
            Dict: 资产信息（asset_tag / activate_date），不可用时返回 None
        """
        try:
            from app.services.asset_service import AssetService

            asset_svc = AssetService()
            # 资产服务目前通过资产编号查询，设备名无法直接查
            # 这里先返回 None，后续需要联软 devassetno → 资产编号 → 查询
            # 或者资产Excel按计算机名匹配
            return None
        except Exception as e:
            logger.warning(f"资产服务查询失败: {e}")
            return None

    # ==========================================================================
    # 数据聚合
    # ==========================================================================

    def _build_current_device(
        self,
        device_info: Dict[str, Any],
        security_info: Optional[Dict[str, Any]],
        asset_info: Optional[Dict[str, Any]],
    ) -> Dict[str, Any]:
        """聚合联软+火绒+资产数据为前端期望的格式。

        前端 BasicInfoCard.vue 期望的数据结构：
        - device_name / is_online / asset_tag / activate_date
        - ip_address / public_ip / location / os / mac / uptime
        - cpu / memory / disks (进度条)
        - security_checks / compliance_checks (状态数组)

        Args:
            device_info: 联软设备信息
            security_info: 火绒安全状态（可能为 None）
            asset_info: 资产信息（可能为 None）

        Returns:
            Dict: 前端期望的设备数据结构
        """
        # CPU 使用率（联软不提供实时使用率，用硬件型号代替）
        cpu_model = ""
        cpu_usage = 0
        if device_info.get("cpu_list"):
            cpu = device_info["cpu_list"][0]
            cpu_model = f"{cpu.get('vendor', '')} {cpu.get('name', '')}".strip()
            cpu_usage = 0  # 联软不提供实时使用率

        # 内存总量
        memory_total = ""
        memory_usage = 0
        if device_info.get("memory_list"):
            mem = device_info["memory_list"][0]
            memory_total = mem.get("capacity", "") or ""
            # 联软返回的是硬件容量，不是使用率

        # 磁盘分区
        disks = []
        for disk in device_info.get("logical_disks", []):
            try:
                usage = int(float(disk.get("usage_percent", "0").replace("%", "").strip() or "0"))
            except (ValueError, TypeError):
                usage = 0

            total = disk.get("total", "0")
            free = disk.get("free", "0")
            # 计算已用空间
            used = self._calc_used_space(total, free)

            disks.append({
                "label": f"硬盘{disk.get('label', 'C盘')}",
                "usage": usage,
                "used": used,
                "total": total,
            })

        # 安全检查状态数组（6项）
        security_checks = self._build_security_checks(security_info)

        # 合规检查状态数组（2项）
        compliance_checks = self._build_compliance_checks(device_info, security_info)

        # 健康评分
        health_score = self._calc_health_score(security_checks, compliance_checks, device_info.get("is_online", False))

        return {
            "device_name": device_info.get("device_name", ""),
            "is_online": device_info.get("is_online", False),
            "asset_tag": asset_info.get("asset_tag", "") if asset_info else "",
            "activate_date": asset_info.get("activate_date", "") if asset_info else "",
            "ip_address": device_info.get("ip_address", ""),
            "public_ip": "",  # 公网出口IP需要额外查询
            "location": device_info.get("location", ""),
            "os": device_info.get("os", ""),
            "mac": device_info.get("mac", ""),
            "uptime": device_info.get("uptime", ""),
            "cpu": {"usage": cpu_usage, "model": cpu_model},
            "memory": {"usage": memory_usage, "total": memory_total},
            "disks": disks,
            "security_checks": security_checks,
            "compliance_checks": compliance_checks,
            "health_score": health_score,
        }

    def _build_security_checks(
        self, security_info: Optional[Dict[str, Any]]
    ) -> List[Dict[str, str]]:
        """构建安全检查状态数组（6项）。

        前端期望6个检查项：
        0. 火绒安装状态
        1. 系统补丁（高危漏洞）
        2. 高危软件
        3. 病毒状态
        4. 内部攻击（接入中）
        5. 网络代理（接入中）

        Args:
            security_info: 火绒安全状态（可能为 None）

        Returns:
            List: 6个状态对象 [{status: "pass"|"warning"|"danger"|"pending"}]
        """
        if security_info is None:
            # 火绒未配置 → 全部 pending
            return [{"status": "pending"}] * 6

        checks = []

        # 0. 火绒安装
        if security_info.get("huorong_installed"):
            checks.append({"status": "pass"})
        else:
            checks.append({"status": "danger"})  # 未安装火绒 = 危险

        # 1. 系统补丁（高危漏洞）
        if security_info.get("high_risk_leaks", 0) > 0:
            checks.append({"status": "danger"})
        else:
            checks.append({"status": "pass"})

        # 2. 高危软件（火绒不直接提供，暂返回 pass）
        checks.append({"status": "pass"})

        # 3. 病毒状态
        uncleaned = security_info.get("virus_uncleaned", 0)
        if uncleaned > 0:
            checks.append({"status": "danger"})
        elif security_info.get("virus_count", 0) > 0:
            checks.append({"status": "warning"})
        else:
            checks.append({"status": "pass"})

        # 4. 内部攻击（接入中 — 联软尚未对接此数据源）
        checks.append({"status": "pending"})

        # 5. 网络代理（接入中 — 联软尚未对接此数据源）
        checks.append({"status": "pending"})

        return checks

    def _build_compliance_checks(
        self, device_info: Dict[str, Any], security_info: Optional[Dict[str, Any]]
    ) -> List[Dict[str, str]]:
        """构建合规检查状态数组（2项）。

        前端期望2个检查项：
        0. 自备电脑检查
        1. 未审批商业软件检查

        Args:
            device_info: 联软设备信息
            security_info: 火绒安全状态

        Returns:
            List: 2个状态对象
        """
        # 自备电脑：联软设备类型中如果有"自备"标记则 danger
        device_type = device_info.get("device_type", "")
        if "自备" in device_type:
            return [{"status": "danger"}, {"status": "pass"}]

        # 默认通过
        return [{"status": "pass"}, {"status": "pass"}]

    def _calc_health_score(
        self,
        security_checks: List[Dict[str, str]],
        compliance_checks: List[Dict[str, str]],
        is_online: bool,
    ) -> int:
        """计算 IT 健康评分（0-100）。

        评分算法（与前端 BasicInfoCard.vue 一致）：
        - 安全项 danger: -15 / warning: -8 / pending: 0
        - 合规项 danger: -10 / warning: -5
        - 离线设备权重 60%

        Args:
            security_checks: 安全检查状态数组
            compliance_checks: 合规检查状态数组
            is_online: 设备是否在线

        Returns:
            int: 健康评分 0-100
        """
        score = 100

        for check in security_checks:
            status = check.get("status", "pending")
            if status == "danger":
                score -= 15
            elif status == "warning":
                score -= 8

        for check in compliance_checks:
            status = check.get("status", "pass")
            if status == "danger":
                score -= 10
            elif status == "warning":
                score -= 5

        if not is_online:
            score = round(score * 0.6)

        return max(0, score)

    # ==========================================================================
    # 工具方法
    # ==========================================================================

    def _format_uptime(self, last_online_time: str) -> str:
        """格式化运行时长。

        联软返回的是最近上线时间字符串，计算距现在的时长。
        如果无法解析则返回空字符串。

        Args:
            last_online_time: 联软返回的上线时间字符串

        Returns:
            str: 如 "12天3小时" 或空字符串
        """
        if not last_online_time:
            return ""

        try:
            # 联软时间格式可能是 "2026-07-12 08:30:00" 或类似
            dt = datetime.strptime(last_online_time.replace("T", " "), "%Y-%m-%d %H:%M:%S")
            now = datetime.now()
            delta = now - dt

            days = delta.days
            hours = delta.seconds // 3600

            if days > 0:
                return f"{days}天{hours}小时"
            else:
                minutes = delta.seconds // 60
                return f"{minutes}分钟"
        except (ValueError, TypeError):
            return ""

    def _calc_used_space(self, total: str, free: str) -> str:
        """计算已用空间。

        Args:
            total: 总容量字符串（如 "256GB"）
            free: 可用空间字符串（如 "86GB"）

        Returns:
            str: 已用空间（如 "170GB"）
        """
        try:
            # 尝试提取数字部分
            total_num = float("".join(c for c in total if c.isdigit() or c == "."))
            free_num = float("".join(c for c in free if c.isdigit() or c == "."))

            used_num = total_num - free_num
            if used_num < 0:
                used_num = 0

            # 保留单位
            unit = "".join(c for c in total if c.isalpha())
            if unit:
                return f"{int(used_num)}{unit}"
            return str(int(used_num))
        except (ValueError, TypeError):
            return ""

    # ==========================================================================
    # Mock 数据（联软/火绒未配置时降级）
    # ==========================================================================

    def _get_mock_data(self, employee_id: str) -> Dict[str, Any]:
        """返回 Mock 数据（联软不可用时降级）。

        Args:
            employee_id: 员工ID（用于日志）

        Returns:
            Dict: 与真实数据结构一致的 Mock 数据
        """
        return {
            "current_device": {
                "device_name": "DESKTOP-MOCK",
                "is_online": True,
                "asset_tag": "",
                "activate_date": "",
                "ip_address": "10.90.5.x",
                "public_ip": "218.75.34.87",
                "location": "待获取",
                "os": "待获取",
                "mac": "",
                "uptime": "",
                "cpu": {"usage": 0, "model": ""},
                "memory": {"usage": 0, "total": ""},
                "disks": [],
                "security_checks": [{"status": "pending"}] * 6,
                "compliance_checks": [{"status": "pass"}, {"status": "pass"}],
                "health_score": 100,
            },
            "other_devices": [],
            "data_source": "mock",
            "generated_at": datetime.now(timezone.utc).isoformat(),
        }