FPSMS-backend/scripts/generate_item_default_shelf_life_liquibase.py

#!/usr/bin/env python3
"""Generate Liquibase seed SQL from 工埸產品保質期 Excel (joExpiry.xlsx).

Stores both chilled defaultDays and minus18Days.
useMinus18 = 1 only when the row has no non-18 shelf life (frozen-only).
Flip useMinus18 in DB later for items that should print -18 days.

Example:
  py scripts/generate_item_default_shelf_life_liquibase.py
  py scripts/generate_item_default_shelf_life_liquibase.py "C:\\Users\\Administrator\\Downloads\\joExpiry.xlsx"
"""
from __future__ import annotations

import argparse
import datetime as dt
from pathlib import Path

import openpyxl

PREFERRED_SHEET = "XXXXXXXX"
OUT_SQL = Path(
    "src/main/resources/db/changelog/changes/20260820_item_default_shelf_life/"
    "05_seed_item_default_shelf_life_minus18_flag.sql"
)


def sql_str(value: str | None) -> str:
    if value is None:
        return "NULL"
    return "'" + value.replace("\\", "\\\\").replace("'", "''") + "'"


def sql_int(value: int | None) -> str:
    return "NULL" if value is None else str(value)


def norm_temp(raw) -> str | None:
    if raw is None:
        return None
    s = str(raw).strip().replace("℃", "").replace("°C", "").replace(" ", "")
    if not s or s.lower() in ("n/a", "none", "null", "-", "nan"):
        return None
    return s


def is_minus18(temp: str | None) -> bool:
    return temp in ("-18", "-18.0")


def to_days(raw) -> int | None:
    if raw is None or raw == "":
        return None
    if isinstance(raw, str) and raw.strip().lower() in ("n/a", "-", "none", "null"):
        return None
    try:
        n = int(round(float(raw)))
    except (TypeError, ValueError):
        return None
    return n if n > 0 else None


def pick_days(t1, d1, t2, d2) -> tuple[int | None, str | None, int | None]:
    """Return (chilledDays, chilledTemp, minus18Days)."""
    chilled_days = None
    chilled_temp = None
    minus18_days = None
    for temp, days in ((norm_temp(t1), to_days(d1)), (norm_temp(t2), to_days(d2))):
        if not temp or not days:
            continue
        if is_minus18(temp):
            if minus18_days is None:
                minus18_days = days
        elif chilled_days is None:
            chilled_days = days
            chilled_temp = temp
    return chilled_days, chilled_temp, minus18_days


def load_rows(xlsx: Path) -> tuple[str, list[dict]]:
    wb = openpyxl.load_workbook(xlsx, data_only=True)
    sheet_name = PREFERRED_SHEET if PREFERRED_SHEET in wb.sheetnames else wb.sheetnames[0]
    ws = wb[sheet_name]
    kept: list[dict] = []
    seen: set[str] = set()
    for row in ws.iter_rows(min_row=4, values_only=True):
        code_raw = row[3]
        if not code_raw:
            continue
        code = str(code_raw).strip().upper()
        if not code or code in seen:
            continue
        chilled, storage, minus18 = pick_days(row[5], row[6], row[7], row[8])
        if chilled is None and minus18 is None:
            continue
        seen.add(code)
        use_minus18 = 1 if chilled is None else 0
        name = str(row[4]).strip() if row[4] else ""
        cat = str(row[2]).strip() if row[2] else ""
        remark = f"{cat} {name}".strip()[:255]
        kept.append(
            {
                "itemCode": code,
                "defaultDays": chilled,
                "minus18Days": minus18,
                "useMinus18": use_minus18,
                "storageC": "-18" if use_minus18 else storage,
                "remarks": remark,
            }
        )
    return sheet_name, kept


def render_sql(sheet_name: str, kept: list[dict], source_name: str) -> str:
    today = dt.date.today().isoformat()
    n_flag = sum(1 for r in kept if r["useMinus18"] == 1)
    lines = [
        "--liquibase formatted sql",
        "",
        "--changeset fpsms:seed_item_default_shelf_life_minus18_flag",
        f"--comment: Upsert joExpiry.xlsx sheet {sheet_name}: defaultDays + minus18Days + useMinus18 flag",
        f"-- generated {today}; rows={len(kept)} useMinus18=1 (frozen-only default)={n_flag}",
        "INSERT INTO `item_default_shelf_life`",
        "(`created`, `createdBy`, `version`, `modified`, `modifiedBy`, `deleted`,",
        " `itemCode`, `defaultDays`, `minus18Days`, `useMinus18`, `openedDays`, `storageC`, `remarks`)",
        "VALUES",
    ]
    value_rows = []
    for r in kept:
        value_rows.append(
            "("
            + ", ".join(
                [
                    "NOW()",
                    "'system'",
                    "0",
                    "NOW()",
                    "'system'",
                    "0",
                    sql_str(r["itemCode"]),
                    sql_int(r["defaultDays"]),
                    sql_int(r["minus18Days"]),
                    str(r["useMinus18"]),
                    "NULL",
                    sql_str(r["storageC"]),
                    sql_str(r["remarks"]),
                ]
            )
            + ")"
        )
    lines.append(",\n".join(value_rows))
    lines.append("ON DUPLICATE KEY UPDATE")
    lines.append("  `defaultDays` = VALUES(`defaultDays`),")
    lines.append("  `minus18Days` = VALUES(`minus18Days`),")
    lines.append("  `useMinus18` = VALUES(`useMinus18`),")
    lines.append("  `openedDays` = VALUES(`openedDays`),")
    lines.append("  `storageC` = VALUES(`storageC`),")
    lines.append("  `remarks` = VALUES(`remarks`),")
    lines.append("  `modified` = NOW(),")
    lines.append("  `modifiedBy` = 'system',")
    lines.append("  `deleted` = 0;")
    lines.append("")
    return "\n".join(lines)


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "xlsx",
        nargs="?",
        default=str(Path.home() / "Downloads" / "joExpiry.xlsx"),
    )
    args = parser.parse_args()
    xlsx = Path(args.xlsx)
    if not xlsx.is_file():
        raise SystemExit(f"Excel not found: {xlsx}")
    sheet_name, kept = load_rows(xlsx)
    sql = render_sql(sheet_name, kept, xlsx.name)
    OUT_SQL.parent.mkdir(parents=True, exist_ok=True)
    OUT_SQL.write_text(sql, encoding="utf-8")
    n_flag = sum(1 for r in kept if r["useMinus18"] == 1)
    print(f"sheet={sheet_name} rows={len(kept)} useMinus18=1={n_flag} useMinus18=0={len(kept) - n_flag}")
    print(f"wrote {OUT_SQL}")


if __name__ == "__main__":
    main()