cleaned, fix test.yaml, add exceptions and checks in .py
This commit is contained in:
@ -1,13 +1,23 @@
|
||||
from calendar import c
|
||||
import os
|
||||
import shutil
|
||||
from tabnanny import check
|
||||
import yaml
|
||||
import re
|
||||
import sys
|
||||
|
||||
|
||||
# Global variable to track errors
|
||||
errorList = {"copies": 0, "modifications": 0}
|
||||
|
||||
# Initialize results
|
||||
results = {"copies": 0, "modifications": 0}
|
||||
|
||||
|
||||
def loadConfig(configPath):
|
||||
"""Load YAML configuration."""
|
||||
try:
|
||||
print(f"Loading configuration from {configPath}")
|
||||
print(f"Loading configuration from {configPath} ...")
|
||||
with open(configPath, 'r', encoding='utf-8') as file:
|
||||
config = yaml.safe_load(file)
|
||||
|
||||
@ -15,10 +25,10 @@ def loadConfig(configPath):
|
||||
if not config:
|
||||
raise ValueError("Empty configuration file")
|
||||
|
||||
print("Configuration loaded successfully.")
|
||||
print("Configuration loaded successfully.\n")
|
||||
return config
|
||||
except FileNotFoundError:
|
||||
print(f"Error: Configuration file not found at {configPath}")
|
||||
print(f"Error: Configuration file not found at ./{configPath}")
|
||||
sys.exit(1)
|
||||
except yaml.YAMLError as e:
|
||||
print(f"Error parsing YAML configuration: {e}")
|
||||
@ -30,11 +40,11 @@ def loadConfig(configPath):
|
||||
|
||||
def ensureDirectory(path):
|
||||
"""Ensure that a directory exists."""
|
||||
print(f"Checking/creating directory: {path}")
|
||||
print(f"\nChecking/creating directory: {path}")
|
||||
os.makedirs(path, exist_ok=True)
|
||||
|
||||
|
||||
def copySources(config, destinationDirectory, results):
|
||||
def copySources(config, destinationDirectory):
|
||||
"""Copy files and folders according to copy rules."""
|
||||
print("Starting source file/folder copy process...")
|
||||
for rule in config.get('copy_rules', []):
|
||||
@ -43,17 +53,30 @@ def copySources(config, destinationDirectory, results):
|
||||
# Distinguish between sources with explicit target and without
|
||||
if isinstance(source, dict):
|
||||
sourcePath = source['source']
|
||||
targetPath = os.path.join(destinationDirectory, source['target'])
|
||||
#targetPath = os.path.join(destinationDirectory, source['target'])
|
||||
checkTargetPath = source['target']
|
||||
|
||||
# Check for absolute paths in target and correct them if necessary
|
||||
if os.path.isabs(checkTargetPath):
|
||||
targetPath = os.path.normpath(os.path.join(destinationDirectory, checkTargetPath.lstrip('./')))
|
||||
raise ValueError(f"Absolute path incorrect: Corrected {targetPath} to {targetPath}.")
|
||||
|
||||
|
||||
else:
|
||||
sourcePath = source
|
||||
targetPath = os.path.join(destinationDirectory, os.path.basename(sourcePath))
|
||||
#targetPath = os.path.normpath(os.path.join(destinationDirectory, os.path.basename(sourcePath).lstrip('./')))
|
||||
|
||||
# Check if source exists
|
||||
if not os.path.exists(sourcePath):
|
||||
raise FileNotFoundError(f"Source path does not exist: {sourcePath}")
|
||||
|
||||
# Create target directory
|
||||
ensureDirectory(os.path.dirname(targetPath))
|
||||
|
||||
# Copy or replace mode
|
||||
if rule.get('mode') == 'replace' and os.path.exists(targetPath):
|
||||
print(f"Replacing existing path: {targetPath}")
|
||||
print(f"Replacing existing file/folder: {targetPath}")
|
||||
|
||||
# Explicitly handle directory or file deletion
|
||||
if os.path.isdir(targetPath):
|
||||
@ -69,6 +92,8 @@ def copySources(config, destinationDirectory, results):
|
||||
if srcMtime <= destMtime:
|
||||
print(f"Skipping {sourcePath}: Destination is up to date")
|
||||
break # Skip this source file/folder
|
||||
else:
|
||||
print(f"Should update {sourcePath} to {targetPath}, but {sourcePath} is not (yet) existing.")
|
||||
|
||||
# Copy files or directories
|
||||
if os.path.isdir(sourcePath):
|
||||
@ -79,15 +104,30 @@ def copySources(config, destinationDirectory, results):
|
||||
print(f"Copying file: {sourcePath} -> {targetPath}")
|
||||
shutil.copy2(sourcePath, targetPath)
|
||||
results['copies'] += 1
|
||||
except FileNotFoundError as e:
|
||||
print(f"Error: {e}")
|
||||
errorList["copies"] += 1
|
||||
continue
|
||||
|
||||
except ValueError as e:
|
||||
print(f"Configuration error: {e}")
|
||||
errorList["copies"] += 1
|
||||
continue
|
||||
|
||||
except PermissionError:
|
||||
print(f"Error: Permission denied when copying {sourcePath}")
|
||||
errorList["copies"] += 1
|
||||
continue
|
||||
|
||||
except OSError as e:
|
||||
print(f"Error copying {sourcePath}: {e}")
|
||||
errorList["copies"] += 1
|
||||
continue
|
||||
|
||||
print("Source file/folder copy process completed.")
|
||||
print("\nSource file & folder copy/replace process completed.\n")
|
||||
|
||||
|
||||
def modifyFiles(config, destinationDirectory, results):
|
||||
def modifyFiles(config, destinationDirectory):
|
||||
"""Modify files according to modification rules."""
|
||||
print("Starting file modification process...")
|
||||
for rule in config.get('modification_rules', []):
|
||||
@ -136,10 +176,15 @@ def modifyFiles(config, destinationDirectory, results):
|
||||
results['modifications'] += 1
|
||||
except PermissionError:
|
||||
print(f"Error: Permission denied when modifying {filePath}")
|
||||
errorList["modifications"] += 1
|
||||
continue
|
||||
|
||||
except IOError as e:
|
||||
print(f"Error reading/writing file {filePath}: {e}")
|
||||
errorList["modifications"] += 1
|
||||
continue
|
||||
|
||||
print("File modification process completed.")
|
||||
print("\nFile modification process completed.\n")
|
||||
|
||||
|
||||
def main():
|
||||
@ -153,24 +198,27 @@ def main():
|
||||
|
||||
# Load configuration
|
||||
config = loadConfig(configPath)
|
||||
|
||||
# Initialize results
|
||||
results = {"copies": 0, "modifications": 0}
|
||||
|
||||
# Ensure destination directory
|
||||
destinationDirectory = config.get('destination_directory', './web')
|
||||
ensureDirectory(destinationDirectory)
|
||||
|
||||
# Copy files and folders
|
||||
copySources(config, destinationDirectory, results)
|
||||
copySources(config, destinationDirectory)
|
||||
|
||||
# Modify files
|
||||
modifyFiles(config, destinationDirectory, results)
|
||||
modifyFiles(config, destinationDirectory)
|
||||
|
||||
# Print results
|
||||
print(f'\nTotal successful copies: {results["copies"]}')
|
||||
print(f'Total file modifications: {results["modifications"]}')
|
||||
print(f"All operations in {destinationDirectory} completed successfully.")
|
||||
|
||||
if errorList["copies"] > 0 or errorList["modifications"] > 0:
|
||||
print("Errors occurred during the process. Check the output for details.")
|
||||
print(f"{errorList['copies']} copy errors.")
|
||||
print(f"{errorList['modifications']} modification errors.")
|
||||
else:
|
||||
print(f"All operations in {destinationDirectory} completed successfully.")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
Reference in New Issue
Block a user