本文目录导读：
要实现迁移进度的实时百分比展示,核心在于 “分阶段计数” 和 “实时反馈”，具体实现方式取决于你迁移的数据类型（文件、数据库记录、对象存储等）和使用的脚本语言。
以下提供几种常见场景的脚本实现方案,并附带核心逻辑说明。
核心原理
- 知道总量：预先获取需要迁移的总条目数（总文件数、总记录数、总字节数）。
- 统计已量：每次成功迁移一个单元后，累计计数器。
- 计算并刷新：
进度 = (已处理 / 总量) * 100，然后在终端输出。
基础命令行脚本（适用于文件/对象迁移）
适用场景：拷贝文件、上传对象到云存储（如S3、OSS）。
核心技巧：使用 \r（回车符）让光标回到行首，实现原地刷新，而不是每行打印一条新消息。
Python 示例 (带 tqdm 进度条库，推荐)
- 优点：代码极简，自带美观的进度条、速率、剩余时间。
- 安装：
pip install tqdm
from tqdm import tqdm
import time
import os
def migrate_files(file_list):
# tqdm 自动处理百分比、速率、ETA
for file in tqdm(file_list, desc="迁移进度", unit="file"):
# 模拟迁移操作 (如 shutil.copy, boto3.upload)
time.sleep(0.01) # 替换为实际迁移逻辑
# 这里不需要手动更新，tqdm 自动基于迭代次数计算
# 假设 list_files() 返回所有待迁移的文件路径列表
files_to_migrate = list_files()
migrate_files(files_to_migrate)
- 输出效果：
迁移进度: 45%|████▌ | 45/100 [00:02<00:03, 15.23file/s]
纯 Shell (Bash) 脚本
- 适用：简单文件复制或
rsync/curl场景。 - 注意：
cp命令本身不提供进度，需要配合其他工具。
脚本示例（使用 rsync + 解析输出）：
#!/bin/bash
total=$(ls -1 /source/dir | wc -l) # 获取总文件数
count=0
for file in /source/dir/*; do
# 执行迁移 (此处使用 cp，实际可用 rsync)
cp "$file" /dest/dir/
count=$((count + 1))
# 计算百分比，保留整数
pct=$(( count * 100 / total ))
# \r 回车无换行，\033[K 清空行尾
printf "\r进度: %3d%% (%d/%d)" "$pct" "$count" "$total"
done
echo # 完成后换行
懒人版（使用 pv 命令）：
# 将 tar 打包并通过 pv 管道传输到目标解压
tar cf - /source/dir | pv -s $(du -sb /source/dir | awk '{print $1}') | tar xf - -C /dest/dir
数据库或数据流迁移
适用场景：迁移数据库表记录、清洗转换数据。
关键：需要能查询到 总行数。
Python 示例（基于数据库迁移）
import psycopg2
from tqdm import tqdm
def migrate_table():
conn_src = psycopg2.connect("...source_db")
conn_dst = psycopg2.connect("...dest_db")
cursor = conn_src.cursor()
cursor.execute("SELECT COUNT(*) FROM large_table")
total = cursor.fetchone()[0] # 获取总量
cursor.execute("SELECT * FROM large_table")
# tqdm 直接包装迭代器
for row in tqdm(cursor, total=total, desc="Migrating rows"):
# 插入到目标数据库
conn_dst.cursor().execute("INSERT INTO ... VALUES (%s)", row)
# 每 N 行提交一次，避免内存溢出
if row_count % 1000 == 0:
conn_dst.commit()
conn_dst.commit()
cursor.close()
大文件分段迁移（断点续传）
适用场景：迁移单个超大文件（GB/TB级别），需要基于字节数而非文件数计算。
原理：分段读取源文件 -> 分块写入目标 -> 根据已写字节 / 总字节算进度。
Python 示例 (带实时百分比)
import os
def migrate_large_file(src_path, dst_path, chunk_size=1024*1024):
total_size = os.path.getsize(src_path)
transferred = 0
with open(src_path, 'rb') as f_src, open(dst_path, 'wb') as f_dst:
while True:
chunk = f_src.read(chunk_size)
if not chunk:
break
f_dst.write(chunk)
transferred += len(chunk)
# 计算并显示进度
percent = (transferred / total_size) * 100
print(f"\r迁移中: {percent:.2f}% ({transferred}/{total_size} bytes)", end='')
print() # 换行
进阶技巧：图形化界面（Web / 桌面）
如果你需要更高级的展示（如带有平滑动画的Web仪表盘），通常采用以下架构：
-
后端（Flask / FastAPI）：
- 启动一个后台线程执行迁移任务。
- 维护一个全局变量或 Redis 存储进度（
{ "total": 1000, "done": 450 }）。 - 提供一个 API 端点
GET /progress返回当前JSON。
-
前端（HTML + JavaScript）：
- 使用
fetch或WebSocket定时（如每秒）请求进度。 - 使用 Chart.js 或原生
<progress>标签渲染进度条。
- 使用
快速实现思路（后端部分）：
from flask import Flask, jsonify
import threading
app = Flask(__name__)
progress_data = {"total": 0, "done": 0}
def background_migration():
files = ["..."]
progress_data["total"] = len(files)
for i, f in enumerate(files):
# 执行迁移逻辑
time.sleep(0.5)
progress_data["done"] = i + 1
@app.route('/progress')
def get_progress():
pct = (progress_data["done"] / progress_data["total"]) * 100 if progress_data["total"] else 0
return jsonify({"percent": round(pct, 2), "done": progress_data["done"], "total": progress_data["total"]})
if __name__ == '__main__':
threading.Thread(target=background_migration, daemon=True).start()
app.run()
避坑指南
- 性能问题：不要在每次处理一个小单元（如一条数据库记录）时都刷新界面，这本身会成为瓶颈，建议批量处理（如每处理100条记录刷新一次）。
- 总量未知：有时无法提前知道总数（例如流式数据），这种情况下，可以只显示已处理数量、速率、持续时长，而不显示百分比（
已处理: 1,234 行 | 速率: 500行/s）。 - 多线程安全：在多线程脚本中，需使用线程安全的计数器（如
threading.Lock）保护进度变量的读写。
对于大部分开发场景,方案一的 Python + tqdm 是最快、最优雅的方式，如果需要集成到系统或提供远程监控，则使用 进阶技巧的 Web 方案。
