Python, 原创

Python 实现服务器日志备份到阿里云OSS

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)

Related Post