261 lines
11 KiB
Python
261 lines
11 KiB
Python
import os
|
|
import shutil
|
|
import yaml
|
|
import re
|
|
import sys
|
|
|
|
|
|
# Global variable to track errors
|
|
errorList = {"copies": 0, "modifications": 0, "copyWarnings": 0, "modWarnings": 0}
|
|
|
|
# Initialize results
|
|
results = {"copies": 0, "modifications": 0}
|
|
|
|
|
|
def loadConfig(configPath):
|
|
"""Load YAML configuration."""
|
|
try:
|
|
print(f"Loading configuration from {configPath} ...")
|
|
with open(configPath, 'r', encoding='utf-8') as file:
|
|
config = yaml.safe_load(file)
|
|
|
|
# validate configuration
|
|
if not config:
|
|
raise ValueError("Empty configuration file")
|
|
|
|
print("Configuration loaded successfully.")
|
|
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):
|
|
"""Ensure that a directory exists."""
|
|
print(f"\nChecking for or creating directory: {path}")
|
|
os.makedirs(path, exist_ok=True)
|
|
|
|
|
|
def copySources(config, destinationDirectory):
|
|
"""Copy files and folders according to copy rules."""
|
|
print("Starting source file/folder copy and replace process...")
|
|
for rule in config.get('copy_rules', []):
|
|
for source in rule.get('sources', []):
|
|
try:
|
|
# Distinguish between sources with explicit target and without
|
|
if isinstance(source, dict):
|
|
sourcePath = source['source']
|
|
|
|
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 {checkTargetPath} to {targetPath}.")
|
|
else:
|
|
#targetPath = os.path.join(destinationDirectory, source['target']) # old code
|
|
targetPath = os.path.normpath(os.path.join(destinationDirectory, checkTargetPath))
|
|
|
|
|
|
else:
|
|
sourcePath = source
|
|
#targetPath = os.path.join(destinationDirectory, os.path.basename(sourcePath)) # old code
|
|
targetPath = os.path.normpath(os.path.join(destinationDirectory, os.path.basename(sourcePath)))
|
|
|
|
# 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 file/folder: {targetPath}")
|
|
|
|
# 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
|
|
elif rule.get('mode') != 'update' and not os.path.exists(sourcePath):
|
|
print(f"Should update {sourcePath} to {targetPath}, but {sourcePath} is not (yet) existing.")
|
|
|
|
# Copy files or directories
|
|
if os.path.isdir(sourcePath):
|
|
if not os.path.exists(targetPath):
|
|
print(f"Copying directory: {sourcePath} -> {targetPath}")
|
|
shutil.copytree(sourcePath, targetPath)
|
|
results['copies'] += 1
|
|
else:
|
|
print(f"Skipping directory: {sourcePath} -> {targetPath} (already exists)")
|
|
errorList['copyWarnings'] += 1
|
|
else:
|
|
if not os.path.exists(targetPath):
|
|
print(f"Copying file: {sourcePath} -> {targetPath}")
|
|
shutil.copy2(sourcePath, targetPath)
|
|
results['copies'] += 1
|
|
else:
|
|
print(f"Skipping file: {sourcePath} -> {targetPath} (already exists)")
|
|
errorList['copyWarnings'] += 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(f"\nSource file & folder copy/replace process completed with {errorList['copies']} errors and {errorList['copyWarnings']} warnings.\n")
|
|
|
|
|
|
def modifyFiles(config, destinationDirectory):
|
|
"""Modify files according to modification rules."""
|
|
print("Starting file modification process...")
|
|
for rule in config.get('modification_rules', []):
|
|
filePattern = rule.get('file_pattern', '*')
|
|
print(f"Processing files matching pattern: {filePattern}")
|
|
|
|
# Recursively search the destination directory
|
|
for root, _, files in os.walk(destinationDirectory):
|
|
for filename in files:
|
|
try:
|
|
if re.match(filePattern, filename):
|
|
filePath = os.path.join(root, filename)
|
|
print(f"Modifying file: {filePath}")
|
|
|
|
# Read file content
|
|
with open(filePath, 'r', encoding='utf-8') as f:
|
|
content = f.read()
|
|
|
|
# Perform text replacements
|
|
for insertRule in rule.get('insert_rules', []):
|
|
if 'after_text' in insertRule:
|
|
after_text = insertRule['after_text']
|
|
insert_text = insertRule['insert_text']
|
|
if after_text not in content:
|
|
raise ValueError(f"Text '{after_text}' not found in file: {filePath}")
|
|
elif insert_text not in content:
|
|
print(f" Inserting text after: {after_text}")
|
|
content = content.replace(after_text, after_text + insert_text)
|
|
else:
|
|
print(f" Text already present after: {after_text}")
|
|
errorList['modWarnings'] += 1
|
|
|
|
|
|
if 'before_text' in insertRule:
|
|
before_text = insertRule['before_text']
|
|
insert_text = insertRule['insert_text']
|
|
if before_text not in content:
|
|
raise ValueError(f"Text '{before_text}' not found in file: {filePath}")
|
|
elif insert_text not in content:
|
|
print(f" Inserting text before: {before_text}")
|
|
content = content.replace(before_text, insert_text + before_text)
|
|
else:
|
|
print(f" Text already present before: {before_text}")
|
|
errorList['modWarnings'] += 1
|
|
|
|
|
|
if 'old_text' in insertRule:
|
|
old_text = insertRule['old_text']
|
|
new_text = insertRule['new_text']
|
|
if old_text not in content:
|
|
raise ValueError(f"Text '{old_text}' not found in file: {filePath}")
|
|
elif new_text not in content:
|
|
print(f" Replacing text: {old_text} -> {new_text}")
|
|
content = content.replace(old_text, new_text)
|
|
else:
|
|
print(f" Text already replaced: {old_text} -> {new_text}")
|
|
errorList['modWarnings'] += 1
|
|
|
|
|
|
# Write modified contents
|
|
with open(filePath, 'w', encoding='utf-8') as f:
|
|
f.write(content)
|
|
|
|
results['modifications'] += 1
|
|
|
|
except ValueError as e:
|
|
print(f"Error: {e}")
|
|
errorList["modifications"] += 1
|
|
continue
|
|
|
|
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(f"\nFile modification process completed with {errorList['modifications']} errors and {errorList['modWarnings']} warnings.\n")
|
|
|
|
|
|
def main():
|
|
"""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
|
|
config = loadConfig(configPath)
|
|
|
|
# Ensure destination directory
|
|
destinationDirectory = config.get('destination_directory', './web')
|
|
ensureDirectory(destinationDirectory)
|
|
|
|
# Copy files and folders
|
|
copySources(config, destinationDirectory)
|
|
|
|
# Modify files
|
|
modifyFiles(config, destinationDirectory)
|
|
|
|
# Print results
|
|
print(f'\nTotal successful copies: {results["copies"]}')
|
|
print(f'Total file modifications: {results["modifications"]}')
|
|
|
|
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.")
|
|
elif errorList["copyWarnings"] > 0 or errorList["modWarnings"] > 0:
|
|
print(f"All operations in {destinationDirectory} from {configPath} completed successfully! But there were {errorList['modWarnings'] + errorList['copyWarnings']} warnings. Mybe you should check them.")
|
|
else:
|
|
print(f"All operations in {destinationDirectory} from {configPath} completed successfully!")
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main() |