本文目录导读：
我来帮你编写一个监控数据归档进度的脚本，我会提供几种不同场景的示例,你可以根据实际需要选择使用。
通用文件归档监控脚本
#!/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
注意事项
- 权限问题：确保脚本有足够的权限访问源和目标目录
- 性能影响：监控间隔不要太短，避免影响归档性能
- 错误处理：建议添加重试机制和错误日志
- 资源清理：长时间运行的监控要注意内存使用
- 安全考虑：不要在脚本中硬编码密码
根据你的具体需求选择合适的脚本,或组合使用多个脚本来满足复杂场景。
