"""系统管理服务(契约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, }