supabase/generate_dummy_csv.py

"""Generate dummy CSV data for fact tables based on existing dimension data.

Usage:
  python supabase/generate_dummy_csv.py

Optional environment variables:
  SALES_COUNT
  Order_COUNT
  OVERDUE_AR_COUNT
  INVENTORY_COUNT
  DRY_RUN
"""

from __future__ import annotations

import csv
import os
import random
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import Iterable

ROOT = Path(__file__).resolve().parent.parent
DATA_DIR = ROOT / "public" / "data"
MASTERDATA_DIR = DATA_DIR / "masterdata"
FACT_DIR = DATA_DIR / "fact_tables"

OUT = {
    "SALES": FACT_DIR / "fact_sales.csv",
    "Order": FACT_DIR / "fact_orders.csv",
    "OVERDUE_AR": FACT_DIR / "fact_overdues.csv",
    "INVENTORY": FACT_DIR / "fact_inventory.csv",
}

DEFAULT_COUNTS = {
    "SALES": 1_000_000,
    "Order": 300,
    "OVERDUE_AR": 300,
    "INVENTORY": 300,
}


def read_text_safe(path: Path) -> str:
    return path.read_text(encoding="utf-8").replace("\r\n", "\n").replace("\r", "\n")


def rand_between(min_value: float, max_value: float) -> float:
    return random.uniform(min_value, max_value)


def rand_int(min_value: int, max_value: int) -> int:
    return random.randint(min_value, max_value)


def choice(values: list[str]) -> str:
    return random.choice(values)


def format_fixed(value: float, digits: int) -> str:
    return f"{value:.{digits}f}"


def pad(value: int, width: int) -> str:
    return f"{value:0{width}d}"


def random_date_iso(start_date: date, end_date: date) -> str:
    delta_days = (end_date - start_date).days
    random_days = random.randint(0, delta_days)
    return (start_date + timedelta(days=random_days)).isoformat()


def to_iso_date_string(value: object) -> str:
    if isinstance(value, datetime):
        return value.date().isoformat()
    if isinstance(value, date):
        return value.isoformat()

    text = str(value).strip()
    if len(text) == 10 and text[4] == "-" and text[7] == "-":
        return text

    for sep in ("/", "."):
        parts = text.split(sep)
        if len(parts) == 3 and len(parts[2]) == 4:
            day_part, month_part, year_part = parts
            if day_part.isdigit() and month_part.isdigit() and year_part.isdigit():
                return f"{year_part}-{month_part.zfill(2)}-{day_part.zfill(2)}"

    for fmt in ("%Y-%m-%d", "%d/%m/%Y", "%d.%m.%Y", "%m/%d/%Y"):
        try:
            return datetime.strptime(text, fmt).date().isoformat()
        except ValueError:
            continue

    return text


def read_csv_dict_rows(path: Path) -> list[dict[str, str]]:
    content = read_text_safe(path)
    lines = [line for line in content.split("\n") if line.strip()]
    if len(lines) < 2:
        return []
    reader = csv.DictReader(lines)
    normalized_rows: list[dict[str, str]] = []
    for row in reader:
        if not row:
            continue
        normalized_rows.append({(k or "").strip().lower(): (v or "") for k, v in row.items()})
    return normalized_rows


def load_country_codes() -> list[str]:
    rows = read_csv_dict_rows(MASTERDATA_DIR / "m_country.csv")
    codes: set[str] = set()
    for row in rows:
        code = (row.get("country_code") or "").strip()
        if len(code) == 2 and code.isalpha() and code.upper() == code:
            codes.add(code)
    return list(codes)


def load_customer_soldto_keys() -> list[str]:
    rows = read_csv_dict_rows(MASTERDATA_DIR / "m_customer_soldto.csv")
    keys: list[str] = []
    for row in rows:
        key = (row.get("customer_soldto_key") or "").strip()
        if key:
            keys.append(key)
    return keys


def load_customer_shipto_keys() -> list[str]:
    rows = read_csv_dict_rows(MASTERDATA_DIR / "m_customer_shipto.csv")
    keys: list[str] = []
    for row in rows:
        key = (row.get("customer_shipto_key") or "").strip()
        if key:
            keys.append(key)
    return keys


def load_product_keys() -> list[str]:
    path = MASTERDATA_DIR / "m_product.csv"
    content = read_text_safe(path)
    lines = [line for line in content.split("\n") if line.strip()]
    if len(lines) < 2:
        return []

    first_header = lines[0].split(",")[0].strip()
    keys: list[str] = []

    if first_header.lower() != "product_article_key":
        reader = csv.DictReader(lines)
        for row in reader:
            key = (row.get("product_article_key") or row.get("Product_article_key") or "").strip()
            if key:
                keys.append(key)
        return keys

    for line in lines[1:]:
        first = line.split(",")[0].strip()
        if first:
            keys.append(first)
    return keys


def generate_net_sales(
    countries: list[str],
    customers_soldto: list[str],
    customers_shipto: list[str],
    products: list[str],
    count: int,
    start_date: date,
    end_date: date,
) -> list[str]:
    rows: list[str] = ["date,country_key,customer_soldto_key,customer_shipto_key,product_article_key,sales_volume,sales_eur,sales_lc,sales_usd,source"]
    for _ in range(count):
        row_date = to_iso_date_string(random_date_iso(start_date, end_date))
        sales_volume = max(1.0, rand_between(100, 5000))
        sales_eur = rand_between(100, 5000) * rand_between(0.8, 8.0)
        sales_lc = sales_eur * rand_between(0.9, 1.3)
        sales_usd = sales_eur * rand_between(1.05, 1.2)
        rows.append(
            ",".join(
                [
                    row_date,
                    choice(countries),
                    choice(customers_soldto),
                    choice(customers_shipto),
                    choice(products),
                    format_fixed(sales_volume, 3),
                    format_fixed(sales_eur, 2),
                    format_fixed(sales_lc, 2),
                    format_fixed(sales_usd, 2),
                    "Sales",
                ]
            )
        )
    return rows


def generate_open_orders(
    countries: list[str],
    customers_soldto: list[str],
    customers_shipto: list[str],
    products: list[str],
    count: int,
    start_date: date,
    end_date: date,
) -> list[str]:
    rows: list[str] = ["date,country_key,document_number,customer_soldto_key,customer_shipto_key,product_article_key,po_number,order_volume,order_eur,order_lc,order_usd,source"]
    for index in range(1, count + 1):
        row_date = to_iso_date_string(random_date_iso(start_date, end_date))
        document_number = f"SO-{row_date[:4]}-{pad(index, 6)}"
        po_number = f"PO-{pad(rand_int(1, 999999), 6)}"
        qty_volume = max(1.0, rand_between(50, 10000))
        order_eur = qty_volume * rand_between(0.5, 10.0)
        rows.append(
            ",".join(
                [
                    row_date,
                    choice(countries),
                    document_number,
                    choice(customers_soldto),
                    choice(customers_shipto),
                    choice(products),
                    po_number,
                    format_fixed(qty_volume, 3),
                    format_fixed(order_eur, 2),
                    format_fixed(order_eur * rand_between(0.9, 1.3), 2),
                    format_fixed(order_eur * rand_between(1.05, 1.2), 2),
                    "Order",
                ]
            )
        )
    return rows


def generate_overdue_ar(
    countries: list[str],
    customers_soldto: list[str],
    customers_shipto: list[str],
    products: list[str],
    count: int,
    start_date: date,
    end_date: date,
) -> list[str]:
    rows: list[str] = ["country_key,customer_soldto_key,customer_shipto_key,product_article_key,document_number,due_date,overdue_eur,overdue_lc,overdue_usd,source"]
    for index in range(1, count + 1):
        due_date = to_iso_date_string(random_date_iso(start_date, end_date))
        base = max(50.0, rand_between(100, 20000))
        overdue_eur = base * rand_between(0.8, 1.2)
        rows.append(
            ",".join(
                [
                    choice(countries),
                    choice(customers_soldto),
                    choice(customers_shipto),
                    choice(products),
                    f"AR-{due_date[:4]}-{pad(index, 7)}",
                    due_date,
                    format_fixed(overdue_eur, 2),
                    format_fixed(overdue_eur * rand_between(0.9, 1.3), 2),
                    format_fixed(overdue_eur * rand_between(1.05, 1.2), 2),
                    "Overdue",
                ]
            )
        )
    return rows


def generate_inventory(countries: list[str], products: list[str], count: int) -> list[str]:
    rows: list[str] = ["country_key,product_article_key,inventory_volume,source"]
    seen: set[tuple[str, str]] = set()
    attempts = 0
    max_attempts = count * 10
    while len(rows) - 1 < count and attempts < max_attempts:
        attempts += 1
        country = choice(countries)
        product = choice(products)
        key = (country, product)
        if key in seen:
            continue
        seen.add(key)
        rows.append(
            ",".join(
                [
                    country,
                    product,
                    format_fixed(max(0.0, rand_between(0, 20000)), 3),
                    "Inventory",
                ]
            )
        )
    return rows


def build_sql_schema_ddl() -> str:
    return """-- SQL schema for master data and fact tables inferred from public/data CSVs
-- Conventions:
-- - Keys with leading zeros are VARCHAR to preserve formatting
-- - Currency fields NUMERIC(18,2); volumes NUMERIC(18,3)
-- - Fact tables have FKs to master tables and helpful indexes

CREATE TABLE m_country (
  idx INTEGER NULL,
  country_code CHAR(2) NOT NULL,
  country VARCHAR(100) NOT NULL,
  region VARCHAR(100) NULL,
  subregion VARCHAR(100) NULL,
  country_key VARCHAR(64) NOT NULL,
  CONSTRAINT pk_m_country PRIMARY KEY (country_key)
);

CREATE INDEX ix_m_country_code ON m_country(country_code);
CREATE INDEX ix_m_country_region ON m_country(region);

CREATE TABLE m_customer_shipto (
  customer_shipto_key VARCHAR(10) NOT NULL,
  customer_shipto VARCHAR(255) NOT NULL,
  CONSTRAINT pk_m_customer_shipto PRIMARY KEY (customer_shipto_key)
);

CREATE TABLE m_customer_soldto (
  customer_soldto_key VARCHAR(10) NOT NULL,
  customer_soldto VARCHAR(255) NOT NULL,
  customer_group VARCHAR(100) NULL,
  CONSTRAINT pk_m_customer_soldto PRIMARY KEY (customer_soldto_key)
);

CREATE TABLE m_product (
  product_article_key VARCHAR(8) NOT NULL,
  product_article VARCHAR(255) NOT NULL,
  product_group_key VARCHAR(8) NOT NULL,
  product_group VARCHAR(255) NOT NULL,
  CONSTRAINT pk_m_product PRIMARY KEY (product_article_key)
);

CREATE INDEX ix_m_product_group_key ON m_product(product_group_key);

CREATE TABLE fact_inventory (
  country_key VARCHAR(64) NOT NULL,
  product_article_key VARCHAR(8) NOT NULL,
  inventory_volume NUMERIC(18,3) NOT NULL,
  source VARCHAR(32) NOT NULL,
  CONSTRAINT pk_fact_inventory PRIMARY KEY (country_key, product_article_key),
  CONSTRAINT fk_inv_country FOREIGN KEY (country_key) REFERENCES m_country(country_key),
  CONSTRAINT fk_inv_product FOREIGN KEY (product_article_key) REFERENCES m_product(product_article_key)
);

CREATE INDEX ix_inv_country ON fact_inventory(country_key);
CREATE INDEX ix_inv_product ON fact_inventory(product_article_key);

CREATE TABLE fact_orders (
  date DATE NOT NULL,
  country_key VARCHAR(64) NOT NULL,
  document_number VARCHAR(40) NOT NULL,
  customer_soldto_key VARCHAR(10) NOT NULL,
  customer_shipto_key VARCHAR(10) NOT NULL,
  product_article_key VARCHAR(8) NOT NULL,
  po_number VARCHAR(40) NULL,
  order_volume NUMERIC(18,3) NOT NULL,
  order_eur NUMERIC(18,2) NULL,
  order_lc NUMERIC(18,2) NULL,
  order_usd NUMERIC(18,2) NULL,
  source VARCHAR(32) NOT NULL,
  CONSTRAINT pk_fact_orders PRIMARY KEY (document_number),
  CONSTRAINT fk_ord_country FOREIGN KEY (country_key) REFERENCES m_country(country_key),
  CONSTRAINT fk_ord_soldto FOREIGN KEY (customer_soldto_key) REFERENCES m_customer_soldto(customer_soldto_key),
  CONSTRAINT fk_ord_shipto FOREIGN KEY (customer_shipto_key) REFERENCES m_customer_shipto(customer_shipto_key),
  CONSTRAINT fk_ord_product FOREIGN KEY (product_article_key) REFERENCES m_product(product_article_key)
);

CREATE INDEX ix_ord_date ON fact_orders(date);
CREATE INDEX ix_ord_country ON fact_orders(country_key);
CREATE INDEX ix_ord_soldto ON fact_orders(customer_soldto_key);
CREATE INDEX ix_ord_shipto ON fact_orders(customer_shipto_key);
CREATE INDEX ix_ord_product ON fact_orders(product_article_key);

CREATE TABLE fact_overdues (
  country_key VARCHAR(64) NOT NULL,
  customer_soldto_key VARCHAR(10) NOT NULL,
  customer_shipto_key VARCHAR(10) NOT NULL,
  product_article_key VARCHAR(8) NOT NULL,
  document_number VARCHAR(40) NOT NULL,
  due_date DATE NOT NULL,
  overdue_eur NUMERIC(18,2) NULL,
  overdue_lc NUMERIC(18,2) NULL,
  overdue_usd NUMERIC(18,2) NULL,
  source VARCHAR(32) NOT NULL,
  CONSTRAINT pk_fact_overdues PRIMARY KEY (document_number),
  CONSTRAINT fk_ovd_country FOREIGN KEY (country_key) REFERENCES m_country(country_key),
  CONSTRAINT fk_ovd_soldto FOREIGN KEY (customer_soldto_key) REFERENCES m_customer_soldto(customer_soldto_key),
  CONSTRAINT fk_ovd_shipto FOREIGN KEY (customer_shipto_key) REFERENCES m_customer_shipto(customer_shipto_key),
  CONSTRAINT fk_ovd_product FOREIGN KEY (product_article_key) REFERENCES m_product(product_article_key)
);

CREATE INDEX ix_ovd_due_date ON fact_overdues(due_date);
CREATE INDEX ix_ovd_country ON fact_overdues(country_key);
CREATE INDEX ix_ovd_soldto ON fact_overdues(customer_soldto_key);
CREATE INDEX ix_ovd_shipto ON fact_overdues(customer_shipto_key);
CREATE INDEX ix_ovd_product ON fact_overdues(product_article_key);

CREATE TABLE fact_sales (
  date DATE NOT NULL,
  country_key VARCHAR(64) NOT NULL,
  customer_soldto_key VARCHAR(10) NOT NULL,
  customer_shipto_key VARCHAR(10) NOT NULL,
  product_article_key VARCHAR(8) NOT NULL,
  sales_volume NUMERIC(18,3) NOT NULL,
  sales_eur NUMERIC(18,2) NULL,
  sales_lc NUMERIC(18,2) NULL,
  sales_usd NUMERIC(18,2) NULL,
  source VARCHAR(32) NOT NULL,
  CONSTRAINT pk_fact_sales PRIMARY KEY (date, country_key, customer_soldto_key, customer_shipto_key, product_article_key),
  CONSTRAINT fk_sal_country FOREIGN KEY (country_key) REFERENCES m_country(country_key),
  CONSTRAINT fk_sal_soldto FOREIGN KEY (customer_soldto_key) REFERENCES m_customer_soldto(customer_soldto_key),
  CONSTRAINT fk_sal_shipto FOREIGN KEY (customer_shipto_key) REFERENCES m_customer_shipto(customer_shipto_key),
  CONSTRAINT fk_sal_product FOREIGN KEY (product_article_key) REFERENCES m_product(product_article_key)
);

CREATE INDEX ix_sal_date ON fact_sales(date);
CREATE INDEX ix_sal_country ON fact_sales(country_key);
CREATE INDEX ix_sal_soldto ON fact_sales(customer_soldto_key);
CREATE INDEX ix_sal_shipto ON fact_sales(customer_shipto_key);
CREATE INDEX ix_sal_product ON fact_sales(product_article_key);
"""


def env_int(name: str, default: int) -> int:
    raw = os.getenv(name)
    if raw is None:
        return default
    try:
        value = int(raw)
        return value if value > 0 else default
    except ValueError:
        return default


def write_rows(path: Path, rows: Iterable[str]) -> None:
    path.write_text("\n".join(rows), encoding="utf-8")


def main() -> None:
    countries = load_country_codes()
    customers_soldto = load_customer_soldto_keys()
    customers_shipto = load_customer_shipto_keys()
    products = load_product_keys()

    if not countries:
        raise RuntimeError("No country codes found")
    if not customers_soldto:
        raise RuntimeError("No sold-to customers found")
    if not customers_shipto:
        raise RuntimeError("No ship-to customers found")
    if not products:
        raise RuntimeError("No products found")

    today = date.today()
    end_date = today
    start_date = date(today.year - 2, 1, 1)

    counts = dict(DEFAULT_COUNTS)
    counts["SALES"] = env_int("SALES_COUNT", counts["SALES"])
    counts["Order"] = env_int("Order_COUNT", counts["Order"])
    counts["OVERDUE_AR"] = env_int("OVERDUE_AR_COUNT", counts["OVERDUE_AR"])
    counts["INVENTORY"] = env_int("INVENTORY_COUNT", counts["INVENTORY"])

    if os.getenv("DRY_RUN"):
        print("DRY_RUN enabled. Validated masterdata inputs:")
        print(f"Countries: {len(countries)}")
        print(f"Customers (sold-to): {len(customers_soldto)}")
        print(f"Customers (ship-to): {len(customers_shipto)}")
        print(f"Products: {len(products)}")
        print("Skipping file writes.")
        return

    FACT_DIR.mkdir(parents=True, exist_ok=True)

    net_sales_rows = generate_net_sales(
        countries,
        customers_soldto,
        customers_shipto,
        products,
        counts["SALES"],
        start_date,
        end_date,
    )
    write_rows(OUT["SALES"], net_sales_rows)

    open_orders_rows = generate_open_orders(
        countries,
        customers_soldto,
        customers_shipto,
        products,
        counts["Order"],
        start_date,
        end_date,
    )
    write_rows(OUT["Order"], open_orders_rows)

    overdue_rows = generate_overdue_ar(
        countries,
        customers_soldto,
        customers_shipto,
        products,
        counts["OVERDUE_AR"],
        start_date,
        end_date,
    )
    write_rows(OUT["OVERDUE_AR"], overdue_rows)

    inventory_rows = generate_inventory(countries, products, counts["INVENTORY"])
    write_rows(OUT["INVENTORY"], inventory_rows)

    schema_dir = ROOT / "public" / "data" / "schema_sql"
    schema_dir.mkdir(parents=True, exist_ok=True)
    schema_path = schema_dir / "sql_schema.sql"
    schema_path.write_text(build_sql_schema_ddl(), encoding="utf-8")

    print(f"Wrote {counts['SALES']} rows -> {OUT['SALES']}")
    print(f"Wrote {counts['Order']} rows -> {OUT['Order']}")
    print(f"Wrote {counts['OVERDUE_AR']} rows -> {OUT['OVERDUE_AR']}")
    print(f"Wrote {counts['INVENTORY']} rows -> {OUT['INVENTORY']}")
    print(f"Wrote SQL schema -> {schema_path}")


if __name__ == "__main__":
    try:
        main()
    except Exception as err:
        print(err)
        raise SystemExit(1) from err