"""Build SAP ERP ontology tables for Postgres-backed ontology import."""

import polars as pl

RELATION_ID_SEPARATOR = "__"
UNKNOWN_CLASSIFICATION_ID = "UNKNOWN"
UNKNOWN_MATERIAL_TYPE = "Unknown Material Type"
UNKNOWN_MATERIAL_INDUSTRY = "Unknown Material Industry"


def _relation_id(source: pl.Expr, relation: str, target: pl.Expr) -> pl.Expr:
    return source + pl.lit(f"{RELATION_ID_SEPARATOR}{relation}{RELATION_ID_SEPARATOR}") + target


def _as_text(column: str) -> pl.Expr:
    return pl.col(column).cast(pl.Utf8)


def _unknown_if_empty(column: str) -> pl.Expr:
    value = pl.col(column).cast(pl.Utf8).str.strip_chars()
    return pl.when(value.is_null() | (value == "")).then(
        pl.lit(UNKNOWN_CLASSIFICATION_ID)
    ).otherwise(value)


def _require_non_empty(frame: pl.DataFrame, columns: list[str], label: str) -> None:
    missing = frame.select(
        pl.any_horizontal(
            [
                pl.col(column).is_null()
                | (pl.col(column).cast(pl.Utf8).str.strip_chars() == "")
                for column in columns
            ]
        ).any()
    ).item()
    if missing:
        joined_columns = ", ".join(columns)
        raise ValueError(
            f"{label} contains null or empty values in required columns: {joined_columns}"
        )


def run(
    tb_customer_summary,
    tb_customer_country_summary,
    tb_customer_continent_summary,
    tb_date_info,
    tb_industry_summary,
    tb_material_summary,
    tb_month_info,
    tb_sales,
    tb_sales_master,
    tb_year_info,
    options,
    contexts,
):
    _require_non_empty(
        tb_customer_summary,
        ["customer_id", "country_name"],
        "tb_customer_summary",
    )
    _require_non_empty(
        tb_customer_country_summary,
        ["country_name", "continent_name"],
        "tb_customer_country_summary",
    )
    _require_non_empty(
        tb_customer_continent_summary,
        ["continent_name"],
        "tb_customer_continent_summary",
    )
    _require_non_empty(
        tb_date_info,
        ["order_date", "order_month", "order_year"],
        "tb_date_info",
    )
    _require_non_empty(
        tb_material_summary,
        ["material_type_code"],
        "tb_material_summary",
    )
    _require_non_empty(
        tb_industry_summary,
        ["industry_type_code"],
        "tb_industry_summary",
    )
    _require_non_empty(tb_sales, ["customer_id", "order_date"], "tb_sales")
    _require_non_empty(
        tb_sales_master,
        ["order_id", "customer_id", "order_date", "order_key"],
        "tb_sales_master",
    )
    _require_non_empty(tb_year_info, ["order_year"], "tb_year_info")
    _require_non_empty(tb_month_info, ["order_month"], "tb_month_info")

    customers = tb_customer_summary.select(
        [
            pl.col("customer_id").alias("id"),
            "customer_id",
            "customer_name",
            "customer_type",
            "city",
            "country_name",
            "continent_name",
            "total_orders",
            "total_order_dates",
            "first_order_date",
            "last_order_date",
            "unique_material_count",
        ]
    ).unique(subset=["id"], maintain_order=True)

    countries = tb_customer_country_summary.select(
        [
            pl.col("country_name").alias("id"),
            "country_name",
            "continent_name",
            "customer_count",
            "customer_type_count",
        ]
    ).unique(subset=["id"], maintain_order=True)

    continents = tb_customer_continent_summary.select(
        [
            pl.col("continent_name").alias("id"),
            "continent_name",
            "customer_count",
        ]
    ).unique(subset=["id"], maintain_order=True)

    order_dates = tb_date_info.select(
        [
            pl.col("order_date").alias("id"),
            "order_date",
            "order_month_name",
            "order_month",
            "order_quarter",
            "order_year",
            "order_count",
        ]
    ).unique(subset=["id"], maintain_order=True)

    order_months = tb_month_info.with_columns(
        _as_text("order_month").alias("month_id")
    ).select(
        [
            pl.col("month_id").alias("id"),
            "order_month",
            "order_month_name",
            "order_quarter",
        ]
    ).unique(
        subset=["id"], maintain_order=True
    )

    order_years = tb_year_info.with_columns(
        _as_text("order_year").alias("year_id")
    ).select(
        [
            pl.col("year_id").alias("id"),
            "order_year",
        ]
    ).unique(
        subset=["id"], maintain_order=True
    )

    material_types = pl.concat(
        [
            tb_material_summary.select(
                [
                    pl.col("material_type_code").alias("id"),
                    "material_type",
                    "material_type_code",
                ]
            ),
            pl.DataFrame(
                {
                    "id": [UNKNOWN_CLASSIFICATION_ID],
                    "material_type": [UNKNOWN_MATERIAL_TYPE],
                    "material_type_code": [UNKNOWN_CLASSIFICATION_ID],
                }
            ),
        ],
        how="vertical_relaxed",
    ).unique(subset=["id"], maintain_order=True)

    material_industries = pl.concat(
        [
            tb_industry_summary.select(
                [
                    pl.col("industry_type_code").alias("id"),
                    pl.col("industry_type").alias("string"),
                ]
            ),
            pl.DataFrame(
                {
                    "id": [UNKNOWN_CLASSIFICATION_ID],
                    "string": [UNKNOWN_MATERIAL_INDUSTRY],
                }
            ),
        ],
        how="vertical_relaxed",
    ).unique(subset=["id"], maintain_order=True)

    sales_master = tb_sales_master.with_columns(
        _as_text("order_id").alias("order_id_text"),
        _as_text("order_key").alias("line_id"),
        _unknown_if_empty("material_type_code").alias("material_type_code"),
        _unknown_if_empty("industry_type_code").alias("industry_type_code"),
    )

    sales_orders = sales_master.group_by("order_id_text", maintain_order=True).agg(
        [
            pl.col("order_id").first(),
            pl.col("order_type").first(),
            pl.col("order_date").first(),
            pl.len().alias("line_count"),
            pl.col("material_id").n_unique().alias("material_count"),
            pl.col("quantity").sum().alias("total_quantity"),
            pl.col("net_sales").sum().alias("net_sales"),
            pl.col("order_year").first(),
            pl.col("order_month").first(),
            pl.col("order_quarter").first(),
        ]
    ).rename({"order_id_text": "id"})

    sales_order_lines = sales_master.select(
        [
            pl.col("line_id").alias("id"),
            "order_id",
            "item_number",
            "material_id",
            "item_name",
            "quantity",
            "sales_unit",
            "net_sales",
            "material_type_code",
            "industry_type_code",
        ]
    ).unique(subset=["id"], maintain_order=True)

    located_in = tb_customer_summary.with_columns(
        _relation_id(
            pl.col("customer_id"), "sap_located_in", pl.col("country_name")
        ).alias("id")
    ).select(
        [
            "id",
            pl.col("customer_id").alias("sap_customer_id"),
            pl.col("country_name").alias("sap_country_id"),
        ]
    ).unique(
        subset=["id"], maintain_order=True
    )

    part_of = tb_customer_country_summary.with_columns(
        _relation_id(
            pl.col("country_name"), "sap_part_of", pl.col("continent_name")
        ).alias("id")
    ).select(
        [
            "id",
            pl.col("country_name").alias("sap_country_id"),
            pl.col("continent_name").alias("sap_continent_id"),
        ]
    ).unique(
        subset=["id"], maintain_order=True
    )

    customer_placed_order = sales_master.with_columns(
        _relation_id(
            pl.col("customer_id"), "sap_customer_placed_order", pl.col("order_id_text")
        ).alias("id")
    ).select(
        [
            "id",
            pl.col("customer_id").alias("sap_customer_id"),
            pl.col("order_id_text").alias("sap_sales_order_id"),
        ]
    ).unique(
        subset=["id"], maintain_order=True
    )

    order_on_date = sales_master.with_columns(
        _relation_id(
            pl.col("order_id_text"), "sap_order_on_date", pl.col("order_date")
        ).alias("id")
    ).select(
        [
            "id",
            pl.col("order_id_text").alias("sap_sales_order_id"),
            pl.col("order_date").alias("sap_order_date_id"),
        ]
    ).unique(
        subset=["id"], maintain_order=True
    )

    order_has_line = sales_master.with_columns(
        _relation_id(
            pl.col("order_id_text"), "sap_order_has_line", pl.col("line_id")
        ).alias("id")
    ).select(
        [
            "id",
            pl.col("order_id_text").alias("sap_sales_order_id"),
            pl.col("line_id").alias("sap_sales_order_line_id"),
        ]
    ).unique(
        subset=["id"], maintain_order=True
    )

    line_material_type = sales_master.with_columns(
        _relation_id(
            pl.col("line_id"), "sap_line_material_type", pl.col("material_type_code")
        ).alias("id")
    ).select(
        [
            "id",
            pl.col("line_id").alias("sap_sales_order_line_id"),
            pl.col("material_type_code").alias("sap_material_type_id"),
        ]
    ).unique(
        subset=["id"], maintain_order=True
    )

    line_material_industry = sales_master.with_columns(
        _relation_id(
            pl.col("line_id"),
            "sap_line_material_industry",
            pl.col("industry_type_code"),
        ).alias("id")
    ).select(
        [
            "id",
            pl.col("line_id").alias("sap_sales_order_line_id"),
            pl.col("industry_type_code").alias("sap_material_industry_id"),
        ]
    ).unique(
        subset=["id"], maintain_order=True
    )

    order_facts = tb_sales.group_by(
        ["customer_id", "order_date"], maintain_order=True
    ).agg(
        [
            pl.col("order_id").n_unique().alias("order_count"),
        ]
    )

    placed_order = order_facts.with_columns(
        _relation_id(
            pl.col("customer_id"), "sap_placed_order", pl.col("order_date")
        ).alias("id")
    ).select(
        [
            "id",
            pl.col("customer_id").alias("sap_customer_id"),
            pl.col("order_date").alias("sap_order_date_id"),
            "order_count",
        ]
    ).unique(
        subset=["id"], maintain_order=True
    )

    part_of_month = tb_date_info.with_columns(
        _as_text("order_month").alias("month_id"),
    ).with_columns(
        _relation_id(pl.col("order_date"), "sap_partof_month", pl.col("month_id")).alias(
            "id"
        )
    ).select(
        [
            "id",
            pl.col("order_date").alias("sap_order_date_id"),
            pl.col("month_id").alias("sap_order_month_id"),
        ]
    ).unique(
        subset=["id"], maintain_order=True
    )

    part_of_year = tb_date_info.with_columns(
        _as_text("order_year").alias("year_id"),
    ).with_columns(
        _relation_id(pl.col("order_date"), "sap_partof_year", pl.col("year_id")).alias(
            "id"
        )
    ).select(
        [
            "id",
            pl.col("order_date").alias("sap_order_date_id"),
            pl.col("year_id").alias("sap_order_year_id"),
        ]
    ).unique(
        subset=["id"], maintain_order=True
    )

    return {
        "sap_customer": customers,
        "sap_sales_order": sales_orders,
        "sap_sales_order_line": sales_order_lines,
        "sap_country": countries,
        "sap_continent": continents,
        "sap_order_date": order_dates,
        "sap_order_month": order_months,
        "sap_order_year": order_years,
        "sap_material_type": material_types,
        "sap_material_industry": material_industries,
        "sap_located_in": located_in,
        "sap_customer_placed_order": customer_placed_order,
        "sap_order_on_date": order_on_date,
        "sap_order_has_line": order_has_line,
        "sap_line_material_type": line_material_type,
        "sap_line_material_industry": line_material_industry,
        "sap_part_of": part_of,
        "sap_placed_order": placed_order,
        "sap_partof_month": part_of_month,
        "sap_partof_year": part_of_year,
    }
