"""Build retail inventory ontology tables for Postgres-backed ontology import."""

import polars as pl

RELATION_ID_SEPARATOR = "__"


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 _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(retail_inventory, options, contexts):
    _require_non_empty(
        retail_inventory,
        ["item_code", "branch_code", "region_name", "supplier_name"],
        "retail_inventory",
    )

    base = retail_inventory.with_columns(
        _as_text("item_code").alias("item_id"),
        _as_text("branch_code").alias("branch_id"),
        _as_text("region_name").alias("region_id"),
        _as_text("supplier_name").alias("supplier_id"),
    )

    products = base.group_by("item_id", maintain_order=True).agg(
        [
            pl.col("PROD_GRP").first(),
            pl.col("list_price_krw_synth").first().alias("price"),
            pl.col("product_name").first().alias("name"),
        ]
    ).rename({"item_id": "id"})

    branches = base.group_by("branch_id", maintain_order=True).agg(
        [
            pl.col("warehouse_location").first().alias("location"),
            pl.col("branch_name").first().alias("name"),
            pl.col("latitude").first(),
            pl.col("longitude").first(),
        ]
    ).rename({"branch_id": "id"})

    regions = base.select(
        pl.col("region_id").alias("id"),
        pl.col("region_id").alias("name"),
    ).unique(
        subset=["id"], maintain_order=True
    )

    suppliers = base.select(
        pl.col("supplier_id").alias("id"),
        pl.col("supplier_id").alias("name"),
    ).unique(
        subset=["id"], maintain_order=True
    )

    located_in = base.with_columns(
        _relation_id(pl.col("branch_id"), "ri_located_in", pl.col("region_id")).alias(
            "id"
        )
    ).select(
        [
            "id",
            pl.col("branch_id").alias("ri_branch_id"),
            pl.col("region_id").alias("ri_region_id"),
            pl.lit("located_in").alias("relation_kind"),
        ]
    ).unique(
        subset=["id"], maintain_order=True
    )

    stock_facts = base.group_by(["item_id", "branch_id"], maintain_order=True).agg(
        [
            pl.col("inventory_qty").sum(),
            pl.col("inventory_turnover_rate").mean().round(2),
            pl.col("selling_price_krw_synth").mean().round(0).cast(pl.Int64),
            pl.col("competitor_price_krw_synth").mean().round(0).cast(pl.Int64),
            pl.col("discount_rate_synth").mean().round(4),
            pl.col("promo_flag_synth").max(),
            pl.col("safety_stock_level").max(),
            pl.col("unit_cost_krw_synth").mean().round(0).cast(pl.Int64),
            pl.col("list_price_krw_synth").mean().round(0).cast(pl.Int64),
            pl.col("outbound_qty").sum(),
            pl.col("stockout_risk_flag_synth").max(),
        ]
    )

    stocked_at = stock_facts.with_columns(
        _relation_id(pl.col("item_id"), "ri_stocked_at", pl.col("branch_id")).alias(
            "id"
        )
    ).select(
        [
            "id",
            pl.col("item_id").alias("ri_product_id"),
            pl.col("branch_id").alias("ri_branch_id"),
            "inventory_qty",
            "inventory_turnover_rate",
            "selling_price_krw_synth",
            "competitor_price_krw_synth",
            "discount_rate_synth",
            "promo_flag_synth",
            "safety_stock_level",
            "unit_cost_krw_synth",
            "list_price_krw_synth",
            "outbound_qty",
            "stockout_risk_flag_synth",
        ]
    ).unique(
        subset=["id"], maintain_order=True
    )

    supplied_by = base.with_columns(
        _relation_id(pl.col("item_id"), "ri_supplied_by", pl.col("supplier_id")).alias(
            "id"
        )
    ).select(
        [
            "id",
            pl.col("item_id").alias("ri_product_id"),
            pl.col("supplier_id").alias("ri_supplier_id"),
            pl.lit("supplied_by").alias("relation_kind"),
        ]
    ).unique(
        subset=["id"], maintain_order=True
    )

    return {
        "ri_product": products,
        "ri_branch": branches,
        "ri_region": regions,
        "ri_supplier": suppliers,
        "ri_located_in": located_in,
        "ri_stocked_at": stocked_at,
        "ri_supplied_by": supplied_by,
    }
