1  import sys
   2  import json
   3  from datetime import datetime
   4  from config_validator import ConfigValidator
   5  from logger_setup import setup_logger
   6  from alerting_handler import AlertingHandler
   7  from data_loader import DataLoader
   8  from filter_engine import FilterEngine
   9  from sampler import Sampler
  10  from data_writer import DataWriter
  11  from utils import calculate_config_hash
  12  
  13  def load_config(config_path):
  14      with open(config_path, "r") as f:
  15          return json.load(f)
  16  
  17  def main(config_path, dry_run=False):
  18      logger = setup_logger()
  19  
  20      try:
  21          logger.info(f"Loading config from {config_path}")
  22          tenant_config = load_config(config_path)
  23          
  24          # Validate Config
  25          validator = ConfigValidator(tenant_config)
  26          validator.validate()
  27          logger.info("Config validation successful.")
  28  
  29          # Setup Alerting
  30          alert_handler = AlertingHandler(tenant_config.get("alerting", {}))
  31  
  32          customer_name = tenant_config["customer_name"]
  33          logger.info(f"Starting Recall Audit Sampling for tenant: {customer_name}")
  34  
  35          # Load Data
  36          data_loader = DataLoader(tenant_config["data_sources"]["matches"], tenant_config["data_sources"]["catalog"] , customer_name)
  37          matches_df, catalog_df = data_loader.load()
  38  
  39          logger.info(f"Loaded matches records: {len(matches_df)}")
  40          logger.info(f"Loaded catalog records: {len(catalog_df)}")
  41  
  42          # Filter Data
  43          filter_engine = FilterEngine(tenant_config)
  44          filtered_matches_df, filtered_catalog_df = filter_engine.apply(matches_df, catalog_df)
  45  
  46          logger.info(f"Post-filter matches: {len(filtered_matches_df)}")
  47          logger.info(f"Post-filter catalog: {len(filtered_catalog_df)}")
  48  
  49          # Recall candidates = catalog SKUs not found in matches
  50          matched_base_skus = filtered_matches_df['base_sku'].unique()
  51          recall_candidates_df = filtered_catalog_df[~filtered_catalog_df['sku'].isin(matched_base_skus)]
  52          logger.info(f"Recall candidates after subtracting matched SKUs: {len(recall_candidates_df)}")
  53  
  54          if recall_candidates_df.empty:
  55              raise ValueError("No recall candidates available after filtering and subtraction!")
  56  
  57          # Sampling
  58          sampler = Sampler(tenant_config["sampling_strategy"])
  59          sampled_df = sampler.sample(recall_candidates_df)
  60  
  61          logger.info(f"Total samples generated: {len(sampled_df)}")
  62  
  63          if sampled_df.empty:
  64              raise ValueError("Sampling resulted in 0 samples! Check your strategy or available candidates.")
  65  
  66  
  67          # Initialize DataWriter
  68          data_writer = DataWriter(output_config=tenant_config["output"], customer_name=customer_name)
  69  
  70          # Load audit log
  71          audit_log_df = data_writer.load_audit_log()
  72  
  73          # Exclude already sampled SKUs
  74          recall_candidates_df = data_writer.exclude_already_sampled(recall_candidates_df, audit_log_df)
  75          logger.info(f"Recall candidates after excluding already-sampled SKUs: {len(recall_candidates_df)}")
  76  
  77          if recall_candidates_df.empty:
  78              raise ValueError("No recall candidates left after excluding previously sampled SKUs!")
  79  
  80          if dry_run:
  81              logger.info("Dry-run mode active. Exiting after data loading and filtering.")
  82              logger.info(f"Dry-run summary: {len(recall_candidates_df)} candidates available for sampling.")
  83              return
  84  
  85          
  86          # Metadata Preparation
  87          metadata = {
  88              "customer_name": customer_name,
  89              "run_date": datetime.now().strftime("%Y-%m-%d"),
  90              "sampling_strategy": tenant_config["sampling_strategy"],
  91              "total_candidates_before_sampling": len(recall_candidates_df),
  92              "total_samples_after_sampling": len(sampled_df),
  93              "filters_applied": tenant_config.get("filters", {}),
  94              "config_hash": calculate_config_hash(tenant_config)
  95          }
  96  
  97          # Write output
  98          data_writer.write_samples(sampled_df, metadata)
  99  
 100          # Update audit log
 101          data_writer.update_audit_log(sampled_df['base_sku'].tolist(), metadata['sampling_strategy']['type'])
 102  
 103          logger.info("Recall sampling run completed successfully.")
 104          alert_handler.send_alert(
 105              subject=f"✅ Recall Sampler Success - {customer_name}",
 106              message_body=f"Recall sampling completed successfully for {customer_name} with {len(sampled_df)} samples."
 107          )
 108  
 109      except Exception as e:
 110          logger.error(f"Recall Audit Sampler FAILED. Error: {str(e)}", exc_info=True)
 111          try:
 112              alert_handler.send_alert(
 113                  subject=f"❌ Recall Sampler FAILED - {tenant_config.get('customer_name', 'Unknown')}",
 114                  message_body=f"Recall sampler failed. Reason: {str(e)}"
 115              )
 116          except Exception as alert_error:
 117              logger.error(f"Failed to send alert: {str(alert_error)}", exc_info=True)
 118          raise
 119  
 120  if __name__ == "__main__":
 121      if len(sys.argv) < 2:
 122          print("Usage: python run_recall_sampler.py <path_to_config.json> [--dry-run]")
 123          sys.exit(1)
 124  
 125      config_path = sys.argv[1]
 126      dry_run = False
 127      if len(sys.argv) > 2 and sys.argv[2] == "--dry-run":
 128          dry_run = True
 129  
 130      main(config_path, dry_run=dry_run)
 131