社群 學習 工具庫 休閒
搜尋
繁体中文
首頁 > 後端開發 > Python教學 > 超硬核！ 11個非常實用的 Python 和 Shell 拿來就用腳本實例！

超硬核！ 11個非常實用的 Python 和 Shell 拿來就用腳本實例！

WBOY
發布： 2023-04-12 13:52:05
轉載
1649 人瀏覽過

超硬核！ 11個非常實用的 Python 和 Shell 拿來就用腳本實例！

Python 脚本部分实例：企业微信告警、FTP 客户端、SSH 客户端、Saltstack 客户端、vCenter 客户端、获取域名 ssl 证书过期时间、发送今天的天气预报以及未来的天气趋势图；

Shell 脚本部分实例：SVN 完整备份、Zabbix 监控用户密码过期、构建本地 YUM 以及上篇文章中有读者的需求（负载高时，查出占用比较高的进程脚本并存储或推送通知）；

篇幅有些长，还请大家耐心翻到文末，毕竟有彩蛋。

超硬核！ 11個非常實用的 Python 和 Shell 拿來就用腳本實例！

Python 脚本部分

企业微信告警

此脚本通过企业微信应用，进行微信告警，可用于 Zabbix 监控。

# -*- coding: utf-8 -*-
import requests
import json
class DLF:
 def __init__(self, corpid, corpsecret):
 self.url = "https://qyapi.weixin.qq.com/cgi-bin"
 self.corpid = corpid
 self.corpsecret = corpsecret
 self._token = self._get_token()
 def _get_token(self):
 '''
 获取企业微信API接口的access_token
 :return:
 '''
 token_url = self.url + "/gettoken?corpid=%s&corpsecret=%s" %(self.corpid, self.corpsecret)
 try:
 res = requests.get(token_url).json()
 token = res['access_token']
 return token
 except Exception as e:
 return str(e)
 def _get_media_id(self, file_obj):
 get_media_url = self.url + "/media/upload?access_token={}&type=file".format(self._token)
 data = {"media": file_obj}
 try:
 res = requests.post(url=get_media_url, files=data)
 media_id = res.json()['media_id']
 return media_id
 except Exception as e:
 return str(e)
 def send_text(self, agentid, content, touser=None, toparty=None):
 send_msg_url = self.url + "/message/send?access_token=%s" % (self._token)
 send_data = {
 "touser": touser,
 "toparty": toparty,
 "msgtype": "text",
 "agentid": agentid,
 "text": {
 "content": content
 }
 }
 try:
 res = requests.post(send_msg_url, data=json.dumps(send_data))
 except Exception as e:
 return str(e)
 def send_image(self, agentid, file_obj, touser=None, toparty=None):
 media_id = self._get_media_id(file_obj)
 send_msg_url = self.url + "/message/send?access_token=%s" % (self._token)
 send_data = {
 "touser": touser,
 "toparty": toparty,
 "msgtype": "image",
 "agentid": agentid,
 "image": {
 "media_id": media_id
}
 }
 try:
 res = requests.post(send_msg_url, data=json.dumps(send_data))
 except Exception as e:
 return str(e)
登入後複製

FTP 客户端

通过 ftplib 模块操作 ftp 服务器，进行上传下载等操作。

# -*- coding: utf-8 -*-
from ftplib import FTP
from os import path
import copy
class FTPClient:
 def __init__(self, host, user, passwd, port=21):
 self.host = host
 self.user = user
 self.passwd = passwd
 self.port = port
 self.res = {'status': True, 'msg': None}
 self._ftp = None
 self._login()
 def _login(self):
 '''
 登录FTP服务器
 :return: 连接或登录出现异常时返回错误信息
 '''
 try:
 self._ftp = FTP()
 self._ftp.connect(self.host, self.port, timeout=30)
 self._ftp.login(self.user, self.passwd)
 except Exception as e:
 return e
 def upload(self, localpath, remotepath=None):
 '''
 上传ftp文件
 :param localpath: local file path
 :param remotepath: remote file path
 :return:
 '''
 if not localpath: return 'Please select a local file. '
 # 读取本地文件
 # fp = open(localpath, 'rb')
 # 如果未传递远程文件路径，则上传到当前目录，文件名称同本地文件
 if not remotepath:
 remotepath = path.basename(localpath)
 # 上传文件
 self._ftp.storbinary('STOR ' + remotepath, localpath)
 # fp.close()
 def download(self, remotepath, localpath=None):
 '''
 localpath
 :param localpath: local file path
 :param remotepath: remote file path
 :return:
 '''
 if not remotepath: return 'Please select a remote file. '
 # 如果未传递本地文件路径，则下载到当前目录，文件名称同远程文件
 if not localpath:
 localpath = path.basename(remotepath)
 # 如果localpath是目录的话就和remotepath的basename拼接
 if path.isdir(localpath):
 localpath = path.join(localpath, path.basename(remotepath))
 # 写入本地文件
 fp = open(localpath, 'wb')
 # 下载文件
 self._ftp.retrbinary('RETR ' + remotepath, fp.write)
 fp.close()
 def nlst(self, dir='/'):
 '''
 查看目录下的内容
 :return: 以列表形式返回目录下的所有内容
 '''
 files_list = self._ftp.nlst(dir)
 return files_list
 def rmd(self, dir=None):
 '''
 删除目录
 :param dir: 目录名称
 :return: 执行结果
 '''
 if not dir: return 'Please input dirname'
 res = copy.deepcopy(self.res)
 try:
 del_d = self._ftp.rmd(dir)
 res['msg'] = del_d
 except Exception as e:
 res['status'] = False
 res['msg'] = str(e)
 return res
 def mkd(self, dir=None):
 '''
 创建目录
 :param dir: 目录名称
 :return: 执行结果
 '''
 if not dir: return 'Please input dirname'
 res = copy.deepcopy(self.res)
 try:
 mkd_d = self._ftp.mkd(dir)
 res['msg'] = mkd_d
 except Exception as e:
 res['status'] = False
 res['msg'] = str(e)
 return res
 def del_file(self, filename=None):
 '''
 删除文件
 :param filename: 文件名称
 :return: 执行结果
 '''
 if not filename: return 'Please input filename'
 res = copy.deepcopy(self.res)
 try:
 del_f = self._ftp.delete(filename)
 res['msg'] = del_f
 except Exception as e:
 res['status'] = False
 res['msg'] = str(e)
 return res
 def get_file_size(self, filenames=[]):
 '''
 获取文件大小,单位是字节
 判断文件类型
 :param filename: 文件名称
 :return: 执行结果
 '''
 if not filenames: return {'msg': 'This is an empty directory'}
 res_l = []
 for file in filenames:
 res_d = {}
 # 如果是目录或者文件不存在就会报错
 try:
 size = self._ftp.size(file)
 type = 'f'
 except:
 # 如果是路径的话size显示 - , file末尾加/ （/dir/）
 size = '-'
 type = 'd'
 file = file + '/'
 res_d['filename'] = file
 res_d['size'] = size
 res_d['type'] = type
 res_l.append(res_d)
 return res_l
 def rename(self, old_name=None, new_name=None):
 '''
 重命名
 :param old_name: 旧的文件或者目录名称
 :param new_name: 新的文件或者目录名称
 :return: 执行结果
 '''
 if not old_name or not new_name: return 'Please input old_name and new_name'
 res = copy.deepcopy(self.res)
 try:
 rename_f = self._ftp.rename(old_name, new_name)
 res['msg'] = rename_f
 except Exception as e:
 res['status'] = False
 res['msg'] = str(e)
 return res
 def close(self):
 '''
 退出ftp连接
 :return:
 '''
 try:
 # 向服务器发送quit命令
 self._ftp.quit()
 except Exception:
 return 'No response from server'
 finally:
 # 客户端单方面关闭连接
 self._ftp.close()
登入後複製

SSH 客户端

此脚本仅用于通过 key 连接，如需要密码连接，简单修改下即可。

# -*- coding: utf-8 -*-
import paramiko
class SSHClient:
 def __init__(self, host, port, user, pkey):
 self.ssh_host = host
 self.ssh_port = port
 self.ssh_user = user
 self.private_key = paramiko.RSAKey.from_private_key_file(pkey)
 self.ssh = None
 self._connect()
 def _connect(self):
 self.ssh = paramiko.SSHClient()
 self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
 try:
 self.ssh.connect(hostname=self.ssh_host, port=self.ssh_port, username=self.ssh_user, pkey=self.private_key, timeout=10)
 except:
 return 'ssh connect fail'
 def execute_command(self, command):
 stdin, stdout, stderr = self.ssh.exec_command(command)
 out = stdout.read()
 err = stderr.read()
 return out, err
 def close(self):
 self.ssh.close()
登入後複製

Saltstack 客户端

通过 api 对 Saltstack 服务端进行操作，执行命令。

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import requests
import json
import copy
class SaltApi:
 """
 定义salt api接口的类
 初始化获得token
 """
 def __init__(self):
 self.url = "http://172.85.10.21:8000/"
 self.username = "saltapi"
 self.password = "saltapi"
 self.headers = {"Content-type": "application/json"}
 self.params = {'client': 'local', 'fun': None, 'tgt': None, 'arg': None}
 self.login_url = self.url + "login"
 self.login_params = {'username': self.username, 'password': self.password, 'eauth': 'pam'}
 self.token = self.get_data(self.login_url, self.login_params)['token']
 self.headers['X-Auth-Token'] = self.token
 def get_data(self, url, params):
 '''
 请求url获取数据
 :param url: 请求的url地址
 :param params: 传递给url的参数
 :return: 请求的结果
 '''
 send_data = json.dumps(params)
 request = requests.post(url, data=send_data, headers=self.headers)
 response = request.json()
 result = dict(response)
 return result['return'][0]
 def get_auth_keys(self):
 '''
 获取所有已经认证的key
 :return:
 '''
 data = copy.deepcopy(self.params)
 data['client'] = 'wheel'
 data['fun'] = 'key.list_all'
 result = self.get_data(self.url, data)
 try:
 return result['data']['return']['minions']
 except Exception as e:
 return str(e)
 def get_grains(self, tgt, arg='id'):
 """
 获取系统基础信息
 :tgt: 目标主机
 :return:
 """
 data = copy.deepcopy(self.params)
 if tgt:
 data['tgt'] = tgt
 else:
 data['tgt'] = '*'
 data['fun'] = 'grains.item'
 data['arg'] = arg
 result = self.get_data(self.url, data)
 return result
 def execute_command(self, tgt, fun='cmd.run', arg=None, tgt_type='list', salt_async=False):
 """
 执行saltstack 模块命令，类似于salt '*' cmd.run 'command'
 :param tgt: 目标主机
 :param fun: 模块方法 可为空
 :param arg: 传递参数 可为空
 :return: 执行结果
 """
 data = copy.deepcopy(self.params)
 if not tgt: return {'status': False, 'msg': 'target host not exist'}
 if not arg:
 data.pop('arg')
 else:
 data['arg'] = arg
 if tgt != '*':
 data['tgt_type'] = tgt_type
 if salt_async: data['client'] = 'local_async'
 data['fun'] = fun
 data['tgt'] = tgt
 result = self.get_data(self.url, data)
 return result
 def jobs(self, fun='detail', jid=None):
 """
 任务
 :param fun: active, detail
 :param jod: Job ID
 :return: 任务执行结果
 """
 data = {'client': 'runner'}
 data['fun'] = fun
 if fun == 'detail':
 if not jid: return {'success': False, 'msg': 'job id is none'}
 data['fun'] = 'jobs.lookup_jid'
 data['jid'] = jid
 else:
 return {'success': False, 'msg': 'fun is active or detail'}
 result = self.get_data(self.url, data)
 return result
登入後複製

vCenter 客户端

通过官方 SDK 对 vCenter 进行日常操作，此脚本是我用于 cmdb 平台的，自动获取主机信息，存入数据库。

from pyVim.connect import SmartConnect, Disconnect, SmartConnectNoSSL
from pyVmomi import vim
from asset import models
import atexit
class Vmware:
 def __init__(self, ip, user, password, port, idc, vcenter_id):
 self.ip = ip
 self.user = user
 self.password = password
 self.port = port
 self.idc_id = idc
 self.vcenter_id = vcenter_id
 def get_obj(self, content, vimtype, name=None):
 '''
 列表返回,name 可以指定匹配的对象
 '''
 container = content.viewManager.CreateContainerView(content.rootFolder, vimtype, True)
 obj = [ view for view in container.view ]
 return obj
 def get_esxi_info(self):
 # 宿主机信息
 esxi_host = {}
 res = {"connect_status": True, "msg": None}
 try:
 # connect this thing
 si = SmartConnectNoSSL(host=self.ip, user=self.user, pwd=self.password, port=self.port, connectionPoolTimeout=60)
 except Exception as e:
 res['connect_status'] = False
 try:
 res['msg'] = ("%s Caught vmodl fault : " + e.msg) % (self.ip)
 except Exception as e:
 res['msg'] = '%s: connection error' % (self.ip)
 return res
 # disconnect this thing
 atexit.register(Disconnect, si)
 content = si.RetrieveContent()
 esxi_obj = self.get_obj(content, [vim.HostSystem])
 for esxi in esxi_obj:
 esxi_host[esxi.name] = {}
 esxi_host[esxi.name]['idc_id'] = self.idc_id
 esxi_host[esxi.name]['vcenter_id'] = self.vcenter_id
 esxi_host[esxi.name]['server_ip'] = esxi.name
 esxi_host[esxi.name]['manufacturer'] = esxi.summary.hardware.vendor
 esxi_host[esxi.name]['server_model'] = esxi.summary.hardware.model
 for i in esxi.summary.hardware.otherIdentifyingInfo:
 if isinstance(i, vim.host.SystemIdentificationInfo):
 esxi_host[esxi.name]['server_sn'] = i.identifierValue
 # 系统名称
 esxi_host[esxi.name]['system_name'] = esxi.summary.config.product.fullName
 # cpu总核数
 esxi_cpu_total = esxi.summary.hardware.numCpuThreads
 # 内存总量 GB
 esxi_memory_total = esxi.summary.hardware.memorySize / 1024 / 1024 / 1024
 # 获取硬盘总量 GB
 esxi_disk_total = 0
 for ds in esxi.datastore:
 esxi_disk_total += ds.summary.capacity / 1024 / 1024 / 1024
 # 默认配置4核8G100G，根据这个配置计算剩余可分配虚拟机
 default_configure = {
 'cpu': 4,
 'memory': 8,
 'disk': 100
 }
 esxi_host[esxi.name]['vm_host'] = []
 vm_usage_total_cpu = 0
 vm_usage_total_memory = 0
 vm_usage_total_disk = 0
 # 虚拟机信息
 for vm in esxi.vm:
 host_info = {}
 host_info['vm_name'] = vm.name
 host_info['power_status'] = vm.runtime.powerState
 host_info['cpu_total_kernel'] = str(vm.config.hardware.numCPU) + '核'
 host_info['memory_total'] = str(vm.config.hardware.memoryMB) + 'MB'
 host_info['system_info'] = vm.config.guestFullName
 disk_info = ''
 disk_total = 0
 for d in vm.config.hardware.device:
 if isinstance(d, vim.vm.device.VirtualDisk):
 disk_total += d.capacityInKB / 1024 / 1024
 disk_info += d.deviceInfo.label + ": " +str((d.capacityInKB) / 1024 / 1024) + ' GB' + ','
 host_info['disk_info'] = disk_info
 esxi_host[esxi.name]['vm_host'].append(host_info)
 # 计算当前宿主机可用容量：总量 - 已分配的
 if host_info['power_status'] == 'poweredOn':
 vm_usage_total_cpu += vm.config.hardware.numCPU
 vm_usage_total_disk += disk_total
 vm_usage_total_memory += (vm.config.hardware.memoryMB / 1024)
 esxi_cpu_free = esxi_cpu_total - vm_usage_total_cpu
 esxi_memory_free = esxi_memory_total - vm_usage_total_memory
 esxi_disk_free = esxi_disk_total - vm_usage_total_disk
 esxi_host[esxi.name]['cpu_info'] = 'Total: %d核, Free: %d核' % (esxi_cpu_total, esxi_cpu_free)
 esxi_host[esxi.name]['memory_info'] = 'Total: %dGB, Free: %dGB' % (esxi_memory_total, esxi_memory_free)
 esxi_host[esxi.name]['disk_info'] = 'Total: %dGB, Free: %dGB' % (esxi_disk_total, esxi_disk_free)
 # 计算cpu 内存 磁盘按照默认资源分配的最小值，即为当前可分配资源
 if esxi_cpu_free < 4 or esxi_memory_free < 8 or esxi_disk_free < 100:
 free_allocation_vm_host = 0
 else:
 free_allocation_vm_host = int(min(
 [
 esxi_cpu_free / default_configure['cpu'],
 esxi_memory_free / default_configure['memory'],
 esxi_disk_free / default_configure['disk']
 ]
 ))
 esxi_host[esxi.name]['free_allocation_vm_host'] = free_allocation_vm_host
 esxi_host['connect_status'] = True
 return esxi_host
 def write_to_db(self):
 esxi_host = self.get_esxi_info()
 # 连接失败
 if not esxi_host['connect_status']:
 return esxi_host
 del esxi_host['connect_status']
 for machine_ip in esxi_host:
 # 物理机信息
 esxi_host_dict = esxi_host[machine_ip]
 # 虚拟机信息
 virtual_host = esxi_host[machine_ip]['vm_host']
 del esxi_host[machine_ip]['vm_host']
 obj = models.EsxiHost.objects.create(**esxi_host_dict)
 obj.save()
 for host_info in virtual_host:
 host_info['management_host_id'] = obj.id
 obj2 = models.virtualHost.objects.create(**host_info)
 obj2.save()
登入後複製

获取域名 ssl 证书过期时间

用于 zabbix 告警

import re
import sys
import time
import subprocess
from datetime import datetime
from io import StringIO
def main(domain):
 f = StringIO()
 comm = f"curl -Ivs https://{domain} --connect-timeout 10"
 result = subprocess.getstatusoutput(comm)
 f.write(result[1])
 try:
 m = re.search('start date: (.*?)n.*?expire date: (.*?)n.*?common name: (.*?)n.*?issuer: CN=(.*?)n', f.getvalue(), re.S)
 start_date = m.group(1)
 expire_date = m.group(2)
 common_name = m.group(3)
 issuer = m.group(4)
 except Exception as e:
 return 999999999
 # time 字符串转时间数组
 start_date = time.strptime(start_date, "%b %d %H:%M:%S %Y GMT")
 start_date_st = time.strftime("%Y-%m-%d %H:%M:%S", start_date)
 # datetime 字符串转时间数组
 expire_date = datetime.strptime(expire_date, "%b %d %H:%M:%S %Y GMT")
 expire_date_st = datetime.strftime(expire_date,"%Y-%m-%d %H:%M:%S")
 # 剩余天数
 remaining = (expire_date-datetime.now()).days
return remaining
if __name__ == "__main__":
domain = sys.argv[1]
 remaining_days = main(domain)
 print(remaining_days)
登入後複製

发送今天的天气预报以及未来的天气趋势图

超硬核！ 11個非常實用的 Python 和 Shell 拿來就用腳本實例！

此脚本用于给老婆大人发送今天的天气预报以及未来的天气趋势图，现在微信把网页端禁止了，没法发送到微信了，我是通过企业微信进行通知的，需要把你老婆大人拉到企业微信，无兴趣的小伙伴跳过即可。

# -*- coding: utf-8 -*-
 import requests
 import json
 import datetime
 def weather(city):
 url = "http://wthrcdn.etouch.cn/weather_mini?city=%s" % city
 try:
 data = requests.get(url).json()['data']
 city = data['city']
 ganmao = data['ganmao']
 today_weather = data['forecast'][0]
 res = "老婆今天是{}n今天天气概况n城市: {:<10}n时间: {:<10}n高温: {:<10}n低温: {:<10}n风力: {:<10}n风向: {:<10}n天气: {:<10}nn稍后会发送近期温度趋势图，请注意查看。
 ".format(
 ganmao,
 city,
 datetime.datetime.now().strftime('%Y-%m-%d'),
 today_weather['high'].split()[1],
 today_weather['low'].split()[1],
 today_weather['fengli'].split('[')[2].split(']')[0],
 today_weather['fengxiang'],today_weather['type'],
 )
 return {"source_data": data, "res": res}
 except Exception as e:
 return str(e)
 ```
 + 获取天气预报趋势图
 ```python
 # -*- coding: utf-8 -*-
 import matplotlib.pyplot as plt
 import re
 import datetime
 def Future_weather_states(forecast, save_path, day_num=5):
 '''
 展示未来的天气预报趋势图
 :param forecast: 天气预报预测的数据
 :param day_num: 未来几天
 :return: 趋势图
 '''
 future_forecast = forecast
 dict={}
 for i in range(day_num):
 data = []
 date = future_forecast[i]["date"]
 date = int(re.findall("d+",date)[0])
 data.append(int(re.findall("d+", future_forecast[i]["high"])[0]))
 data.append(int(re.findall("d+", future_forecast[i]["low"])[0]))
 data.append(future_forecast[i]["type"])
 dict[date] = data
 data_list = sorted(dict.items())
 date=[]
 high_temperature = []
 low_temperature = []
 for each in data_list:
 date.append(each[0])
 high_temperature.append(each[1][0])
 low_temperature.append(each[1][1])
 fig = plt.plot(date,high_temperature,"r",date,low_temperature,"b")
 current_date = datetime.datetime.now().strftime('%Y-%m')
 plt.rcParams['font.sans-serif'] = ['SimHei']
 plt.rcParams['axes.unicode_minus'] = False
 plt.xlabel(current_date)
 plt.ylabel("℃")
 plt.legend(["高温", "低温"])
 plt.xticks(date)
 plt.title("最近几天温度变化趋势")
 plt.savefig(save_path)
 ```
 + 发送到企业微信
 ```python
 # -*- coding: utf-8 -*-
 import requests
 import json
 class DLF:
 def __init__(self, corpid, corpsecret):
 self.url = "https://qyapi.weixin.qq.com/cgi-bin"
 self.corpid = corpid
 self.corpsecret = corpsecret
 self._token = self._get_token()
 def _get_token(self):
 '''
 获取企业微信API接口的access_token
 :return:
 '''
 token_url = self.url + "/gettoken?corpid=%s&corpsecret=%s" %(self.corpid, self.corpsecret)
 try:
 res = requests.get(token_url).json()
 token = res['access_token']
 return token
 except Exception as e:
 return str(e)
 def _get_media_id(self, file_obj):
 get_media_url = self.url + "/media/upload?access_token={}&type=file".format(self._token)
 data = {"media": file_obj}
 try:
 res = requests.post(url=get_media_url, files=data)
 media_id = res.json()['media_id']
 return media_id
 except Exception as e:
 return str(e)
 def send_text(self, agentid, content, touser=None, toparty=None):
 send_msg_url = self.url + "/message/send?access_token=%s" % (self._token)
 send_data = {
 "touser": touser,
 "toparty": toparty,
 "msgtype": "text",
 "agentid": agentid,
 "text": {
 "content": content
 }
 }
 try:
 res = requests.post(send_msg_url, data=json.dumps(send_data))
 except Exception as e:
 return str(e)
 def send_image(self, agentid, file_obj, touser=None, toparty=None):
 media_id = self._get_media_id(file_obj)
 send_msg_url = self.url + "/message/send?access_token=%s" % (self._token)
 send_data = {
 "touser": touser,
 "toparty": toparty,
 "msgtype": "image",
 "agentid": agentid,
 "image": {
 "media_id": media_id
}
 }
 try:
 res = requests.post(send_msg_url, data=json.dumps(send_data))
 except Exception as e:
 return str(e)
+ main脚本
# -*- coding: utf-8 -*-
from plugins.weather_forecast import weather
from plugins.trend_chart import Future_weather_states
from plugins.send_wechat import DLF
import os
# 企业微信相关信息
corpid = "xxx"
corpsecret = "xxx"
agentid = "xxx"
# 天气预报趋势图保存路径
_path = os.path.dirname(os.path.abspath(__file__))
save_path = os.path.join(_path ,'./tmp/weather_forecast.jpg')
# 获取天气预报信息
content = weather("大兴")
# 发送文字消息
dlf = DLF(corpid, corpsecret)
dlf.send_text(agentid=agentid, content=content['res'], toparty='1')
# 生成天气预报趋势图
Future_weather_states(content['source_data']['forecast'], save_path)
# 发送图片消息
file_obj = open(save_path, 'rb')
dlf.send_image(agentid=agentid, toparty='1', file_obj=file_obj)
登入後複製

Shell 脚本部分

SVN 完整备份

通过 hotcopy 进行 SVN 完整备份，备份保留 7 天。

#!/bin/bash
# Filename :svn_backup_repos.sh
# Date :2020/12/14
# Author :JakeTian
# Email:JakeTian@***.com
# Crontab:59 23 * * * /bin/bash $BASE_PATH/svn_backup_repos.sh >/dev/null 2>&1
# Notes:将脚本加入crontab中，每天定时执行
# Description:SVN完全备份
set -e
SRC_PATH="/opt/svndata"
DST_PATH="/data/svnbackup"
LOG_FILE="$DST_PATH/logs/svn_backup.log"
SVN_BACKUP_C="/bin/svnadmin hotcopy"
SVN_LOOK_C="/bin/svnlook youngest"
TODAY=$(date +'%F')
cd $SRC_PATH
ALL_REPOS=$(find ./ -maxdepth 1 -type d ! -name 'httpd' -a ! -name 'bak' | tr -d './')
# 创建备份目录，备份脚本日志目录
test -d $DST_PATH || mkdir -p $DST_PATH
test -d $DST_PATH/logs || mkdir $DST_PATH/logs
test -d $DST_PATH/$TODAY || mkdir $DST_PATH/$TODAY
# 备份repos文件
for repo in $ALL_REPOS
do
 $SVN_BACKUP_C $SRC_PATH/$repo $DST_PATH/$TODAY/$repo
 # 判断备份是否完成
 if $SVN_LOOK_C $DST_PATH/$TODAY/$repo;then
echo "$TODAY: $repo Backup Success" >> $LOG_FILE
 else
 echo "$TODAY: $repo Backup Fail" >> $LOG_FILE
 fi
done
# # 备份用户密码文件和权限文件
cp -p authz access.conf $DST_PATH/$TODAY
# 日志文件转储
mv $LOG_FILE $LOG_FILE-$TODAY
# 删除七天前的备份
seven_days_ago=$(date -d "7 days ago" +'%F')
rm -rf $DST_PATH/$seven_days_ago
登入後複製

zabbix 监控用户密码过期

用于 Zabbix 监控 Linux 系统用户（shell 为 /bin/bash 和 /bin/sh）密码过期，密码有效期剩余 7 天触发加自动发现用户。

#!/bin/bash
diskarray=(`awk -F':' '$NF ~ //bin/bash/||//bin/sh/{print $1}' /etc/passwd`)
length=${#diskarray[@]}
printf "{n"
printf't'""data":["
for ((i=0;i<$length;i++))
do
 printf 'ntt{'
 printf ""{#USER_NAME}":"${diskarray[$i]}"}"
 if [ $i -lt $[$length-1] ];then
 printf ','
 fi
done
printf"nt]n"
printf "}n"
检查用户密码过期
#!/bin/bash
export LANG=en_US.UTF-8
SEVEN_DAYS_AGO=$(date -d '-7 day' +'%s')
user="$1"
# 将Sep 09, 2018格式的时间转换成unix时间
expires_date=$(sudo chage -l $user | awk -F':' '/Password expires/{print $NF}' | sed -n 's/^ //p')
if [[ "$expires_date" != "never" ]];then
 expires_date=$(date -d "$expires_date" +'%s')
 if [ "$expires_date" -le "$SEVEN_DAYS_AGO" ];then
 echo "1"
 else
 echo "0"
 fi
else
 echo "0"
fi
登入後複製

构建本地YUM

通过 rsync 的方式同步 yum，通过 nginx 只做 http yum 站点；

但是 centos6 的镜像最近都不能用了，国内貌似都禁用了，如果找到合适的自行更换地址。

#!/bin/bash
# 更新yum镜像
RsyncCommand="rsync -rvutH -P --delete --delete-after --delay-updates --bwlimit=1000"
DIR="/app/yumData"
LogDir="$DIR/logs"
Centos6Base="$DIR/Centos6/x86_64/Base"
Centos7Base="$DIR/Centos7/x86_64/Base"
Centos6Epel="$DIR/Centos6/x86_64/Epel"
Centos7Epel="$DIR/Centos7/x86_64/Epel"
Centos6Salt="超硬核！ 11個非常實用的 Python 和 Shell 拿來就用腳本實例！"
Centos7Salt="$DIR/Centos7/x86_64/Salt"
Centos6Update="$DIR/Centos6/x86_64/Update"
Centos7Update="$DIR/Centos7/x86_64/Update"
Centos6Docker="$DIR/Centos6/x86_64/Docker"
Centos7Docker="$DIR/Centos7/x86_64/Docker"
Centos6Mysql5_7="$DIR/Centos6/x86_64/Mysql/Mysql5.7"
Centos7Mysql5_7="$DIR/Centos7/x86_64/Mysql/Mysql5.7"
Centos6Mysql8_0="$DIR/Centos6/x86_64/Mysql/Mysql8.0"
Centos7Mysql8_0="$DIR/Centos7/x86_64/Mysql/Mysql8.0"
MirrorDomain="rsync://rsync.mirrors.ustc.edu.cn"
# 目录不存在就创建
check_dir(){
 for dir in $*
 do
 test -d $dir || mkdir -p $dir
 done
}
# 检查rsync同步结果
check_rsync_status(){
 if [ $? -eq 0 ];then
 echo "rsync success" >> $1
 else
 echo "rsync fail" >> $1
 fi
}
check_dir $DIR $LogDir $Centos6Base $Centos7Base $Centos6Epel $Centos7Epel $Centos6Salt $Centos7Salt $Centos6Update $Centos7Update $Centos6Docker $Centos7Docker $Centos6Mysql5_7 $Centos7Mysql5_7 $Centos6Mysql8_0 $Centos7Mysql8_0
# Base yumrepo
#$RsyncCommand "$MirrorDomain"/repo/centos/6/os/x86_64/ $Centos6Base >> "$LogDir/centos6Base.log" 2>&1
# check_rsync_status "$LogDir/centos6Base.log"
$RsyncCommand "$MirrorDomain"/repo/centos/7/os/x86_64/ $Centos7Base >> "$LogDir/centos7Base.log" 2>&1
check_rsync_status "$LogDir/centos7Base.log"
# Epel yumrepo
# $RsyncCommand "$MirrorDomain"/repo/epel/6/x86_64/ $Centos6Epel >> "$LogDir/centos6Epel.log" 2>&1
# check_rsync_status "$LogDir/centos6Epel.log"
$RsyncCommand "$MirrorDomain"/repo/epel/7/x86_64/ $Centos7Epel >> "$LogDir/centos7Epel.log" 2>&1
check_rsync_status "$LogDir/centos7Epel.log"
# SaltStack yumrepo
# $RsyncCommand "$MirrorDomain"/repo/salt/yum/redhat/6/x86_64/ $Centos6Salt >> "$LogDir/centos6Salt.log" 2>&1
# ln -s $Centos6Salt/archive/$(ls $Centos6Salt/archive | tail -1) $Centos6Salt/latest
# check_rsync_status "$LogDir/centos6Salt.log"
$RsyncComman "$MirrorDomain"/repo/salt/yum/redhat/7/x86_64/ $Centos7Salt >> "$LogDir/centos7Salt.log" 2>&1
check_rsync_status "$LogDir/centos7Salt.log"
# ln -s $Centos7Salt/archive/$(ls $Centos7Salt/archive | tail -1) $Centos7Salt/latest
# Docker yumrepo
$RsyncCommand "$MirrorDomain"/repo/docker-ce/linux/centos/7/x86_64/stable/ $Centos7Docker >> "$LogDir/centos7Docker.log" 2>&1
check_rsync_status "$LogDir/centos7Docker.log"
# centos update yumrepo
# $RsyncCommand "$MirrorDomain"/repo/centos/6/updates/x86_64/ $Centos6Update >> "$LogDir/centos6Update.log" 2>&1
# check_rsync_status "$LogDir/centos6Update.log"
$RsyncCommand "$MirrorDomain"/repo/centos/7/updates/x86_64/ $Centos7Update >> "$LogDir/centos7Update.log" 2>&1
check_rsync_status "$LogDir/centos7Update.log"
# mysql 5.7 yumrepo
# $RsyncCommand "$MirrorDomain"/repo/mysql-repo/yum/mysql-5.7-community/el/6/x86_64/ "$Centos6Mysql5_7" >> "$LogDir/centos6Mysql5.7.log" 2>&1
# check_rsync_status "$LogDir/centos6Mysql5.7.log"
$RsyncCommand "$MirrorDomain"/repo/mysql-repo/yum/mysql-5.7-community/el/7/x86_64/ "$Centos7Mysql5_7" >> "$LogDir/centos7Mysql5.7.log" 2>&1
check_rsync_status "$LogDir/centos7Mysql5.7.log"
# mysql 8.0 yumrepo
# $RsyncCommand "$MirrorDomain"/repo/mysql-repo/yum/mysql-8.0-community/el/6/x86_64/ "$Centos6Mysql8_0" >> "$LogDir/centos6Mysql8.0.log" 2>&1
# check_rsync_status "$LogDir/centos6Mysql8.0.log"
$RsyncCommand "$MirrorDomain"/repo/mysql-repo/yum/mysql-8.0-community/el/7/x86_64/ "$Centos7Mysql8_0" >> "$LogDir/centos7Mysql8.0.log" 2>&1
check_rsync_status "$LogDir/centos7Mysql8.0.log"
登入後複製

读者需求解答

负载高时，查出占用比较高的进程脚本并存储或推送通知

这部分内容是上篇 Shell 脚本实例中底部读者留言的需求，如下：

#!/bin/bash
# 物理cpu个数
physical_cpu_count=$(egrep 'physical id' /proc/cpuinfo | sort | uniq | wc -l)
# 单个物理cpu核数
physical_cpu_cores=$(egrep 'cpu cores' /proc/cpuinfo | uniq | awk '{print $NF}')
# 总核数
total_cpu_cores=$((physical_cpu_count*physical_cpu_cores))
# 分别是一分钟、五分钟、十五分钟负载的阈值，其中有一项超过阈值才会触发
one_min_load_threshold="$total_cpu_cores"
five_min_load_threshold=$(awk 'BEGIN {print '"$total_cpu_cores"' * "0.8"}')
fifteen_min_load_threshold=$(awk 'BEGIN {print '"$total_cpu_cores"' * "0.7"}')
# 分别是分钟、五分钟、十五分钟负载平均值
one_min_load=$(uptime | awk '{print $(NF-2)}' | tr -d ',')
five_min_load=$(uptime | awk '{print $(NF-1)}' | tr -d ',')
fifteen_min_load=$(uptime | awk '{print $NF}' | tr -d ',')
# 获取当前cpu 内存 磁盘io信息，并写入日志文件
# 如果需要发送消息或者调用其他，请自行编写函数即可
get_info(){
 log_dir="cpu_high_script_log"
 test -d "$log_dir" || mkdir "$log_dir"
 ps -eo user,pid,%cpu,stat,time,command --sort -%cpu | head -10 > "$log_dir"/cpu_top10.log
 ps -eo user,pid,%mem,rss,vsz,stat,time,command --sort -%mem | head -10 > "$log_dir"/mem_top10.log
 iostat -dx 1 10 > "$log_dir"/disk_io_10.log
}
export -f get_info
echo "$one_min_load $one_min_load_threshold $five_min_load $five_min_load_threshold $fifteen_min_load $fifteen_min_load_threshold" | 
awk '{ if ($1>=$2 || $3>=$4 || $5>=$6) system("get_info") }'
登入後複製

以上，就是今天分享的全部内容了。

希望大家通过这些案例能够学以致用，结合自身的实际场景进行运用，从而提高自己的工作效率。

以上是超硬核！ 11個非常實用的 Python 和 Shell 拿來就用腳本實例！的詳細內容。更多資訊請關注PHP中文網其他相關文章！

相關標籤：
python 腳本 shell
來源：51cto.com
上一篇：這個好玩！用Python辨識花種類，並自動整理分類！ 下一篇：太實用了！ Schedule模組， Python 週期任務神器！
本網站聲明
本文內容由網友自願投稿，版權歸原作者所有。本站不承擔相應的法律責任。如發現涉嫌抄襲或侵權的內容，請聯絡admin@php.cn
作者最新文章
最新問題
Python/MySQL無法正確持久化整數數據 在這裡不需要任何程式碼。我想要保存一個非常長的數字，因為我正在製作一個遊戲，需要保存分數。但是我測試了一下，將分數設定為25000000000，但在mysql中儲存為2147483...
來自於 2024-04-04 19:09:44
0
1
367
使用selenium想要點擊並在類別中定義URL 今天我需要另一個提示。我正在嘗試建立Python/Selenium程式碼，想法是點擊www.thewebsiteIwantoclickon下面是我正在處理的HTML範例。類別ent...
來自於 2024-04-04 14:14:44
0
1
3492
Selenium + Python - 透過execute_script檢查映像 我需要使用python中的selenium驗證圖片是否顯示在頁面上。例如，讓我們檢查https://openweathermap.org/頁面左上角的標誌。我使用execute_s...
來自於 2024-04-03 09:32:15
0
1
375
保留前X行，刪除表格行的方法 我在MySQLincident_archive中有一個包含數百萬筆記錄的大表，我想按created列對行進行排序，並保留前X行並刪除其餘行，最有效的方法是什麼。到目前為止，我用Py...
來自於 2024-04-01 18:32:54
0
1
347
如何使用 BeautifulSoup 抓取特定的Google天氣文字？ 如何使用BeautifulSoup在Python中找到課程文本“美國紐約市”？嘗試複製影片進行練習，但不再有效。嘗試在官方文件中找到一些內容，但沒有成功。或者我的get_html_...
來自於 2024-04-01 14:06:14
0
1
308
相關專題
更多>
熱門推薦
熱門教學
更多>
相關教學
熱門推薦
最新課程
最新下載
更多>
網站特效
網站源碼
網站素材
前端模板