#!/bin/python3 import os import shutil import datetime import sys import zipfile import time import json import urllib import threading import subprocess import logging import http.client from ftplib import FTP CONFIG_FILE = "/app/config/history_archiver.json" temp = """ { "history_dir": "/mnt/ssd/history", "archive_after_years": 1, "delete_after_years": 3, "disk_threshold": 90.0, "upload_method": "ftp", "upload_retry_times": 3, "ftp": { "host": "ftp.lnxall.com", "port": 60001, "user": "devicedata", "password": "lnxall123", "remote_dir": "/log" }, "http": { "url": "http://192.168.22.130:9090/browser/http/", "argv": { "username": "admin", "password": "password123" } } } """ # 需根据客户服务器定制, 此处进提供未验证的示例, 尽量避免使用第三方库 class HTTPUploader: def __init__(self, config): self.config = config def upload(self, file_path): if not os.path.exists(file_path): return False # 解析配置 try: url = self.config.get('url', '') argv = self.config.get('argv', {}) username = argv.get('username', '') password = argv.get('password', '') if not url: raise ValueError("URL不能为空") parsed_url = urllib.parse.urlparse(url) host = parsed_url.netloc path = parsed_url.path with open(file_path, 'rb') as f: file_content = f.read() # 构建请求 boundary = '----WebKitFormBoundary7MA4YWxkTrZu0gW' body = [] # 添加表单字段 if username: body.append(f'--{boundary}\r\n') body.append('Content-Disposition: form-data; name="username"\r\n\r\n') body.append(f'{username}\r\n') if password: body.append(f'--{boundary}\r\n') body.append('Content-Disposition: form-data; name="password"\r\n\r\n') body.append(f'{password}\r\n') # 添加文件 body.append(f'--{boundary}\r\n') body.append(f'Content-Disposition: form-data; name="file"; filename="{os.path.basename(file_path)}"\r\n') body.append('Content-Type: application/octet-stream\r\n\r\n') body.append(file_content) body.append(f'\r\n--{boundary}--\r\n') # 连接服务器 try: if parsed_url.scheme == 'https': conn = http.client.HTTPSConnection(host) else: conn = http.client.HTTPConnection(host) headers = { 'Content-Type': f'multipart/form-data; boundary={boundary}', } # 发送请求 conn.request("POST", path, ''.join(body), headers) response = conn.getresponse() print(f"状态码: {response.status}") print(f"响应头: {response.getheaders()}") print(f"响应体: {response.read().decode()}") conn.close() return True except Exception as e: logging.error(f"HTTP upload failed: {str(e)}") return False except Exception as e: logging.error(f"HTTP upload failed: {str(e)}") return False #ftp 使用公司文件服务器验证,支持4G class FTPUploader: def __init__(self, config): self.host = config.get('host') self.port = config.get('port', 21) self.user = config.get('user') self.password = config.get('password') self.remote_dir = config.get('remote_dir') def upload(self, file_path): try: with FTP() as ftp: ftp.connect(self.host, self.port) ftp.login(self.user, self.password) if self.remote_dir: ftp.cwd(self.remote_dir) with open(file_path, 'rb') as f: ftp.storbinary(f'STOR {os.path.basename(file_path)}', f) logging.info(f"Successfully uploaded {file_path} via FTP") os.remove(file_path) return True except Exception as e: logging.error(f"FTP upload failed: {str(e)}") return False class HistoryArchiver: def __init__(self): self.load_config() self.ensure_directories() self.running = True self.ftp_uploader = FTPUploader(self.ftp_config) if self.upload_method == 'ftp' else None self.http_uploader = HTTPUploader(self.http_config) if self.upload_method == 'http' else None def load_config(self): """从JSON文件加载配置""" config = {} if os.path.exists(CONFIG_FILE): with open(CONFIG_FILE) as f: config = json.load(f) self.history_dir = config.get('history_dir', '/mnt/ssd/history') self.compress_dir = os.path.join(self.history_dir, 'compressed') self.retention_years = config.get('archive_after_years', 1) self.delete_after_years = config.get('delete_after_years', 3) self.disk_threshold = config.get('disk_threshold', 85.0) self.upload_method = config.get('upload_method', 'delete') self.ftp_config = config.get('ftp', {}) self.http_config = config.get('http', {}) self.retrys = {} def ensure_directories(self): """确保所需目录存在""" os.makedirs(self.compress_dir, exist_ok=True) logging.info(f"Ensured directory exists: {self.compress_dir}") def get_old_folders(self): """获取超过保留期限的文件夹""" old_folders = [] cutoff_date = datetime.datetime.now() - datetime.timedelta(days=365 * self.retention_years) for folder in os.listdir(self.history_dir): folder_path = os.path.join(self.history_dir, folder) if os.path.isdir(folder_path) and folder != 'compressed': try: folder_date = datetime.datetime.strptime(folder, '%Y-%m-%d') if folder_date < cutoff_date: old_folders.append(folder_path) except ValueError: continue return old_folders def compress_old_folders(self): """压缩旧文件夹并按月份归档""" old_folders = self.get_old_folders() if not old_folders: logging.info("No old folders to compress") return month_groups = {} for folder in old_folders: folder_name = os.path.basename(folder) month_key = folder_name[:7] # 获取年月部分，如2024-09 if month_key not in month_groups: month_groups[month_key] = [] month_groups[month_key].append(folder) for month, folders in month_groups.items(): archive_name = f"{month}.zip" archive_path = os.path.join(self.compress_dir, archive_name) # 检查压缩包是否存在，如果存在则删除 if os.path.exists(archive_path): logging.warning(f"Archive {archive_name} already exists, delete!") os.remove(archive_path) logging.info(f"Creating archive for {month} with {len(folders)} folders") with zipfile.ZipFile(archive_path, "w", zipfile.ZIP_DEFLATED) as zipf: for folder in folders: # 遍历文件夹中的所有文件并添加到zip文件中 for root, dirs, files in os.walk(folder): for file in files: file_path = os.path.join(root, file) # 计算在zip文件中的相对路径 arcname = os.path.relpath(file_path, os.path.dirname(folder)) zipf.write(file_path, arcname) # 删除文件夹 shutil.rmtree(folder) logging.info(f"Archive {archive_name} created successfully") def check_disk_space(self): """通过df命令检查磁盘空间""" try: df_output = subprocess.check_output(['df', self.history_dir]).decode() usage_line = df_output.split('\n')[1] usage_percent = int(usage_line.split()[4].replace('%', '')) return usage_percent >= self.disk_threshold except Exception as e: logging.error(f"Disk space check failed: {str(e)}") return False def get_old_archives(self): """获取超过保留期限的压缩包""" old_archives = [] cutoff_date = datetime.datetime.now() - datetime.timedelta(days=((365) * self.delete_after_years - 30)) for archive in os.listdir(self.compress_dir): if archive.endswith('.zip'): month_key = archive[:7] file_date = datetime.datetime.strptime(month_key, "%Y-%m") if file_date < cutoff_date: archive_path = os.path.join(self.compress_dir, archive) old_archives.append(archive_path) return old_archives # 获取文件夹下最早的压缩文件 def get_oldest_archive(self): old_archives = os.listdir(self.compress_dir) if not old_archives: return None old_archives = [os.path.join(self.compress_dir, archive) for archive in old_archives] oldest_archive = min(old_archives, key=os.path.getmtime) return oldest_archive def handle_old_compressed_folders(self, archive): if self.upload_method == 'ftp' and self.ftp_uploader: return self.ftp_uploader.upload(archive) elif self.upload_method == 'http' and self.http_uploader: return self.http_uploader.upload(archive) else: logging.info(f"Deleted old archive: {archive}") return True def handle_old_archives(self): """处理旧压缩包""" while self.check_disk_space(): oldest_archive = self.get_oldest_archive() if oldest_archive: try: result = self.handle_old_compressed_folders(oldest_archive) logging.info(f"Processed old archive: {oldest_archive}, result: {result}, delete") os.remove(oldest_archive) # 磁盘容量不足时不做保留删除压缩包 if oldest_archive in self.retrys: del self.retrys[archive] except Exception as e: logging.error(f"Failed to process old archive: {oldest_archive}") # 检查磁盘空间后, 正常处理超期的压缩包 old_archives = self.get_old_archives() if not old_archives: return for archive in old_archives: if self.handle_old_compressed_folders(archive): try: logging.info(f"Processed old archive: {archive}, delete") os.remove(archive) if archive in self.retrys: del self.retrys[archive] except Exception as e: logging.error(f"Failed to process old archive: {archive}") else: logging.error(f"Failed to process old archive: {archive}") try: if archive not in self.retrys: self.retrys[archive] = 1 logging.info(f"Retry {self.retrys[archive]} times") elif self.retrys[archive] < 3: self.retrys[archive] = self.retrys[archive] + 1 logging.info(f"Retry {self.retrys[archive]} times") else: # 如果重试3次后仍然失败，则删除该压缩包 logging.warning(f"Failed to process old archive: {archive} after 3 retries, delete") os.remove(archive) del self.retrys[archive] except Exception as e: logging.error(f"Failed to process old archive: {archive}") def run_monthly_task(self): """每月1号执行的任务""" now = datetime.datetime.now() #if now.day == 1 and now.hour == 0 and now.minute == 0: self.compress_old_folders() def run_daily_task(self): """每天执行的任务""" now = datetime.datetime.now() # if now.hour == 1 and now.minute == 0: self.handle_old_archives() def start(self): try: while True: self.run_monthly_task() self.run_daily_task() time.sleep(60) except KeyboardInterrupt: print("Stopping...") if __name__ == '__main__': logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', filename='/tmp/ems/history_archiver.log' ) console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(logging.INFO) logging.getLogger().addHandler(console_handler) try: archiver = HistoryArchiver() archiver.start() except Exception as e: logging.error(f"Fatal error: {str(e)}", exc_info=True)