Spaces:
Building
Building
| # /home/bk_anupam/code/LLM_agents/RAG_BOT/file_manager.py | |
| import os | |
| import shutil | |
| from RAG_BOT.logger import logger | |
| from RAG_BOT.config import Config | |
| class FileManager: | |
| def __init__(self): | |
| self.config = Config() | |
| def move_indexed_file(self, source_path: str, language: str) -> bool: | |
| """ | |
| Moves a successfully indexed file to the 'indexed_data' directory, | |
| replicating the exact directory structure from the original source path. | |
| For example: | |
| source_path = RAG_BOT/data/hindi/1969/filename.pdf | |
| indexed_data_path = RAG_BOT/indexed_data | |
| Will result in a destination of: RAG_BOT/indexed_data/hindi/1969/filename.pdf | |
| """ | |
| if self.config.INDEXED_DATA_PATH is None: | |
| logger.error("INDEXED_DATA_PATH not configured. Cannot move indexed files.") | |
| return False | |
| if not source_path or not os.path.exists(source_path): | |
| logger.warning(f"Source file path invalid or file does not exist: {source_path}. Cannot move.") | |
| return False | |
| try: | |
| file_name = os.path.basename(source_path) | |
| # Define the root folder for source files | |
| source_data_root = os.path.join("RAG_BOT", "data") | |
| # Get the directory of the source file and extract subdirectories relative to source_data_root | |
| source_dir = os.path.dirname(source_path) | |
| relative_subdirs = os.path.relpath(source_dir, start=source_data_root) | |
| # If no subdirectories are found, relative_subdirs might be '.'; in that case, ignore it | |
| if relative_subdirs == '.': | |
| relative_subdirs = '' | |
| # Build the destination directory path by concatenating the indexed_data path and the subdirectories | |
| destination_dir = os.path.join(self.config.INDEXED_DATA_PATH, relative_subdirs) | |
| # Normalize the destination directory to remove any trailing slashes | |
| destination_dir = os.path.normpath(destination_dir) | |
| destination_path = os.path.join(destination_dir, file_name) | |
| # Ensure that the destination directory exists | |
| os.makedirs(destination_dir, exist_ok=True) | |
| shutil.move(source_path, destination_path) | |
| logger.info(f"Successfully moved indexed file '{file_name}' to: {destination_path}") | |
| return True | |
| except Exception as e: | |
| logger.error(f"Failed to move file {source_path} to destination: {e}", exc_info=True) | |
| return False | |