Step 3: RSP Processing¶

required_columns = ['EAN', 'Market', 'SRP', 'Valid_From']
optional_columns = ['House', 'Item_Name', 'LP']

def load_and_clean_source(self, source_path):
    # Read as binary
    binary_df = spark.read.format("binaryFile").load(source_path)

    # Convert to pandas with EAN as string
    dtype_spec = {'EAN': str, 'Ean': str, 'ean': str}
    df_pd = pd.read_excel(excel_buffer, dtype=dtype_spec)

    # Clean column names
    df_pd = self._clean_column_names(df_pd)

    # Convert to Spark and validate
    return self._convert_to_spark(df_pd)

class MarketMappingProcessor:
    def apply_mapping(self, df):
        # Join on Market column
        df_joined = df.join(
            F.broadcast(mapping_df),
            on="Market",
            how="left"
        )
        return df_joined

def apply_fx_rate_conversion(self, df):
    # Extract fiscal year
    df = df.withColumn(
        "fiscal_year", 
        F.year("Valid_From").cast(StringType())
    )

    # Query FX rates
    fx_rates = execute_sql_query("""
        SELECT Fiscal_Year, Reporting_Unit_Key, Toeur 
        FROM bronze_sellin_fx_rate
    """)

    # Convert to EUR
    df = df.withColumn(
        "SRP_eur",
        F.col("SRP") / F.col("Toeur")
    )

    # Also convert LP if exists
    if "Lp" in df.columns:
        df = df.withColumn(
            "LP_eur",
            F.col("Lp") / F.col("Toeur")
        )

def apply_inheritance(self, df):
    inherited_dfs = []

    for rule in inheritance_rules:
        inheritor = rule["Reporting_Unit_Key"]
        parent = rule["Inherits_From"]

        # Filter parent rows
        parent_rows = df.filter(
            F.col("Reporting_Unit_Key") == parent
        )

        # Create inherited rows
        inherited = parent_rows.withColumn(
            "Reporting_Unit_Key", F.lit(inheritor)
        ).withColumn(
            "is_inherited", F.lit(True)
        )

        inherited_dfs.append(inherited)

    # Union all DataFrames
    return union_all([df] + inherited_dfs)

def expand_date_ranges(self, df):
    # Find next Valid_From for each EAN/RU
    df_with_next = df.withColumn(
        "next_valid_from",
        F.lead("Valid_From").over(
            Window.partitionBy("reporting_unit_key", "EAN")
                  .orderBy("Valid_From")
        )
    )

    # Set end_date
    df = df.withColumn(
        "end_date",
        F.coalesce(
            F.date_sub("next_valid_from", 1),
            F.lit(current_date)
        )
    )

    # Create date sequence
    date_df = spark.sql(f"""
        SELECT explode(sequence(
            to_date('{min_date}'),
            to_date('{max_date}'),
            interval 1 day
        )) as Date
    """)

    # Cross join with date range filter
    expanded = df.join(date_df).where(
        (date_df.Date >= df.Valid_From) & 
        (date_df.Date <= df.end_date)
    )

# Cache before expensive operations
df = df.cache()

# Unpersist after use
df.unpersist()