cnphpbb/deploy.stack
1
0
Fork 0
forked from DevOps/deploy.stack
Code Pull Requests Activity

feat: 添加MCP时间服务器及相关工具

Browse Source 
实现一个基于Golang的MCP时间服务器，提供获取当前时间和日期功能
包含客户端示例、安装脚本和详细文档

refactor: 优化磁盘巡检脚本以支持SAS和SSD硬盘

增强磁盘巡检脚本的兼容性，改进SMART信息解析逻辑
添加硬盘类型检测和更全面的错误处理

docs: 更新README和安装说明

添加MCP时间服务器的使用文档和API说明
完善磁盘巡检报告格式和内容
This commit is contained in:
cnphpbb
2025-09-12 13:45:08 +08:00
parent 45d8181205
commit 648faac289
12 changed files with 1160 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

140
crontab/disk_inspection.py
View File

@@ -52,28 +52,47 @@ class DiskInspection:
return None
def parse_smart_info(self, output, controller_id):
"""解析smartctl输出的信息，根据实际smartctl_log格式调整"""
"""解析smartctl输出的信息，兼容SAS和SSD硬盘格式"""
if not output:
return {"error": "No data available"}
parsed_info = {}
# 提取基本信息
# 提取基本信息 - 兼容SAS和SSD格式
# 厂商信息 (SAS格式)
vendor_match = re.search(r"Vendor:\s+(.*)", output)
if vendor_match:
parsed_info["vendor"] = vendor_match.group(1)
# 产品型号 (SAS格式)
product_match = re.search(r"Product:\s+(.*)", output)
if product_match:
parsed_info["device_model"] = product_match.group(1)
# 设备型号 (SSD格式)
elif not parsed_info.get("device_model"):
device_model_match = re.search(r"Device Model:\s+(.*)", output)
if device_model_match:
parsed_info["device_model"] = device_model_match.group(1)
# 固件版本 (SAS格式)
revision_match = re.search(r"Revision:\s+(.*)", output)
if revision_match:
parsed_info["firmware_version"] = revision_match.group(1)
# 固件版本 (SSD格式)
elif not parsed_info.get("firmware_version"):
firmware_match = re.search(r"Firmware Version:\s+(.*)", output)
if firmware_match:
parsed_info["firmware_version"] = firmware_match.group(1)
# 序列号 (SAS格式)
serial_match = re.search(r"Serial number:\s+(.*)", output)
if serial_match:
parsed_info["serial_number"] = serial_match.group(1)
# 序列号 (SSD格式)
elif not parsed_info.get("serial_number"):
serial_ssd_match = re.search(r"Serial Number:\s+(.*)", output)
if serial_ssd_match:
parsed_info["serial_number"] = serial_ssd_match.group(1)
# 提取容量信息
capacity_match = re.search(r"User Capacity:\s+(.*?)\s+bytes", output)
@@ -85,27 +104,69 @@ class DiskInspection:
if rotation_match:
parsed_info["rotation_rate"] = rotation_match.group(1)
# 提取SMART健康状态
# 提取SMART健康状态 - 兼容SAS和SSD格式
# SAS格式
health_match = re.search(r"SMART Health Status:\s+(.*)", output)
if health_match:
parsed_info["health_status"] = health_match.group(1)
# SSD格式
elif not parsed_info.get("health_status"):
health_ssd_match = re.search(r"SMART overall-health self-assessment test result:\s+(.*)", output)
if health_ssd_match:
parsed_info["health_status"] = health_ssd_match.group(1)
# 提取温度信息
# 提取温度信息 - 兼容SAS和SSD格式
# SAS格式
temp_match = re.search(r"Current Drive Temperature:\s+([0-9]+)", output)
if temp_match:
parsed_info["temperature"] = temp_match.group(1)
# SSD格式 - 增强版，能处理更多格式变化
elif not parsed_info.get("temperature"):
# 尝试匹配194 Temperature_Celsius行，精确提取RAW_VALUE
temp_ssd_match = re.search(r"194 Temperature_Celsius.*?\b(\d+)\b\s*$", output)
if temp_ssd_match:
parsed_info["temperature"] = temp_ssd_match.group(1)
else:
# 尝试匹配其他可能的温度表示方式
temp_ssd_alt_match = re.search(r"Temperature_Celsius.*?\b(\d+)\b", output)
if temp_ssd_alt_match:
parsed_info["temperature"] = temp_ssd_alt_match.group(1)
# 提取通电时间
# 提取通电时间 - 兼容SAS和SSD格式
# SAS格式
power_on_match = re.search(r"Accumulated power on time, hours:minutes\s+([0-9:]+)", output)
if power_on_match:
parsed_info["power_on_time"] = power_on_match.group(1)
# 提取制造日期
parsed_info["power_on_time"] = f"{ power_on_match.group(1)}"
# SSD格式 - 增强版，能处理更多格式变化
elif not parsed_info.get("power_on_time"):
# 尝试匹配9 Power_On_Hours行，精确提取RAW_VALUE
# 改进的正则：添加\s*匹配行首空格，使用re.MULTILINE匹配多行
power_on_ssd_match = re.search(r"\s*9 Power_On_Hours.*?\b(\d+)\b\s*$", output, re.MULTILINE)
if power_on_ssd_match:
parsed_info["power_on_time"] = f"{power_on_ssd_match.group(1)}"
else:
# 尝试其他可能的通电时间表示方式，使用更宽松的正则
power_on_ssd_alt_match = re.search(r"Power_On_Hours.*?\b(\d+)\b", output, re.MULTILINE)
if power_on_ssd_alt_match:
parsed_info["power_on_time"] = f"{power_on_ssd_alt_match.group(1)}"
# 提取制造日期 - 增强版，支持多种格式
# 格式1: Manufactured in week XX of year XXXX
manufactured_match = re.search(r"Manufactured in week (\d+) of year (\d+)", output)
if manufactured_match:
parsed_info["manufactured_date"] = f"{manufactured_match.group(2)}-W{manufactured_match.group(1)}"
else:
# 格式2: 尝试从固件版本或其他字段提取制造年份信息
# 例如：固件版本通常包含年份信息 SN14546 -> 2024年
firmware = parsed_info.get("firmware_version", "")
if firmware:
# 尝试从固件版本中提取制造年份信息
year_match = re.search(r"(\d{2})\d{2}", firmware)
if year_match:
year = year_match.group(1)
# 假设年份是21世纪
parsed_info["manufactured_date"] = f"20{year}"
# 提取错误计数信息
# 提取错误计数信息 (SAS格式)
error_read_match = re.search(r"read:\s+(\d+)\s+(\d+)\s+(\d+)\s+(\d+)\s+(\d+)\s+(\d+\.\d+)\s+(\d+)", output)
if error_read_match:
parsed_info["read_errors"] = {
@@ -122,7 +183,8 @@ class DiskInspection:
"data_processed": f"{error_write_match.group(6)} GB"
}
# 提取SMART自检日志信息
# 提取SMART自检日志信息 - 兼容SAS和SSD格式
# SAS格式
self_test_match = re.search(r"SMART Self-test log\nNum\s+Test\s+Status\s+segment\s+LifeTime\s+LBA_first_err \[SK ASC ASQ\]\n(.*?)(\n\s*\n|$)", output, re.DOTALL)
if self_test_match:
test_results = []
@@ -140,11 +202,38 @@ class DiskInspection:
test_results.append(test_info)
if test_results:
parsed_info["self_test_results"] = test_results
# SSD格式
elif not parsed_info.get("self_test_results"):
self_test_ssd_match = re.search(r"SMART Self-test log structure revision number \d+\nNum\s+Test_Description\s+Status\s+Remaining\s+LifeTime\(hours\)\s+LBA_of_first_error\n(.*?)(\n\s*\n|$)", output, re.DOTALL)
if self_test_ssd_match:
test_results = []
lines = self_test_ssd_match.group(1).strip().split("\n")
for line in lines:
if line.strip() and line.startswith("#"):
parts = re.split(r"\s+", line.strip())
if len(parts) >= 5:
test_info = {
"id": parts[0].replace("#", ""),
"type": parts[1],
"status": " ".join(parts[2:4]) if parts[2] == "Completed" and parts[3] == "without" else parts[2],
"lifetime_hours": parts[4]
}
test_results.append(test_info)
if test_results:
parsed_info["self_test_results"] = test_results
# 检测硬盘类型 (根据旋转速率或特定属性)
if "rotation_rate" in parsed_info and "Solid State Device" in parsed_info["rotation_rate"]:
parsed_info["disk_type"] = "SSD"
elif "rotation_rate" in parsed_info and "rpm" in parsed_info["rotation_rate"]:
parsed_info["disk_type"] = "HDD"
else:
parsed_info["disk_type"] = "Unknown"
return parsed_info
def generate_md_report(self):
"""生成Markdown格式的报告，根据实际smartctl_log格式调整"""
"""生成Markdown格式的报告，兼容SAS和SSD硬盘格式"""
with open(self.md_report, "w") as f:
f.write(f"# 硬盘巡检报告\n\n")
f.write(f"**检查日期**: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
@@ -153,24 +242,34 @@ class DiskInspection:
# 添加健康状态概览
healthy_count = sum(1 for cid, info in self.results.items()
if "health_status" in info and info["health_status"] == "OK")
if "health_status" in info and (info["health_status"] == "OK" or info["health_status"] == "PASSED"))
detected_count = sum(1 for cid, info in self.results.items()
if "health_status" in info)
# 统计硬盘类型
hdd_count = sum(1 for cid, info in self.results.items()
if "disk_type" in info and info["disk_type"] == "HDD")
ssd_count = sum(1 for cid, info in self.results.items()
if "disk_type" in info and info["disk_type"] == "SSD")
f.write(f"### 健康状态概览\n")
f.write(f"- 总控制器数: {self.controller_count}\n")
f.write(f"- 检测到的控制器数: {detected_count}\n")
f.write(f"- 健康控制器数: {healthy_count}\n")
f.write(f"- 异常控制器数: {detected_count - healthy_count}\n\n")
f.write(f"- 异常控制器数: {detected_count - healthy_count}\n")
f.write(f"- HDD硬盘数: {hdd_count}\n")
f.write(f"- SSD硬盘数: {ssd_count}\n\n")
# 详细信息表格 - 简化版
# 详细信息表格 - 简化版，添加硬盘类型列
f.write("### 基本信息概览\n")
f.write("| 控制器ID | 厂商 | 设备型号 | 序列号 | 健康状态 | 温度(°C) | 通电时间 |\n")
f.write("|----------|------|----------|--------|----------|----------|----------|\n")
f.write("| 控制器ID | 硬盘类型 | 厂商 | 设备型号 | 序列号 | 健康状态 | 温度(°C) | 通电时间 |\n")
f.write("|----------|----------|------|----------|--------|----------|----------|----------|\n")
for controller_id in range(self.controller_count):
info = self.results.get(controller_id, {})
if "health_status" in info: # 只显示有数据的控制器
f.write(f"| {controller_id} ")
f.write(f"| {info.get('disk_type', 'N/A')} ")
f.write(f"| {info.get('vendor', 'N/A')} ")
f.write(f"| {info.get('device_model', 'N/A')} ")
f.write(f"| {info.get('serial_number', 'N/A')} ")
@@ -192,12 +291,13 @@ class DiskInspection:
f.write(f"- **固件版本**: {info.get('firmware_version', 'N/A')}\n")
f.write(f"- **容量**: {info.get('capacity', 'N/A')}\n")
f.write(f"- **旋转速率**: {info.get('rotation_rate', 'N/A')}\n")
f.write(f"- **硬盘类型**: {info.get('disk_type', 'N/A')}\n")
f.write(f"- **制造日期**: {info.get('manufactured_date', 'N/A')}\n")
f.write(f"- **健康状态**: **{info.get('health_status', 'N/A')}**\n")
f.write(f"- **当前温度**: {info.get('temperature', 'N/A')}°C\n")
f.write(f"- **累计通电时间**: {info.get('power_on_time', 'N/A')}\n")
# 错误计数信息
# 错误计数信息 (主要适用于SAS硬盘)
if "read_errors" in info:
read_err = info["read_errors"]
f.write("- **读取错误**:\n")
@@ -218,8 +318,8 @@ class DiskInspection:
for test in info["self_test_results"]:
f.write(f" - 测试 #{test['id']}: {test['type']} - {test['status']} (运行时间: {test['lifetime_hours']}小时)\n")
# 检查是否有异常
if info.get("health_status", "") != "OK":
# 检查是否有异常 - 兼容SAS(OK)和SSD(PASSED)的健康状态表示
if info.get("health_status", "") not in ["OK", "PASSED"]:
has_issues = True
f.write("\n**⚠️ 警告：此控制器状态异常，请及时关注！**\n")
@@ -230,7 +330,7 @@ class DiskInspection:
f.write("**发现异常控制器，请关注以下问题：**\n\n")
for controller_id in range(self.controller_count):
info = self.results.get(controller_id, {})
if info.get("health_status", "") != "OK":
if info.get("health_status", "") not in ["OK", "PASSED"]:
f.write(f"- **控制器 {controller_id}**: 状态为 {info.get('health_status', '未知')}\n")
f.write("\n")
else: