add exceptions
This commit is contained in:
@ -2,112 +2,176 @@ import os
|
|||||||
import shutil
|
import shutil
|
||||||
import yaml
|
import yaml
|
||||||
import re
|
import re
|
||||||
|
import sys
|
||||||
|
|
||||||
def loadConfig(config_path):
|
def loadConfig(configPath):
|
||||||
"""Load YAML configuration."""
|
"""Load YAML configuration."""
|
||||||
print(f"Loading configuration from {config_path}")
|
try:
|
||||||
with open(config_path, 'r', encoding='utf-8') as file:
|
print(f"Loading configuration from {configPath}")
|
||||||
|
with open(configPath, 'r', encoding='utf-8') as file:
|
||||||
config = yaml.safe_load(file)
|
config = yaml.safe_load(file)
|
||||||
|
|
||||||
|
# validate configuration
|
||||||
|
if not config:
|
||||||
|
raise ValueError("Empty configuration file")
|
||||||
|
|
||||||
print("Configuration loaded successfully.")
|
print("Configuration loaded successfully.")
|
||||||
return config
|
return config
|
||||||
|
except FileNotFoundError:
|
||||||
|
print(f"Error: Configuration file not found at {configPath}")
|
||||||
|
sys.exit(1)
|
||||||
|
except yaml.YAMLError as e:
|
||||||
|
print(f"Error parsing YAML configuration: {e}")
|
||||||
|
sys.exit(1)
|
||||||
|
except ValueError as e:
|
||||||
|
print(f"Configuration error: {e}")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
|
||||||
def ensureDirectory(path):
|
def ensureDirectory(path):
|
||||||
"""Ensure that a directory exists."""
|
"""Ensure that a directory exists."""
|
||||||
print(f"Checking/creating directory: {path}")
|
print(f"Checking/creating directory: {path}")
|
||||||
os.makedirs(path, exist_ok=True)
|
os.makedirs(path, exist_ok=True)
|
||||||
|
|
||||||
def copySources(config, destination_directory):
|
|
||||||
|
def copySources(config, destinationDirectory, results):
|
||||||
"""Copy files and folders according to copy rules."""
|
"""Copy files and folders according to copy rules."""
|
||||||
print("Starting source file/folder copy process...")
|
print("Starting source file/folder copy process...")
|
||||||
for rule in config.get('copy_rules', []):
|
for rule in config.get('copy_rules', []):
|
||||||
for source in rule.get('sources', []):
|
for source in rule.get('sources', []):
|
||||||
|
try:
|
||||||
# Distinguish between sources with explicit target and without
|
# Distinguish between sources with explicit target and without
|
||||||
if isinstance(source, dict):
|
if isinstance(source, dict):
|
||||||
src_path = source['source']
|
sourcePath = source['source']
|
||||||
target_path = os.path.join(destination_directory, source['target'])
|
targetPath = os.path.join(destinationDirectory, source['target'])
|
||||||
else:
|
else:
|
||||||
src_path = source
|
sourcePath = source
|
||||||
target_path = os.path.join(destination_directory, os.path.basename(src_path))
|
targetPath = os.path.join(destinationDirectory, os.path.basename(sourcePath))
|
||||||
|
|
||||||
# Create target directory
|
# Create target directory
|
||||||
ensureDirectory(os.path.dirname(target_path))
|
ensureDirectory(os.path.dirname(targetPath))
|
||||||
|
|
||||||
# Copy or replace mode
|
# Copy or replace mode
|
||||||
if rule.get('mode') == 'replace' and os.path.exists(target_path):
|
if rule.get('mode') == 'replace' and os.path.exists(targetPath):
|
||||||
print(f"Replacing existing path: {target_path}")
|
print(f"Replacing existing path: {targetPath}")
|
||||||
shutil.rmtree(target_path) if os.path.isdir(target_path) else os.remove(target_path)
|
|
||||||
|
# Explicitly handle directory or file deletion
|
||||||
|
if os.path.isdir(targetPath):
|
||||||
|
shutil.rmtree(targetPath)
|
||||||
|
else:
|
||||||
|
os.remove(targetPath)
|
||||||
|
|
||||||
|
# Update mode
|
||||||
|
if rule.get('mode') == 'update' and os.path.exists(targetPath):
|
||||||
|
print(f"Checking if {sourcePath} is newer than {targetPath}")
|
||||||
|
srcMtime = os.path.getmtime(sourcePath)
|
||||||
|
destMtime = os.path.getmtime(targetPath)
|
||||||
|
if srcMtime <= destMtime:
|
||||||
|
print(f"Skipping {sourcePath}: Destination is up to date")
|
||||||
|
break # Skip this source file/folder
|
||||||
|
|
||||||
# Copy files or directories
|
# Copy files or directories
|
||||||
if os.path.isdir(src_path):
|
if os.path.isdir(sourcePath):
|
||||||
print(f"Copying directory: {src_path} -> {target_path}")
|
print(f"Copying directory: {sourcePath} -> {targetPath}")
|
||||||
shutil.copytree(src_path, target_path)
|
shutil.copytree(sourcePath, targetPath)
|
||||||
|
results['copies'] += 1
|
||||||
else:
|
else:
|
||||||
print(f"Copying file: {src_path} -> {target_path}")
|
print(f"Copying file: {sourcePath} -> {targetPath}")
|
||||||
shutil.copy2(src_path, target_path)
|
shutil.copy2(sourcePath, targetPath)
|
||||||
|
results['copies'] += 1
|
||||||
|
except PermissionError:
|
||||||
|
print(f"Error: Permission denied when copying {sourcePath}")
|
||||||
|
except OSError as e:
|
||||||
|
print(f"Error copying {sourcePath}: {e}")
|
||||||
|
|
||||||
print("Source file/folder copy process completed.")
|
print("Source file/folder copy process completed.")
|
||||||
|
|
||||||
def modifyFiles(config, destination_directory):
|
|
||||||
|
def modifyFiles(config, destinationDirectory, results):
|
||||||
"""Modify files according to modification rules."""
|
"""Modify files according to modification rules."""
|
||||||
print("Starting file modification process...")
|
print("Starting file modification process...")
|
||||||
for rule in config.get('modification_rules', []):
|
for rule in config.get('modification_rules', []):
|
||||||
file_pattern = rule.get('file_pattern', '*')
|
filePattern = rule.get('file_pattern', '*')
|
||||||
print(f"Processing files matching pattern: {file_pattern}")
|
print(f"Processing files matching pattern: {filePattern}")
|
||||||
|
|
||||||
# Recursively search the destination directory
|
# Recursively search the destination directory
|
||||||
for root, _, files in os.walk(destination_directory):
|
for root, _, files in os.walk(destinationDirectory):
|
||||||
for filename in files:
|
for filename in files:
|
||||||
if re.match(file_pattern, filename):
|
try:
|
||||||
file_path = os.path.join(root, filename)
|
if re.match(filePattern, filename):
|
||||||
print(f"Modifying file: {file_path}")
|
filePath = os.path.join(root, filename)
|
||||||
|
print(f"Modifying file: {filePath}")
|
||||||
|
|
||||||
# Read file content
|
# Read file content
|
||||||
with open(file_path, 'r', encoding='utf-8') as f:
|
with open(filePath, 'r', encoding='utf-8') as f:
|
||||||
content = f.read()
|
content = f.read()
|
||||||
|
|
||||||
# Perform text replacements
|
# Perform text replacements
|
||||||
for insert_rule in rule.get('insert_rules', []):
|
for insertRule in rule.get('insert_rules', []):
|
||||||
if 'after_text' in insert_rule:
|
if 'after_text' in insertRule:
|
||||||
print(f" Inserting text after: {insert_rule['after_text']}")
|
print(f" Inserting text after: {insertRule['after_text']}")
|
||||||
content = content.replace(
|
content = content.replace(
|
||||||
insert_rule['after_text'],
|
insertRule['after_text'],
|
||||||
insert_rule['after_text'] + insert_rule['insert_text']
|
insertRule['after_text'] + insertRule['insert_text']
|
||||||
)
|
)
|
||||||
|
|
||||||
if 'before_text' in insert_rule:
|
if 'before_text' in insertRule:
|
||||||
print(f" Inserting text before: {insert_rule['before_text']}")
|
print(f" Inserting text before: {insertRule['before_text']}")
|
||||||
content = content.replace(
|
content = content.replace(
|
||||||
insert_rule['before_text'],
|
insertRule['before_text'],
|
||||||
insert_rule['insert_text'] + insert_rule['before_text']
|
insertRule['insert_text'] + insertRule['before_text']
|
||||||
)
|
)
|
||||||
|
|
||||||
if 'old_text' in insert_rule:
|
if 'old_text' in insertRule:
|
||||||
print(f" Replacing text: {insert_rule['old_text']} -> {insert_rule['new_text']}")
|
print(f" Replacing text: {insertRule['old_text']} -> {insertRule['new_text']}")
|
||||||
content = content.replace(
|
content = content.replace(
|
||||||
insert_rule['old_text'],
|
insertRule['old_text'],
|
||||||
insert_rule['new_text']
|
insertRule['new_text']
|
||||||
)
|
)
|
||||||
|
|
||||||
# Write modified contents
|
# Write modified contents
|
||||||
with open(file_path, 'w', encoding='utf-8') as f:
|
with open(filePath, 'w', encoding='utf-8') as f:
|
||||||
f.write(content)
|
f.write(content)
|
||||||
|
|
||||||
|
results['modifications'] += 1
|
||||||
|
except PermissionError:
|
||||||
|
print(f"Error: Permission denied when modifying {filePath}")
|
||||||
|
except IOError as e:
|
||||||
|
print(f"Error reading/writing file {filePath}: {e}")
|
||||||
|
|
||||||
print("File modification process completed.")
|
print("File modification process completed.")
|
||||||
|
|
||||||
def main(config_path):
|
|
||||||
|
def main():
|
||||||
"""Main function to execute all operations."""
|
"""Main function to execute all operations."""
|
||||||
|
# Check command-line argument
|
||||||
|
if len(sys.argv) < 2:
|
||||||
|
print("Please provide the path to the configuration file.")
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
|
configPath = sys.argv[1]
|
||||||
|
|
||||||
# Load configuration
|
# Load configuration
|
||||||
config = loadConfig(config_path)
|
config = loadConfig(configPath)
|
||||||
|
|
||||||
|
# Initialize results
|
||||||
|
results = {"copies": 0, "modifications": 0}
|
||||||
|
|
||||||
# Ensure destination directory
|
# Ensure destination directory
|
||||||
destination_directory = config.get('destination_directory', './web')
|
destinationDirectory = config.get('destination_directory', './web')
|
||||||
ensureDirectory(destination_directory)
|
ensureDirectory(destinationDirectory)
|
||||||
|
|
||||||
# Copy files and folders
|
# Copy files and folders
|
||||||
copySources(config, destination_directory)
|
copySources(config, destinationDirectory, results)
|
||||||
|
|
||||||
# Modify files
|
# Modify files
|
||||||
modifyFiles(config, destination_directory)
|
modifyFiles(config, destinationDirectory, results)
|
||||||
|
|
||||||
|
# Print results
|
||||||
|
print(f'\nTotal successful copies: {results["copies"]}')
|
||||||
|
print(f'Total file modifications: {results["modifications"]}')
|
||||||
|
print(f"All operations in {destinationDirectory} completed successfully.")
|
||||||
|
|
||||||
print(f"All operations in {destination_directory} completed successfully.")
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
main('config.yaml')
|
main()
|
Reference in New Issue
Block a user