Python 腳本

# -*- coding: utf-8 -*-
import subprocessdef get_disk_usage():# 執行 df 命令獲取磁盤使用情況df_process = subprocess.Popen(['df', '-h', '/'], stdout=subprocess.PIPE)output, _ = df_process.communicate()output = output.decode('utf-8')# 解析輸出，獲取磁盤總量和已使用占比total_space = Noneused_percentage = Nonelines = output.split('\n')for line in lines:if line.startswith('/dev'):parts = line.split()total_space = parts[1]used_percentage = parts[4]breakprint(f"磁盤總量: {total_space}, 已使用: {used_percentage}")return total_space, used_percentagedef check_disk_full(used_percentage, threshold_percent=90):# 檢查磁盤是否快滿了if used_percentage is not None:used_percentage = int(used_percentage[:-1])  # 去掉百分號并轉換為整數if used_percentage >= threshold_percent:return Truereturn Falsedef get_memory_usage():try:# 執行 free 命令獲取內存信息free_process = subprocess.Popen(['free', '-h'], stdout=subprocess.PIPE)output, _ = free_process.communicate()output = output.decode('utf-8')# 解析輸出，獲取內存總量和已使用占比lines = output.split('\n')for line in lines:if line.startswith('Mem:'):parts = line.split()total_memory = parts[1]  # 內存總量used_memory = parts[2]   # 已使用內存print(f"內存總量: {total_memory}, 已使用: {used_memory}")return total_memory, used_memoryreturn None, Noneexcept Exception as e:print(f"Error: {e}")return None, Nonedef check_memory_insufficient(total_memory, used_memory, threshold_percent=90):if total_memory is not None and used_memory is not None:# 解析內存值，去除單位，并將字符串轉換為數值total_memory_value = float(total_memory[:-1])used_memory_value = float(used_memory[:-1])# 檢查是否內存不足used_percentage = (used_memory_value / total_memory_value) * 100if used_percentage >= threshold_percent:return Truereturn Falseif __name__ == "__main__":# 獲取磁盤使用情況total_space, used_percentage = get_disk_usage()if total_space and used_percentage:# 檢查磁盤是否快滿了（閾值默認為90%）if check_disk_full(used_percentage, threshold_percent=90):print("磁盤快滿了！")else:print("未能獲取磁盤使用情況。")# 獲取內存使用情況total_memory, used_memory = get_memory_usage()if total_memory is not None and used_memory is not None:# 檢查是否內存不足（默認閾值為90%）if check_memory_insufficient(total_memory, used_memory, threshold_percent=90):print("內存不足！")else:print("未能獲取內存使用情況。")

• 輸出結果

磁盤總量: 36G, 已使用: 65%
內存總量: 4.7G, 已使用: 1.0G

釘釘推送通知

• 釘釘自定義機器人Python腳本推送通知
• 釘釘推送配置與閾值

- ding-talk:secret: 'xxx'access-token: 'xxx'
- project:name: '項目名稱'disk-threshold: 80memory-threshold: 90

• Python

pip3 install pyyaml
pip3 install requests

# -*- coding: utf-8 -*-
import subprocess
import yaml
import time
import hashlib
import base64
import hmac
import requests
from urllib.parse import quote
import socketdef get_disk_usage():# 執行 df 命令獲取磁盤使用情況df_process = subprocess.Popen(['df', '-h', '/'], stdout=subprocess.PIPE)output, _ = df_process.communicate()output = output.decode('utf-8')# 解析輸出，獲取磁盤總量和已使用占比total_space = Noneused_percentage = Nonelines = output.split('\n')for line in lines:if line.startswith('/dev'):parts = line.split()total_space = parts[1]used_percentage = parts[4]breakprint(f"磁盤總量: {total_space}, 已使用: {used_percentage}")return total_space, used_percentagedef check_disk_full(used_percentage, threshold_percent=90):# 檢查磁盤是否快滿了if used_percentage is not None:used_percentage = int(used_percentage[:-1])  # 去掉百分號并轉換為整數if used_percentage >= threshold_percent:return Truereturn Falsedef get_memory_usage():try:# 執行 free 命令獲取內存信息free_process = subprocess.Popen(['free', '-h'], stdout=subprocess.PIPE)output, _ = free_process.communicate()output = output.decode('utf-8')# 解析輸出，獲取內存總量和已使用占比lines = output.split('\n')for line in lines:if line.startswith('Mem:'):parts = line.split()total_memory = parts[1]  # 內存總量used_memory = parts[2]   # 已使用內存print(f"內存總量: {total_memory}, 已使用: {used_memory}")return total_memory, used_memoryreturn None, Noneexcept Exception as e:print(f"Error: {e}")return None, Nonedef check_memory_insufficient(total_memory, used_memory, threshold_percent=90):if total_memory is not None and used_memory is not None:# 解析內存值，去除單位，并將字符串轉換為數值total_memory_value = float(total_memory[:-1])used_memory_value = float(used_memory[:-1])# 檢查是否內存不足used_percentage = (used_memory_value / total_memory_value) * 100if used_percentage >= threshold_percent:return Truereturn Falsedef read_yaml(file_path):with open(file_path, 'r', encoding='utf-8') as file:try:data = yaml.safe_load(file)return dataexcept yaml.YAMLError as e:print(f"讀取 YAML 文件時出錯：{e}")return Nonedef get_local_ip():try:# 創建一個 UDP 套接字sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)sock.connect(('8.8.8.8', 80))  # 連接 Google DNS# 獲取本地 IP 地址local_ip = sock.getsockname()[0]return local_ipexcept Exception as e:print(f"Error: {e}")return Nonedef dingTalkSign(dingTalkSecret):# 獲取當前時間戳，并將其轉換為毫秒級timestamp = int(time.time() * 1000)# 將時間戳和釘釘應用的密鑰拼接在一起，將拼接后的字符串轉換為字節數組signBefore = ('%s\n%s' % (timestamp, dingTalkSecret)).encode('utf-8')# 用HMAC-SHA256算法對字節數組進行簽名hsha256 = hmac.new(dingTalkSecret.encode('utf-8'), signBefore, hashlib.sha256)# 將簽名進行Base64編碼，將編碼后的簽名進行URL編碼sign = quote(base64.b64encode(hsha256.digest()))return {"timestamp": timestamp, "sign": sign}def sendMessage(dingTalkUrl='', dingTalkSecret=None, message='', atMobiles=[], isAtAll=False):print("發送內容:", message, atMobiles, isAtAll)json = {"msgtype": "text","text": {"content": message,},"at": {"atMobiles": atMobiles,"isAtAll": isAtAll}}sign = dingTalkSign(dingTalkSecret)response = requests.post(url=dingTalkUrl, params=sign, json=json)print("響應內容:", response.json())if __name__ == "__main__":local_ip = get_local_ip()file_data = read_yaml("config.yml")if file_data:for entry in file_data:if 'ding-talk' in entry:dingTalkSecret = entry['ding-talk']['secret']access_token = entry['ding-talk']['access-token']if dingTalkSecret is None:print("未配置釘釘機器人密鑰")if access_token is None:print("未配置釘釘機器人Token")else:# 自定義機器人推送地址dingTalkUrl = f"https://oapi.dingtalk.com/robot/send?access_token={access_token}"if dingTalkSecret is not None and 'project' in entry:project_name = entry['project']['name']disk_threshold = entry['project']['disk-threshold']memory_threshold = entry['project']['memory-threshold']# 獲取磁盤使用情況total_space, used_percentage = get_disk_usage()total_memory, used_memory = get_memory_usage()if (total_space and used_percentage) or (total_memory and used_memory):# 檢查磁盤是否快滿了（閾值默認為90%）if check_disk_full(used_percentage, threshold_percent=disk_threshold):sendMessage(dingTalkSecret=dingTalkSecret, dingTalkUrl=dingTalkUrl, isAtAll=True,message=f'項目：{project_name}\n內網：{local_ip}\n磁盤：{total_space} / {used_percentage} (總/已用)\n內存：{total_memory} / {used_memory} (總/已用)\n磁盤不足，請及時擴容！')# 檢查是否內存不足（默認閾值為90%）if check_memory_insufficient(total_memory, used_memory, threshold_percent=memory_threshold):sendMessage(dingTalkSecret=dingTalkSecret, dingTalkUrl=dingTalkUrl, isAtAll=True,message=f'項目：{project_name}\n內網：{local_ip}\n磁盤：{total_space} / {used_percentage} (總/已用)\n內存：{total_memory} / {used_memory} (總/已用)\n內存不足，請及時擴容！')else:print("未能獲取磁盤使用情況。")

定時任務

• 在線生成CRON：https://tool.lu/crontab

# 查看python3安裝位置
which python3
# 添加定時任務
crontab -e

# 定時執行Python腳本,根據自己需求配置執行時間
20 9 * * * /usr/bin/python3 /u01/setup.py