document-management/backend/dms/services/system_management_service.py

"""系统管理服务（契约1.23）：资源监控、系统信息、健康检查与维护操作。

对齐 Java 版 com.dms.system.SystemManagementService：
- MetricSampler 后台线程按固定间隔采集 CPU/内存写 sys_metric_sample；
- 查询类接口实时读取（psutil）；psutil 缺失时监控接口返回 METRIC_UNAVAILABLE；
- 维护操作（预览缓存清理/强制下线）由 API 层记审计。
"""

from __future__ import annotations

import logging
import os
import platform
import shutil
import sys
import threading
from collections import deque
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any

import sqlalchemy as sa
from flask import current_app

from dms.common.errors import MetricUnavailableError
from dms.extensions import db
from dms.models import MetricSample
from dms.storage.paths import STORAGE_SUBDIRECTORIES

logger = logging.getLogger(__name__)

API_CONTRACT_VERSION = "1.30.0"
BACKEND_KIND = "PYTHON"

try:
    import psutil

    PSUTIL_AVAILABLE = True
except ImportError:  # pragma: no cover - 部署环境缺依赖时优雅降级
    psutil = None  # type: ignore[assignment]
    PSUTIL_AVAILABLE = False


# ─── 采样器 ──────────────────────────────────────────────────────────────────


class MetricSampler:
    """CPU/内存周期采样线程（30 秒间隔，7 天保留，物理删除过期行）。"""

    def __init__(
        self,
        app,
        interval_seconds: int = 30,
        retention_days: int = 7,
    ) -> None:
        self._app = app
        self._interval_seconds = max(10, interval_seconds)
        self._retention_days = max(1, retention_days)
        self._stop_event = threading.Event()
        self._thread: threading.Thread | None = None

    @property
    def running(self) -> bool:
        return self._thread is not None and self._thread.is_alive()

    def start(self) -> None:
        if not PSUTIL_AVAILABLE:
            logger.warning("psutil 未安装，资源采样器未启动")
            return
        if self._thread is not None and self._thread.is_alive():
            return
        # 预热 CPU 百分比基线（首次调用返回 0.0 无意义，丢弃）
        try:
            psutil.cpu_percent(interval=None)
        except Exception:  # noqa: BLE001
            pass
        self._stop_event.clear()
        self._thread = threading.Thread(
            target=self._run_loop, name="dms-metric-sampler", daemon=True
        )
        self._thread.start()

    def stop(self) -> None:
        self._stop_event.set()

    def _run_loop(self) -> None:
        while not self._stop_event.wait(self._interval_seconds):
            try:
                self._tick_once()
            except Exception:  # noqa: BLE001 - 单次采样失败不终止线程
                logger.warning("资源采样失败", exc_info=True)

    def _tick_once(self) -> None:
        cpu_percent = psutil.cpu_percent(interval=None)
        memory = psutil.virtual_memory()
        with self._app.app_context():
            db.session.add(
                MetricSample(
                    sampled_at=datetime.now(timezone.utc).replace(tzinfo=None),
                    cpu_percent=round(cpu_percent, 2),
                    mem_used_bytes=memory.used,
                    mem_total_bytes=memory.total,
                )
            )
            threshold = datetime.now(timezone.utc).replace(
                tzinfo=None
            ) - timedelta(days=self._retention_days)
            db.session.execute(
                sa.delete(MetricSample).where(MetricSample.sampled_at < threshold)
            )
            db.session.commit()


# ─── 查询类 ──────────────────────────────────────────────────────────────────


def _require_psutil() -> None:
    if not PSUTIL_AVAILABLE:
        raise MetricUnavailableError()


def _iso(moment: datetime) -> str:
    return (
        moment.replace(tzinfo=timezone.utc)
        .isoformat(timespec="milliseconds")
        .replace("+00:00", "Z")
    )


def current_metrics() -> dict[str, Any]:
    """实时资源快照：CPU/内存/交换分区/磁盘分区/开机时长。"""
    _require_psutil()
    disks: list[dict[str, Any]] = []
    for partition in psutil.disk_partitions():
        try:
            usage = psutil.disk_usage(partition.mountpoint)
        except (PermissionError, OSError):
            continue  # Windows 无介质光驱等不可读分区
        disks.append(
            {
                "device": partition.device,
                "mountpoint": partition.mountpoint,
                "fstype": partition.fstype,
                "totalBytes": usage.total,
                "usedBytes": usage.used,
                "freeBytes": usage.free,
                "usedPercent": round(usage.percent, 2),
            }
        )
    memory = psutil.virtual_memory()
    swap = psutil.swap_memory()
    boot_time = datetime.fromtimestamp(psutil.boot_time(), tz=timezone.utc)
    return {
        "cpu": {
            "percent": round(psutil.cpu_percent(interval=0.3), 2),
            "coreCount": psutil.cpu_count(logical=True) or 0,
            "loadAvgOneMinute": (
                round(os.getloadavg()[0], 2) if hasattr(os, "getloadavg") else None
            ),
        },
        "memory": {
            "totalBytes": memory.total,
            "usedBytes": memory.used,
            "availableBytes": memory.available,
            "usedPercent": round(memory.percent, 2),
        },
        "swap": {
            "totalBytes": swap.total,
            "usedBytes": swap.used,
            "usedPercent": round(swap.percent, 2),
        },
        "disks": disks,
        "bootAt": _iso(boot_time.replace(tzinfo=None)),
    }


def metric_history(hours: int) -> dict[str, Any]:
    """近 N 小时采样序列（升序，供趋势图）。"""
    _require_psutil()
    threshold = datetime.now(timezone.utc).replace(tzinfo=None) - timedelta(
        hours=hours
    )
    samples = db.session.scalars(
        sa.select(MetricSample)
        .where(MetricSample.sampled_at >= threshold)
        .order_by(MetricSample.sampled_at.asc())
        .limit(24 * 3600)  # 防御性上限（30秒间隔×24小时=2880）
    ).all()
    return {
        "hours": hours,
        "entries": [
            {
                "sampledAt": _iso(sample.sampled_at),
                "cpuPercent": sample.cpu_percent,
                "memUsedBytes": sample.mem_used_bytes,
                "memTotalBytes": sample.mem_total_bytes,
            }
            for sample in samples
        ],
    }


def process_metrics() -> dict[str, Any]:
    """应用进程监控：CPU/内存/线程数/运行时长。"""
    _require_psutil()
    process = psutil.Process()
    memory_info = process.memory_info()
    started = datetime.fromtimestamp(process.create_time(), tz=timezone.utc)
    return {
        "pid": process.pid,
        "cpuPercent": round(
            sum(process.cpu_percent(interval=0.3) for _ in range(1)) / max(
                1, (psutil.cpu_count(logical=True) or 1)
            ),
            2,
        ),
        "memoryRssBytes": memory_info.rss,
        "memoryVmsBytes": getattr(memory_info, "vms", 0),
        "threadCount": process.num_threads(),
        "startedAt": _iso(started.replace(tzinfo=None)),
        "uptimeSeconds": max(
            0, int((datetime.now(tz=timezone.utc) - started).total_seconds())
        ),
        "pythonVersion": platform.python_version(),
    }


def storage_usage() -> list[dict[str, Any]]:
    """七个存储目录的大小与文件数统计（实时遍历）。"""
    root = Path(current_app.config["DMS_STORAGE_ROOT"])
    entries: list[dict[str, Any]] = []
    for name in STORAGE_SUBDIRECTORIES:
        directory = root / name
        total_bytes = 0
        file_count = 0
        if directory.is_dir():
            for dirpath, _dirnames, filenames in os.walk(directory):
                for filename in filenames:
                    try:
                        total_bytes += (Path(dirpath) / filename).stat().st_size
                        file_count += 1
                    except OSError:
                        continue
        entries.append(
            {
                "directory": name,
                "totalBytes": total_bytes,
                "fileCount": file_count,
            }
        )
    return entries


def converter_status() -> dict[str, Any]:
    """LibreOffice 转换器状态（存活/并发占用/累计计数）。"""
    converter = current_app.extensions["dms_document_converter"]
    snapshot = (
        converter.status_snapshot()
        if hasattr(converter, "status_snapshot")
        else {"available": getattr(converter, "available", False)}
    )
    snapshot["previewEnabled"] = current_app.config.get(
        "DMS_OFFICE_PREVIEW_ENABLED", True
    )
    snapshot["timeoutSeconds"] = current_app.config.get(
        "DMS_LIBREOFFICE_TIMEOUT_SECONDS", 60
    )
    return snapshot


def health_check() -> list[dict[str, Any]]:
    """一键体检：数据库/存储可写/转换器/备份目标/信封加密。"""
    items: list[dict[str, Any]] = []

    database = {"item": "DATABASE", "ok": False, "detail": ""}
    try:
        db.session.execute(sa.text("SELECT 1"))
        database["ok"] = True
    except Exception as exc:  # noqa: BLE001
        database["detail"] = str(exc)[:200]
    items.append(database)

    storage = {"item": "STORAGE_WRITABLE", "ok": False, "detail": ""}
    try:
        root = Path(current_app.config["DMS_STORAGE_ROOT"])
        probe = root / ".dms-health-probe"
        probe.write_text("ok", encoding="utf-8")
        probe.unlink(missing_ok=True)
        storage["ok"] = True
    except OSError as exc:
        storage["detail"] = str(exc)[:200]
    items.append(storage)

    converter = current_app.extensions["dms_document_converter"]
    items.append(
        {
            "item": "LIBREOFFICE",
            "ok": bool(getattr(converter, "available", False)),
            "detail": "",
        }
    )

    backup = {"item": "BACKUP_TARGET", "ok": False, "detail": ""}
    try:
        engine = current_app.extensions["dms_backup_engine"]
        target = engine.build_target(engine.load_config())
        target.check_writable()
        backup["ok"] = True
    except Exception as exc:  # noqa: BLE001
        backup["detail"] = str(exc)[:200]
    items.append(backup)

    kek_configured = bool(current_app.config.get("DMS_STORAGE_KEK_BYTES"))
    items.append(
        {
            "item": "STORAGE_ENCRYPTION",
            "ok": True,
            "detail": "KEK已配置" if kek_configured else "KEK未配置（明文落盘）",
        }
    )
    return items


def system_info() -> dict[str, Any]:
    """系统信息卡片：版本/后端类型/运行环境/数据库/启动时间。"""
    from dms.config import resolve_backup_config_path, resolve_env_path

    database_url = str(current_app.config["SQLALCHEMY_DATABASE_URI"])
    if database_url.startswith("dm"):
        database_type = "DM (达梦)"
    elif database_url.startswith("mysql"):
        database_type = "MySQL"
    else:
        database_type = database_url.split(":", 1)[0]
    uptime_seconds = 0
    started_at = ""
    if PSUTIL_AVAILABLE:
        process = psutil.Process()
        started = datetime.fromtimestamp(process.create_time(), tz=timezone.utc)
        started_at = _iso(started.replace(tzinfo=None))
        uptime_seconds = max(
            0, int((datetime.now(tz=timezone.utc) - started).total_seconds())
        )
    return {
        "contractVersion": API_CONTRACT_VERSION,
        "backendKind": BACKEND_KIND,
        "runtimeVersion": platform.python_version(),
        "operatingSystem": platform.platform(),
        "hostname": platform.node(),
        "machineArch": platform.machine(),
        "databaseType": database_type,
        "startedAt": started_at,
        "uptimeSeconds": uptime_seconds,
        "workingDirectory": os.getcwd(),
        "storageRoot": current_app.config["DMS_STORAGE_ROOT"],
        "backupConfigPath": str(resolve_backup_config_path()),
        "envFilePath": str(resolve_env_path()),
        "logFilePath": str(log_file_path()) if log_file_path() else "",
        # 契约1.26：监控信息展示开关（false 时前端隐藏维护操作之上的监控区块）
        "monitoringEnabled": bool(
            current_app.config.get("DMS_SYSTEM_MONITOR_ENABLED", False)
        ),
    }


def runtime_config() -> list[dict[str, Any]]:
    """运行参数只读总览（脱敏：密钥类仅显示是否已配置）。"""

    def item(
        category: str, key: str, value: Any, unit: str = ""
    ) -> dict[str, Any]:
        return {"category": category, "key": key, "value": value, "unit": unit}

    config = current_app.config
    cors_configured = bool(os.environ.get("DMS_CORS_ORIGINS", "").strip())
    return [
        item(
            "认证",
            "访问令牌有效期",
            config.get("DMS_JWT_EXPIRES_SECONDS", 3600),
            "秒",
        ),
        item("认证", "保持登录有效期", 604800, "秒"),
        item("认证", "JWT密钥", "已配置" if config.get("DMS_JWT_SECRET") else "未配置"),
        item("存储", "存储根目录", config["DMS_STORAGE_ROOT"]),
        item(
            "存储",
            "信封加密KEK",
            "已配置" if config.get("DMS_STORAGE_KEK_BYTES") else "未配置（明文落盘）",
        ),
        item(
            "上传",
            "单请求上限",
            config.get("DMS_MAX_UPLOAD_SIZE", 0),
            "字节",
        ),
        item("上传", "单文件上限", config.get("DMS_MAX_FILE_SIZE", 0), "字节"),
        item("上传", "批量导入上限", config.get("DMS_BATCH_MAX_FILES", 0), "个"),
        item("业务", "业务时区", config.get("DMS_BUSINESS_TIMEZONE", "")),
        item(
            "转换",
            "LibreOffice预览",
            "启用" if config.get("DMS_OFFICE_PREVIEW_ENABLED", True) else "停用",
        ),
        item(
            "转换",
            "LibreOffice路径",
            config.get("DMS_LIBREOFFICE_EXECUTABLE", "") or "未配置",
        ),
        item(
            "转换",
            "转换超时",
            config.get("DMS_LIBREOFFICE_TIMEOUT_SECONDS", 60),
            "秒",
        ),
        item(
            "转换",
            "转换并发上限",
            config.get("DMS_LIBREOFFICE_MAX_CONCURRENCY", 2),
            "",
        ),
        item("安全", "CORS跨域白名单", "已配置" if cors_configured else "未配置（同源）"),
        item("安全", "日志文件", str(log_file_path()) if log_file_path() else "未启用"),
        item(
            "系统",
            "监控信息展示",
            "启用" if config.get("DMS_SYSTEM_MONITOR_ENABLED", False) else "停用",
        ),
    ]


# ─── 维护操作 ────────────────────────────────────────────────────────────────


def clear_preview_cache() -> dict[str, int]:
    """清空预览缓存目录（可重建），返回删除文件数与释放字节数。"""
    root = Path(current_app.config["DMS_STORAGE_ROOT"])
    preview_root = root / "preview"
    removed_files = 0
    removed_bytes = 0
    if preview_root.is_dir():
        for dirpath, _dirnames, filenames in os.walk(preview_root):
            for filename in filenames:
                target = Path(dirpath) / filename
                try:
                    removed_bytes += target.stat().st_size
                    target.unlink()
                    removed_files += 1
                except OSError:
                    continue
        # 清理空的前缀分层目录
        for child in preview_root.iterdir():
            if child.is_dir():
                try:
                    child.rmdir()
                except OSError:
                    pass
    return {"removedFiles": removed_files, "removedBytes": removed_bytes}


# ─── 运行日志 ────────────────────────────────────────────────────────────────


def log_file_path() -> Path | None:
    """运行日志文件路径（app.py 已配置文件 handler 时存在）。"""
    configured = current_app.config.get("DMS_LOG_FILE", "") if current_app else ""
    value = configured or os.environ.get("DMS_LOG_FILE", "").strip()
    if not value or value.upper() in {"OFF", "NONE"}:
        return None
    return Path(value)


_LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL")


def read_log_tail(lines: int = 500, level: str | None = None) -> dict[str, Any]:
    """读取运行日志尾部 N 行，按级别过滤（[LEVEL] 标记匹配）。"""
    path = log_file_path()
    if path is None or not path.is_file():
        return {"file": str(path) if path else "", "lines": [], "truncated": False}
    normalized_level = (level or "").strip().upper()
    if normalized_level and normalized_level not in _LOG_LEVELS:
        normalized_level = None
    collected: deque[str] = deque(maxlen=lines)
    truncated = False
    try:
        with path.open("r", encoding="utf-8", errors="replace") as handle:
            if normalized_level:
                marker = f"[{normalized_level}]"
                for raw_line in handle:
                    if marker in raw_line:
                        collected.append(raw_line.rstrip("\r\n"))
                truncated = True  # 级别过滤为全量扫描后再截尾
            else:
                for raw_line in handle:
                    collected.append(raw_line.rstrip("\r\n"))
    except OSError as exc:
        return {"file": str(path), "lines": [], "truncated": False, "error": str(exc)}
    return {
        "file": str(path),
        "lines": list(collected)[-lines:],
        "truncated": truncated,
    }