-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathsync_dirs.py
More file actions
156 lines (130 loc) · 5.4 KB
/
Copy pathsync_dirs.py
File metadata and controls
156 lines (130 loc) · 5.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
#!/usr/bin/env python3
# Most code here was mostly generated by Grok 4.5, then cobbled together by human.
"""
Compare files under directory A against the same relative paths under directory B.
For each file found under A:
- "duplicate: <path>" if the file exists under B and is byte-identical
- "different: <path>" if the file exists under B but differs
- "missing: <path>" if the file does not exist under B
Binary comparison is performed by reading both files in blocks of L MiB
(default: 16 MiB).
"""
import argparse
import os
import sys, shutil
DEFAULT_BLOCK_SIZE_MB = 16
def move_missing_items(dir_a: str, dir_b: str) -> None:
"""
Recursively scan under A. For every file or subdirectory that does
not exist at the same relative path under B, move it from A to B.
The roots A and B themselves are never moved.
"""
dir_a = os.path.abspath(dir_a)
dir_b = os.path.abspath(dir_b)
if not os.path.isdir(dir_a):
raise NotADirectoryError(f"'{dir_a}' is not a directory")
if not os.path.isdir(dir_b):
raise NotADirectoryError(f"'{dir_b}' is not a directory")
for root, dirs, files in os.walk(dir_a, topdown=True):
# --- directories ---
# Iterate over a copy so we can safely modify dirs (for pruning)
for name in list(dirs):
rel = os.path.relpath(os.path.join(root, name), dir_a)
path_a = os.path.join(dir_a, rel)
path_b = os.path.join(dir_b, rel)
if not os.path.exists(path_b):
# Parent of path_b is guaranteed to exist (top-down processing)
print(f"moving dir {path_a} -> {path_b}")
shutil.move(path_a, path_b)
# Prevent os.walk from descending into the moved tree
dirs.remove(name)
# --- files ---
for name in files:
rel = os.path.relpath(os.path.join(root, name), dir_a)
path_a = os.path.join(dir_a, rel)
path_b = os.path.join(dir_b, rel)
if not os.path.exists(path_b):
print(f"moving file {path_a} -> {path_b}")
# Ensure the target parent directory exists
os.makedirs(os.path.dirname(path_b), exist_ok=True)
shutil.move(path_a, path_b)
def remove_empty_dirs(path: str) -> None:
"""
Recursively remove all empty subdirectories under the given path.
The root directory itself is never removed.
"""
path = os.path.abspath(path)
if not os.path.isdir(path):
raise NotADirectoryError(f"'{path}' is not a directory")
# Walk bottom-up so children are processed before parents
for root, dirs, _files in os.walk(path, topdown=False):
for name in dirs:
dir_path = os.path.join(root, name)
try:
print("removing empty dir:",dir_path)
os.rmdir(dir_path) # succeeds only if the directory is empty
except OSError:
# Directory is not empty (or permission error, etc.) — leave it
pass
def files_identical(path_a: str, path_b: str, block_size: int) -> bool:
"""Return True if the two files are byte-for-byte identical."""
try:
size_a = os.path.getsize(path_a)
size_b = os.path.getsize(path_b)
except OSError:
return False
if size_a != size_b:
return False
with open(path_a, "rb") as fa, open(path_b, "rb") as fb:
while True:
block_a = fa.read(block_size)
block_b = fb.read(block_size)
if block_a != block_b:
return False
if not block_a: # both files exhausted
return True
def compare_directories(dir_a: str, dir_b: str, block_size: int) -> None:
dir_a = os.path.abspath(dir_a)
dir_b = os.path.abspath(dir_b)
if not os.path.isdir(dir_a):
print(f"Error: '{dir_a}' is not a directory", file=sys.stderr)
sys.exit(1)
if not os.path.isdir(dir_b):
print(f"Error: '{dir_b}' is not a directory", file=sys.stderr)
sys.exit(1)
for root, _dirs, files in os.walk(dir_a):
for name in files:
path_a = os.path.join(root, name)
rel = os.path.relpath(path_a, dir_a)
path_b = os.path.join(dir_b, rel)
if not os.path.isfile(path_b):
print(f"missing: {rel}")
elif files_identical(path_a, path_b, block_size):
print(f"duplicate: {rel}")
os.remove(path_a)
else:
print(f"different: {rel}")
def main() -> None:
parser = argparse.ArgumentParser(
description="Recursively compare files under directory A with the same relative paths under B."
)
parser.add_argument("A", help="Source directory to scan")
parser.add_argument("B", help="Directory to compare against")
parser.add_argument(
"-L",
"--block-size",
type=int,
default=DEFAULT_BLOCK_SIZE_MB,
metavar="MB",
help=f"Block size for binary comparison in MiB (default: {DEFAULT_BLOCK_SIZE_MB})",
)
args = parser.parse_args()
if args.block_size <= 0:
print("Error: block size must be positive", file=sys.stderr)
sys.exit(1)
move_missing_items(args.A, args.B)
block_size_bytes = args.block_size * 1024 * 1024
compare_directories(args.A, args.B, block_size_bytes)
remove_empty_dirs(args.A)
if __name__ == "__main__":
main()