python依赖
# requirements.txt
setuptools-rust>=0.11.6 # 1.1.2
requests>==2.25.0 # 2.27.1
oss2>=2.18.5
pyyaml>=5.3.1 # 6.0
zipfile37>=0.1.3
主要代码
# main.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import logging
from datetime import datetime
from utility import ip, zip_helper, oss_helper
log = logging.getLogger('info')
SERVER_LOG_PATH = "/var/log"
BACKUP_DATA_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'data')
if __name__ == '__main__':
log.info("start backup logs")
is_internal_endpoint = False
# 获取当前内网IP
ip_local = ip.get_ip_local()
if '10.0.' in ip_local:
ip_public = ip_local
else:
ip_public = ip.get_ip_public()
if '172.31.' in ip_local:
is_internal_endpoint = True
else:
ip_local = ip_public
# 文件名拼装
datetime_now = datetime.now().strftime('%Y%m%d%H%M%S')
oss_path_pre = ip_public
zip_file_name = 'var_log_' + ip_public + "-" + ip_local + "-" + datetime_now + '.zip'
local_object_path = os.path.join(BACKUP_DATA_PATH, zip_file_name)
# 压缩文件
zip_helper.zip_dir(SERVER_LOG_PATH, local_object_path)
# OSS 初始化信息
oss_object_path = oss_path_pre + '/' + zip_file_name
log.info("start upload file %s to oss path oss://%s/%s", local_object_path, oss_helper.get_oss_config().bucket(), oss_path_pre)
oss_bucket = oss_helper.get_bucket_client(is_internal_endpoint)
# 上传到OSS
oss_bucket.put_object_from_file(oss_object_path, local_object_path)
配置文件
# oss.yaml
oss:
ACCESS_KEY_ID: key
ACCESS_KEY_SECRET: secret
BUCKET: server-log
ENDPOINT: oss-cn-shanghai.aliyuncs.com
ENDPOINT_INTERNAL: oss-cn-shanghai-internal.aliyuncs.com
# utility/yaml_helper.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import yaml
def read_yaml_file(fileName):
"""
读取当前目录下的配置文件
"""
# 获取当前脚本的目录
script_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
# 构建文件路径
config_file_path = os.path.join(script_dir, fileName)
with open(config_file_path, 'r') as file:
config_params = yaml.safe_load(file)
return config_params
# utility/oss_helper.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import oss2
from utility import yaml_helper
class OssConfig:
def __init__(self, default_path='config/oss.yaml'):
config_params = yaml_helper.read_yaml_file(default_path)
if config_params['oss']:
self.oss_params = config_params['oss']
else:
self.oss_params = config_params
def access_key_id(self):
return self.oss_params['ACCESS_KEY_ID']
def access_key_secret(self):
return self.oss_params['ACCESS_KEY_SECRET']
def bucket(self):
return self.oss_params['BUCKET']
def endpoint(self):
return self.oss_params['ENDPOINT']
def endpoint_internal(self):
return self.oss_params['ENDPOINT_INTERNAL']
def get_oss_config():
return OssConfig()
def get_oss_auth(oss_config = None):
if oss_config is None:
oss_config = get_oss_config()
return oss2.Auth(oss_config.access_key_id(), oss_config.access_key_secret())
def get_bucket_client(is_internal_endpoint = False):
oss_config = get_oss_config()
auth = get_oss_auth(oss_config)
endpoint = oss_config.endpoint_internal() if is_internal_endpoint else oss_config.endpoint()
return oss2.Bucket(auth, endpoint, oss_config.bucket())
# utility/ip_helper.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import socket
import requests
import re
class IP:
@staticmethod
def get_ip_public():
"""
获取本机公网IP
:return:
"""
try:
ip = requests.get('https://ident.me').text.strip()
print('IP public is ' + ip)
ip_public = re.findall(r'\d{1,3}\.\d{,3}\.\d{1,3}\.\d{1,3}', ip)[0]
return ip_public
except:
return IP.get_ip_public2()
@staticmethod
def get_ip_public2():
"""
获取本机公网IP 第二种方式
:return:
"""
try:
text = requests.get("http://txt.go.sohu.com/ip/soip").text
ip_public = re.findall(r'\d{1,3}\.\d{,3}\.\d{1,3}\.\d{1,3}', text)[0]
print('IP public2 is ' + ip_public)
return ip_public
except:
return None
@staticmethod
def get_ip_local():
"""
获取本机内网IP
:return:
"""
try:
ip = socket.gethostbyname(socket.gethostname())
if re.match(r'\d{1,3}\.\d{,3}\.\d{1,3}\.\d{1,3}', ip):
print('IP local is ' + ip)
return ip
return IP.get_ip_local2()
except:
return IP.get_ip_local2()
@staticmethod
def get_ip_local2():
"""
获取本机内网IP 第二种方法
:return:
"""
try:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))
ip = s.getsockname()[0]
print('IP local2 is ' + ip)
finally:
s.close()
return ip
def get_ip_helper():
return IP()
# utility/zip_helper.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
import zipfile
def zip_dir(dirpath, outFullName):
"""
压缩指定文件夹
:param dirpath: 目标文件夹路径
:param outFullName: 压缩文件保存路径+xxxx.zip
:return: 无
"""
print('打包ZIP文件开始')
zip = zipfile.ZipFile(outFullName, "w", zipfile.ZIP_DEFLATED)
for path, dirnames, filenames in os.walk(dirpath):
# 去掉目标跟路径,只对目标文件夹下边的文件及文件夹进行压缩
fpath = path.replace(dirpath, '')
for filename in filenames:
zip.write(os.path.join(path, filename), os.path.join(fpath, filename))
zip.close()
print('打包ZIP文件完成')
(0)