-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathfile_copy_new_folder.py
More file actions
96 lines (78 loc) · 3.5 KB
/
Copy pathfile_copy_new_folder.py
File metadata and controls
96 lines (78 loc) · 3.5 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
import os
import shutil
import pandas as pd
from datetime import datetime
# Configuration
csv_path = r"S:\qfl_australia_tls\Curtain Fig\trees_within_plot.csv"
source_folder = r"S:\qfl_australia_tls\Curtain Fig\output"
destination_folder = r"S:\qfl_australia_tls\Curtain Fig\selected_trees2.0"
leftover_folder = r"S:\qfl_australia_tls\Curtain Fig\leftover_trees"
def setup_folders():
"""Create destination folders if they don't exist."""
os.makedirs(destination_folder, exist_ok=True)
os.makedirs(leftover_folder, exist_ok=True)
def get_processed_files():
"""Get sets of already processed files for resume functionality."""
selected = set(os.listdir(destination_folder)) if os.path.exists(destination_folder) else set()
leftover = set(os.listdir(leftover_folder)) if os.path.exists(leftover_folder) else set()
return selected, leftover
def copy_file(source_path, dest_path, file_type):
"""Copy a single file and return result status."""
try:
# Check path length for Windows
if len(dest_path) > 260:
return f"skipped_long_path"
shutil.copy2(source_path, dest_path)
return f"copied_to_{file_type}"
except Exception as e:
return f"error: {e}"
def main():
setup_folders()
# Load CSV and get file lists
df = pd.read_csv(csv_path)
if 'filename' not in df.columns:
raise ValueError("CSV file must contain a column named 'filename'")
csv_filenames = set(df['filename'].tolist())
all_files = [f for f in os.listdir(source_folder) if os.path.isfile(os.path.join(source_folder, f))]
already_selected, already_leftover = get_processed_files()
# Statistics
stats = {'processed': 0, 'skipped': 0, 'errors': 0}
total_files = len(all_files)
print(f"Starting: {total_files} files total")
print(f"Already processed: {len(already_selected)} selected, {len(already_leftover)} leftover")
# Process files
for i, file_name in enumerate(all_files, 1):
# Progress indicator
if i % 500 == 0 or i == total_files:
print(f"Progress: {i}/{total_files}")
# Skip if already processed
if file_name in already_selected or file_name in already_leftover:
stats['skipped'] += 1
continue
source_path = os.path.join(source_folder, file_name)
# Determine destination based on CSV membership
if file_name in csv_filenames:
dest_path = os.path.join(destination_folder, file_name)
result = copy_file(source_path, dest_path, "selected")
else:
dest_path = os.path.join(leftover_folder, file_name)
result = copy_file(source_path, dest_path, "leftover")
# Update statistics
if result.startswith("copied"):
stats['processed'] += 1
elif result.startswith("error") or result.startswith("skipped"):
stats['errors'] += 1
if result.startswith("error"):
print(f"Error with {file_name}: {result}")
# Summary
print(f"\n=== SUMMARY ===")
print(f"Total files: {total_files}")
print(f"Newly processed: {stats['processed']}")
print(f"Skipped (already done): {stats['skipped']}")
print(f"Errors/long paths: {stats['errors']}")
# Check for missing files from CSV
missing = [f for f in csv_filenames if f not in all_files]
if missing:
print(f"Files in CSV but not in source folder: {len(missing)}")
if __name__ == "__main__":
main()