import argparse
import os
from concurrent.futures import ThreadPoolExecutor
import subprocess
import sys
import platform
# 在文件开头添加平台检测
IS_WINDOWS = platform.system() == 'Windows'
RSYNC_PATH = '/usr/bin/rsync' if not IS_WINDOWS else 'rsync.exe'
def change_default_encoding():
'''判断是否在 windows git-bash 下运行,是则使用 utf-8 编码'''
if platform.system() == 'Windows':
terminal = os.environ.get('TERM')
if terminal and 'xterm' in terminal:
sys.stdout.reconfigure(encoding='utf-8')
def read_backup_list(file_path):
"""读取备份目录列表文件,过滤空行和注释"""
directories = []
with open(file_path, 'r') as f:
for line in f:
line = line.strip()
if line and not line.startswith('#'):
directories.append(line)
return directories
def run_rsync(src_dir, target_root):
"""执行rsync备份(终极路径保留版)"""
try:
# 保留原始路径结构
src = src_dir.rstrip('/') + '/'
dst = target_root.rstrip('/') + '/'
cmd = [
RSYNC_PATH,
'-avhn',
src,
dst
]
print(cmd, flush=True)
# Windows特殊处理
if IS_WINDOWS:
cmd.insert(1, '-rt') # 替换归档模式参数
# 执行原始命令
subprocess.run(
cmd,
check=True,
shell=IS_WINDOWS,
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE
)
return True
except subprocess.CalledProcessError as e:
print(f"Failed: {e.stderr.decode().strip()}")
return False
def main():
parser = argparse.ArgumentParser(
description='Parallel rsync backup tool',
formatter_class=argparse.ArgumentDefaultsHelpFormatter
)
parser.add_argument(
'-l', '--list',
required=True,
help='Path to file containing directory list to backup'
)
parser.add_argument(
'-t', '--target',
required=True,
help='Root directory for backups'
)
parser.add_argument(
'-j', '--jobs',
type=int,
default=4,
help='Number of parallel jobs'
)
args = parser.parse_args()
# 读取备份目录列表
try:
directories = read_backup_list(args.list)
print(f"📄 Found {len(directories)} directories to backup")
except Exception as e:
print(f"🚨 Error reading list file: {str(e)}")
return
# 创建线程池执行任务
with ThreadPoolExecutor(max_workers=args.jobs) as executor:
futures = []
for idx, src_dir in enumerate(directories, 1):
print(f"🔄 Queueing ({idx}/{len(directories)}): {src_dir}")
futures.append(executor.submit(run_rsync, src_dir, args.target))
# 等待所有任务完成
success = 0
for future in futures:
if future.result():
success += 1
print(f"n📊 Backup complete! Success: {success}/{len(directories)}")
if __name__ == '__main__':
change_default_encoding()
main()
bash$ python sync.py -h
usage: sync.py [-h] -l LIST -t TARGET [-j JOBS]
Parallel rsync backup tool
options:
-h, --help show this help message and exit
-l LIST, --list LIST Path to file containing directory list to backup
(default: None)
-t TARGET, --target TARGET
Root directory for backups (default: None)
-j JOBS, --jobs JOBS Number of parallel jobs (default: 4)
python sync.py -l file.list -t /d/test1/
bash$ cat test2.py
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--path") # 默认 type=str
args = parser.parse_args()
print(f"Raw path: {args.path}")
import sys
print(sys.argv)
python test2.py --path /d/test1
export MSYS_NO_PATHCONV=1
unset MSYS_NO_PATHCONV
原文始发于微信公众号(生有可恋):Windows 下Git Bash的目录转换
免责声明:文章中涉及的程序(方法)可能带有攻击性,仅供安全研究与教学之用,读者将其信息做其他用途,由读者承担全部法律及连带责任,本站不承担任何法律及连带责任;如有问题可邮件联系(建议使用企业邮箱或有效邮箱,避免邮件被拦截,联系方式见首页),望知悉。
- 左青龙
- 微信扫一扫
-
- 右白虎
- 微信扫一扫
-
评论