|
|
import os |
|
|
import yaml |
|
|
import logging |
|
|
from typing import Dict, List |
|
|
|
|
|
|
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') |
|
|
|
|
|
def load_extract_config(config_name: str) -> Dict: |
|
|
"""Load a YAML configuration file from the configs directory. |
|
|
|
|
|
Args: |
|
|
config_name (str): Name of the config file (e.g., 'model_config.yml'). |
|
|
|
|
|
Returns: |
|
|
Dict: Parsed configuration dictionary. |
|
|
|
|
|
Raises: |
|
|
FileNotFoundError: If the configuration file does not exist. |
|
|
ValueError: If config_name is empty or not a string. |
|
|
""" |
|
|
if not isinstance(config_name, str) or not config_name.strip(): |
|
|
raise ValueError("config_name must be a non-empty string") |
|
|
|
|
|
config_path = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', 'configs', config_name)) |
|
|
logging.debug(f"Attempting to load config from: {config_path}") |
|
|
|
|
|
if not os.path.exists(config_path): |
|
|
logging.error(f"Configuration file not found: {config_path}") |
|
|
raise FileNotFoundError(f"Configuration file not found: {config_path}") |
|
|
|
|
|
try: |
|
|
with open(config_path, 'r') as f: |
|
|
config = yaml.safe_load(f) |
|
|
if config is None: |
|
|
logging.warning(f"Configuration file {config_name} is empty") |
|
|
return {} |
|
|
logging.info(f"Successfully loaded config: {config_name}") |
|
|
return config |
|
|
except yaml.YAMLError as e: |
|
|
logging.error(f"Failed to parse YAML in {config_name}: {e}") |
|
|
raise |
|
|
|
|
|
def get_parquet_file_names() -> List[str]: |
|
|
"""Retrieve Parquet file names from the extract_data.yml configuration. |
|
|
|
|
|
Returns: |
|
|
List[str]: List of Parquet file names derived from CSV file names. |
|
|
|
|
|
Raises: |
|
|
FileNotFoundError: If extract_data.yml is missing. |
|
|
ValueError: If no files are specified in the configuration. |
|
|
""" |
|
|
config = load_extract_config('extract_data.yml') |
|
|
files = config.get('files', []) |
|
|
if not files: |
|
|
logging.error("No files specified in extract_data.yml") |
|
|
raise ValueError("No files specified in extract_data.yml") |
|
|
|
|
|
parquet_files = [f.replace(".csv", ".parquet") for f in files] |
|
|
logging.debug(f"Derived Parquet file names: {parquet_files}") |
|
|
return parquet_files |
|
|
|
|
|
def load_pipeline_config() -> Dict: |
|
|
"""Load pipeline configuration from pipeline_config.yml. |
|
|
|
|
|
Returns: |
|
|
Dict: Pipeline configuration dictionary. |
|
|
|
|
|
Raises: |
|
|
FileNotFoundError: If pipeline_config.yml is missing. |
|
|
""" |
|
|
config = load_extract_config('pipeline_config.yml') |
|
|
logging.debug(f"Pipeline config loaded: {config}") |
|
|
return config |
|
|
|
|
|
def define_server_filenames(**kwargs) -> List[str]: |
|
|
"""Extract base filenames from client file paths using Airflow XCom. |
|
|
|
|
|
Args: |
|
|
kwargs: Airflow task instance arguments containing 'ti' for XCom. |
|
|
|
|
|
Returns: |
|
|
List[str]: List of base filenames. |
|
|
|
|
|
Raises: |
|
|
KeyError: If 'ti' is not provided in kwargs. |
|
|
ValueError: If no files are retrieved from XCom. |
|
|
""" |
|
|
if 'ti' not in kwargs: |
|
|
logging.error("Task instance 'ti' not provided in kwargs") |
|
|
raise KeyError("Task instance 'ti' not provided in kwargs") |
|
|
|
|
|
ti = kwargs['ti'] |
|
|
client_files = ti.xcom_pull(task_ids='download_binance_csv') |
|
|
if client_files is None: |
|
|
logging.error("No files retrieved from XCom for task 'download_binance_csv'") |
|
|
raise ValueError("No files retrieved from XCom for task 'download_binance_csv'") |
|
|
|
|
|
if not isinstance(client_files, list): |
|
|
client_files = [client_files] |
|
|
|
|
|
server_files = [os.path.basename(p) for p in client_files] |
|
|
logging.debug(f"Extracted server filenames: {server_files}") |
|
|
return server_files |
