监控数据归档进度的脚本如何编写

wen 实用脚本 2026-07-04 182

本文目录导读：

监控数据归档进度的脚本如何编写

通用文件归档监控脚本
数据库归档监控脚本
压缩归档进度监控
Web界面监控（使用Flask）
日志文件归档监控
使用方法
注意事项

我来帮你编写一个监控数据归档进度的脚本，我会提供几种不同场景的示例,你可以根据实际需要选择使用。

通用文件归档监控脚本

#!/usr/bin/env python3
"""
数据归档进度监控脚本
支持监控文件复制、压缩、传输等归档操作
"""
import os
import time
import psutil
import json
from datetime import datetime
from pathlib import Path
class ArchiveMonitor:
    def __init__(self, source_dir, target_dir, check_interval=5):
        self.source_dir = Path(source_dir)
        self.target_dir = Path(target_dir)
        self.check_interval = check_interval  # 检查间隔（秒）
        self.start_time = None
        self.total_files = 0
        self.total_size = 0
        self.processed_files = 0
        self.processed_size = 0
    def scan_source(self):
        """扫描源目录，获取总文件数和总大小"""
        print(f"正在扫描源目录: {self.source_dir}")
        for file_path in self.source_dir.rglob('*'):
            if file_path.is_file():
                self.total_files += 1
                self.total_size += file_path.stat().st_size
        print(f"发现 {self.total_files} 个文件，总大小: {self.format_size(self.total_size)}")
    def check_progress(self):
        """检查当前归档进度"""
        current_files = 0
        current_size = 0
        if self.target_dir.exists():
            for file_path in self.target_dir.rglob('*'):
                if file_path.is_file():
                    current_files += 1
                    current_size += file_path.stat().st_size
        return current_files, current_size
    def calculate_speed(self, size, elapsed):
        """计算传输速度"""
        if elapsed > 0:
            return size / elapsed
        return 0
    def format_size(self, bytes):
        """格式化文件大小显示"""
        for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
            if bytes < 1024.0:
                return f"{bytes:.2f} {unit}"
            bytes /= 1024.0
        return f"{bytes:.2f} PB"
    def estimate_time(self, processed, total, speed):
        """预估剩余时间"""
        if speed > 0 and processed < total:
            remaining = (total - processed) / speed
            return remaining
        return 0
    def monitor(self):
        """主监控循环"""
        print(f"开始监控归档进度...")
        print(f"源目录: {self.source_dir}")
        print(f"目标目录: {self.target_dir}")
        print("-" * 60)
        self.start_time = time.time()
        self.scan_source()
        try:
            while True:
                time.sleep(self.check_interval)
                elapsed = time.time() - self.start_time
                # 获取当前进度
                self.processed_files, self.processed_size = self.check_progress()
                # 计算指标
                file_percent = (self.processed_files / self.total_files * 100) if self.total_files > 0 else 0
                size_percent = (self.processed_size / self.total_size * 100) if self.total_size > 0 else 0
                speed = self.calculate_speed(self.processed_size, elapsed)
                eta = self.estimate_time(self.processed_size, self.total_size, speed)
                # 显示进度
                os.system('clear' if os.name == 'posix' else 'cls')  # 清屏
                print(f"数据归档进度监控 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
                print("=" * 60)
                print(f"文件进度: {self.processed_files}/{self.total_files} ({file_percent:.2f}%)")
                print(f"大小进度: {self.format_size(self.processed_size)}/{self.format_size(self.total_size)} ({size_percent:.2f}%)")
                print(f"传输速度: {self.format_size(speed)}/s")
                print(f"已用时间: {self.format_time(elapsed)}")
                print(f"预计剩余: {self.format_time(eta)}")
                print("-" * 60)
                # 进度条
                bar_length = 50
                filled = int(bar_length * size_percent / 100)
                bar = '█' * filled + '░' * (bar_length - filled)
                print(f"进度: |{bar}| {size_percent:.1f}%")
                # 检查是否完成
                if size_percent >= 100:
                    print("\n✅ 归档完成!")
                    break
        except KeyboardInterrupt:
            print("\n\n监控已停止")
    def format_time(self, seconds):
        """格式化时间显示"""
        if seconds <= 0:
            return "0秒"
        hours = int(seconds // 3600)
        minutes = int((seconds % 3600) // 60)
        secs = int(seconds % 60)
        parts = []
        if hours > 0:
            parts.append(f"{hours}小时")
        if minutes > 0:
            parts.append(f"{minutes}分")
        parts.append(f"{secs}秒")
        return "".join(parts)
# 使用示例
if __name__ == "__main__":
    # 配置路径
    SOURCE_DIR = "/path/to/source"  # 源数据目录
    TARGET_DIR = "/path/to/archive"  # 归档目标目录
    # 创建监控器并开始监控
    monitor = ArchiveMonitor(SOURCE_DIR, TARGET_DIR, check_interval=2)
    monitor.monitor()

数据库归档监控脚本

#!/usr/bin/env python3
"""
数据库归档进度监控脚本
支持MySQL/PostgreSQL数据库备份进度监控
"""
import subprocess
import re
import time
from datetime import datetime
class DatabaseArchiveMonitor:
    def __init__(self, db_type='mysql', host='localhost', port=3306, 
                 user='root', password='', database=''):
        self.db_type = db_type
        self.host = host
        self.port = port
        self.user = user
        self.password = password
        self.database = database
    def monitor_mysql_dump(self, output_file):
        """监控MySQL数据库导出进度"""
        print(f"开始监控MySQL数据库归档: {self.database}")
        print(f"输出文件: {output_file}")
        print("-" * 60)
        start_time = time.time()
        last_size = 0
        # 执行mysqldump命令
        cmd = f"mysqldump -h{self.host} -P{self.port} -u{self.user}"
        if self.password:
            cmd += f" -p{self.password}"
        cmd += f" {self.database} > {output_file}"
        process = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, 
                                   stderr=subprocess.PIPE)
        try:
            while process.poll() is None:
                time.sleep(2)
                elapsed = time.time() - start_time
                # 检查输出文件大小
                try:
                    current_size = os.path.getsize(output_file)
                    speed = (current_size - last_size) / 2  # 每秒速度
                    # 显示进度
                    print(f"\r时间: {self.format_time(elapsed)} | "
                          f"大小: {self.format_size(current_size)} | "
                          f"速度: {self.format_size(speed)}/s", end='')
                    last_size = current_size
                except FileNotFoundError:
                    print("\r等待文件创建...", end='')
            # 检查是否成功
            if process.returncode == 0:
                total_size = os.path.getsize(output_file)
                print(f"\n✅ MySQL数据库归档完成!")
                print(f"最终大小: {self.format_size(total_size)}")
                print(f"总用时: {self.format_time(time.time() - start_time)}")
            else:
                error = process.stderr.read().decode()
                print(f"\n❌ 归档失败: {error}")
        except KeyboardInterrupt:
            process.terminate()
            print("\n监控已停止")
    def monitor_postgresql_dump(self, output_file):
        """监控PostgreSQL数据库导出进度"""
        print(f"开始监控PostgreSQL数据库归档: {self.database}")
        print(f"输出文件: {output_file}")
        print("-" * 60)
        start_time = time.time()
        # 使用pg_dump并获取进度
        cmd = f"pg_dump -h{self.host} -p{self.port} -U{self.user}"
        if self.password:
            os.environ['PGPASSWORD'] = self.password
        cmd += f" {self.database} > {output_file}"
        process = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, 
                                   stderr=subprocess.PIPE)
        # 监控进度
        while process.poll() is None:
            time.sleep(2)
            elapsed = time.time() - start_time
            try:
                current_size = os.path.getsize(output_file)
                print(f"\r时间: {self.format_time(elapsed)} | "
                      f"已备份: {self.format_size(current_size)}", end='')
            except FileNotFoundError:
                print("\r等待备份开始...", end='')
        if process.returncode == 0:
            total_size = os.path.getsize(output_file)
            print(f"\n✅ PostgresQL数据库归档完成!")
            print(f"最终大小: {self.format_size(total_size)}")
            print(f"总用时: {self.format_time(time.time() - start_time)}")
        else:
            error = process.stderr.read().decode()
            print(f"\n❌ 归档失败: {error}")
    def format_size(self, bytes):
        """格式化文件大小"""
        for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
            if bytes < 1024.0:
                return f"{bytes:.2f} {unit}"
            bytes /= 1024.0
        return f"{bytes:.2f} PB"
    def format_time(self, seconds):
        """格式化时间"""
        hours = int(seconds // 3600)
        minutes = int((seconds % 3600) // 60)
        secs = int(seconds % 60)
        return f"{hours:02d}:{minutes:02d}:{secs:02d}"
# 使用示例
if __name__ == "__main__":
    # MySQL监控示例
    mysql_monitor = DatabaseArchiveMonitor(
        db_type='mysql',
        host='localhost',
        user='root',
        password='password',
        database='mydatabase'
    )
    mysql_monitor.monitor_mysql_dump("backup.sql")
    # PostgreSQL监控示例
    pg_monitor = DatabaseArchiveMonitor(
        db_type='postgresql',
        host='localhost',
        port=5432,
        user='postgres',
        password='password',
        database='mydb'
    )
    pg_monitor.monitor_postgresql_dump("pg_backup.sql")

压缩归档进度监控

#!/usr/bin/env python3
"""
压缩归档进度监控脚本
监控tar/zip等压缩操作进度
"""
import os
import time
import threading
from pathlib import Path
class CompressionMonitor:
    def __init__(self, source_dir, archive_file, archive_type='tar.gz'):
        self.source_dir = Path(source_dir)
        self.archive_file = Path(archive_file)
        self.archive_type = archive_type
        self.total_size = 0
        self.monitoring = False
    def get_total_size(self):
        """计算源目录总大小"""
        total = 0
        for file_path in self.source_dir.rglob('*'):
            if file_path.is_file():
                total += file_path.stat().st_size
        return total
    def monitor(self):
        """启动监控"""
        print(f"开始监控压缩归档: {self.source_dir} -> {self.archive_file}")
        print("-" * 60)
        self.total_size = self.get_total_size()
        start_time = time.time()
        self.monitoring = True
        # 在后台启动压缩进程
        compress_thread = threading.Thread(target=self.compress_files)
        compress_thread.start()
        try:
            while compress_thread.is_alive():
                time.sleep(2)
                elapsed = time.time() - start_time
                if self.archive_file.exists():
                    current_size = self.archive_file.stat().st_size
                    # 压缩比大约在2-5倍之间，这里估算为3倍
                    estimated_progress = min((current_size * 3) / self.total_size * 100, 99)
                    speed = current_size / elapsed if elapsed > 0 else 0
                    print(f"\r进度: {self.format_size(current_size)}/{self.format_size(self.total_size)} | "
                          f"耗时: {self.format_time(elapsed)} | "
                          f"速度: {self.format_size(speed)}/s", end='')
            # 压缩完成
            final_size = self.archive_file.stat().st_size if self.archive_file.exists() else 0
            total_time = time.time() - start_time
            compression_ratio = self.total_size / final_size if final_size > 0 else 0
            print(f"\n✅ 压缩归档完成!")
            print(f"原始大小: {self.format_size(self.total_size)}")
            print(f"压缩大小: {self.format_size(final_size)}")
            print(f"压缩比: {compression_ratio:.2f}:1")
            print(f"总用时: {self.format_time(total_time)}")
        except KeyboardInterrupt:
            print("\n监控已停止")
            self.monitoring = False
    def compress_files(self):
        """执行文件压缩"""
        import tarfile
        import gzip
        if self.archive_type == 'tar.gz':
            with tarfile.open(self.archive_file, "w:gz") as tar:
                tar.add(self.source_dir, arcname=self.source_dir.name)
        elif self.archive_type == 'zip':
            import zipfile
            with zipfile.ZipFile(self.archive_file, 'w', zipfile.ZIP_DEFLATED) as zipf:
                for file_path in self.source_dir.rglob('*'):
                    if file_path.is_file():
                        zipf.write(file_path, file_path.relative_to(self.source_dir.parent))
    def format_size(self, bytes):
        """格式化文件大小"""
        for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
            if bytes < 1024.0:
                return f"{bytes:.2f} {unit}"
            bytes /= 1024.0
        return f"{bytes:.2f} PB"
    def format_time(self, seconds):
        """格式化时间"""
        hours = int(seconds // 3600)
        minutes = int((seconds % 3600) // 60)
        secs = int(seconds % 60)
        return f"{hours:02d}:{minutes:02d}:{secs:02d}"
# 使用示例
if __name__ == "__main__":
    source = "/path/to/data"  # 要压缩的数据目录
    archive = "/path/to/archive.tar.gz"  # 输出的归档文件
    monitor = CompressionMonitor(source, archive, 'tar.gz')
    monitor.monitor()

Web界面监控（使用Flask）

#!/usr/bin/env python3
"""
Web版归档监控界面
提供实时进度显示
"""
from flask import Flask, jsonify, render_template_string
import os
import time
import threading
from pathlib import Path
app = Flask(__name__)
# 全局状态
archive_status = {
    'total_files': 0,
    'processed_files': 0,
    'total_size': 0,
    'processed_size': 0,
    'speed': 0,
    'start_time': None,
    'is_running': False,
    'percentage': 0
}
HTML_TEMPLATE = '''
<!DOCTYPE html>
<html>
<head>数据归档监控</title>
    <style>
        body {
            font-family: Arial, sans-serif;
            margin: 40px;
            background: #f5f5f5;
        }
        .container {
            max-width: 800px;
            margin: 0 auto;
            background: white;
            padding: 30px;
            border-radius: 10px;
            box-shadow: 0 2px 10px rgba(0,0,0,0.1);
        }
        h1 {
            color: #333;
            text-align: center;
        }
        .progress-bar {
            width: 100%;
            height: 30px;
            background: #e0e0e0;
            border-radius: 15px;
            overflow: hidden;
            margin: 20px 0;
        }
        .progress-fill {
            height: 100%;
            background: linear-gradient(90deg, #4CAF50, #45a049);
            transition: width 1s ease;
            display: flex;
            align-items: center;
            justify-content: center;
            color: white;
            font-weight: bold;
        }
        .stats {
            display: grid;
            grid-template-columns: 1fr 1fr;
            gap: 10px;
            margin: 20px 0;
        }
        .stat-item {
            background: #f9f9f9;
            padding: 15px;
            border-radius: 5px;
            text-align: center;
        }
        .stat-label {
            color: #666;
            font-size: 14px;
        }
        .stat-value {
            color: #333;
            font-size: 24px;
            font-weight: bold;
            margin-top: 5px;
        }
        .status {
            text-align: center;
            padding: 10px;
            border-radius: 5px;
            margin: 20px 0;
        }
        .running {
            background: #e3f2fd;
            color: #1976D2;
        }
        .completed {
            background: #e8f5e9;
            color: #388E3C;
        }
    </style>
</head>
<body>
    <div class="container">
        <h1>📊 数据归档监控</h1>
        <div class="progress-bar">
            <div class="progress-fill" style="width: {{ status.percentage }}%">
                {{ "%.1f"|format(status.percentage) }}%
            </div>
        </div>
        <div class="stats">
            <div class="stat-item">
                <div class="stat-label">文件进度</div>
                <div class="stat-value">{{ status.processed_files }}/{{ status.total_files }}</div>
            </div>
            <div class="stat-item">
                <div class="stat-label">大小进度</div>
                <div class="stat-value">{{ format_size(status.processed_size) }}/{{ format_size(status.total_size) }}</div>
            </div>
            <div class="stat-item">
                <div class="stat-label">传输速度</div>
                <div class="stat-value">{{ format_size(status.speed) }}/s</div>
            </div>
            <div class="stat-item">
                <div class="stat-label">运行时间</div>
                <div class="stat-value">{{ format_time(status.elapsed) }}</div>
            </div>
        </div>
        <div class="status {{ 'running' if status.is_running else 'completed' }}">
            {{ '🔄 归档进行中...' if status.is_running else '✅ 归档完成!' }}
        </div>
    </div>
    <script>
        function updateStatus() {
            fetch('/status')
                .then(response => response.json())
                .then(data => {
                    document.querySelector('.progress-fill').style.width = data.percentage + '%';
                    document.querySelector('.progress-fill').textContent = data.percentage.toFixed(1) + '%';
                    // 更新其他统计信息
                    document.querySelectorAll('.stat-value')[0].textContent = 
                        data.processed_files + '/' + data.total_files;
                    document.querySelectorAll('.stat-value')[1].textContent = 
                        formatBytes(data.processed_size) + '/' + formatBytes(data.total_size);
                    document.querySelectorAll('.stat-value')[2].textContent = 
                        formatBytes(data.speed) + '/s';
                    if (data.percentage >= 100) {
                        document.querySelector('.status').className = 'status completed';
                        document.querySelector('.status').textContent = '✅ 归档完成!';
                    }
                });
        }
        function formatBytes(bytes) {
            if (bytes === 0) return '0 B';
            const k = 1024;
            const sizes = ['B', 'KB', 'MB', 'GB', 'TB'];
            const i = Math.floor(Math.log(bytes) / Math.log(k));
            return parseFloat((bytes / Math.pow(k, i)).toFixed(2)) + ' ' + sizes[i];
        }
        // 每2秒更新一次
        setInterval(updateStatus, 2000);
    </script>
</body>
</html>
'''
@app.route('/')
def index():
    """主页面"""
    return render_template_string(HTML_TEMPLATE, status=archive_status)
@app.route('/status')
def status():
    """返回JSON状态"""
    if archive_status['start_time']:
        elapsed = time.time() - archive_status['start_time']
    else:
        elapsed = 0
    return jsonify({
        'total_files': archive_status['total_files'],
        'processed_files': archive_status['processed_files'],
        'total_size': archive_status['total_size'],
        'processed_size': archive_status['processed_size'],
        'speed': archive_status['speed'],
        'percentage': archive_status['percentage'],
        'elapsed': elapsed,
        'is_running': archive_status['is_running']
    })
def update_status():
    """更新归档状态（示例）"""
    # 这里应该实现实际的归档进度检查逻辑
    # 以下为模拟数据
    while archive_status['is_running']:
        time.sleep(2)
        if archive_status['processed_size'] < archive_status['total_size']:
            # 模拟进度增加
            increase = min(1024*1024*100, archive_status['total_size'] - archive_status['processed_size'])
            archive_status['processed_size'] += increase
            archive_status['percentage'] = (archive_status['processed_size'] / archive_status['total_size']) * 100
        else:
            archive_status['is_running'] = False
            archive_status['percentage'] = 100
            break
if __name__ == '__main__':
    # 初始化示例数据
    archive_status['total_files'] = 1000
    archive_status['total_size'] = 1024 * 1024 * 1024 * 10  # 10GB
    archive_status['start_time'] = time.time()
    archive_status['is_running'] = True
    # 启动后台更新线程
    update_thread = threading.Thread(target=update_status, daemon=True)
    update_thread.start()
    # 启动Web服务器
    app.run(host='0.0.0.0', port=5000, debug=True)

日志文件归档监控

#!/usr/bin/env python3
"""
日志归档监控脚本
适合监控日志文件的轮转和归档
"""
import time
import os
from datetime import datetime, timedelta
from pathlib import Path
class LogArchiveMonitor:
    def __init__(self, log_dir, archive_dir, rotation_size='100M'):
        self.log_dir = Path(log_dir)
        self.archive_dir = Path(archive_dir)
        self.rotation_size = self.parse_size(rotation_size)
        # 确保目录存在
        self.archive_dir.mkdir(parents=True, exist_ok=True)
    def parse_size(self, size_str):
        """解析大小字符串"""
        size_str = size_str.upper()
        if size_str.endswith('K'):
            return int(float(size_str[:-1]) * 1024)
        elif size_str.endswith('M'):
            return int(float(size_str[:-1]) * 1024 * 1024)
        elif size_str.endswith('G'):
            return int(float(size_str[:-1]) * 1024 * 1024 * 1024)
        else:
            return int(size_str)
    def format_size(self, bytes):
        """格式化大小"""
        for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
            if bytes < 1024.0:
                return f"{bytes:.2f} {unit}"
            bytes /= 1024.0
        return f"{bytes:.2f} PB"
    def get_log_files(self):
        """获取所有日志文件"""
        log_files = []
        for file_path in self.log_dir.glob('*.log'):
            if file_path.is_file():
                log_files.append(file_path)
        return log_files
    def check_rotation_needed(self, file_path):
        """检查是否需要轮转"""
        if file_path.exists():
            size = file_path.stat().st_size
            return size >= self.rotation_size
        return False
    def rotate_log(self, file_path):
        """轮转日志文件"""
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        archive_name = f"{file_path.stem}_{timestamp}{file_path.suffix}.gz"
        archive_path = self.archive_dir / archive_name
        # 压缩并归档
        import gzip
        import shutil
        with open(file_path, 'rb') as f_in:
            with gzip.open(archive_path, 'wb') as f_out:
                shutil.copyfileobj(f_in, f_out)
        # 清空原日志文件
        file_path.write_text('')
        return archive_path
    def monitor(self):
        """启动监控循环"""
        print(f"开始监控日志归档...")
        print(f"日志目录: {self.log_dir}")
        print(f"归档目录: {self.archive_dir}")
        print(f"轮转阈值: {self.format_size(self.rotation_size)}")
        print("-" * 60)
        stats = {
            'rotated_files': 0,
            'total_size': 0,
            'start_time': time.time()
        }
        try:
            while True:
                time.sleep(10)  # 每10秒检查一次
                log_files = self.get_log_files()
                for log_file in log_files:
                    if self.check_rotation_needed(log_file):
                        archived = self.rotate_log(log_file)
                        stats['rotated_files'] += 1
                        stats['total_size'] += archived.stat().st_size
                        print(f"[{datetime.now().strftime('%H:%M:%S')}] "
                              f"轮转: {log_file.name} -> {archived.name} "
                              f"({self.format_size(archived.stat().st_size)})")
                # 显示汇总信息
                elapsed = time.time() - stats['start_time']
                sys.stdout.write(f"\r运行时间: {self.format_time(int(elapsed))} | "
                                 f"已轮转: {stats['rotated_files']} 个文件 | "
                                 f"总归档大小: {self.format_size(stats['total_size'])}")
                sys.stdout.flush()
        except KeyboardInterrupt:
            print("\n\n监控已停止")
            print(f"总计轮转: {stats['rotated_files']} 个文件")
            print(f"总归档大小: {self.format_size(stats['total_size'])}")
    def format_time(self, seconds):
        """格式化时间"""
        hours = seconds // 3600
        minutes = (seconds % 3600) // 60
        secs = seconds % 60
        return f"{hours:02d}:{minutes:02d}:{secs:02d}"
# 使用示例
if __name__ == "__main__":
    monitor = LogArchiveMonitor(
        log_dir="/var/log/myapp",
        archive_dir="/data/logs/archive",
        rotation_size="100M"  # 100MB触发轮转
    )
    monitor.monitor()

使用方法

通用文件归档监控
```
python file_archive_monitor.py
```
数据库归档监控
```
python db_archive_monitor.py
```
压缩归档监控
```
python compression_monitor.py
```

Web监控界面

pip install flask
python web_monitor.py
# 打开浏览器访问 http://localhost:5000

注意事项

权限问题：确保脚本有足够的权限访问源和目标目录
性能影响：监控间隔不要太短，避免影响归档性能
错误处理：建议添加重试机制和错误日志
资源清理：长时间运行的监控要注意内存使用
安全考虑：不要在脚本中硬编码密码

根据你的具体需求选择合适的脚本,或组合使用多个脚本来满足复杂场景。

本文地址： https://www.bmjup.com/post/15461.html