current_date
import yaml
import os
import logging
from datetime import datetime
import shutil
from concurrent.futures import ThreadPoolExecutor
import multiprocessing
def setup_logger(log_file):
logger = logging.getLogger()
logger.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
file_handler = logging.FileHandler(log_file, mode='a')
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
def read_yaml_file(file_path):
try:
with open(file_path, 'r') as file:
return yaml.safe_load(file)
except FileNotFoundError:
logging.error(f"Error: File '{file_path}' not found.")
return None
except Exception as e:
logging.error(f"An error occurred while reading the file: {e}")
return None
def move_files(source_path, backup_path, mlz_path):
try:
moved_files = []
for root, _, files in os.walk(source_path):
for file_name in files:
source_file_path = os.path.join(root, file_name)
destination_file_path = os.path.join(backup_path, file_name)
try:
shutil.move(source_file_path, destination_file_path)
moved_files.append(file_name)
logging.info(f"File '{file_name}' moved to '{destination_file_path}' successfully.")
except Exception as e:
logging.error(f"Error moving file '{file_name}': {e}")
# Apply renaming logic for each moved file
for file_name in moved_files:
source_file_path = os.path.join(backup_path, file_name)
if os.path.exists(source_file_path):
base_name, extension = os.path.splitext(file_name)
file_number = int(base_name[-1]) # Extract last digit of base_name
if file_number == 9:
file_number -= 1 # Decrement by 1 if file_number is 9
else:
file_number += 1 # Increment by 1 otherwise
new_base_name = base_name[:-1] + str(file_number) # Update base_name
new_file_name = f"{new_base_name}{extension}"
new_file_path = os.path.join(backup_path, new_file_name)
os.rename(source_file_path, new_file_path)
logging.info(f"File '{file_name}' renamed to '{new_file_name}' successfully.")
# Copy renamed file to MLANDINGZONE_PATH
mlz_file_path = os.path.join(mlz_path, new_file_name)
shutil.copy(new_file_path, mlz_file_path)
logging.info(f"File '{new_file_name}' copied to MLANDINGZONE_PATH successfully.")
return moved_files
except Exception as e:
logging.error(f"Error moving files from {source_path} to {backup_path}: {e}")
def main():
# Path to the directory containing the master_config.yaml file
directory_path = 'D:\\Python_pro\\Feeds_recovery\\'
# Name of the YAML file
yaml_file_name = 'master_config.yaml'
# Full path to the YAML file
yaml_file_path = os.path.join(directory_path, yaml_file_name)
# Logging setup
log_file_name = datetime.now().strftime('%Y-%m-%d_%H-%M-%S.log')
log_file_path = os.path.join(directory_path, log_file_name)
setup_logger(log_file_path)
# Check if YAML file exists
if not os.path.exists(yaml_file_path):
logging.error(f"Error: File '{yaml_file_path}' does not exist.")
return
# Read YAML file
data = read_yaml_file(yaml_file_path)
if data is None:
logging.error("Exiting due to errors.")
return
# Filter sections where APP_MODE is 'YES'
app_yes_configs = [config for config in data if config.get('APP_MODE', '') == 'YES']
# Print configurations and check file existence for filtered sections
for config in app_yes_configs:
logging.info(f"APP: {config.get('APP')}")
logging.info(f"APP_MODE: {config.get('APP_MODE')}")
# Check if only one mode is set to 'YES' or one of the flags ['P', 'L', 'E']
modes = ['LANDINGZONE_MODE', 'PROCESSZONE_MODE', 'ERROR_FILE_MODE', 'LPE_MODE']
num_yes_modes = sum(config.get(mode, '') in ['YES', 'P', 'L', 'E'] for mode in modes)
if num_yes_modes != 1:
logging.warning("Only one mode (LANDINGZONE_MODE, PROCESSZONE_MODE, ERROR_FILE_MODE, or LPE_MODE) should be set to 'YES' or one of the flags ['P', 'L', 'E'].")
continue
# Determine which mode is set to 'YES' or one of the flags ['P', 'L', 'E']
yes_mode = next(mode for mode in modes if config.get(mode, '') in ['YES', 'P', 'L', 'E'])
# If all modes except LPE_MODE are set to 'NO', perform LPE_MODE operation by default
if yes_mode == 'LPE_MODE' and all(config.get(mode, '') == 'NO' for mode in modes if mode != 'LPE_MODE'):
lpe_flag = config.get('LPE_MODE')
if lpe_flag in ['P', 'L', 'E']:
logging.info(f"LPE_MODE: {lpe_flag}")
# Read file names from LANDINGZONE_FILES.txt, PROCESSZONE_FILES.txt, or ERROR_FILE.txt
if lpe_flag == 'P':
file_list_path = config.get('PROCESSZONE_FILES')
elif lpe_flag == 'L':
file_list_path = config.get('LANDINGZONE_FILES')
elif lpe_flag == 'E':
file_list_path = config.get('ERROR_FILE')
if not os.path.exists(file_list_path):
logging.warning(f"File '{file_list_path}' not found.")
continue
file_list = read_file_content(file_list_path)
# Perform file operations based on LPE_MODE
source_path = ""
if lpe_flag == 'P':
source_path = config.get('PROCESSZONE_PATH')
elif lpe_flag == 'L':
source_path = config.get('LANDINGZONE_PATH')
elif lpe_flag == 'E':
source_path = config.get('ERROR_FILE_PATH')
backup_path = config.get('BACKUP_PATH')
if not os.path.exists(source_path):
logging.warning(f"Source path '{source_path}' not found.")
continue
if not os.path.exists(backup_path):
logging.warning(f"Backup path '{backup_path}' not found.")
continue
mlz_path = config.get('MLANDINGZONE_PATH')
if not os.path.exists(mlz_path):
logging.warning(f"MLANDINGZONE path '{mlz_path}' not found.")
continue
# Move files in parallel using ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=multiprocessing.cpu_count()) as executor:
moved_files = executor.submit(move_files, source_path, backup_path, mlz_path).result()
logging.info(f"Files moved: {moved_files}")
else:
logging.warning(f"Invalid LPE_MODE '{lpe_flag}' found in configuration.")
else:
# Get the path and backup path based on the mode that is set to 'YES' or one of the flags ['P', 'L', 'E']
mode_mapping = {
'LANDINGZONE_MODE': ('LANDINGZONE_PATH', 'BACKUP_PATH'),
'PROCESSZONE_MODE': ('PROCESSZONE_PATH', 'BACKUP_PATH'),
'ERROR_FILE_MODE': ('ERROR_FILE_PATH', 'BACKUP_PATH')
}
path_key, backup_path_key = mode_mapping[yes_mode]
source_path = config.get(path_key)
backup_path = config.get(backup_path_key)
mlz_path = config.get('MLANDINGZONE_PATH')
# Move only files from the source path to the backup path and store the list of moved files
if not os.path.exists(source_path):
logging.warning(f"Source path '{source_path}' not found.")
continue
if not os.path.exists(backup_path):
logging.warning(f"Backup path '{backup_path}' not found.")
continue
if not os.path.exists(mlz_path):
logging.warning(f"MLANDINGZONE path '{mlz_path}' not found.")
continue
# Move files in parallel using ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=multiprocessing.cpu_count()) as executor:
moved_files = executor.submit(move_files, source_path, backup_path, mlz_path).result()
logging.info(f"Files moved: {moved_files}")
if __name__ == "__main__":
main()
- APP: BIS
APP_MODE: 'NO'
LANDINGZONE_PATH: 'D:\Python_pro\Feeds_recovery\BIS\LANDINGZONE'
PROCESSZONE_PATH: 'D:\Python_pro\Feeds_recovery\BIS\PROCESSZONE'
BACKUP_PATH: 'D:\Python_pro\Feeds_recovery\BIS\BIS_BACKUP'
LANDINGZONE_MODE: 'NO'
PROCE
2024-04-13 17:05:31,706 - INFO - APP: RFDT
2024-04-13 17:05:31,706 - INFO - APP_MODE: YES
2024-04-13 17:05:31,706 - INFO - File 'Pnew_test_file_13.txt' moved to 'D:\Python_pro\Feeds_recovery\RFDT\RFDT_BACKUP\Pnew_test_file_13.txt' successfully.
2024-04-13 17:05:31,706 - INFO - File 'Pnew_test_file_20240413035958.txt' moved to 'D:\Python_pro\Feeds_recovery\RFDT\RFDT_BACKUP\Pnew_test_file_20240413035958.txt' successfully.
2024-04-13 17:05:31,706 - INFO - File 'Pnew_test_file_24.txt' moved to 'D:\Python_pro\Feeds_recovery\RFDT\RFDT_BACKUP\Pnew_test_file_24.txt' successfully.
2024-04-13 17:05:31,721 - INFO - File 'Pnew_test_file_46.txt' moved to 'D:\Python_pro\Feeds_recovery\RFDT\RFDT_BACKUP\Pnew_test_file_46.txt' successfully.
2024-04-13 17:05:31,721 - INFO - File 'Pnew_test_file_13.txt' renamed to 'Pnew_test_file_14.txt' successfully.
2024-04-13 17:05:31,721 - INFO - File 'Pnew_test_file_14.txt' copied to MLANDINGZONE_PATH successfully.
2024-04-13 17:05:31,721 - INFO - File 'Pnew_test_file_20240413035958.txt' renamed to 'Pnew_test_file_20240413035959.txt' successfully.
2024-04-13 17:05:31,721 - INFO - File 'Pnew_test_file_20240413035959.txt' copied to MLANDINGZONE_PATH successfully.
2024-04-13 17:05:31,721 - INFO - File 'Pnew_test_file_24.txt' renamed to 'Pnew_test_file_25.txt' successfully.
2024-04-13 17:05:31,735 - INFO - File 'Pnew_test_file_25.txt' copied to MLANDINGZONE_PATH successfully.
2024-04-13 17:05:31,738 - INFO - File 'Pnew_test_file_46.txt' renamed to 'Pnew_test_file_47.txt' successfully.
2024-04-13 17:05:31,738 - INFO - File 'Pnew_test_file_47.txt' copied to MLANDINGZONE_PATH successfully.
2024-04-13 17:05:31,738 - INFO - Files moved: ['Pnew_test_file_13.txt', 'Pnew_test_file_20240413035958.txt', 'Pnew_test_file_24.txt', 'Pnew_test_file_46.txt']
Subject: Automated Bulk Recovery DAG Proposal
Hi Nisha and Team,
As discussed during our last MCR, approximately 600 feeds across multiple applications were moved to the REJECTED folder due to the EMR1 version upgrade. Over the past few days, we've observed unforeseen events where multiple feeds or files were either stuck or moved from the landing zone to the process zone without the File Ingestion DAG performing the necessary cleanup actions. Additionally, there have been instances of application glitches, resulting in feeds getting stuck in the process zone.
In such cases, we've had to manually recover these feeds, which has been both time-consuming and labor-intensive. To address this issue and streamline the process, we propose creating an Automated Bulk Recovery DAG.
We are providing a detailed explanation of the Bulk_Recovery_Dag.py Python script, focusing on its functionality for bulk recovery purposes. This explanation is divided into two sections: Functionality Overview and Code Structure. This structure is intended to give you both a high-level understanding and a deeper insight into how the DAG operates.
________________________________________
Section 1: Functionality Overview
The Bulk_Recovery_Dag.py script is specifically designed to facilitate the bulk recovery of files that have become stuck during processing, particularly in the process zone or REJECT folder, across predefined applications like BIS, FRANK, or RFDT. Below is a summary of how the script achieves this:
1. Configuration Management:
o The script operates using settings defined in an external YAML configuration file (master_config.yaml). This file includes crucial information such as directory paths, file handling rules, and a list of feed names associated with various applications like BIS, FRANK, or RFDT. This approach allows the script to be easily adapted to different environments without the need for code changes.
2. Bulk Recovery Process:
o The bulk recovery process begins by identifying the application feeds that are stuck in the process zone or REJECT folder. These stuck feeds may have been moved to the REJECT folder for various reasons, such as errors during processing or specific actions taken by the Prepare Flow DAG. Feeds that were moved from the landing zone to the process zone by the File Arrival DAG or experienced issues during file ingestion are also considered.
o Once these stuck feeds are identified, their names are recorded into specific application files, such as ERROR_FILE.txt or PROCESSZONE_FILES.txt.
o Following this, the Bulk_Recovery_Dag is triggered. This DAG automates the recovery process, efficiently handling multiple feeds in bulk and ensuring they are moved and processed correctly.
3. File Transfer and Renaming:
o The script transfers the identified stuck feeds from the source directory to a designated backup directory, preventing potential data loss.
o After moving the files, the script applies a consistent renaming logic: the last digit of each filename is adjusted (decremented if it is 9, otherwise incremented). This step is crucial for maintaining an organized and conflict-free file system.
4. Logging and Error Handling:
o Detailed logs are generated for every action, including successful file moves and any errors encountered. This comprehensive logging is essential for monitoring the recovery process and troubleshooting any issues. The script is designed to handle errors gracefully, ensuring that the recovery process continues even if some files encounter problems.
By automating these steps, the Bulk_Recovery_Dag.py DAG ensures a systematic and reliable recovery of multiple feeds in bulk, reducing the likelihood of data loss or operational disruptions.
________________________________________
Section 2: Code Structure
Now, let’s delve into the specific components of the script:
1. Imports:
o The script imports several modules essential for its operations:
 yaml: For reading configuration settings.
 os and shutil: For file operations.
 logging: For setting up a comprehensive logging system.
 datetime: Used for timestamps in logs.
 ThreadPoolExecutor and multiprocessing: For potential parallel processing, enhancing efficiency.
2. Logging Setup:
o The setup_logger(log_file) function initializes the logging system:
 Level: Set to DEBUG to capture detailed logs.
 Format: Each log entry includes the timestamp, severity level, and message.
 File Handling: Logs are stored in a specified file, which is appended to with each run.
3. Configuration Loading:
o The read_yaml_file(file_path) function loads configuration data from the YAML file:
 It converts the file's contents into a Python dictionary, making it easy to reference configuration settings throughout the script.
 Error handling is included to manage scenarios where the file might be missing or unreadable, with errors being logged appropriately.
4. File Management Logic:
o The move_files(source_path, backup_path, mlz_path) function is the core of the script:
 File Movement: It scans the source directory, identifies the feeds related to the specified applications (BIS, FRANK, RFDT), and moves each file to the backup directory, logging each action.
 Renaming: After moving, it renames files based on the last digit of their filenames, ensuring a systematic approach to file management.
5. Error Resilience:
o The script is designed to log errors without interrupting the process. This means that even if a file encounters an issue, the script will continue to process the remaining files, ensuring maximum efficiency and reliability.
________________________________________
By combining configuration-driven design, detailed logging, and robust error handling, it ensures that multiple feeds are recovered efficiently and with minimal disruption to operations.
Please Note: The recovery DAG has been thoroughly tested on my local PC. We plan to perform additional testing in the Pre-prod environment, where adjustments related to AWS keys will be made. The success log from the local test has been captured and shared via email.
Please feel free to reach out if you need further clarification or if there are specific aspects of the script you would like to discuss.
Thank you for your time and attention.
Best regards,