from __future__ import annotations

import hashlib
import hmac
import json
import os
import re
import secrets
import sqlite3
from datetime import date, datetime
from pathlib import Path

import psycopg2
from psycopg2.extensions import connection as PGNativeConnection

from app_core.permissions import DEFAULT_ROLE_DEFAULTS, normalize_permission_settings
from app_core.paths import PERMISSIONS_SETTINGS_PATH, REPORT_CONFIG_PATH, SERVER_SETTINGS_PATH, remap_runtime_path


PURCHASE_ORDERS_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS purchase_orders (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    po_number TEXT NOT NULL,
    po_date TEXT,
    po_time TEXT,
    simple_domestic_pt TEXT,
    dept_number TEXT,
    vendor_name TEXT,
    vendor_address TEXT,
    vendor_attention TEXT,
    vendor_email TEXT,
    buyer_name TEXT,
    buyer_email TEXT,
    shipping_address TEXT,
    mailing_address TEXT,
    ship_to_name TEXT,
    ship_to_address TEXT,
    mark_for TEXT,
    store_number TEXT,
    order_date TEXT,
    ship_date TEXT,
    cancel_date TEXT,
    duns TEXT,
    bealls_code TEXT,
    purchase_disc TEXT,
    post_dist TEXT,
    co_op TEXT,
    fob_point TEXT,
    terms TEXT,
    as_of TEXT,
    rtv TEXT,
    reference_number TEXT,
    new_store_disc TEXT,
    freight_allowance TEXT,
    total_packs_ordered TEXT,
    total_units TEXT,
    total_cost TEXT,
    total_qty TEXT,
    special_instructions TEXT,
    source_pdf TEXT,
    status TEXT NOT NULL DEFAULT 'draft',
    is_active INTEGER NOT NULL DEFAULT 1,
    raw_json TEXT NOT NULL,
    created_at TEXT DEFAULT CURRENT_TIMESTAMP,
    updated_at TEXT DEFAULT CURRENT_TIMESTAMP
);
"""

PURCHASE_ORDER_ITEMS_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS purchase_order_items (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    purchase_order_id INTEGER NOT NULL,
    sku TEXT,
    mfg_style TEXT,
    mfg_color TEXT,
    size_desc TEXT,
    description TEXT,
    upc TEXT,
    cost_unit TEXT,
    comp TEXT,
    retail TEXT,
    pack_qty TEXT,
    total_packs TEXT,
    total_units TEXT,
    FOREIGN KEY (purchase_order_id) REFERENCES purchase_orders(id) ON DELETE CASCADE
);
"""

PURCHASE_ORDER_IDENTITY_INDEX_SQL = """
CREATE UNIQUE INDEX IF NOT EXISTS idx_purchase_orders_identity
ON purchase_orders (po_number, simple_domestic_pt, source_pdf);
"""

COSTING_IMPORTS_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS costing_imports (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    collection TEXT,
    season TEXT,
    source_pdf TEXT NOT NULL,
    debug_dir TEXT,
    raw_json TEXT NOT NULL,
    created_at TEXT DEFAULT CURRENT_TIMESTAMP,
    updated_at TEXT DEFAULT CURRENT_TIMESTAMP
);
"""

COSTING_VARIANTS_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS costing_variants (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    costing_import_id INTEGER NOT NULL,
    style_code TEXT,
    color_label TEXT,
    description TEXT,
    image_path TEXT,
    page_image_path TEXT,
    page_number INTEGER,
    crop_meta_json TEXT,
    collection TEXT,
    season TEXT,
    created_at TEXT DEFAULT CURRENT_TIMESTAMP,
    updated_at TEXT DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (costing_import_id) REFERENCES costing_imports(id) ON DELETE CASCADE
);
"""

COSTING_WS_IMPORTS_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS costing_ws_imports (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    collection TEXT,
    season TEXT,
    source_workbook TEXT NOT NULL,
    source_sheet TEXT,
    raw_json TEXT NOT NULL,
    created_at TEXT DEFAULT CURRENT_TIMESTAMP,
    updated_at TEXT DEFAULT CURRENT_TIMESTAMP
);
"""

COSTING_WS_ROWS_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS costing_ws_rows (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    costing_ws_import_id INTEGER NOT NULL,
    dept_number TEXT,
    mj_class TEXT,
    sub_class TEXT,
    brand TEXT,
    style_code TEXT,
    description TEXT,
    fabrication TEXT,
    composition TEXT,
    manf_color TEXT,
    color_code TEXT,
    product_type TEXT,
    ticket_type TEXT,
    packing_type TEXT,
    total_units INTEGER,
    allocation_multiple INTEGER,
    min_qty INTEGER,
    max_qty INTEGER,
    inner_pack INTEGER,
    outer_pack INTEGER,
    size_range TEXT,
    size_scale TEXT,
    est_cbm REAL,
    hanger_supplier TEXT,
    hanger_type TEXT,
    item_type_for_freight TEXT,
    item_retail REAL,
    po_number TEXT,
    color_label TEXT,
    collection TEXT,
    season TEXT,
    source_row_number INTEGER,
    status TEXT NOT NULL DEFAULT 'draft',
    is_active INTEGER NOT NULL DEFAULT 1,
    created_at TEXT DEFAULT CURRENT_TIMESTAMP,
    updated_at TEXT DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (costing_ws_import_id) REFERENCES costing_ws_imports(id) ON DELETE CASCADE
);
"""

WORKFLOW_STATUS_VALUES = {"draft", "checking", "approved"}


def normalize_workflow_status(value: object, default: str = "draft") -> str:
    normalized_default = str(default or "draft").strip().lower() or "draft"
    status = str(value or "").strip().lower()
    if status not in WORKFLOW_STATUS_VALUES:
        return normalized_default if normalized_default in WORKFLOW_STATUS_VALUES else "draft"
    return status


def parse_active_flag(value: object, default: bool = True) -> bool:
    if isinstance(value, bool):
        return value
    if isinstance(value, (int, float)):
        return bool(value)
    text = str(value or "").strip().lower()
    if not text:
        return bool(default)
    if text in {"1", "true", "yes", "y", "active"}:
        return True
    if text in {"0", "false", "no", "n", "inactive"}:
        return False
    return bool(default)

USERS_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    email TEXT NOT NULL UNIQUE,
    password_hash TEXT NOT NULL,
    display_name TEXT,
    role TEXT NOT NULL DEFAULT 'user',
    is_active INTEGER NOT NULL DEFAULT 1,
    approval_status TEXT NOT NULL DEFAULT 'approved',
    approved_by TEXT,
    approved_at TEXT,
    created_at TEXT DEFAULT CURRENT_TIMESTAMP,
    updated_at TEXT DEFAULT CURRENT_TIMESTAMP
);
"""

PASSWORD_RESET_TOKENS_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS password_reset_tokens (
    token TEXT PRIMARY KEY,
    email TEXT NOT NULL,
    expires_at TEXT NOT NULL,
    used_at TEXT,
    created_at TEXT DEFAULT CURRENT_TIMESTAMP
);
"""

APP_SETTINGS_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS app_settings (
    setting_key TEXT PRIMARY KEY,
    value_json TEXT NOT NULL,
    updated_at TEXT DEFAULT CURRENT_TIMESTAMP
);
"""

TECH_PACK_RECORDS_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS tech_pack_records (
    style_code TEXT PRIMARY KEY,
    record_json TEXT NOT NULL,
    created_at TEXT DEFAULT CURRENT_TIMESTAMP,
    updated_at TEXT DEFAULT CURRENT_TIMESTAMP
);
"""

SAMPLE_RECORDS_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS sample_records (
    record_id TEXT PRIMARY KEY,
    style_code TEXT,
    request_number TEXT,
    record_json TEXT NOT NULL,
    created_at TEXT DEFAULT CURRENT_TIMESTAMP,
    updated_at TEXT DEFAULT CURRENT_TIMESTAMP
);
"""

SAMPLE_RECORDS_STYLE_INDEX_SQL = """
CREATE INDEX IF NOT EXISTS idx_sample_records_style_code
ON sample_records (style_code);
"""

BOM_HEADERS_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS bom_headers (
    bom_id TEXT PRIMARY KEY,
    style_code TEXT NOT NULL,
    bom_version TEXT NOT NULL,
    bom_name TEXT,
    season TEXT,
    collection TEXT,
    parent_type TEXT,
    source_type TEXT,
    source_style_code TEXT,
    source_note TEXT,
    status TEXT NOT NULL DEFAULT 'draft',
    is_active INTEGER NOT NULL DEFAULT 1,
    created_at TEXT DEFAULT CURRENT_TIMESTAMP,
    updated_at TEXT DEFAULT CURRENT_TIMESTAMP
);
"""

BOM_HEADERS_STYLE_INDEX_SQL = """
CREATE INDEX IF NOT EXISTS idx_bom_headers_style_code
ON bom_headers (style_code);
"""

BOM_HEADERS_STYLE_VERSION_INDEX_SQL = """
CREATE INDEX IF NOT EXISTS idx_bom_headers_style_version
ON bom_headers (style_code, bom_version);
"""

BOM_ITEMS_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS bom_items (
    item_id TEXT PRIMARY KEY,
    bom_id TEXT NOT NULL,
    style_code TEXT NOT NULL,
    item_code TEXT NOT NULL,
    item_name TEXT NOT NULL,
    item_type TEXT NOT NULL,
    section TEXT,
    article_ref TEXT,
    color TEXT,
    color_code TEXT,
    description TEXT,
    composition TEXT,
    position_of_use TEXT,
    spec_summary TEXT,
    uom TEXT,
    qty_per_garment TEXT,
    trim_size_um TEXT,
    trim_size_unit TEXT,
    length_value TEXT,
    length_unit TEXT,
    width_value TEXT,
    width_unit TEXT,
    gsm_value TEXT,
    gsm_unit TEXT,
    waste_percent TEXT,
    tech_comment TEXT,
    default_supplier TEXT,
    supplier_item_code TEXT,
    moq TEXT,
    moq_unit TEXT,
    moq_charge TEXT,
    moq_charge_unit TEXT,
    mcq TEXT,
    mcq_unit TEXT,
    mcq_charge TEXT,
    mcq_charge_unit TEXT,
    lead_time TEXT,
    lead_time_unit TEXT,
    sample_lead_time TEXT,
    sample_lead_time_unit TEXT,
    purchase_uom TEXT,
    standard_price TEXT,
    currency TEXT,
    purchase_note TEXT,
    fds_date TEXT,
    image_path TEXT,
    status TEXT NOT NULL DEFAULT 'draft',
    source_style_code TEXT,
    source_section TEXT,
    source_section_index TEXT,
    source_item_index TEXT,
    source_payload_json TEXT,
    created_at TEXT DEFAULT CURRENT_TIMESTAMP,
    updated_at TEXT DEFAULT CURRENT_TIMESTAMP,
    FOREIGN KEY (bom_id) REFERENCES bom_headers(bom_id) ON DELETE CASCADE
);
"""

BOM_ITEMS_BOM_ITEM_CODE_INDEX_SQL = """
CREATE UNIQUE INDEX IF NOT EXISTS idx_bom_items_bom_item_code
ON bom_items (bom_id, item_code);
"""

BOM_ITEMS_STYLE_INDEX_SQL = """
CREATE INDEX IF NOT EXISTS idx_bom_items_style_code
ON bom_items (style_code);
"""

BOM_ITEMS_TYPE_INDEX_SQL = """
CREATE INDEX IF NOT EXISTS idx_bom_items_type
ON bom_items (item_type);
"""

BOM_ITEMS_SUPPLIER_INDEX_SQL = """
CREATE INDEX IF NOT EXISTS idx_bom_items_supplier
ON bom_items (default_supplier);
"""

BOM_ITEMS_STATUS_INDEX_SQL = """
CREATE INDEX IF NOT EXISTS idx_bom_items_status
ON bom_items (status);
"""

SUPPLIERS_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS suppliers (
    supplier_id TEXT PRIMARY KEY,
    supplier_code TEXT NOT NULL,
    supplier_name TEXT NOT NULL,
    short_name TEXT,
    supplier_type TEXT,
    status TEXT NOT NULL DEFAULT 'Active',
    contact_person TEXT,
    phone TEXT,
    email TEXT,
    country TEXT,
    city TEXT,
    address_line TEXT,
    tax_code TEXT,
    payment_term TEXT,
    currency TEXT,
    note TEXT,
    created_at TEXT DEFAULT CURRENT_TIMESTAMP,
    updated_at TEXT DEFAULT CURRENT_TIMESTAMP
);
"""

SUPPLIERS_CODE_INDEX_SQL = """
CREATE UNIQUE INDEX IF NOT EXISTS idx_suppliers_code
ON suppliers (supplier_code);
"""

SUPPLIERS_NAME_INDEX_SQL = """
CREATE INDEX IF NOT EXISTS idx_suppliers_name
ON suppliers (supplier_name);
"""

SUPPLIERS_STATUS_INDEX_SQL = """
CREATE INDEX IF NOT EXISTS idx_suppliers_status
ON suppliers (status);
"""

HR_EMPLOYEES_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS hr_employees (
    employee_id TEXT PRIMARY KEY,
    user_email TEXT,
    full_name TEXT,
    employee_json TEXT NOT NULL,
    created_at TEXT DEFAULT CURRENT_TIMESTAMP,
    updated_at TEXT DEFAULT CURRENT_TIMESTAMP
);
"""

HR_EMPLOYEES_USER_EMAIL_INDEX_SQL = """
CREATE UNIQUE INDEX IF NOT EXISTS idx_hr_employees_user_email
ON hr_employees (user_email);
"""

HR_ATTENDANCE_RECORDS_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS hr_attendance_records (
    attendance_id TEXT PRIMARY KEY,
    employee_id TEXT NOT NULL,
    user_email TEXT,
    work_date TEXT NOT NULL,
    record_json TEXT NOT NULL,
    created_at TEXT DEFAULT CURRENT_TIMESTAMP,
    updated_at TEXT DEFAULT CURRENT_TIMESTAMP
);
"""

HR_ATTENDANCE_EMPLOYEE_DATE_INDEX_SQL = """
CREATE UNIQUE INDEX IF NOT EXISTS idx_hr_attendance_employee_date
ON hr_attendance_records (employee_id, work_date);
"""

TECH_PACK_HIDDEN_STYLES_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS tech_pack_hidden_styles (
    style_code TEXT PRIMARY KEY,
    updated_at TEXT DEFAULT CURRENT_TIMESTAMP
);
"""

PERMISSION_ROLE_DEFAULTS_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS permission_role_defaults (
    role TEXT PRIMARY KEY,
    modules_json TEXT NOT NULL,
    actions_json TEXT NOT NULL,
    updated_at TEXT DEFAULT CURRENT_TIMESTAMP
);
"""

PERMISSION_USER_OVERRIDES_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS permission_user_overrides (
    email TEXT PRIMARY KEY,
    role_hint TEXT NOT NULL,
    modules_json TEXT NOT NULL,
    actions_json TEXT NOT NULL,
    updated_at TEXT DEFAULT CURRENT_TIMESTAMP
);
"""

DEFAULT_ADMIN_EMAIL = "rs-spl-pt@royalstitches.com.vn"
DEFAULT_ADMIN_PASSWORD = "Admin123@"
DEFAULT_ADMIN_DISPLAY_NAME = "System Admin"
DEFAULT_MODERATE_EMAIL = "mod@royalstitches.com.vn"
DEFAULT_MODERATE_PASSWORD = "Mod123@"
DEFAULT_MODERATE_DISPLAY_NAME = "Department Moderator"
DEFAULT_USER_EMAIL = "user@royalstitches.com.vn"
DEFAULT_USER_PASSWORD = "User123@"
DEFAULT_USER_DISPLAY_NAME = "Standard User"

POSTGRES_SERIAL_ID_TABLES = {
    "purchase_orders",
    "purchase_order_items",
    "costing_imports",
    "costing_variants",
    "costing_ws_imports",
    "costing_ws_rows",
    "users",
}


class CompatRow(dict):
    def __init__(self, columns: list[str], values: tuple[object, ...]) -> None:
        normalized_values = [_normalize_db_value(value) for value in values]
        super().__init__(zip(columns, normalized_values))
        self._ordered_values = normalized_values

    def __getitem__(self, key: object) -> object:
        if isinstance(key, int):
            return self._ordered_values[key]
        return super().__getitem__(key)


class PostgresCompatCursor:
    def __init__(
        self,
        cursor: object,
        *,
        lastrowid: int | None = None,
        prefetched_rows: list[tuple[object, ...]] | None = None,
    ) -> None:
        self._cursor = cursor
        self.lastrowid = lastrowid
        self._prefetched_rows = prefetched_rows or []
        self.rowcount = int(getattr(cursor, "rowcount", 0) or 0)

    def _columns(self) -> list[str]:
        description = getattr(self._cursor, "description", None) or []
        return [str(item[0]) for item in description]

    def fetchone(self) -> CompatRow | None:
        if self._prefetched_rows:
            row = self._prefetched_rows.pop(0)
        else:
            row = self._cursor.fetchone()
        if row is None:
            return None
        return CompatRow(self._columns(), tuple(row))

    def fetchall(self) -> list[CompatRow]:
        rows = list(self._prefetched_rows)
        self._prefetched_rows.clear()
        rows.extend(self._cursor.fetchall())
        columns = self._columns()
        return [CompatRow(columns, tuple(row)) for row in rows]


class PostgresCompatConnection:
    def __init__(self, dsn: str) -> None:
        self._conn: PGNativeConnection = psycopg2.connect(dsn)

    def __enter__(self) -> PostgresCompatConnection:
        return self

    def __exit__(self, exc_type, exc, tb) -> None:
        if exc_type is None:
            self._conn.commit()
        else:
            self._conn.rollback()
        self._conn.close()

    def close(self) -> None:
        self._conn.close()

    def execute(self, sql: str, params: tuple[object, ...] | list[object] = ()) -> PostgresCompatCursor:
        translated_sql = _translate_sql_for_postgres(sql)
        prefetched_rows: list[tuple[object, ...]] | None = None
        lastrowid: int | None = None

        if _should_append_returning_id(translated_sql):
            translated_sql = f"{translated_sql.rstrip().rstrip(';')} RETURNING id"

        cursor = self._conn.cursor()
        cursor.execute(translated_sql, tuple(params))

        if translated_sql.rstrip().upper().endswith("RETURNING ID"):
            returned = cursor.fetchone()
            prefetched_rows = [tuple(returned)] if returned is not None else []
            lastrowid = int(returned[0]) if returned is not None else None

        return PostgresCompatCursor(cursor, lastrowid=lastrowid, prefetched_rows=prefetched_rows)

    def executemany(self, sql: str, seq_of_params: list[tuple[object, ...]] | tuple[tuple[object, ...], ...]) -> None:
        translated_sql = _translate_sql_for_postgres(sql)
        cursor = self._conn.cursor()
        cursor.executemany(translated_sql, list(seq_of_params))


def _normalize_db_value(value: object) -> object:
    if isinstance(value, datetime):
        return value.isoformat(sep=" ")
    if isinstance(value, date):
        return value.isoformat()
    return value


def _postgres_dsn() -> str:
    return str(
        os.environ.get("SUNNY_ERP_DATABASE_URL")
        or os.environ.get("DATABASE_URL")
        or ""
    ).strip()


def using_postgres() -> bool:
    return bool(_postgres_dsn())


def get_database_mode() -> str:
    return "postgresql" if using_postgres() else "sqlite"


def purchase_order_style_summary_sql() -> str:
    if using_postgres():
        return """
                (
                    SELECT STRING_AGG(style_value, ', ' ORDER BY first_item_id)
                    FROM (
                        SELECT UPPER(TRIM(COALESCE(poi.mfg_style, ''))) AS style_value, MIN(poi.id) AS first_item_id
                        FROM purchase_order_items poi
                        WHERE poi.purchase_order_id = po.id
                          AND TRIM(COALESCE(poi.mfg_style, '')) <> ''
                        GROUP BY UPPER(TRIM(COALESCE(poi.mfg_style, '')))
                    ) style_rows
                )
        """
    return """
                (
                    SELECT GROUP_CONCAT(style_value, ', ')
                    FROM (
                        SELECT UPPER(TRIM(IFNULL(poi.mfg_style, ''))) AS style_value, MIN(poi.id) AS first_item_id
                        FROM purchase_order_items poi
                        WHERE poi.purchase_order_id = po.id
                          AND TRIM(IFNULL(poi.mfg_style, '')) <> ''
                        GROUP BY UPPER(TRIM(IFNULL(poi.mfg_style, '')))
                        ORDER BY first_item_id
                    ) style_rows
                )
    """


def _translate_sql_for_postgres(sql: str) -> str:
    translated = str(sql)
    translated = translated.replace("IFNULL(", "COALESCE(")
    translated = translated.replace("INSERT OR REPLACE INTO permission_role_defaults", "INSERT INTO permission_role_defaults")
    translated = translated.replace("INSERT OR REPLACE INTO permission_user_overrides", "INSERT INTO permission_user_overrides")
    translated = translated.replace("INSERT OR REPLACE INTO sample_records", "INSERT INTO sample_records")
    translated = translated.replace("?", "%s")
    return translated


def _translate_schema_sql_for_postgres(sql: str) -> str:
    translated = _translate_sql_for_postgres(sql)
    translated = translated.replace("INTEGER PRIMARY KEY AUTOINCREMENT", "BIGSERIAL PRIMARY KEY")
    translated = re.sub(r"\b(created_at|updated_at|approved_at|expires_at|used_at)\b TEXT", r"\1 TIMESTAMPTZ", translated)
    return translated


def _extract_insert_table_name(sql: str) -> str:
    match = re.search(r"INSERT\s+INTO\s+([a-zA-Z_][a-zA-Z0-9_]*)", sql, flags=re.IGNORECASE)
    return str(match.group(1)).strip().lower() if match else ""


def _should_append_returning_id(sql: str) -> bool:
    normalized = sql.lstrip().upper()
    if not normalized.startswith("INSERT INTO "):
        return False
    if " RETURNING " in normalized:
        return False
    return _extract_insert_table_name(sql) in POSTGRES_SERIAL_ID_TABLES


def normalize_style_identifier(value: object) -> str:
    text = str(value or "").strip().upper()
    text = "".join(ch for ch in text if ch.isalnum())
    return text


def default_sample_record_payload() -> dict:
    return {
        "record_id": "",
        "created_at": "",
        "updated_at": "",
        "request_number": "",
        "style": "",
        "style_code": "",
        "style_name": "",
        "baby_styles": "",
        "image": "",
        "description": "",
        "mr_pic": "",
        "pattern_pic": "",
        "sewer_pic": "",
        "ie_pic": "",
        "material_date": "",
        "bom_item_id": "",
        "bom_item_name": "",
        "bom_composition": "",
        "deadline": "",
        "sample_type": "",
        "color": "",
        "size": "",
        "quantity": "",
        "cutting_date": "",
        "sewing_date": "",
        "finishing_date": "",
        "sending_date": "",
        "fit_evals_status": "",
        "upload_comment_date": "",
        "buyer_comment_date": "",
        "remark": "",
    }


def normalize_sample_record_rows(payload: object) -> list[dict]:
    rows = payload if isinstance(payload, list) else []
    normalized: list[dict] = []
    for row in rows:
        if not isinstance(row, dict):
            continue
        item = default_sample_record_payload()
        for key in item:
            item[key] = str(row.get(key, item[key]) or "").strip()
        if not item["record_id"]:
            item["record_id"] = secrets.token_hex(16)
        item["style_code"] = normalize_style_identifier(item.get("style_code") or item.get("style"))
        if not item["style"] and item["style_code"]:
            item["style"] = item["style_code"]
        normalized.append(item)
    return normalized


def connect(db_path: Path) -> sqlite3.Connection | PostgresCompatConnection:
    if using_postgres():
        return PostgresCompatConnection(_postgres_dsn())

    db_path.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    return conn


def is_database_ready(db_path: Path) -> bool:
    if using_postgres():
        try:
            with connect(db_path) as conn:
                row = conn.execute(
                    """
                    SELECT 1
                    FROM information_schema.tables
                    WHERE table_schema = 'public' AND table_name = 'users'
                    """
                ).fetchone()
            return row is not None
        except Exception:
            return False
    return db_path.exists()


def normalize_email(email: str) -> str:
    return str(email or "").strip().lower()


def hash_password(password: str, *, salt_hex: str | None = None, iterations: int = 200_000) -> str:
    normalized_password = str(password or "")
    salt = bytes.fromhex(salt_hex) if salt_hex else secrets.token_bytes(16)
    digest = hashlib.pbkdf2_hmac("sha256", normalized_password.encode("utf-8"), salt, iterations)
    return f"pbkdf2_sha256${iterations}${salt.hex()}${digest.hex()}"


def verify_password(password: str, stored_hash: str) -> bool:
    try:
        algorithm, iterations_raw, salt_hex, digest_hex = str(stored_hash or "").split("$", 3)
        if algorithm != "pbkdf2_sha256":
            return False
        expected = hash_password(password, salt_hex=salt_hex, iterations=int(iterations_raw))
        return hmac.compare_digest(expected, stored_hash)
    except Exception:
        return False


def ensure_default_admin_user(conn: sqlite3.Connection) -> None:
    default_users = [
        (DEFAULT_ADMIN_EMAIL, DEFAULT_ADMIN_PASSWORD, DEFAULT_ADMIN_DISPLAY_NAME, "admin"),
        (DEFAULT_MODERATE_EMAIL, DEFAULT_MODERATE_PASSWORD, DEFAULT_MODERATE_DISPLAY_NAME, "moderate"),
        (DEFAULT_USER_EMAIL, DEFAULT_USER_PASSWORD, DEFAULT_USER_DISPLAY_NAME, "user"),
    ]
    for email_raw, password, display_name, role in default_users:
        email = normalize_email(email_raw)
        existing = conn.execute("SELECT id FROM users WHERE email = ?", (email,)).fetchone()
        if existing is not None:
            continue
        conn.execute(
            """
            INSERT INTO users (email, password_hash, display_name, role, is_active, approval_status, approved_by, approved_at)
            VALUES (?, ?, ?, ?, 1, 'approved', ?, CURRENT_TIMESTAMP)
            """,
            (
                email,
                hash_password(password),
                display_name,
                role,
                DEFAULT_ADMIN_EMAIL,
            ),
        )


def ensure_users_auth_columns(conn: sqlite3.Connection) -> None:
    columns = {
        str(row["name"]).strip().lower()
        for row in conn.execute("PRAGMA table_info('users')").fetchall()
    }
    if "approval_status" not in columns:
        conn.execute("ALTER TABLE users ADD COLUMN approval_status TEXT NOT NULL DEFAULT 'approved'")
    if "approved_by" not in columns:
        conn.execute("ALTER TABLE users ADD COLUMN approved_by TEXT")
    if "approved_at" not in columns:
        conn.execute("ALTER TABLE users ADD COLUMN approved_at TEXT")
    conn.execute(
        """
        UPDATE users
        SET approval_status = 'approved'
        WHERE IFNULL(TRIM(approval_status), '') = ''
        """
    )


def get_table_columns(conn: sqlite3.Connection | PostgresCompatConnection, table_name: str) -> set[str]:
    normalized_table = str(table_name or "").strip().lower()
    if not normalized_table:
        return set()
    if using_postgres():
        rows = conn.execute(
            """
            SELECT column_name
            FROM information_schema.columns
            WHERE table_schema = 'public' AND table_name = ?
            """,
            (normalized_table,),
        ).fetchall()
        return {str(row["column_name"]).strip().lower() for row in rows}
    rows = conn.execute(f"PRAGMA table_info('{table_name}')").fetchall()
    return {str(row["name"]).strip().lower() for row in rows}


def ensure_table_column(conn: sqlite3.Connection | PostgresCompatConnection, table_name: str, column_name: str, column_sql: str) -> None:
    columns = get_table_columns(conn, table_name)
    if column_name.strip().lower() not in columns:
        conn.execute(f"ALTER TABLE {table_name} ADD COLUMN {column_name} {column_sql}")


def ensure_record_timestamp_columns(conn: sqlite3.Connection) -> None:
    for table_name in ("purchase_orders", "costing_imports", "costing_variants", "costing_ws_imports", "costing_ws_rows"):
        ensure_table_column(conn, table_name, "updated_at", "TEXT")
        conn.execute(
            f"""
            UPDATE {table_name}
            SET updated_at = COALESCE(NULLIF(updated_at, ''), created_at, CURRENT_TIMESTAMP)
            WHERE updated_at IS NULL OR TRIM(updated_at) = ''
            """
        )

    ensure_table_column(conn, "tech_pack_records", "created_at", "TEXT")
    conn.execute(
        """
        UPDATE tech_pack_records
        SET created_at = COALESCE(NULLIF(created_at, ''), updated_at, CURRENT_TIMESTAMP)
        WHERE created_at IS NULL OR TRIM(created_at) = ''
        """
    )
    ensure_workflow_columns(conn)


def ensure_workflow_columns(conn: sqlite3.Connection | PostgresCompatConnection) -> None:
    ensure_table_column(conn, "purchase_orders", "status", "TEXT NOT NULL DEFAULT 'draft'")
    ensure_table_column(conn, "purchase_orders", "is_active", "INTEGER NOT NULL DEFAULT 1")
    conn.execute(
        """
        UPDATE purchase_orders
        SET status = 'draft'
        WHERE status IS NULL OR TRIM(CAST(status AS TEXT)) = ''
        """
    )
    conn.execute(
        """
        UPDATE purchase_orders
        SET is_active = 1
        WHERE is_active IS NULL
        """
    )
    ensure_table_column(conn, "costing_ws_rows", "status", "TEXT NOT NULL DEFAULT 'draft'")
    ensure_table_column(conn, "costing_ws_rows", "is_active", "INTEGER NOT NULL DEFAULT 1")
    for column_name, column_type in {
        "color_code": "TEXT",
        "product_type": "TEXT",
        "ticket_type": "TEXT",
        "packing_type": "TEXT",
        "allocation_multiple": "INTEGER",
        "min_qty": "INTEGER",
        "max_qty": "INTEGER",
        "hanger_supplier": "TEXT",
        "hanger_type": "TEXT",
        "item_retail": "REAL",
        "po_number": "TEXT",
    }.items():
        ensure_table_column(conn, "costing_ws_rows", column_name, column_type)
    conn.execute(
        """
        UPDATE costing_ws_rows
        SET status = 'draft'
        WHERE status IS NULL OR TRIM(CAST(status AS TEXT)) = ''
        """
    )
    conn.execute(
        """
        UPDATE costing_ws_rows
        SET is_active = 1
        WHERE is_active IS NULL
        """
    )


def ensure_bom_item_purchase_columns(conn: sqlite3.Connection | PostgresCompatConnection) -> None:
    ensure_table_column(conn, "bom_items", "trim_size_unit", "TEXT")
    ensure_table_column(conn, "bom_items", "length_value", "TEXT")
    ensure_table_column(conn, "bom_items", "length_unit", "TEXT")
    ensure_table_column(conn, "bom_items", "width_unit", "TEXT")
    ensure_table_column(conn, "bom_items", "gsm_unit", "TEXT")
    ensure_table_column(conn, "bom_items", "moq_charge", "TEXT")
    ensure_table_column(conn, "bom_items", "moq_unit", "TEXT")
    ensure_table_column(conn, "bom_items", "moq_charge_unit", "TEXT")
    ensure_table_column(conn, "bom_items", "mcq", "TEXT")
    ensure_table_column(conn, "bom_items", "mcq_unit", "TEXT")
    ensure_table_column(conn, "bom_items", "mcq_charge", "TEXT")
    ensure_table_column(conn, "bom_items", "mcq_charge_unit", "TEXT")
    ensure_table_column(conn, "bom_items", "lead_time_unit", "TEXT")
    ensure_table_column(conn, "bom_items", "sample_lead_time", "TEXT")
    ensure_table_column(conn, "bom_items", "sample_lead_time_unit", "TEXT")
    ensure_table_column(conn, "bom_items", "fds_date", "TEXT")


def _read_json_file(path: Path) -> dict | list | None:
    if not path.exists():
        return None
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return None


def ensure_default_permission_rows(conn: sqlite3.Connection) -> None:
    existing_roles = {
        str(row["role"]).strip().lower()
        for row in conn.execute("SELECT role FROM permission_role_defaults").fetchall()
    }
    for role, payload in DEFAULT_ROLE_DEFAULTS.items():
        if role in existing_roles:
            continue
        conn.execute(
            """
            INSERT INTO permission_role_defaults (role, modules_json, actions_json, updated_at)
            VALUES (?, ?, ?, CURRENT_TIMESTAMP)
            """,
            (
                role,
                json.dumps(payload.get("modules", {}), ensure_ascii=False),
                json.dumps(payload.get("actions", {}), ensure_ascii=False),
            ),
        )


def migrate_legacy_permission_settings(conn: sqlite3.Connection, legacy_path: Path) -> None:
    has_role_rows = conn.execute("SELECT 1 FROM permission_role_defaults LIMIT 1").fetchone() is not None
    has_override_rows = conn.execute("SELECT 1 FROM permission_user_overrides LIMIT 1").fetchone() is not None
    if has_role_rows or has_override_rows:
        return

    legacy_payload = _read_json_file(legacy_path)
    normalized = normalize_permission_settings(legacy_payload if isinstance(legacy_payload, dict) else None)

    for role, payload in dict(normalized.get("role_defaults", {})).items():
        conn.execute(
            """
            INSERT INTO permission_role_defaults (role, modules_json, actions_json, updated_at)
            VALUES (?, ?, ?, CURRENT_TIMESTAMP)
            ON CONFLICT(role) DO UPDATE SET
                modules_json = excluded.modules_json,
                actions_json = excluded.actions_json,
                updated_at = CURRENT_TIMESTAMP
            """,
            (
                role,
                json.dumps(dict(payload.get("modules", {})), ensure_ascii=False),
                json.dumps(dict(payload.get("actions", {})), ensure_ascii=False),
            ),
        )

    for email, payload in dict(normalized.get("user_overrides", {})).items():
        conn.execute(
            """
            INSERT INTO permission_user_overrides (email, role_hint, modules_json, actions_json, updated_at)
            VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP)
            ON CONFLICT(email) DO UPDATE SET
                role_hint = excluded.role_hint,
                modules_json = excluded.modules_json,
                actions_json = excluded.actions_json,
                updated_at = CURRENT_TIMESTAMP
            """,
            (
                normalize_email(email),
                str(payload.get("role_hint", "user")).strip().lower() or "user",
                json.dumps(dict(payload.get("modules", {})), ensure_ascii=False),
                json.dumps(dict(payload.get("actions", {})), ensure_ascii=False),
            ),
        )


def migrate_legacy_app_setting(conn: sqlite3.Connection, setting_key: str, legacy_path: Path) -> None:
    existing = conn.execute("SELECT 1 FROM app_settings WHERE setting_key = ?", (setting_key,)).fetchone()
    if existing is not None:
        return
    legacy_payload = _read_json_file(legacy_path)
    if legacy_payload is None:
        return
    conn.execute(
        """
        INSERT INTO app_settings (setting_key, value_json, updated_at)
        VALUES (?, ?, CURRENT_TIMESTAMP)
        """,
        (setting_key, json.dumps(legacy_payload, ensure_ascii=False)),
    )


def needs_purchase_order_migration(conn: sqlite3.Connection) -> bool:
    table_sql_row = conn.execute(
        "SELECT sql FROM sqlite_master WHERE type = 'table' AND name = 'purchase_orders'"
    ).fetchone()
    if table_sql_row is None:
        return False

    table_sql = table_sql_row[0] or ""
    if "po_number TEXT NOT NULL UNIQUE" in table_sql:
        return True
    if "simple_domestic_pt TEXT" not in table_sql:
        return True

    indexes = conn.execute("PRAGMA index_list('purchase_orders')").fetchall()
    for index in indexes:
        if index["name"] == "idx_purchase_orders_identity":
            columns = conn.execute("PRAGMA index_info('idx_purchase_orders_identity')").fetchall()
            column_names = [column["name"] for column in columns]
            return column_names != ["po_number", "simple_domestic_pt", "source_pdf"]
    return True


def ensure_purchase_order_identity_index(conn: sqlite3.Connection) -> None:
    conn.execute("DROP INDEX IF EXISTS idx_purchase_orders_identity")
    conn.execute(PURCHASE_ORDER_IDENTITY_INDEX_SQL)


def backfill_simple_domestic_pt(conn: sqlite3.Connection) -> None:
    rows = conn.execute(
        """
        SELECT id, source_pdf, raw_json
        FROM purchase_orders
        WHERE IFNULL(simple_domestic_pt, '') = ''
        """
    ).fetchall()
    if not rows:
        return

    try:
        import pdfplumber
        from parser import extract_simple_domestic_pt
    except Exception:
        return

    for row in rows:
        simple_domestic_pt = ""
        raw_json = row["raw_json"] or ""
        if raw_json:
            try:
                payload = json.loads(raw_json)
                simple_domestic_pt = str(payload.get("simple_domestic_pt", "")).strip()
            except Exception:
                simple_domestic_pt = ""

        source_pdf = Path(str(row["source_pdf"] or "").strip())
        if not simple_domestic_pt and source_pdf.exists():
            try:
                with pdfplumber.open(str(source_pdf)) as pdf:
                    simple_domestic_pt = extract_simple_domestic_pt(pdf)
            except Exception:
                simple_domestic_pt = ""

        if simple_domestic_pt:
            conn.execute(
                "UPDATE purchase_orders SET simple_domestic_pt = ? WHERE id = ?",
                (simple_domestic_pt, row["id"]),
            )


def migrate_purchase_orders_table(conn: sqlite3.Connection) -> None:
    conn.execute("PRAGMA foreign_keys = OFF")
    conn.execute("BEGIN")
    try:
        conn.execute(PURCHASE_ORDERS_TABLE_SQL.replace("purchase_orders", "purchase_orders_new", 1))
        conn.execute(
            """
            INSERT INTO purchase_orders_new (
                id, po_number, po_date, po_time, simple_domestic_pt, dept_number, vendor_name, vendor_address,
                vendor_attention, vendor_email, buyer_name, buyer_email, shipping_address,
                mailing_address, ship_to_name, ship_to_address, mark_for, store_number,
                order_date, ship_date, cancel_date, duns, bealls_code, purchase_disc,
                post_dist, co_op, fob_point, terms, as_of, rtv, reference_number,
                new_store_disc, freight_allowance, total_packs_ordered, total_units,
                total_cost, total_qty, special_instructions, source_pdf, status, is_active, raw_json, created_at
            )
            SELECT
                id, po_number, po_date, po_time, NULL, dept_number, vendor_name, vendor_address,
                vendor_attention, vendor_email, buyer_name, buyer_email, shipping_address,
                mailing_address, ship_to_name, ship_to_address, mark_for, store_number,
                order_date, ship_date, cancel_date, duns, bealls_code, purchase_disc,
                post_dist, co_op, fob_point, terms, as_of, rtv, reference_number,
                new_store_disc, freight_allowance, total_packs_ordered, total_units,
                total_cost, total_qty, special_instructions, source_pdf, 'draft', 1, raw_json, created_at
            FROM purchase_orders
            """
        )
        conn.execute("DROP TABLE purchase_orders")
        conn.execute("ALTER TABLE purchase_orders_new RENAME TO purchase_orders")
        conn.execute("COMMIT")
    except Exception:
        conn.execute("ROLLBACK")
        raise
    finally:
        conn.execute("PRAGMA foreign_keys = ON")


def _initialize_postgres_schema(conn: sqlite3.Connection | PostgresCompatConnection) -> None:
    for statement in (
        PURCHASE_ORDERS_TABLE_SQL,
        PURCHASE_ORDER_ITEMS_TABLE_SQL,
        COSTING_IMPORTS_TABLE_SQL,
        COSTING_VARIANTS_TABLE_SQL,
        COSTING_WS_IMPORTS_TABLE_SQL,
        COSTING_WS_ROWS_TABLE_SQL,
        USERS_TABLE_SQL,
        APP_SETTINGS_TABLE_SQL,
        TECH_PACK_RECORDS_TABLE_SQL,
        SAMPLE_RECORDS_TABLE_SQL,
        SAMPLE_RECORDS_STYLE_INDEX_SQL,
        BOM_HEADERS_TABLE_SQL,
        BOM_HEADERS_STYLE_INDEX_SQL,
        BOM_HEADERS_STYLE_VERSION_INDEX_SQL,
        BOM_ITEMS_TABLE_SQL,
        BOM_ITEMS_BOM_ITEM_CODE_INDEX_SQL,
        BOM_ITEMS_STYLE_INDEX_SQL,
        BOM_ITEMS_TYPE_INDEX_SQL,
        BOM_ITEMS_SUPPLIER_INDEX_SQL,
        BOM_ITEMS_STATUS_INDEX_SQL,
        SUPPLIERS_TABLE_SQL,
        SUPPLIERS_CODE_INDEX_SQL,
        SUPPLIERS_NAME_INDEX_SQL,
        SUPPLIERS_STATUS_INDEX_SQL,
        HR_EMPLOYEES_TABLE_SQL,
        HR_EMPLOYEES_USER_EMAIL_INDEX_SQL,
        HR_ATTENDANCE_RECORDS_TABLE_SQL,
        HR_ATTENDANCE_EMPLOYEE_DATE_INDEX_SQL,
        TECH_PACK_HIDDEN_STYLES_TABLE_SQL,
        PERMISSION_ROLE_DEFAULTS_TABLE_SQL,
        PERMISSION_USER_OVERRIDES_TABLE_SQL,
        PASSWORD_RESET_TOKENS_TABLE_SQL,
        PURCHASE_ORDER_IDENTITY_INDEX_SQL,
    ):
        conn.execute(_translate_schema_sql_for_postgres(statement))


def init_db(db_path: Path) -> None:
    if using_postgres():
        with connect(db_path) as conn:
            _initialize_postgres_schema(conn)
            ensure_workflow_columns(conn)
            ensure_bom_item_purchase_columns(conn)
            ensure_default_admin_user(conn)
            migrate_legacy_permission_settings(conn, PERMISSIONS_SETTINGS_PATH)
            ensure_default_permission_rows(conn)
            migrate_legacy_app_setting(conn, "server_settings", SERVER_SETTINGS_PATH)
            migrate_legacy_app_setting(conn, "report_templates", REPORT_CONFIG_PATH)
            ensure_tech_pack_storage_migrated(conn)
            ensure_sample_records_storage_migrated(conn)
        return

    with connect(db_path) as conn:
        table_exists = (
            conn.execute(
                "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'purchase_orders'"
            ).fetchone()
            is not None
        )
        conn.execute(PURCHASE_ORDERS_TABLE_SQL)
        conn.execute(PURCHASE_ORDER_ITEMS_TABLE_SQL)
        if table_exists and needs_purchase_order_migration(conn):
            migrate_purchase_orders_table(conn)
        conn.execute(PURCHASE_ORDER_ITEMS_TABLE_SQL)
        conn.execute(COSTING_IMPORTS_TABLE_SQL)
        conn.execute(COSTING_VARIANTS_TABLE_SQL)
        conn.execute(COSTING_WS_IMPORTS_TABLE_SQL)
        conn.execute(COSTING_WS_ROWS_TABLE_SQL)
        conn.execute(USERS_TABLE_SQL)
        ensure_users_auth_columns(conn)
        conn.execute(APP_SETTINGS_TABLE_SQL)
        conn.execute(TECH_PACK_RECORDS_TABLE_SQL)
        conn.execute(SAMPLE_RECORDS_TABLE_SQL)
        conn.execute(SAMPLE_RECORDS_STYLE_INDEX_SQL)
        conn.execute(BOM_HEADERS_TABLE_SQL)
        conn.execute(BOM_HEADERS_STYLE_INDEX_SQL)
        conn.execute(BOM_HEADERS_STYLE_VERSION_INDEX_SQL)
        conn.execute(BOM_ITEMS_TABLE_SQL)
        conn.execute(BOM_ITEMS_BOM_ITEM_CODE_INDEX_SQL)
        conn.execute(BOM_ITEMS_STYLE_INDEX_SQL)
        conn.execute(BOM_ITEMS_TYPE_INDEX_SQL)
        conn.execute(BOM_ITEMS_SUPPLIER_INDEX_SQL)
        conn.execute(BOM_ITEMS_STATUS_INDEX_SQL)
        ensure_bom_item_purchase_columns(conn)
        conn.execute(SUPPLIERS_TABLE_SQL)
        conn.execute(SUPPLIERS_CODE_INDEX_SQL)
        conn.execute(SUPPLIERS_NAME_INDEX_SQL)
        conn.execute(SUPPLIERS_STATUS_INDEX_SQL)
        conn.execute(HR_EMPLOYEES_TABLE_SQL)
        conn.execute(HR_EMPLOYEES_USER_EMAIL_INDEX_SQL)
        conn.execute(HR_ATTENDANCE_RECORDS_TABLE_SQL)
        conn.execute(HR_ATTENDANCE_EMPLOYEE_DATE_INDEX_SQL)
        ensure_record_timestamp_columns(conn)
        conn.execute(TECH_PACK_HIDDEN_STYLES_TABLE_SQL)
        conn.execute(PERMISSION_ROLE_DEFAULTS_TABLE_SQL)
        conn.execute(PERMISSION_USER_OVERRIDES_TABLE_SQL)
        conn.execute(PASSWORD_RESET_TOKENS_TABLE_SQL)
        backfill_simple_domestic_pt(conn)
        ensure_purchase_order_identity_index(conn)
        ensure_default_admin_user(conn)
        migrate_legacy_permission_settings(conn, PERMISSIONS_SETTINGS_PATH)
        ensure_default_permission_rows(conn)
        migrate_legacy_app_setting(conn, "server_settings", SERVER_SETTINGS_PATH)
        migrate_legacy_app_setting(conn, "report_templates", REPORT_CONFIG_PATH)
        ensure_tech_pack_storage_migrated(conn)
        ensure_sample_records_storage_migrated(conn)


def _load_json_setting_from_conn(conn: sqlite3.Connection, setting_key: str, default: object = None) -> object:
    row = conn.execute(
        "SELECT value_json FROM app_settings WHERE setting_key = ?",
        (str(setting_key).strip(),),
    ).fetchone()
    if row is None:
        return default
    try:
        return json.loads(row["value_json"])
    except Exception:
        return default


def ensure_tech_pack_storage_migrated(conn: sqlite3.Connection) -> None:
    has_records = conn.execute("SELECT 1 FROM tech_pack_records LIMIT 1").fetchone() is not None
    if not has_records:
        legacy_records = _load_json_setting_from_conn(conn, "tech_pack_library", [])
        if isinstance(legacy_records, list):
            normalized_rows: dict[str, dict] = {}
            for item in legacy_records:
                if not isinstance(item, dict):
                    continue
                style_code = normalize_style_identifier(item.get("style_code"))
                if not style_code:
                    continue
                normalized = dict(item)
                normalized["style_code"] = style_code
                normalized_rows[style_code] = normalized
            for style_code, row in normalized_rows.items():
                conn.execute(
                    """
                    INSERT INTO tech_pack_records (style_code, record_json, created_at, updated_at)
                    VALUES (?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
                    ON CONFLICT(style_code) DO UPDATE SET
                        record_json = excluded.record_json,
                        updated_at = CURRENT_TIMESTAMP
                    """,
                    (style_code, json.dumps(row, ensure_ascii=False)),
                )

    has_hidden = conn.execute("SELECT 1 FROM tech_pack_hidden_styles LIMIT 1").fetchone() is not None
    if not has_hidden:
        legacy_hidden = _load_json_setting_from_conn(conn, "tech_pack_hidden_styles", [])
        if isinstance(legacy_hidden, list):
            normalized_hidden = sorted(
                {
                    style_code
                    for style_code in (normalize_style_identifier(item) for item in legacy_hidden)
                    if style_code
                }
            )
            for style_code in normalized_hidden:
                conn.execute(
                    """
                    INSERT INTO tech_pack_hidden_styles (style_code, updated_at)
                    VALUES (?, CURRENT_TIMESTAMP)
                    ON CONFLICT(style_code) DO UPDATE SET
                        updated_at = CURRENT_TIMESTAMP
                    """,
                    (style_code,),
                )


def ensure_sample_records_storage_migrated(conn: sqlite3.Connection) -> None:
    has_records = conn.execute("SELECT 1 FROM sample_records LIMIT 1").fetchone() is not None
    if has_records:
        return
    for item in normalize_sample_record_rows(_load_json_setting_from_conn(conn, "sample_tracking_records", [])):
        conn.execute(
            """
            INSERT INTO sample_records (
                record_id, style_code, request_number, record_json, created_at, updated_at
            )
            VALUES (?, ?, ?, ?, COALESCE(NULLIF(?, ''), CURRENT_TIMESTAMP), COALESCE(NULLIF(?, ''), CURRENT_TIMESTAMP))
            ON CONFLICT(record_id) DO UPDATE SET
                style_code = excluded.style_code,
                request_number = excluded.request_number,
                record_json = excluded.record_json,
                created_at = COALESCE(sample_records.created_at, excluded.created_at),
                updated_at = COALESCE(excluded.updated_at, CURRENT_TIMESTAMP)
            """,
            (
                item["record_id"],
                item["style_code"],
                item["request_number"],
                json.dumps(item, ensure_ascii=False),
                item.get("created_at", ""),
                item.get("updated_at", ""),
            ),
        )


def save_purchase_order(db_path: Path, po: dict) -> int:
    with connect(db_path) as conn:
        po_number = str(po.get("po_number", "")).strip()
        po["po_number"] = po_number
        po_date = str(po.get("po_date", "")).strip()
        po_time = str(po.get("po_time", "")).strip()
        simple_domestic_pt = str(po.get("simple_domestic_pt", "")).strip()
        source_pdf = str(po.get("source_pdf", "")).strip()
        status_value = normalize_workflow_status(po.get("status"), "draft")
        is_active_value = parse_active_flag(po.get("is_active"), True)
        if simple_domestic_pt:
            existing = conn.execute(
                """
                SELECT id
                FROM purchase_orders
                WHERE po_number = ? AND IFNULL(simple_domestic_pt, '') = ? AND IFNULL(source_pdf, '') = ?
                """,
                (po_number, simple_domestic_pt, source_pdf),
            ).fetchone()
        else:
            existing = conn.execute(
                """
                SELECT id
                FROM purchase_orders
                WHERE po_number = ? AND IFNULL(po_date, '') = ? AND IFNULL(po_time, '') = ? AND IFNULL(source_pdf, '') = ?
                """,
                (po_number, po_date, po_time, source_pdf),
            ).fetchone()

        payload = (
            po_number,
            po_date,
            po_time,
            simple_domestic_pt,
            po.get("dept_number"),
            po.get("vendor_name"),
            po.get("vendor_address"),
            po.get("vendor_attention"),
            po.get("vendor_email"),
            po.get("buyer_name"),
            po.get("buyer_email"),
            po.get("shipping_address"),
            po.get("mailing_address"),
            po.get("ship_to_name"),
            po.get("ship_to_address"),
            po.get("mark_for"),
            po.get("store_number"),
            po.get("order_date"),
            po.get("ship_date"),
            po.get("cancel_date"),
            po.get("duns"),
            po.get("bealls_code"),
            po.get("purchase_disc"),
            po.get("post_dist"),
            po.get("co_op"),
            po.get("fob_point"),
            po.get("terms"),
            po.get("as_of"),
            po.get("rtv"),
            po.get("reference_number"),
            po.get("new_store_disc"),
            po.get("freight_allowance"),
            po.get("total_packs_ordered"),
            po.get("total_units"),
            po.get("total_cost"),
            po.get("total_qty"),
            po.get("special_instructions"),
            source_pdf,
            status_value,
            1 if is_active_value else 0,
            json.dumps(po, ensure_ascii=True),
        )

        if existing:
            order_id = existing["id"]
            conn.execute(
                """
                UPDATE purchase_orders
                SET po_number = ?, po_date = ?, po_time = ?, simple_domestic_pt = ?, dept_number = ?, vendor_name = ?,
                    vendor_address = ?, vendor_attention = ?, vendor_email = ?, buyer_name = ?,
                    buyer_email = ?, shipping_address = ?, mailing_address = ?, ship_to_name = ?,
                    ship_to_address = ?, mark_for = ?, store_number = ?, order_date = ?, ship_date = ?,
                    cancel_date = ?, duns = ?, bealls_code = ?, purchase_disc = ?, post_dist = ?,
                    co_op = ?, fob_point = ?, terms = ?, as_of = ?, rtv = ?, reference_number = ?,
                    new_store_disc = ?, freight_allowance = ?, total_packs_ordered = ?, total_units = ?,
                    total_cost = ?, total_qty = ?, special_instructions = ?, source_pdf = ?, status = ?, is_active = ?, raw_json = ?,
                    updated_at = CURRENT_TIMESTAMP
                WHERE id = ?
                """,
                payload + (order_id,),
            )
            conn.execute(
                "DELETE FROM purchase_order_items WHERE purchase_order_id = ?",
                (order_id,),
            )
        else:
            cursor = conn.execute(
                """
                INSERT INTO purchase_orders (
                    po_number, po_date, po_time, simple_domestic_pt, dept_number, vendor_name, vendor_address,
                    vendor_attention, vendor_email, buyer_name, buyer_email, shipping_address,
                    mailing_address, ship_to_name, ship_to_address, mark_for, store_number,
                    order_date, ship_date, cancel_date, duns, bealls_code, purchase_disc,
                    post_dist, co_op, fob_point, terms, as_of, rtv, reference_number,
                    new_store_disc, freight_allowance, total_packs_ordered, total_units,
                    total_cost, total_qty, special_instructions, source_pdf, status, is_active, raw_json, created_at, updated_at
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
                """,
                payload,
            )
            order_id = int(cursor.lastrowid)

        conn.executemany(
            """
            INSERT INTO purchase_order_items (
                purchase_order_id, sku, mfg_style, mfg_color, size_desc, description,
                upc, cost_unit, comp, retail, pack_qty, total_packs, total_units
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            [
                (
                    order_id,
                    item.get("sku"),
                    item.get("mfg_style"),
                    item.get("mfg_color"),
                    item.get("size_desc"),
                    item.get("description"),
                    item.get("upc"),
                    item.get("cost_unit"),
                    item.get("comp"),
                    item.get("retail"),
                    item.get("pack_qty"),
                    item.get("total_packs"),
                    item.get("total_units"),
                )
                for item in po["items"]
            ],
        )
        ensure_tech_pack_records_for_purchase_order(conn, po)
        return int(order_id)


def ensure_tech_pack_records_for_purchase_order(
    conn: sqlite3.Connection | PostgresCompatConnection,
    po: dict,
) -> None:
    if not isinstance(po, dict):
        return
    items = po.get("items", [])
    if not isinstance(items, list):
        return

    style_rows: dict[str, dict] = {}
    for item in items:
        if not isinstance(item, dict):
            continue
        style_code = normalize_style_identifier(item.get("mfg_style") or item.get("style_code"))
        if not style_code:
            continue
        row = style_rows.setdefault(
            style_code,
            {
                "style_code": style_code,
                "style_name": "",
                "season": "",
                "collection": "",
                "overview_data": {},
                "source_files": [],
                "preview_images": [],
                "consumption_versions": [],
                "bom_data": [],
                "construction_data": [],
                "stitching_data": [],
                "pom_data": [],
                "design_intent_data": [],
                "attachments_data": [],
                "detail_tab_overrides": {},
            },
        )
        if not row["style_name"]:
            row["style_name"] = str(item.get("description", "") or "").strip()

    if not style_rows:
        return

    ensure_tech_pack_storage_migrated(conn)
    for style_code, payload in style_rows.items():
        conn.execute(
            """
            INSERT INTO tech_pack_records (style_code, record_json, created_at, updated_at)
            VALUES (?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
            ON CONFLICT(style_code) DO NOTHING
            """,
            (style_code, json.dumps(payload, ensure_ascii=False)),
        )
        conn.execute(
            "DELETE FROM tech_pack_hidden_styles WHERE style_code = ?",
            (style_code,),
        )


def backfill_tech_pack_records_from_existing_purchase_orders(db_path: Path) -> list[str]:
    created_styles: list[str] = []
    with connect(db_path) as conn:
        order_rows = conn.execute(
            """
            SELECT po.raw_json
            FROM purchase_orders po
            ORDER BY po.id
            """
        ).fetchall()
        existing_styles = {
            str(row["style_code"] or "").strip()
            for row in conn.execute(
                """
                SELECT style_code
                FROM tech_pack_records
                """
            ).fetchall()
        }
        for row in order_rows:
            raw_json = str(row["raw_json"] or "").strip()
            if not raw_json:
                continue
            try:
                po = json.loads(raw_json)
            except Exception:
                continue
            if not isinstance(po, dict):
                continue

            before_styles = set(existing_styles)
            ensure_tech_pack_records_for_purchase_order(conn, po)

            items = po.get("items", [])
            if not isinstance(items, list):
                continue
            for item in items:
                if not isinstance(item, dict):
                    continue
                style_code = normalize_style_identifier(item.get("mfg_style") or item.get("style_code"))
                if not style_code or style_code in before_styles:
                    continue
                existing_styles.add(style_code)
                created_styles.append(style_code)
    return created_styles


def get_user_by_email(db_path: Path, email: str) -> dict | None:
    normalized_email = normalize_email(email)
    if not normalized_email:
        return None
    with connect(db_path) as conn:
        row = conn.execute(
            """
            SELECT id, email, password_hash, display_name, role, is_active, approval_status, approved_by, approved_at, created_at, updated_at
            FROM users
            WHERE email = ?
            """,
            (normalized_email,),
        ).fetchone()
    return dict(row) if row is not None else None


def register_user(
    db_path: Path,
    *,
    email: str,
    password: str,
    display_name: str = "",
    role: str = "user",
) -> dict:
    normalized_email = normalize_email(email)
    normalized_role = str(role or "user").strip().lower() or "user"
    if not normalized_email:
        raise ValueError("Email is required.")
    if not password:
        raise ValueError("Password is required.")

    with connect(db_path) as conn:
        existing = conn.execute("SELECT id FROM users WHERE email = ?", (normalized_email,)).fetchone()
        if existing is not None:
            raise ValueError("This email is already registered.")

        cursor = conn.execute(
            """
            INSERT INTO users (email, password_hash, display_name, role, is_active, approval_status)
            VALUES (?, ?, ?, ?, 0, 'pending')
            """,
            (
                normalized_email,
                hash_password(password),
                str(display_name or "").strip(),
                normalized_role,
            ),
        )
        user_id = int(cursor.lastrowid)
        row = conn.execute(
            """
            SELECT id, email, display_name, role, is_active, approval_status, approved_by, approved_at, created_at, updated_at
            FROM users
            WHERE id = ?
            """,
            (user_id,),
        ).fetchone()
    return dict(row) if row is not None else {}


def authenticate_user(db_path: Path, email: str, password: str) -> dict | None:
    normalized_email = normalize_email(email)
    if not normalized_email or not password:
        return None
    with connect(db_path) as conn:
        row = conn.execute(
            """
            SELECT id, email, password_hash, display_name, role, is_active, approval_status, approved_by, approved_at, created_at, updated_at
            FROM users
            WHERE email = ?
            """,
            (normalized_email,),
        ).fetchone()
    if row is None or int(row["is_active"] or 0) != 1 or str(row["approval_status"] or "approved").strip().lower() != "approved":
        return None
    if not verify_password(password, str(row["password_hash"] or "")):
        return None
    payload = dict(row)
    payload.pop("password_hash", None)
    return payload


def update_user_password(db_path: Path, email: str, new_password: str) -> None:
    normalized_email = normalize_email(email)
    if not normalized_email:
        raise ValueError("Email is required.")
    if not new_password:
        raise ValueError("New password is required.")
    with connect(db_path) as conn:
        existing = conn.execute("SELECT id FROM users WHERE email = ?", (normalized_email,)).fetchone()
        if existing is None:
            raise ValueError("Account not found.")
        conn.execute(
            """
            UPDATE users
            SET password_hash = ?, updated_at = CURRENT_TIMESTAMP
            WHERE email = ?
            """,
            (hash_password(new_password), normalized_email),
        )


def update_user_role(db_path: Path, email: str, role: str) -> None:
    normalized_email = normalize_email(email)
    normalized_role = str(role or "user").strip().lower() or "user"
    if not normalized_email:
        raise ValueError("Email is required.")
    if normalized_role not in {"admin", "moderate", "user"}:
        raise ValueError("Invalid role.")
    with connect(db_path) as conn:
        existing = conn.execute("SELECT id FROM users WHERE email = ?", (normalized_email,)).fetchone()
        if existing is None:
            raise ValueError("Account not found.")
        conn.execute(
            """
            UPDATE users
            SET role = ?, updated_at = CURRENT_TIMESTAMP
            WHERE email = ?
            """,
            (normalized_role, normalized_email),
        )


def load_permission_settings_from_db(db_path: Path) -> dict:
    with connect(db_path) as conn:
        role_rows = conn.execute(
            """
            SELECT role, modules_json, actions_json
            FROM permission_role_defaults
            """
        ).fetchall()
        override_rows = conn.execute(
            """
            SELECT email, role_hint, modules_json, actions_json
            FROM permission_user_overrides
            """
        ).fetchall()

    payload = {
        "role_defaults": {},
        "user_overrides": {},
    }
    for row in role_rows:
        try:
            modules = json.loads(row["modules_json"] or "{}")
        except Exception:
            modules = {}
        try:
            actions = json.loads(row["actions_json"] or "{}")
        except Exception:
            actions = {}
        payload["role_defaults"][str(row["role"]).strip().lower()] = {
            "modules": modules,
            "actions": actions,
        }

    for row in override_rows:
        try:
            modules = json.loads(row["modules_json"] or "{}")
        except Exception:
            modules = {}
        try:
            actions = json.loads(row["actions_json"] or "{}")
        except Exception:
            actions = {}
        payload["user_overrides"][normalize_email(row["email"])] = {
            "role_hint": str(row["role_hint"] or "user").strip().lower() or "user",
            "modules": modules,
            "actions": actions,
        }
    return normalize_permission_settings(payload)


def save_permission_settings_to_db(db_path: Path, payload: dict) -> dict:
    normalized = normalize_permission_settings(payload)
    with connect(db_path) as conn:
        conn.execute("DELETE FROM permission_role_defaults")
        conn.execute("DELETE FROM permission_user_overrides")
        for role, row in dict(normalized.get("role_defaults", {})).items():
            conn.execute(
                """
                INSERT INTO permission_role_defaults (role, modules_json, actions_json, updated_at)
                VALUES (?, ?, ?, CURRENT_TIMESTAMP)
                """,
                (
                    str(role).strip().lower(),
                    json.dumps(dict(row.get("modules", {})), ensure_ascii=False),
                    json.dumps(dict(row.get("actions", {})), ensure_ascii=False),
                ),
            )
        for email, row in dict(normalized.get("user_overrides", {})).items():
            conn.execute(
                """
                INSERT INTO permission_user_overrides (email, role_hint, modules_json, actions_json, updated_at)
                VALUES (?, ?, ?, ?, CURRENT_TIMESTAMP)
                """,
                (
                    normalize_email(email),
                    str(row.get("role_hint", "user")).strip().lower() or "user",
                    json.dumps(dict(row.get("modules", {})), ensure_ascii=False),
                    json.dumps(dict(row.get("actions", {})), ensure_ascii=False),
                ),
            )
    return normalized


def get_app_setting(db_path: Path, setting_key: str, default: object = None) -> object:
    with connect(db_path) as conn:
        row = conn.execute(
            "SELECT value_json FROM app_settings WHERE setting_key = ?",
            (str(setting_key).strip(),),
        ).fetchone()
    if row is None:
        return default
    try:
        return json.loads(row["value_json"])
    except Exception:
        return default


def set_app_setting(db_path: Path, setting_key: str, value: object) -> object:
    key = str(setting_key).strip()
    payload = json.dumps(value, ensure_ascii=False)
    with connect(db_path) as conn:
        conn.execute(
            """
            INSERT INTO app_settings (setting_key, value_json, updated_at)
            VALUES (?, ?, CURRENT_TIMESTAMP)
            ON CONFLICT(setting_key) DO UPDATE SET
                value_json = excluded.value_json,
                updated_at = CURRENT_TIMESTAMP
            """,
            (key, payload),
        )
    return value


def list_tech_pack_records(db_path: Path) -> list[dict]:
    with connect(db_path) as conn:
        ensure_tech_pack_storage_migrated(conn)
        rows = conn.execute(
            """
            SELECT style_code, record_json, created_at, updated_at
            FROM tech_pack_records
            ORDER BY style_code
            """
        ).fetchall()
    records: list[dict] = []
    for row in rows:
        try:
            payload = json.loads(row["record_json"] or "{}")
        except Exception:
            payload = {}
        if isinstance(payload, dict):
            payload["style_code"] = str(row["style_code"] or payload.get("style_code") or "").strip()
            payload["created_at"] = str(row["created_at"] or payload.get("created_at") or "").strip()
            payload["updated_at"] = str(row["updated_at"] or payload.get("updated_at") or "").strip()
            records.append(payload)
    return records


def save_tech_pack_records(db_path: Path, rows: list[dict]) -> list[dict]:
    normalized_rows: dict[str, dict] = {}
    for item in rows:
        if not isinstance(item, dict):
            continue
        style_code = normalize_style_identifier(item.get("style_code"))
        if not style_code:
            continue
        normalized = dict(item)
        normalized["style_code"] = style_code
        normalized_rows[style_code] = normalized

    with connect(db_path) as conn:
        ensure_tech_pack_storage_migrated(conn)
        existing_rows = {
            str(row["style_code"]).strip(): row
            for row in conn.execute(
                """
                SELECT style_code, created_at
                FROM tech_pack_records
                """
            ).fetchall()
        }
        stale_styles = set(existing_rows) - set(normalized_rows)
        for style_code in stale_styles:
            conn.execute("DELETE FROM tech_pack_records WHERE style_code = ?", (style_code,))
        for style_code in sorted(normalized_rows):
            payload = dict(normalized_rows[style_code])
            existing_row = existing_rows.get(style_code)
            created_at = (
                str((existing_row["created_at"] if existing_row is not None else payload.get("created_at")) or "").strip()
                or None
            )
            conn.execute(
                """
                INSERT INTO tech_pack_records (style_code, record_json, created_at, updated_at)
                VALUES (?, ?, COALESCE(?, CURRENT_TIMESTAMP), CURRENT_TIMESTAMP)
                ON CONFLICT(style_code) DO UPDATE SET
                    record_json = excluded.record_json,
                    created_at = COALESCE(tech_pack_records.created_at, excluded.created_at),
                    updated_at = CURRENT_TIMESTAMP
                """,
                (style_code, json.dumps(payload, ensure_ascii=False), created_at),
            )
    return list_tech_pack_records(db_path)


def save_tech_pack_record(db_path: Path, row: dict) -> dict | None:
    if not isinstance(row, dict):
        return None
    style_code = normalize_style_identifier(row.get("style_code"))
    if not style_code:
        return None
    payload = dict(row)
    payload["style_code"] = style_code

    with connect(db_path) as conn:
        ensure_tech_pack_storage_migrated(conn)
        existing_row = conn.execute(
            """
            SELECT style_code, created_at
            FROM tech_pack_records
            WHERE style_code = ?
            """,
            (style_code,),
        ).fetchone()
        created_at = (
            str((existing_row["created_at"] if existing_row is not None else payload.get("created_at")) or "").strip()
            or None
        )
        conn.execute(
            """
            INSERT INTO tech_pack_records (style_code, record_json, created_at, updated_at)
            VALUES (?, ?, COALESCE(?, CURRENT_TIMESTAMP), CURRENT_TIMESTAMP)
            ON CONFLICT(style_code) DO UPDATE SET
                record_json = excluded.record_json,
                created_at = COALESCE(tech_pack_records.created_at, excluded.created_at),
                updated_at = CURRENT_TIMESTAMP
            """,
            (style_code, json.dumps(payload, ensure_ascii=False), created_at),
        )
        saved_row = conn.execute(
            """
            SELECT style_code, record_json, created_at, updated_at
            FROM tech_pack_records
            WHERE style_code = ?
            """,
            (style_code,),
        ).fetchone()
    if saved_row is None:
        return None
    try:
        saved_payload = json.loads(saved_row["record_json"] or "{}")
    except Exception:
        saved_payload = {}
    if not isinstance(saved_payload, dict):
        saved_payload = {}
    saved_payload["style_code"] = str(saved_row["style_code"] or saved_payload.get("style_code") or "").strip()
    saved_payload["created_at"] = str(saved_row["created_at"] or saved_payload.get("created_at") or "").strip()
    saved_payload["updated_at"] = str(saved_row["updated_at"] or saved_payload.get("updated_at") or "").strip()
    return saved_payload


def _merge_nested_patch(base: object, patch: object) -> object:
    if isinstance(base, dict) and isinstance(patch, dict):
        merged = dict(base)
        for key, value in patch.items():
            merged[str(key)] = _merge_nested_patch(merged.get(str(key)), value)
        return merged
    return patch


def save_tech_pack_record_patch(db_path: Path, style_code: object, changes: dict) -> dict | None:
    normalized_style = normalize_style_identifier(style_code)
    if not normalized_style or not isinstance(changes, dict):
        return None
    current_record = None
    for item in list_tech_pack_records(db_path):
        if str(item.get("style_code", "")).strip() == normalized_style:
            current_record = dict(item)
            break
    merged_record = _merge_nested_patch(current_record or {"style_code": normalized_style}, changes)
    if isinstance(merged_record, dict):
        merged_record["style_code"] = normalized_style
        return save_tech_pack_record(db_path, merged_record)
    return None


def list_sample_records(db_path: Path) -> list[dict]:
    with connect(db_path) as conn:
        ensure_sample_records_storage_migrated(conn)
        rows = conn.execute(
            """
            SELECT record_id, style_code, request_number, record_json, created_at, updated_at
            FROM sample_records
            ORDER BY created_at DESC, request_number DESC, record_id DESC
            """
        ).fetchall()
    records: list[dict] = []
    for row in rows:
        try:
            payload = json.loads(row["record_json"] or "{}")
        except Exception:
            payload = {}
        if not isinstance(payload, dict):
            payload = {}
        payload["record_id"] = str(row["record_id"] or payload.get("record_id") or "").strip()
        payload["style_code"] = str(row["style_code"] or payload.get("style_code") or "").strip()
        payload["request_number"] = str(row["request_number"] or payload.get("request_number") or "").strip()
        payload["created_at"] = str(row["created_at"] or payload.get("created_at") or "").strip()
        payload["updated_at"] = str(row["updated_at"] or payload.get("updated_at") or "").strip()
        if not str(payload.get("style", "")).strip() and payload["style_code"]:
            payload["style"] = payload["style_code"]
        records.append(payload)
    return normalize_sample_record_rows(records)


def save_sample_records(db_path: Path, rows: list[dict]) -> list[dict]:
    normalized_rows = normalize_sample_record_rows(rows)
    rows_by_id = {
        str(item.get("record_id", "")).strip(): item
        for item in normalized_rows
        if str(item.get("record_id", "")).strip()
    }
    with connect(db_path) as conn:
        ensure_sample_records_storage_migrated(conn)
        existing_rows = {
            str(row["record_id"]).strip(): row
            for row in conn.execute(
                """
                SELECT record_id, created_at
                FROM sample_records
                """
            ).fetchall()
        }
        stale_ids = set(existing_rows) - set(rows_by_id)
        for record_id in stale_ids:
            conn.execute("DELETE FROM sample_records WHERE record_id = ?", (record_id,))
        for record_id, payload in rows_by_id.items():
            existing_row = existing_rows.get(record_id)
            created_at = (
                str((existing_row["created_at"] if existing_row is not None else payload.get("created_at")) or "").strip()
                or None
            )
            conn.execute(
                """
                INSERT INTO sample_records (
                    record_id, style_code, request_number, record_json, created_at, updated_at
                )
                VALUES (?, ?, ?, ?, COALESCE(?, CURRENT_TIMESTAMP), CURRENT_TIMESTAMP)
                ON CONFLICT(record_id) DO UPDATE SET
                    style_code = excluded.style_code,
                    request_number = excluded.request_number,
                    record_json = excluded.record_json,
                    created_at = COALESCE(sample_records.created_at, excluded.created_at),
                    updated_at = CURRENT_TIMESTAMP
                """,
                (
                    record_id,
                    payload.get("style_code", ""),
                    payload.get("request_number", ""),
                    json.dumps(payload, ensure_ascii=False),
                    created_at,
                ),
            )
    return list_sample_records(db_path)


def save_sample_record(db_path: Path, row: dict) -> dict | None:
    normalized_rows = normalize_sample_record_rows([row])
    if not normalized_rows:
        return None
    payload = normalized_rows[0]
    record_id = str(payload.get("record_id", "")).strip()
    if not record_id:
        return None

    with connect(db_path) as conn:
        ensure_sample_records_storage_migrated(conn)
        existing_row = conn.execute(
            """
            SELECT record_id, created_at
            FROM sample_records
            WHERE record_id = ?
            """,
            (record_id,),
        ).fetchone()
        created_at = (
            str((existing_row["created_at"] if existing_row is not None else payload.get("created_at")) or "").strip()
            or None
        )
        conn.execute(
            """
            INSERT INTO sample_records (
                record_id, style_code, request_number, record_json, created_at, updated_at
            )
            VALUES (?, ?, ?, ?, COALESCE(?, CURRENT_TIMESTAMP), CURRENT_TIMESTAMP)
            ON CONFLICT(record_id) DO UPDATE SET
                style_code = excluded.style_code,
                request_number = excluded.request_number,
                record_json = excluded.record_json,
                created_at = COALESCE(sample_records.created_at, excluded.created_at),
                updated_at = CURRENT_TIMESTAMP
            """,
            (
                record_id,
                str(payload.get("style_code", "")).strip(),
                str(payload.get("request_number", "")).strip(),
                json.dumps(payload, ensure_ascii=False),
                created_at,
            ),
        )
        saved_row = conn.execute(
            """
            SELECT record_id, style_code, request_number, record_json, created_at, updated_at
            FROM sample_records
            WHERE record_id = ?
            """,
            (record_id,),
        ).fetchone()
    if saved_row is None:
        return None
    try:
        saved_payload = json.loads(saved_row["record_json"] or "{}")
    except Exception:
        saved_payload = {}
    if not isinstance(saved_payload, dict):
        saved_payload = {}
    saved_payload["record_id"] = str(saved_row["record_id"] or saved_payload.get("record_id") or "").strip()
    saved_payload["style_code"] = str(saved_row["style_code"] or saved_payload.get("style_code") or "").strip()
    saved_payload["request_number"] = str(saved_row["request_number"] or saved_payload.get("request_number") or "").strip()
    saved_payload["created_at"] = str(saved_row["created_at"] or saved_payload.get("created_at") or "").strip()
    saved_payload["updated_at"] = str(saved_row["updated_at"] or saved_payload.get("updated_at") or "").strip()
    if not str(saved_payload.get("style", "")).strip() and saved_payload["style_code"]:
        saved_payload["style"] = saved_payload["style_code"]
    normalized_saved = normalize_sample_record_rows([saved_payload])
    return normalized_saved[0] if normalized_saved else saved_payload


def default_bom_header_payload() -> dict:
    return {
        "bom_id": "",
        "style_code": "",
        "bom_version": "",
        "bom_name": "",
        "season": "",
        "collection": "",
        "parent_type": "",
        "source_type": "",
        "source_style_code": "",
        "source_note": "",
        "status": "draft",
        "is_active": True,
        "created_at": "",
        "updated_at": "",
    }


def normalize_bom_header_row(payload: object) -> dict | None:
    if not isinstance(payload, dict):
        return None
    normalized = default_bom_header_payload()
    for key in normalized:
        if key == "is_active":
            normalized[key] = bool(payload.get(key, normalized[key]))
        else:
            normalized[key] = str(payload.get(key, normalized[key]) or "").strip()
    normalized["bom_id"] = normalized["bom_id"] or secrets.token_hex(16)
    normalized["style_code"] = normalize_style_identifier(normalized.get("style_code"))
    normalized["source_style_code"] = normalize_style_identifier(normalized.get("source_style_code"))
    if not normalized["bom_version"]:
        return None
    normalized["status"] = normalized["status"] or "draft"
    return normalized


def default_bom_item_payload() -> dict:
    return {
        "item_id": "",
        "bom_id": "",
        "style_code": "",
        "bom_version": "",
        "bom_name": "",
        "season": "",
        "collection": "",
        "item_code": "",
        "item_name": "",
        "item_type": "Other",
        "section": "",
        "article_ref": "",
        "color": "",
        "color_code": "",
        "description": "",
        "composition": "",
        "position_of_use": "",
        "spec_summary": "",
        "uom": "",
        "qty_per_garment": "",
        "trim_size_um": "",
        "trim_size_unit": "",
        "length_value": "",
        "length_unit": "",
        "width_value": "",
        "width_unit": "",
        "gsm_value": "",
        "gsm_unit": "",
        "waste_percent": "",
        "tech_comment": "",
        "default_supplier": "",
        "supplier_item_code": "",
        "moq": "",
        "moq_unit": "",
        "moq_charge": "",
        "moq_charge_unit": "",
        "mcq": "",
        "mcq_unit": "",
        "mcq_charge": "",
        "mcq_charge_unit": "",
        "lead_time": "",
        "lead_time_unit": "",
        "sample_lead_time": "",
        "sample_lead_time_unit": "",
        "purchase_uom": "",
        "standard_price": "",
        "currency": "",
        "purchase_note": "",
        "fds_date": "",
        "image_path": "",
        "status": "draft",
        "source_style_code": "",
        "source_section": "",
        "source_section_index": "",
        "source_item_index": "",
        "source_payload_json": "",
        "created_at": "",
        "updated_at": "",
    }


def normalize_bom_item_row(payload: object) -> dict | None:
    if not isinstance(payload, dict):
        return None
    normalized = default_bom_item_payload()
    for key in normalized:
        normalized[key] = str(payload.get(key, normalized[key]) or "").strip()
    normalized["item_id"] = normalized["item_id"] or secrets.token_hex(16)
    normalized["style_code"] = normalize_style_identifier(normalized.get("style_code"))
    normalized["source_style_code"] = normalize_style_identifier(normalized.get("source_style_code"))
    if not normalized["bom_version"] or not normalized["item_code"] or not normalized["item_name"]:
        return None
    normalized["item_type"] = normalized["item_type"] or "Other"
    normalized["status"] = normalize_workflow_status(normalized["status"], "draft")
    normalized["uom"] = str(normalized.get("uom", "")).strip().upper()
    normalized["purchase_uom"] = str(normalized.get("purchase_uom", "")).strip().upper()
    normalized["trim_size_unit"] = str(normalized.get("trim_size_unit", "")).strip().upper()
    normalized["length_unit"] = str(normalized.get("length_unit", "")).strip().upper()
    normalized["width_unit"] = str(normalized.get("width_unit", "")).strip().upper()
    normalized["gsm_unit"] = str(normalized.get("gsm_unit", "")).strip().upper()
    normalized["moq_unit"] = str(normalized.get("moq_unit", "")).strip().upper()
    normalized["moq_charge_unit"] = str(normalized.get("moq_charge_unit", "")).strip().upper()
    normalized["mcq_unit"] = str(normalized.get("mcq_unit", "")).strip().upper()
    normalized["mcq_charge_unit"] = str(normalized.get("mcq_charge_unit", "")).strip().upper()
    normalized["lead_time_unit"] = str(normalized.get("lead_time_unit", "")).strip().upper()
    normalized["sample_lead_time_unit"] = str(normalized.get("sample_lead_time_unit", "")).strip().upper()
    normalized["currency"] = str(normalized.get("currency", "")).strip().upper()
    return normalized


def _resolve_bom_header_id(conn: sqlite3.Connection | PostgresCompatConnection, item: dict) -> str:
    explicit_bom_id = str(item.get("bom_id", "")).strip()
    if explicit_bom_id:
        existing = conn.execute(
            "SELECT bom_id FROM bom_headers WHERE bom_id = ?",
            (explicit_bom_id,),
        ).fetchone()
        if existing is not None:
            return explicit_bom_id
    style_code = normalize_style_identifier(item.get("style_code"))
    bom_version = str(item.get("bom_version", "")).strip()
    existing = conn.execute(
        """
        SELECT bom_id, created_at
        FROM bom_headers
        WHERE style_code = ? AND bom_version = ?
        ORDER BY updated_at DESC, created_at DESC
        LIMIT 1
        """,
        (style_code, bom_version),
    ).fetchone()
    normalized_header = normalize_bom_header_row(
        {
            "bom_id": str((existing["bom_id"] if existing is not None else item.get("bom_id")) or "").strip(),
            "style_code": style_code,
            "bom_version": bom_version,
            "bom_name": item.get("bom_name"),
            "season": item.get("season"),
            "collection": item.get("collection"),
            "source_type": item.get("source_type", "manual"),
            "source_style_code": item.get("source_style_code") or style_code,
            "source_note": item.get("source_note"),
            "status": item.get("header_status") or "draft",
            "created_at": str(existing["created_at"] if existing is not None else item.get("created_at") or "").strip(),
        }
    )
    if normalized_header is None:
        raise ValueError("bom_version is required.")
    conn.execute(
        """
        INSERT INTO bom_headers (
            bom_id, style_code, bom_version, bom_name, season, collection, parent_type,
            source_type, source_style_code, source_note, status, is_active, created_at, updated_at
        )
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, COALESCE(?, CURRENT_TIMESTAMP), CURRENT_TIMESTAMP)
        ON CONFLICT(bom_id) DO UPDATE SET
            style_code = excluded.style_code,
            bom_version = excluded.bom_version,
            bom_name = excluded.bom_name,
            season = excluded.season,
            collection = excluded.collection,
            parent_type = excluded.parent_type,
            source_type = excluded.source_type,
            source_style_code = excluded.source_style_code,
            source_note = excluded.source_note,
            status = excluded.status,
            is_active = excluded.is_active,
            created_at = COALESCE(bom_headers.created_at, excluded.created_at),
            updated_at = CURRENT_TIMESTAMP
        """,
        (
            normalized_header["bom_id"],
            normalized_header["style_code"],
            normalized_header["bom_version"],
            normalized_header["bom_name"],
            normalized_header["season"],
            normalized_header["collection"],
            normalized_header["parent_type"],
            normalized_header["source_type"],
            normalized_header["source_style_code"],
            normalized_header["source_note"],
            normalized_header["status"],
            1 if normalized_header["is_active"] else 0,
            normalized_header["created_at"] or None,
        ),
    )
    return str(normalized_header["bom_id"]).strip()


def list_bom_headers(db_path: Path) -> list[dict]:
    with connect(db_path) as conn:
        rows = conn.execute(
            """
            SELECT bom_id, style_code, bom_version, bom_name, season, collection,
                   parent_type, source_type, source_style_code, source_note,
                   status, is_active, created_at, updated_at
            FROM bom_headers
            ORDER BY updated_at DESC, created_at DESC, style_code
            """
        ).fetchall()
    items: list[dict] = []
    for row in rows:
        payload = default_bom_header_payload()
        for key in payload:
            if key == "is_active":
                payload[key] = bool(row[key])
            else:
                payload[key] = str(row[key] or "").strip()
        items.append(payload)
    return items


def list_bom_items(db_path: Path, bom_id: str | None = None) -> list[dict]:
    query = """
        SELECT bi.item_id, bi.bom_id, bi.style_code, bi.item_code, bi.item_name, bi.item_type,
               bi.section, bi.article_ref, bi.color, bi.color_code, bi.description, bi.composition,
               bi.position_of_use, bi.spec_summary, bi.uom, bi.qty_per_garment, bi.trim_size_um,
               bi.trim_size_unit, bi.length_value, bi.length_unit, bi.width_value, bi.width_unit,
               bi.gsm_value, bi.gsm_unit, bi.waste_percent, bi.tech_comment, bi.default_supplier,
               bi.supplier_item_code, bi.moq, bi.moq_unit, bi.moq_charge, bi.moq_charge_unit,
               bi.mcq, bi.mcq_unit, bi.mcq_charge, bi.mcq_charge_unit, bi.lead_time,
               bi.lead_time_unit, bi.sample_lead_time, bi.sample_lead_time_unit,
               bi.purchase_uom, bi.standard_price,
               bi.currency, bi.purchase_note, bi.fds_date, bi.image_path, bi.status, bi.source_style_code,
               bi.source_section, bi.source_section_index, bi.source_item_index, bi.source_payload_json,
               bi.created_at, bi.updated_at, bh.bom_version, bh.bom_name, bh.season, bh.collection
        FROM bom_items bi
        LEFT JOIN bom_headers bh ON bh.bom_id = bi.bom_id
    """
    params: tuple[object, ...] = ()
    if str(bom_id or "").strip():
        query += " WHERE bi.bom_id = ?"
        params = (str(bom_id or "").strip(),)
    query += " ORDER BY bi.updated_at DESC, bi.created_at DESC, bi.item_code"
    with connect(db_path) as conn:
        rows = conn.execute(query, params).fetchall()
    items: list[dict] = []
    for row in rows:
        payload = default_bom_item_payload()
        for key in payload:
            payload[key] = str(row[key] or "").strip()
        items.append(payload)
    return items


def save_bom_item(db_path: Path, row: dict) -> dict | None:
    payload = normalize_bom_item_row(row)
    if payload is None:
        return None
    with connect(db_path) as conn:
        bom_id = _resolve_bom_header_id(conn, payload)
        existing_row = conn.execute(
            "SELECT item_id, created_at FROM bom_items WHERE item_id = ?",
            (payload["item_id"],),
        ).fetchone()
        created_at = str((existing_row["created_at"] if existing_row is not None else payload.get("created_at")) or "").strip() or None
        conn.execute(
            """
            INSERT INTO bom_items (
                item_id, bom_id, style_code, item_code, item_name, item_type, section, article_ref,
                color, color_code, description, composition, position_of_use, spec_summary, uom,
                qty_per_garment, trim_size_um, trim_size_unit, length_value, length_unit,
                width_value, width_unit, gsm_value, gsm_unit, waste_percent, tech_comment,
                default_supplier, supplier_item_code, moq, moq_unit, moq_charge, moq_charge_unit,
                mcq, mcq_unit, mcq_charge, mcq_charge_unit, lead_time, lead_time_unit,
                sample_lead_time, sample_lead_time_unit, purchase_uom, standard_price,
                currency, purchase_note, fds_date, image_path, status, source_style_code, source_section,
                source_section_index, source_item_index, source_payload_json, created_at, updated_at
            )
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, COALESCE(?, CURRENT_TIMESTAMP), CURRENT_TIMESTAMP)
            ON CONFLICT(item_id) DO UPDATE SET
                bom_id = excluded.bom_id,
                style_code = excluded.style_code,
                item_code = excluded.item_code,
                item_name = excluded.item_name,
                item_type = excluded.item_type,
                section = excluded.section,
                article_ref = excluded.article_ref,
                color = excluded.color,
                color_code = excluded.color_code,
                description = excluded.description,
                composition = excluded.composition,
                position_of_use = excluded.position_of_use,
                spec_summary = excluded.spec_summary,
                uom = excluded.uom,
                qty_per_garment = excluded.qty_per_garment,
                trim_size_um = excluded.trim_size_um,
                trim_size_unit = excluded.trim_size_unit,
                length_value = excluded.length_value,
                length_unit = excluded.length_unit,
                width_value = excluded.width_value,
                width_unit = excluded.width_unit,
                gsm_value = excluded.gsm_value,
                gsm_unit = excluded.gsm_unit,
                waste_percent = excluded.waste_percent,
                tech_comment = excluded.tech_comment,
                default_supplier = excluded.default_supplier,
                supplier_item_code = excluded.supplier_item_code,
                moq = excluded.moq,
                moq_unit = excluded.moq_unit,
                moq_charge = excluded.moq_charge,
                moq_charge_unit = excluded.moq_charge_unit,
                mcq = excluded.mcq,
                mcq_unit = excluded.mcq_unit,
                mcq_charge = excluded.mcq_charge,
                mcq_charge_unit = excluded.mcq_charge_unit,
                lead_time = excluded.lead_time,
                lead_time_unit = excluded.lead_time_unit,
                sample_lead_time = excluded.sample_lead_time,
                sample_lead_time_unit = excluded.sample_lead_time_unit,
                purchase_uom = excluded.purchase_uom,
                standard_price = excluded.standard_price,
                currency = excluded.currency,
                purchase_note = excluded.purchase_note,
                fds_date = excluded.fds_date,
                image_path = excluded.image_path,
                status = excluded.status,
                source_style_code = excluded.source_style_code,
                source_section = excluded.source_section,
                source_section_index = excluded.source_section_index,
                source_item_index = excluded.source_item_index,
                source_payload_json = excluded.source_payload_json,
                created_at = COALESCE(bom_items.created_at, excluded.created_at),
                updated_at = CURRENT_TIMESTAMP
            """,
            (
                payload["item_id"],
                bom_id,
                payload["style_code"],
                payload["item_code"],
                payload["item_name"],
                payload["item_type"],
                payload["section"],
                payload["article_ref"],
                payload["color"],
                payload["color_code"],
                payload["description"],
                payload["composition"],
                payload["position_of_use"],
                payload["spec_summary"],
                payload["uom"],
                payload["qty_per_garment"],
                payload["trim_size_um"],
                payload["trim_size_unit"],
                payload["length_value"],
                payload["length_unit"],
                payload["width_value"],
                payload["width_unit"],
                payload["gsm_value"],
                payload["gsm_unit"],
                payload["waste_percent"],
                payload["tech_comment"],
                payload["default_supplier"],
                payload["supplier_item_code"],
                payload["moq"],
                payload["moq_unit"],
                payload["moq_charge"],
                payload["moq_charge_unit"],
                payload["mcq"],
                payload["mcq_unit"],
                payload["mcq_charge"],
                payload["mcq_charge_unit"],
                payload["lead_time"],
                payload["lead_time_unit"],
                payload["sample_lead_time"],
                payload["sample_lead_time_unit"],
                payload["purchase_uom"],
                payload["standard_price"],
                payload["currency"],
                payload["purchase_note"],
                payload["fds_date"],
                payload["image_path"],
                payload["status"],
                payload["source_style_code"],
                payload["source_section"],
                payload["source_section_index"],
                payload["source_item_index"],
                payload["source_payload_json"],
                created_at,
            ),
        )
    saved_rows = [item for item in list_bom_items(db_path) if str(item.get("item_id", "")).strip() == payload["item_id"]]
    return saved_rows[0] if saved_rows else None


def delete_bom_item(db_path: Path, item_id: str) -> int:
    normalized_item_id = str(item_id or "").strip()
    if not normalized_item_id:
        return 0
    with connect(db_path) as conn:
        row = conn.execute("SELECT bom_id FROM bom_items WHERE item_id = ?", (normalized_item_id,)).fetchone()
        cursor = conn.execute("DELETE FROM bom_items WHERE item_id = ?", (normalized_item_id,))
        deleted_count = int(getattr(cursor, "rowcount", 0) or 0)
        if row is not None:
            bom_id = str(row["bom_id"] or "").strip()
            if bom_id:
                remaining = conn.execute("SELECT 1 FROM bom_items WHERE bom_id = ? LIMIT 1", (bom_id,)).fetchone()
                if remaining is None:
                    conn.execute("DELETE FROM bom_headers WHERE bom_id = ?", (bom_id,))
    return deleted_count


def default_supplier_payload() -> dict:
    return {
        "supplier_id": "",
        "supplier_code": "",
        "supplier_name": "",
        "short_name": "",
        "supplier_type": "",
        "status": "Active",
        "contact_person": "",
        "phone": "",
        "email": "",
        "country": "",
        "city": "",
        "address_line": "",
        "tax_code": "",
        "payment_term": "",
        "currency": "",
        "note": "",
        "created_at": "",
        "updated_at": "",
    }


def normalize_supplier_row(payload: object) -> dict | None:
    if not isinstance(payload, dict):
        return None
    normalized = default_supplier_payload()
    for key in normalized:
        normalized[key] = str(payload.get(key, normalized[key]) or "").strip()
    normalized["supplier_id"] = normalized["supplier_id"] or secrets.token_hex(16)
    normalized["supplier_code"] = normalized["supplier_code"].upper()
    normalized["status"] = normalized["status"] or "Active"
    if not normalized["supplier_code"] or not normalized["supplier_name"]:
        return None
    return normalized


def list_suppliers(db_path: Path) -> list[dict]:
    with connect(db_path) as conn:
        rows = conn.execute(
            """
            SELECT supplier_id, supplier_code, supplier_name, short_name, supplier_type,
                   status, contact_person, phone, email, country, city, address_line,
                   tax_code, payment_term, currency, note, created_at, updated_at
            FROM suppliers
            ORDER BY updated_at DESC, created_at DESC, supplier_code
            """
        ).fetchall()
    items: list[dict] = []
    for row in rows:
        payload = default_supplier_payload()
        for key in payload:
            payload[key] = str(row[key] or "").strip()
        items.append(payload)
    return items


def save_supplier(db_path: Path, row: dict) -> dict | None:
    payload = normalize_supplier_row(row)
    if payload is None:
        return None
    with connect(db_path) as conn:
        existing_row = conn.execute(
            "SELECT supplier_id, created_at FROM suppliers WHERE supplier_id = ?",
            (payload["supplier_id"],),
        ).fetchone()
        created_at = str((existing_row["created_at"] if existing_row is not None else payload.get("created_at")) or "").strip() or None
        conn.execute(
            """
            INSERT INTO suppliers (
                supplier_id, supplier_code, supplier_name, short_name, supplier_type, status,
                contact_person, phone, email, country, city, address_line, tax_code,
                payment_term, currency, note, created_at, updated_at
            )
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, COALESCE(?, CURRENT_TIMESTAMP), CURRENT_TIMESTAMP)
            ON CONFLICT(supplier_id) DO UPDATE SET
                supplier_code = excluded.supplier_code,
                supplier_name = excluded.supplier_name,
                short_name = excluded.short_name,
                supplier_type = excluded.supplier_type,
                status = excluded.status,
                contact_person = excluded.contact_person,
                phone = excluded.phone,
                email = excluded.email,
                country = excluded.country,
                city = excluded.city,
                address_line = excluded.address_line,
                tax_code = excluded.tax_code,
                payment_term = excluded.payment_term,
                currency = excluded.currency,
                note = excluded.note,
                created_at = COALESCE(suppliers.created_at, excluded.created_at),
                updated_at = CURRENT_TIMESTAMP
            """,
            (
                payload["supplier_id"],
                payload["supplier_code"],
                payload["supplier_name"],
                payload["short_name"],
                payload["supplier_type"],
                payload["status"],
                payload["contact_person"],
                payload["phone"],
                payload["email"],
                payload["country"],
                payload["city"],
                payload["address_line"],
                payload["tax_code"],
                payload["payment_term"],
                payload["currency"],
                payload["note"],
                created_at,
            ),
        )
    saved_rows = [item for item in list_suppliers(db_path) if str(item.get("supplier_id", "")).strip() == payload["supplier_id"]]
    return saved_rows[0] if saved_rows else None


def delete_supplier(db_path: Path, supplier_id: str) -> None:
    normalized_supplier_id = str(supplier_id or "").strip()
    if not normalized_supplier_id:
        return
    with connect(db_path) as conn:
        conn.execute("DELETE FROM suppliers WHERE supplier_id = ?", (normalized_supplier_id,))


def default_hr_employee_payload() -> dict:
    return {
        "employee_id": "",
        "user_email": "",
        "full_name": "",
        "display_name": "",
        "department": "",
        "position": "",
        "phone": "",
        "email": "",
        "join_date": "",
        "shift_name": "Office",
        "shift_start": "08:00",
        "shift_end": "17:00",
        "break_time_out": "12:00",
        "break_time_in": "13:00",
        "grace_minutes": 5,
        "is_active": True,
        "note": "",
        "created_at": "",
        "updated_at": "",
    }


def normalize_hr_employee_rows(payload: object) -> list[dict]:
    rows = payload if isinstance(payload, list) else []
    normalized_rows: list[dict] = []
    seen_ids: set[str] = set()
    for item in rows:
        if not isinstance(item, dict):
            continue
        normalized = default_hr_employee_payload()
        normalized.update(
            {
                "employee_id": str(item.get("employee_id", "")).strip(),
                "user_email": normalize_email(item.get("user_email", "")),
                "full_name": str(item.get("full_name", "")).strip(),
                "display_name": str(item.get("display_name", item.get("nick_name", ""))).strip(),
                "department": str(item.get("department", "")).strip(),
                "position": str(item.get("position", "")).strip(),
                "phone": str(item.get("phone", "")).strip(),
                "email": normalize_email(item.get("email", "")),
                "join_date": str(item.get("join_date", "")).strip(),
                "shift_name": str(item.get("shift_name", "Office") or "Office").strip(),
                "shift_start": str(item.get("shift_start", "08:00") or "08:00").strip(),
                "shift_end": str(item.get("shift_end", "17:00") or "17:00").strip(),
                "break_time_out": str(item.get("break_time_out", "12:00") or "12:00").strip(),
                "break_time_in": str(item.get("break_time_in", "13:00") or "13:00").strip(),
                "grace_minutes": int(item.get("grace_minutes", 5) or 0),
                "is_active": bool(item.get("is_active", True)),
                "note": str(item.get("note", "")).strip(),
                "created_at": str(item.get("created_at", "")).strip(),
                "updated_at": str(item.get("updated_at", "")).strip(),
            }
        )
        employee_id = normalized["employee_id"]
        if not employee_id or employee_id in seen_ids:
            continue
        seen_ids.add(employee_id)
        normalized_rows.append(normalized)
    normalized_rows.sort(key=lambda row: (str(row.get("full_name", "")).lower(), str(row.get("employee_id", "")).lower()))
    return normalized_rows


def default_hr_attendance_record_payload() -> dict:
    return {
        "attendance_id": "",
        "employee_id": "",
        "user_email": "",
        "employee_name": "",
        "department": "",
        "position": "",
        "work_date": "",
        "check_in_at": "",
        "check_out_at": "",
        "shift_name": "Office",
        "shift_start": "08:00",
        "shift_end": "17:00",
        "break_time_out": "12:00",
        "break_time_in": "13:00",
        "break_out_at": "",
        "break_in_at": "",
        "grace_minutes": 5,
        "late_minutes": 0,
        "early_leave_minutes": 0,
        "worked_minutes": 0,
        "working_day": 0.0,
        "status": "Absent",
        "note": "",
        "manual_adjusted": False,
        "created_by": "",
        "updated_by": "",
        "created_at": "",
        "updated_at": "",
    }


def normalize_hr_attendance_rows(payload: object) -> list[dict]:
    rows = payload if isinstance(payload, list) else []
    normalized_rows: list[dict] = []
    seen_ids: set[str] = set()
    for item in rows:
        if not isinstance(item, dict):
            continue
        normalized = default_hr_attendance_record_payload()
        normalized.update(
            {
                "attendance_id": str(item.get("attendance_id", "")).strip(),
                "employee_id": str(item.get("employee_id", "")).strip(),
                "user_email": normalize_email(item.get("user_email", "")),
                "employee_name": str(item.get("employee_name", "")).strip(),
                "department": str(item.get("department", "")).strip(),
                "position": str(item.get("position", "")).strip(),
                "work_date": str(item.get("work_date", "")).strip(),
                "check_in_at": str(item.get("check_in_at", "")).strip(),
                "check_out_at": str(item.get("check_out_at", "")).strip(),
                "shift_name": str(item.get("shift_name", "Office") or "Office").strip(),
                "shift_start": str(item.get("shift_start", "08:00") or "08:00").strip(),
                "shift_end": str(item.get("shift_end", "17:00") or "17:00").strip(),
                "break_time_out": str(item.get("break_time_out", "12:00") or "12:00").strip(),
                "break_time_in": str(item.get("break_time_in", "13:00") or "13:00").strip(),
                "break_out_at": str(item.get("break_out_at", "")).strip(),
                "break_in_at": str(item.get("break_in_at", "")).strip(),
                "grace_minutes": int(item.get("grace_minutes", 5) or 0),
                "late_minutes": int(item.get("late_minutes", 0) or 0),
                "early_leave_minutes": int(item.get("early_leave_minutes", 0) or 0),
                "worked_minutes": int(item.get("worked_minutes", 0) or 0),
                "working_day": float(item.get("working_day", 0) or 0),
                "status": str(item.get("status", "Absent") or "Absent").strip(),
                "note": str(item.get("note", "")).strip(),
                "manual_adjusted": bool(item.get("manual_adjusted", False)),
                "created_by": normalize_email(item.get("created_by", "")) or str(item.get("created_by", "")).strip(),
                "updated_by": normalize_email(item.get("updated_by", "")) or str(item.get("updated_by", "")).strip(),
                "created_at": str(item.get("created_at", "")).strip(),
                "updated_at": str(item.get("updated_at", "")).strip(),
            }
        )
        attendance_id = normalized["attendance_id"]
        if not attendance_id or attendance_id in seen_ids:
            continue
        seen_ids.add(attendance_id)
        normalized_rows.append(normalized)
    normalized_rows.sort(
        key=lambda row: (
            str(row.get("work_date", "")),
            str(row.get("employee_name", "")).lower(),
            str(row.get("employee_id", "")).lower(),
        ),
        reverse=True,
    )
    return normalized_rows


def _parse_iso_datetime(value: object) -> datetime | None:
    text = str(value or "").strip()
    if not text:
        return None
    for pattern in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"):
        try:
            return datetime.strptime(text, pattern)
        except Exception:
            continue
    return None


def _parse_time_hhmm(value: object) -> tuple[int, int]:
    text = str(value or "").strip()
    matched = re.fullmatch(r"(\d{1,2}):(\d{2})", text)
    if not matched:
        return (8, 0)
    return (int(matched.group(1)), int(matched.group(2)))


def _build_shift_datetime(work_date: str, shift_time: object) -> datetime | None:
    base_date = str(work_date or "").strip()
    if not base_date:
        return None
    try:
        date_value = datetime.strptime(base_date, "%Y-%m-%d")
    except Exception:
        return None
    hour, minute = _parse_time_hhmm(shift_time)
    return date_value.replace(hour=hour, minute=minute, second=0, microsecond=0)


def _calculate_attendance_metrics(record: dict) -> dict:
    normalized = dict(record)
    work_date = str(normalized.get("work_date", "")).strip()
    shift_start_dt = _build_shift_datetime(work_date, normalized.get("shift_start", "08:00"))
    shift_end_dt = _build_shift_datetime(work_date, normalized.get("shift_end", "17:00"))
    break_time_out_dt = _build_shift_datetime(work_date, normalized.get("break_time_out", "12:00"))
    break_time_in_dt = _build_shift_datetime(work_date, normalized.get("break_time_in", "13:00"))
    check_in_dt = _parse_iso_datetime(normalized.get("check_in_at"))
    check_out_dt = _parse_iso_datetime(normalized.get("check_out_at"))
    break_out_dt = _parse_iso_datetime(normalized.get("break_out_at"))
    break_in_dt = _parse_iso_datetime(normalized.get("break_in_at"))
    late_minutes = 0
    early_leave_minutes = 0
    worked_minutes = 0
    break_minutes = 0
    if check_in_dt and shift_start_dt:
        delta_minutes = int((check_in_dt - shift_start_dt).total_seconds() // 60)
        late_minutes = max(0, delta_minutes)
    if check_in_dt and check_out_dt:
        worked_minutes = max(0, int((check_out_dt - check_in_dt).total_seconds() // 60))
    if break_out_dt and break_in_dt:
        break_minutes = max(0, int((break_in_dt - break_out_dt).total_seconds() // 60))
    elif break_time_out_dt and break_time_in_dt and check_in_dt and check_out_dt:
        if check_in_dt <= break_time_out_dt and check_out_dt >= break_time_in_dt:
            break_minutes = max(0, int((break_time_in_dt - break_time_out_dt).total_seconds() // 60))
    worked_minutes = max(0, worked_minutes - break_minutes)
    if check_out_dt and shift_end_dt:
        early_leave_minutes = max(0, int((shift_end_dt - check_out_dt).total_seconds() // 60))
    normalized["late_minutes"] = late_minutes
    normalized["early_leave_minutes"] = early_leave_minutes
    normalized["worked_minutes"] = worked_minutes
    normalized["working_day"] = 1.0 if worked_minutes >= 480 else 0.5 if worked_minutes >= 240 else 0.0
    if check_in_dt and check_out_dt:
        normalized["status"] = "Present"
    elif check_in_dt:
        normalized["status"] = "Checked in"
    else:
        normalized["status"] = str(normalized.get("status", "Absent") or "Absent").strip() or "Absent"
    return normalized


def list_hr_employees(db_path: Path) -> list[dict]:
    with connect(db_path) as conn:
        rows = conn.execute(
            """
            SELECT employee_id, user_email, full_name, employee_json, created_at, updated_at
            FROM hr_employees
            ORDER BY full_name, employee_id
            """
        ).fetchall()
    employees: list[dict] = []
    for row in rows:
        try:
            payload = json.loads(row["employee_json"] or "{}")
        except Exception:
            payload = {}
        if not isinstance(payload, dict):
            payload = {}
        payload["employee_id"] = str(row["employee_id"] or payload.get("employee_id") or "").strip()
        payload["user_email"] = normalize_email(row["user_email"] or payload.get("user_email") or "")
        payload["full_name"] = str(row["full_name"] or payload.get("full_name") or "").strip()
        payload["created_at"] = str(row["created_at"] or payload.get("created_at") or "").strip()
        payload["updated_at"] = str(row["updated_at"] or payload.get("updated_at") or "").strip()
        employees.append(payload)
    return normalize_hr_employee_rows(employees)


def save_hr_employees(db_path: Path, rows: list[dict]) -> list[dict]:
    normalized_rows = normalize_hr_employee_rows(rows)
    rows_by_id = {str(item.get("employee_id", "")).strip(): item for item in normalized_rows if str(item.get("employee_id", "")).strip()}
    with connect(db_path) as conn:
        conn.execute("UPDATE hr_employees SET user_email = NULL WHERE TRIM(COALESCE(user_email, '')) = ''")
        existing_rows = {
            str(row["employee_id"]).strip(): row
            for row in conn.execute("SELECT employee_id, created_at FROM hr_employees").fetchall()
        }
        stale_ids = set(existing_rows) - set(rows_by_id)
        for employee_id in stale_ids:
            conn.execute("DELETE FROM hr_employees WHERE employee_id = ?", (employee_id,))
        for employee_id, payload in rows_by_id.items():
            existing_row = existing_rows.get(employee_id)
            created_at = str((existing_row["created_at"] if existing_row is not None else payload.get("created_at")) or "").strip() or None
            conn.execute(
                """
                INSERT INTO hr_employees (employee_id, user_email, full_name, employee_json, created_at, updated_at)
                VALUES (?, ?, ?, ?, COALESCE(?, CURRENT_TIMESTAMP), CURRENT_TIMESTAMP)
                ON CONFLICT(employee_id) DO UPDATE SET
                    user_email = excluded.user_email,
                    full_name = excluded.full_name,
                    employee_json = excluded.employee_json,
                    created_at = COALESCE(hr_employees.created_at, excluded.created_at),
                    updated_at = CURRENT_TIMESTAMP
                """,
                (
                    employee_id,
                    normalize_email(payload.get("user_email", "")) or None,
                    str(payload.get("full_name", "")).strip(),
                    json.dumps(payload, ensure_ascii=False),
                    created_at,
                ),
            )
    return list_hr_employees(db_path)


def save_hr_employee(db_path: Path, row: dict) -> dict | None:
    normalized_rows = normalize_hr_employee_rows([row])
    if not normalized_rows:
        return None
    payload = normalized_rows[0]
    employee_id = str(payload.get("employee_id", "")).strip()
    if not employee_id:
        return None
    with connect(db_path) as conn:
        conn.execute("UPDATE hr_employees SET user_email = NULL WHERE TRIM(COALESCE(user_email, '')) = ''")
        existing_row = conn.execute(
            "SELECT employee_id, created_at FROM hr_employees WHERE employee_id = ?",
            (employee_id,),
        ).fetchone()
        created_at = str((existing_row["created_at"] if existing_row is not None else payload.get("created_at")) or "").strip() or None
        conn.execute(
            """
            INSERT INTO hr_employees (employee_id, user_email, full_name, employee_json, created_at, updated_at)
            VALUES (?, ?, ?, ?, COALESCE(?, CURRENT_TIMESTAMP), CURRENT_TIMESTAMP)
            ON CONFLICT(employee_id) DO UPDATE SET
                user_email = excluded.user_email,
                full_name = excluded.full_name,
                employee_json = excluded.employee_json,
                created_at = COALESCE(hr_employees.created_at, excluded.created_at),
                updated_at = CURRENT_TIMESTAMP
            """,
            (
                employee_id,
                normalize_email(payload.get("user_email", "")) or None,
                str(payload.get("full_name", "")).strip(),
                json.dumps(payload, ensure_ascii=False),
                created_at,
            ),
        )
    saved_rows = [row for row in list_hr_employees(db_path) if str(row.get("employee_id", "")).strip() == employee_id]
    return saved_rows[0] if saved_rows else None


def list_hr_attendance_records(db_path: Path) -> list[dict]:
    with connect(db_path) as conn:
        rows = conn.execute(
            """
            SELECT attendance_id, employee_id, user_email, work_date, record_json, created_at, updated_at
            FROM hr_attendance_records
            ORDER BY work_date DESC, employee_id ASC, updated_at DESC
            """
        ).fetchall()
    records: list[dict] = []
    for row in rows:
        try:
            payload = json.loads(row["record_json"] or "{}")
        except Exception:
            payload = {}
        if not isinstance(payload, dict):
            payload = {}
        payload["attendance_id"] = str(row["attendance_id"] or payload.get("attendance_id") or "").strip()
        payload["employee_id"] = str(row["employee_id"] or payload.get("employee_id") or "").strip()
        payload["user_email"] = normalize_email(row["user_email"] or payload.get("user_email") or "")
        payload["work_date"] = str(row["work_date"] or payload.get("work_date") or "").strip()
        payload["created_at"] = str(row["created_at"] or payload.get("created_at") or "").strip()
        payload["updated_at"] = str(row["updated_at"] or payload.get("updated_at") or "").strip()
        records.append(payload)
    return normalize_hr_attendance_rows(records)


def save_hr_attendance_records(db_path: Path, rows: list[dict]) -> list[dict]:
    normalized_rows = [_calculate_attendance_metrics(item) for item in normalize_hr_attendance_rows(rows)]
    rows_by_id = {str(item.get("attendance_id", "")).strip(): item for item in normalized_rows if str(item.get("attendance_id", "")).strip()}
    with connect(db_path) as conn:
        existing_rows = {
            str(row["attendance_id"]).strip(): row
            for row in conn.execute("SELECT attendance_id, created_at FROM hr_attendance_records").fetchall()
        }
        stale_ids = set(existing_rows) - set(rows_by_id)
        for attendance_id in stale_ids:
            conn.execute("DELETE FROM hr_attendance_records WHERE attendance_id = ?", (attendance_id,))
        for attendance_id, payload in rows_by_id.items():
            existing_row = existing_rows.get(attendance_id)
            created_at = str((existing_row["created_at"] if existing_row is not None else payload.get("created_at")) or "").strip() or None
            conn.execute(
                """
                INSERT INTO hr_attendance_records (
                    attendance_id, employee_id, user_email, work_date, record_json, created_at, updated_at
                )
                VALUES (?, ?, ?, ?, ?, COALESCE(?, CURRENT_TIMESTAMP), CURRENT_TIMESTAMP)
                ON CONFLICT(attendance_id) DO UPDATE SET
                    employee_id = excluded.employee_id,
                    user_email = excluded.user_email,
                    work_date = excluded.work_date,
                    record_json = excluded.record_json,
                    created_at = COALESCE(hr_attendance_records.created_at, excluded.created_at),
                    updated_at = CURRENT_TIMESTAMP
                """,
                (
                    attendance_id,
                    str(payload.get("employee_id", "")).strip(),
                    normalize_email(payload.get("user_email", "")),
                    str(payload.get("work_date", "")).strip(),
                    json.dumps(payload, ensure_ascii=False),
                    created_at,
                ),
            )
    return list_hr_attendance_records(db_path)


def save_hr_attendance_record(db_path: Path, row: dict) -> dict | None:
    normalized_rows = normalize_hr_attendance_rows([row])
    if not normalized_rows:
        return None
    payload = _calculate_attendance_metrics(normalized_rows[0])
    attendance_id = str(payload.get("attendance_id", "")).strip()
    if not attendance_id:
        return None
    with connect(db_path) as conn:
        existing_row = conn.execute(
            "SELECT attendance_id, created_at FROM hr_attendance_records WHERE attendance_id = ?",
            (attendance_id,),
        ).fetchone()
        created_at = str((existing_row["created_at"] if existing_row is not None else payload.get("created_at")) or "").strip() or None
        conn.execute(
            """
            INSERT INTO hr_attendance_records (
                attendance_id, employee_id, user_email, work_date, record_json, created_at, updated_at
            )
            VALUES (?, ?, ?, ?, ?, COALESCE(?, CURRENT_TIMESTAMP), CURRENT_TIMESTAMP)
            ON CONFLICT(attendance_id) DO UPDATE SET
                employee_id = excluded.employee_id,
                user_email = excluded.user_email,
                work_date = excluded.work_date,
                record_json = excluded.record_json,
                created_at = COALESCE(hr_attendance_records.created_at, excluded.created_at),
                updated_at = CURRENT_TIMESTAMP
            """,
            (
                attendance_id,
                str(payload.get("employee_id", "")).strip(),
                normalize_email(payload.get("user_email", "")),
                str(payload.get("work_date", "")).strip(),
                json.dumps(payload, ensure_ascii=False),
                created_at,
            ),
        )
    saved_rows = [item for item in list_hr_attendance_records(db_path) if str(item.get("attendance_id", "")).strip() == attendance_id]
    return saved_rows[0] if saved_rows else None


def _build_today_attendance_record(employee_row: dict, *, actor_email: str, existing_row: dict | None = None) -> dict:
    now_text = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    work_date = datetime.now().strftime("%Y-%m-%d")
    payload = default_hr_attendance_record_payload()
    if isinstance(existing_row, dict):
        payload.update(existing_row)
    payload.update(
        {
            "attendance_id": str(payload.get("attendance_id", "")).strip() or secrets.token_hex(16),
            "employee_id": str(employee_row.get("employee_id", "")).strip(),
            "user_email": normalize_email(employee_row.get("user_email", "")),
            "employee_name": str(employee_row.get("full_name", "")).strip(),
            "department": str(employee_row.get("department", "")).strip(),
            "position": str(employee_row.get("position", "")).strip(),
            "work_date": work_date,
            "shift_name": str(employee_row.get("shift_name", "Office") or "Office").strip(),
            "shift_start": str(employee_row.get("shift_start", "08:00") or "08:00").strip(),
            "shift_end": str(employee_row.get("shift_end", "17:00") or "17:00").strip(),
            "break_time_out": str(employee_row.get("break_time_out", "12:00") or "12:00").strip(),
            "break_time_in": str(employee_row.get("break_time_in", "13:00") or "13:00").strip(),
            "grace_minutes": int(employee_row.get("grace_minutes", 5) or 0),
            "updated_by": normalize_email(actor_email) or str(actor_email or "").strip(),
            "updated_at": now_text,
        }
    )
    if not str(payload.get("created_by", "")).strip():
        payload["created_by"] = normalize_email(actor_email) or str(actor_email or "").strip()
    if not str(payload.get("created_at", "")).strip():
        payload["created_at"] = now_text
    return payload


def hr_attendance_check_in(db_path: Path, employee_row: dict, *, actor_email: str = "") -> dict:
    employee_id = str(employee_row.get("employee_id", "")).strip()
    if not employee_id:
        raise ValueError("employee_id is required.")
    work_date = datetime.now().strftime("%Y-%m-%d")
    existing = next(
        (
            row
            for row in list_hr_attendance_records(db_path)
            if str(row.get("employee_id", "")).strip() == employee_id and str(row.get("work_date", "")).strip() == work_date
        ),
        None,
    )
    record = _build_today_attendance_record(employee_row, actor_email=actor_email, existing_row=existing)
    if str(record.get("check_in_at", "")).strip():
        raise ValueError("This employee has already checked in today.")
    record["check_in_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    saved = save_hr_attendance_record(db_path, record)
    if not isinstance(saved, dict):
        raise ValueError("Could not save attendance record.")
    return saved


def hr_attendance_check_out(db_path: Path, employee_row: dict, *, actor_email: str = "") -> dict:
    employee_id = str(employee_row.get("employee_id", "")).strip()
    if not employee_id:
        raise ValueError("employee_id is required.")
    work_date = datetime.now().strftime("%Y-%m-%d")
    existing = next(
        (
            row
            for row in list_hr_attendance_records(db_path)
            if str(row.get("employee_id", "")).strip() == employee_id and str(row.get("work_date", "")).strip() == work_date
        ),
        None,
    )
    if not isinstance(existing, dict):
        raise ValueError("This employee has not checked in today.")
    if not str(existing.get("check_in_at", "")).strip():
        raise ValueError("This employee has not checked in today.")
    if str(existing.get("check_out_at", "")).strip():
        raise ValueError("This employee has already checked out today.")
    record = _build_today_attendance_record(employee_row, actor_email=actor_email, existing_row=existing)
    record["check_out_at"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    saved = save_hr_attendance_record(db_path, record)
    if not isinstance(saved, dict):
        raise ValueError("Could not save attendance record.")
    return saved


def list_tech_pack_hidden_styles(db_path: Path) -> list[str]:
    with connect(db_path) as conn:
        ensure_tech_pack_storage_migrated(conn)
        rows = conn.execute(
            """
            SELECT style_code
            FROM tech_pack_hidden_styles
            ORDER BY style_code
            """
        ).fetchall()
    return [str(row["style_code"]).strip() for row in rows if str(row["style_code"]).strip()]


def save_tech_pack_hidden_styles(db_path: Path, style_codes: list[object]) -> list[str]:
    normalized = sorted(
        {
            style_code
            for style_code in (normalize_style_identifier(item) for item in style_codes)
            if style_code
        }
    )
    with connect(db_path) as conn:
        ensure_tech_pack_storage_migrated(conn)
        conn.execute("DELETE FROM tech_pack_hidden_styles")
        for style_code in normalized:
            conn.execute(
                """
                INSERT INTO tech_pack_hidden_styles (style_code, updated_at)
                VALUES (?, CURRENT_TIMESTAMP)
                """,
                (style_code,),
            )
    return normalized


def update_user_active_status(db_path: Path, email: str, is_active: bool) -> None:
    normalized_email = normalize_email(email)
    if not normalized_email:
        raise ValueError("Email is required.")
    with connect(db_path) as conn:
        existing = conn.execute("SELECT id FROM users WHERE email = ?", (normalized_email,)).fetchone()
        if existing is None:
            raise ValueError("Account not found.")
        conn.execute(
            """
            UPDATE users
            SET is_active = ?, updated_at = CURRENT_TIMESTAMP
            WHERE email = ?
            """,
            (1 if bool(is_active) else 0, normalized_email),
        )


def update_user_approval_status(
    db_path: Path,
    email: str,
    approval_status: str,
    *,
    approved_by: str = "",
) -> None:
    normalized_email = normalize_email(email)
    normalized_status = str(approval_status or "pending").strip().lower() or "pending"
    approver = normalize_email(approved_by)
    if not normalized_email:
        raise ValueError("Email is required.")
    if normalized_status not in {"pending", "approved", "rejected"}:
        raise ValueError("Invalid approval status.")
    with connect(db_path) as conn:
        existing = conn.execute("SELECT id FROM users WHERE email = ?", (normalized_email,)).fetchone()
        if existing is None:
            raise ValueError("Account not found.")
        conn.execute(
            """
            UPDATE users
            SET approval_status = ?,
                approved_by = ?,
                approved_at = CASE WHEN ? = 'approved' THEN CURRENT_TIMESTAMP ELSE approved_at END,
                is_active = CASE
                    WHEN ? = 'approved' THEN 1
                    WHEN ? = 'pending' THEN 0
                    WHEN ? = 'rejected' THEN 0
                    ELSE is_active
                END,
                updated_at = CURRENT_TIMESTAMP
            WHERE email = ?
            """,
            (
                normalized_status,
                approver,
                normalized_status,
                normalized_status,
                normalized_status,
                normalized_status,
                normalized_email,
            ),
        )


def create_password_reset_token(db_path: Path, email: str, expires_at_iso: str) -> str:
    normalized_email = normalize_email(email)
    if not normalized_email:
        raise ValueError("Email is required.")
    token = secrets.token_urlsafe(32)
    with connect(db_path) as conn:
        existing = conn.execute(
            "SELECT id FROM users WHERE email = ? AND is_active = 1 AND approval_status = 'approved'",
            (normalized_email,),
        ).fetchone()
        if existing is None:
            raise ValueError("Account not found.")
        conn.execute("DELETE FROM password_reset_tokens WHERE email = ?", (normalized_email,))
        conn.execute(
            """
            INSERT INTO password_reset_tokens (token, email, expires_at, used_at)
            VALUES (?, ?, ?, NULL)
            """,
            (token, normalized_email, expires_at_iso),
        )
    return token


def get_password_reset_token(db_path: Path, token: str) -> dict | None:
    token_value = str(token or "").strip()
    if not token_value:
        return None
    with connect(db_path) as conn:
        row = conn.execute(
            """
            SELECT token, email, expires_at, used_at, created_at
            FROM password_reset_tokens
            WHERE token = ?
            """,
            (token_value,),
        ).fetchone()
    return dict(row) if row is not None else None


def update_user_password_with_token(db_path: Path, token: str, new_password: str) -> None:
    token_value = str(token or "").strip()
    if not token_value:
        raise ValueError("Reset token is required.")
    if not new_password:
        raise ValueError("New password is required.")
    with connect(db_path) as conn:
        row = conn.execute(
            """
            SELECT token, email, expires_at, used_at
            FROM password_reset_tokens
            WHERE token = ?
            """,
            (token_value,),
        ).fetchone()
        if row is None:
            raise ValueError("Reset token is invalid.")
        if str(row["used_at"] or "").strip():
            raise ValueError("Reset token was already used.")
        expires_at = str(row["expires_at"] or "").strip()
        now_row = conn.execute("SELECT CURRENT_TIMESTAMP AS now_utc").fetchone()
        now_utc = str(now_row["now_utc"] or "").strip() if now_row is not None else ""
        if expires_at and now_utc and expires_at < now_utc:
            raise ValueError("Reset token has expired.")
        normalized_email = normalize_email(row["email"])
        conn.execute(
            """
            UPDATE users
            SET password_hash = ?, updated_at = CURRENT_TIMESTAMP
            WHERE email = ?
            """,
            (hash_password(new_password), normalized_email),
        )
        conn.execute(
            """
            UPDATE password_reset_tokens
            SET used_at = CURRENT_TIMESTAMP
            WHERE token = ?
            """,
            (token_value,),
        )


def delete_users(db_path: Path, emails: list[str]) -> None:
    normalized_emails = [normalize_email(email) for email in emails if normalize_email(email)]
    if not normalized_emails:
        return
    placeholders = ", ".join("?" for _ in normalized_emails)
    with connect(db_path) as conn:
        conn.execute(
            f"DELETE FROM permission_user_overrides WHERE email IN ({placeholders})",
            tuple(normalized_emails),
        )
        conn.execute(
            f"DELETE FROM users WHERE email IN ({placeholders})",
            tuple(normalized_emails),
        )


def list_users(db_path: Path) -> list[dict]:
    with connect(db_path) as conn:
        rows = conn.execute(
            """
            SELECT id, email, display_name, role, is_active, approval_status, approved_by, approved_at, created_at, updated_at
            FROM users
            ORDER BY
                CASE approval_status
                    WHEN 'pending' THEN 0
                    WHEN 'approved' THEN 1
                    ELSE 2
                END,
                CASE role
                    WHEN 'admin' THEN 0
                    WHEN 'moderate' THEN 1
                    ELSE 2
                END,
                email
            """
        ).fetchall()
    return [dict(row) for row in rows]


def list_orders(db_path: Path) -> list[sqlite3.Row]:
    style_summary_sql = purchase_order_style_summary_sql()
    with connect(db_path) as conn:
        rows = conn.execute(
            f"""
            SELECT
                po.id,
                po.po_number,
                po.po_date,
                po.po_time,
                po.simple_domestic_pt,
                {style_summary_sql} AS style,
                po.vendor_name,
                po.order_date,
                po.total_cost,
                po.total_qty,
                po.source_pdf,
                po.created_at,
                po.updated_at
            FROM purchase_orders po
            ORDER BY created_at DESC, id DESC
            """
        ).fetchall()
    return list(rows)


def list_orders_for_web(db_path: Path, search: str = "", limit: int = 100) -> list[sqlite3.Row]:
    normalized_search = f"%{search.strip()}%"
    safe_limit = max(1, min(int(limit), 5000))
    style_summary_sql = purchase_order_style_summary_sql()
    with connect(db_path) as conn:
        rows = conn.execute(
            f"""
            SELECT
                po.id,
                po.po_number,
                po.po_date,
                po.po_time,
                po.simple_domestic_pt,
                po.dept_number,
                po.vendor_name,
                po.order_date,
                po.ship_date,
                po.cancel_date,
                po.total_packs_ordered,
                po.store_number,
                po.total_cost,
                po.total_qty,
                po.source_pdf,
                po.created_at,
                po.updated_at,
                po.raw_json,
                {style_summary_sql} AS style
            FROM purchase_orders po
            WHERE
                ? = '%%'
                OR po.po_number LIKE ?
                OR COALESCE(po.vendor_name, '') LIKE ?
                OR COALESCE(po.simple_domestic_pt, '') LIKE ?
                OR EXISTS (
                    SELECT 1
                    FROM purchase_order_items poi
                    WHERE poi.purchase_order_id = po.id
                      AND COALESCE(poi.mfg_style, '') LIKE ?
                )
            ORDER BY po.created_at DESC, po.id DESC
            LIMIT ?
            """,
            (
                normalized_search,
                normalized_search,
                normalized_search,
                normalized_search,
                normalized_search,
                safe_limit,
            ),
        ).fetchall()
    return list(rows)


def list_order_records(db_path: Path) -> list[sqlite3.Row]:
    style_summary_sql = purchase_order_style_summary_sql()
    with connect(db_path) as conn:
        rows = conn.execute(
            f"""
            SELECT
                po.*,
                {style_summary_sql} AS style
            FROM purchase_orders po
            ORDER BY po.created_at DESC, po.id DESC
            """
        ).fetchall()
    return list(rows)


def get_order(db_path: Path, order_id: int) -> dict | None:
    with connect(db_path) as conn:
        order = conn.execute(
            "SELECT * FROM purchase_orders WHERE id = ?",
            (order_id,),
        ).fetchone()
        if order is None:
            return None

        items = conn.execute(
            """
            SELECT sku, mfg_style, mfg_color, size_desc, description, upc, cost_unit,
                   comp, retail, pack_qty, total_packs, total_units
            FROM purchase_order_items
            WHERE purchase_order_id = ?
            ORDER BY id
            """,
            (order_id,),
        ).fetchall()

    return {
        "header": dict(order),
        "items": [dict(item) for item in items],
    }


def get_dashboard_summary(db_path: Path) -> dict:
    with connect(db_path) as conn:
        order_count = int(conn.execute("SELECT COUNT(*) FROM purchase_orders").fetchone()[0])
        item_count = int(conn.execute("SELECT COUNT(*) FROM purchase_order_items").fetchone()[0])
        costing_import_count = int(conn.execute("SELECT COUNT(*) FROM costing_imports").fetchone()[0])
        costing_variant_count = int(conn.execute("SELECT COUNT(*) FROM costing_variants").fetchone()[0])
        costing_ws_import_count = int(conn.execute("SELECT COUNT(*) FROM costing_ws_imports").fetchone()[0])
        costing_ws_row_count = int(conn.execute("SELECT COUNT(*) FROM costing_ws_rows").fetchone()[0])

        latest_order = conn.execute(
            """
            SELECT id, po_number, vendor_name, total_cost, total_qty, created_at
            FROM purchase_orders
            ORDER BY created_at DESC, id DESC
            LIMIT 1
            """
        ).fetchone()

    return {
        "orders": order_count,
        "line_items": item_count,
        "costing_imports": costing_import_count,
        "costing_variants": costing_variant_count,
        "costing_ws_imports": costing_ws_import_count,
        "costing_ws_rows": costing_ws_row_count,
        "latest_order": dict(latest_order) if latest_order else None,
    }


def find_purchase_orders_by_styles(db_path: Path, style_codes: list[str]) -> dict[str, list[dict]]:
    normalized_styles = sorted(
        {
            normalize_style_identifier(style)
            for style in style_codes
            if normalize_style_identifier(style)
        }
    )
    if not normalized_styles:
        return {}

    placeholders = ",".join("?" for _ in normalized_styles)
    with connect(db_path) as conn:
        rows = conn.execute(
            f"""
            SELECT
                UPPER(IFNULL(poi.mfg_style, '')) AS raw_style_code,
                po.id AS order_id,
                po.po_number
            FROM purchase_order_items poi
            JOIN purchase_orders po ON po.id = poi.purchase_order_id
            WHERE REPLACE(REPLACE(REPLACE(UPPER(IFNULL(poi.mfg_style, '')), ' ', ''), '-', ''), '_', '') IN ({placeholders})
            ORDER BY po.created_at DESC, po.id DESC
            """,
            tuple(normalized_styles),
        ).fetchall()

    mapping: dict[str, list[dict]] = {}
    seen_pairs: set[tuple[str, int]] = set()
    for row in rows:
        style_code = normalize_style_identifier(row["raw_style_code"])
        order_id = int(row["order_id"])
        pair_key = (style_code, order_id)
        if pair_key in seen_pairs:
            continue
        seen_pairs.add(pair_key)
        mapping.setdefault(style_code, [])
        mapping[style_code].append(
            {
                "order_id": order_id,
                "po_number": str(row["po_number"] or "").strip(),
            }
        )
    return mapping


def _parse_numeric_text(value: object) -> float:
    text = str(value or "").strip().replace(",", "")
    if not text:
        return 0.0
    try:
        return float(text)
    except Exception:
        return 0.0


def summarize_purchase_order_requirements_by_style(db_path: Path) -> dict[str, dict]:
    with connect(db_path) as conn:
        rows = conn.execute(
            """
            SELECT
                UPPER(IFNULL(poi.mfg_style, '')) AS raw_style_code,
                po.po_number,
                poi.total_units,
                poi.total_packs,
                poi.pack_qty
            FROM purchase_order_items poi
            JOIN purchase_orders po ON po.id = poi.purchase_order_id
            WHERE LOWER(IFNULL(TRIM(po.status), 'draft')) = 'approved'
            ORDER BY po.created_at DESC, po.id DESC, poi.id DESC
            """
        ).fetchall()

    summary: dict[str, dict] = {}
    for row in rows:
        style_code = normalize_style_identifier(row["raw_style_code"])
        if not style_code:
            continue
        total_units = _parse_numeric_text(row["total_units"])
        if total_units <= 0:
            total_units = _parse_numeric_text(row["total_packs"]) * _parse_numeric_text(row["pack_qty"])
        po_number = str(row["po_number"] or "").strip()
        item = summary.setdefault(
            style_code,
            {
                "style_code": style_code,
                "total_qty": 0.0,
                "po_numbers": [],
            },
        )
        item["total_qty"] = float(item.get("total_qty", 0.0)) + total_units
        if po_number and po_number not in item["po_numbers"]:
            item["po_numbers"].append(po_number)

    for item in summary.values():
        po_numbers = list(item.get("po_numbers", []))
        item["po_count"] = len(po_numbers)
        item["po_summary"] = ", ".join(po_numbers[:3]) + (" ..." if len(po_numbers) > 3 else "")
    return summary


def delete_purchase_order(db_path: Path, order_id: int) -> int:
    with connect(db_path) as conn:
        cursor = conn.execute("DELETE FROM purchase_orders WHERE id = ?", (order_id,))
        return int(cursor.rowcount or 0)


def save_costing_import(db_path: Path, costing_data: dict) -> int:
    with connect(db_path) as conn:
        cursor = conn.execute(
            """
            INSERT INTO costing_imports (collection, season, source_pdf, debug_dir, raw_json, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
            """,
            (
                costing_data.get("collection"),
                costing_data.get("season"),
                costing_data.get("source_pdf"),
                costing_data.get("debug_dir"),
                json.dumps(costing_data, ensure_ascii=True),
            ),
        )
        costing_import_id = int(cursor.lastrowid)

        variants = costing_data.get("items", [])
        conn.executemany(
            """
            INSERT INTO costing_variants (
                costing_import_id,
                style_code,
                color_label,
                description,
                image_path,
                page_image_path,
                page_number,
                crop_meta_json,
                collection,
                season,
                created_at,
                updated_at
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
            """,
            [
                (
                    costing_import_id,
                    item.get("style_code"),
                    item.get("color_label"),
                    item.get("description"),
                    item.get("image_path"),
                    item.get("page_image_path"),
                    item.get("page_number"),
                    json.dumps(item.get("crop_meta") or {}, ensure_ascii=True),
                    item.get("collection"),
                    item.get("season"),
                )
                for item in variants
            ],
        )
        return costing_import_id


def save_costing_ws_import(db_path: Path, worksheet_data: dict) -> int:
    with connect(db_path) as conn:
        conn.execute(
            """
            DELETE FROM costing_ws_imports
            WHERE IFNULL(source_workbook, '') = ? AND IFNULL(source_sheet, '') = ?
            """,
            (
                str(worksheet_data.get("source_workbook") or worksheet_data.get("source_pdf") or "").strip(),
                str(worksheet_data.get("source_sheet") or "").strip(),
            ),
        )
        cursor = conn.execute(
            """
            INSERT INTO costing_ws_imports (collection, season, source_workbook, source_sheet, raw_json, created_at, updated_at)
            VALUES (?, ?, ?, ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
            """,
            (
                worksheet_data.get("collection"),
                worksheet_data.get("season"),
                worksheet_data.get("source_workbook") or worksheet_data.get("source_pdf"),
                worksheet_data.get("source_sheet"),
                json.dumps(worksheet_data, ensure_ascii=True),
            ),
        )
        costing_ws_import_id = int(cursor.lastrowid)

        rows = worksheet_data.get("items", [])
        conn.executemany(
            """
            INSERT INTO costing_ws_rows (
                costing_ws_import_id,
                dept_number,
                mj_class,
                sub_class,
                brand,
                style_code,
                description,
                fabrication,
                composition,
                manf_color,
                color_code,
                product_type,
                ticket_type,
                packing_type,
                total_units,
                allocation_multiple,
                min_qty,
                max_qty,
                inner_pack,
                outer_pack,
                size_range,
                size_scale,
                est_cbm,
                hanger_supplier,
                hanger_type,
                item_type_for_freight,
                item_retail,
                po_number,
                color_label,
                collection,
                season,
                source_row_number,
                status,
                is_active,
                created_at,
                updated_at
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
            """,
            [
                (
                    costing_ws_import_id,
                    item.get("dept_number"),
                    item.get("mj_class"),
                    item.get("sub_class"),
                    item.get("brand"),
                    item.get("style_code"),
                    item.get("description"),
                    item.get("fabrication"),
                    item.get("composition"),
                    item.get("manf_color"),
                    item.get("color_code"),
                    item.get("product_type"),
                    item.get("ticket_type"),
                    item.get("packing_type"),
                    item.get("total_units"),
                    item.get("allocation_multiple"),
                    item.get("min_qty"),
                    item.get("max_qty"),
                    item.get("inner_pack"),
                    item.get("outer_pack"),
                    item.get("size_range"),
                    item.get("size_scale"),
                    item.get("est_cbm"),
                    item.get("hanger_supplier"),
                    item.get("hanger_type"),
                    item.get("item_type_for_freight"),
                    item.get("item_retail"),
                    item.get("po_number"),
                    item.get("color_label"),
                    item.get("collection"),
                    item.get("season"),
                    item.get("source_row_number"),
                    normalize_workflow_status(item.get("status"), "draft"),
                    1 if parse_active_flag(item.get("is_active"), True) else 0,
                )
                for item in rows
            ],
        )
        return costing_ws_import_id


def update_costing_ws_row(
    db_path: Path,
    *,
    source_workbook: str,
    source_sheet: str,
    source_row_number: int,
    updates: dict,
) -> int:
    allowed_fields = {
        "collection",
        "dept_number",
        "mj_class",
        "sub_class",
        "mfg_style",
        "style_code",
        "description",
        "fabrication",
        "mfg_color",
        "manf_color",
        "color_code",
        "product_type",
        "ticket_type",
        "packing_type",
        "total_units",
        "allocation_multiple",
        "min_qty",
        "max_qty",
        "inner_pack",
        "outer_pack",
        "size_range",
        "size_scale",
        "hanger_supplier",
        "hanger_type",
        "item_retail",
        "po_number",
        "status",
        "is_active",
    }
    payload = {key: updates.get(key) for key in allowed_fields if key in updates}
    style_value = payload.get("mfg_style", payload.get("style_code"))
    color_value = payload.get("mfg_color", payload.get("manf_color"))
    if "mfg_style" in payload and "style_code" not in payload:
        payload["style_code"] = style_value
    if "mfg_color" in payload and "manf_color" not in payload:
        payload["manf_color"] = color_value
    if "style_code" in payload and "mfg_style" not in payload:
        payload["mfg_style"] = payload["style_code"]
    if "manf_color" in payload and "mfg_color" not in payload:
        payload["mfg_color"] = payload["manf_color"]

    assignments = {
        "collection": payload.get("collection"),
        "dept_number": payload.get("dept_number"),
        "mj_class": payload.get("mj_class"),
        "sub_class": payload.get("sub_class"),
        "style_code": payload.get("style_code"),
        "description": payload.get("description"),
        "fabrication": payload.get("fabrication"),
        "manf_color": payload.get("manf_color"),
        "color_code": payload.get("color_code"),
        "product_type": payload.get("product_type"),
        "ticket_type": payload.get("ticket_type"),
        "packing_type": payload.get("packing_type"),
        "total_units": payload.get("total_units"),
        "allocation_multiple": payload.get("allocation_multiple"),
        "min_qty": payload.get("min_qty"),
        "max_qty": payload.get("max_qty"),
        "inner_pack": payload.get("inner_pack"),
        "outer_pack": payload.get("outer_pack"),
        "size_range": payload.get("size_range"),
        "size_scale": payload.get("size_scale"),
        "hanger_supplier": payload.get("hanger_supplier"),
        "hanger_type": payload.get("hanger_type"),
        "item_retail": payload.get("item_retail"),
        "po_number": payload.get("po_number"),
        "color_label": payload.get("manf_color") or payload.get("mfg_color"),
        "status": normalize_workflow_status(payload.get("status"), "draft") if "status" in payload else None,
        "is_active": (1 if parse_active_flag(payload.get("is_active"), True) else 0) if "is_active" in payload else None,
        "updated_at": "CURRENT_TIMESTAMP",
    }
    set_parts: list[str] = []
    params: list[object] = []
    for key, value in assignments.items():
        if key == "updated_at" and value == "CURRENT_TIMESTAMP":
            set_parts.append("updated_at = CURRENT_TIMESTAMP")
            continue
        if value is None:
            continue
        set_parts.append(f"{key} = ?")
        params.append(value)
    params.extend(
        [
            int(source_row_number),
            str(source_workbook or "").strip(),
            str(source_sheet or "").strip(),
        ]
    )
    affected_rows = 0
    with connect(db_path) as conn:
        cursor = conn.execute(
            f"""
            UPDATE costing_ws_rows
            SET {", ".join(set_parts)}
            WHERE
                source_row_number = ?
                AND costing_ws_import_id IN (
                    SELECT id
                    FROM costing_ws_imports
                    WHERE IFNULL(source_workbook, '') = ? AND IFNULL(source_sheet, '') = ?
                )
            """,
            tuple(params),
        )
        affected_rows = int(cursor.rowcount or 0)
        if affected_rows <= 0:
            fallback_params = list(params[: len(params) - 3])
            fallback_params.extend(
                [
                    int(source_row_number),
                    str(source_workbook or "").strip(),
                ]
            )
            fallback_cursor = conn.execute(
                f"""
                UPDATE costing_ws_rows
                SET {", ".join(set_parts)}
                WHERE
                    source_row_number = ?
                    AND costing_ws_import_id IN (
                        SELECT id
                        FROM costing_ws_imports
                        WHERE IFNULL(source_workbook, '') = ?
                    )
                """,
                tuple(fallback_params),
            )
            affected_rows = int(fallback_cursor.rowcount or 0)
        conn.execute(
            """
            UPDATE costing_ws_imports
            SET updated_at = CURRENT_TIMESTAMP
            WHERE IFNULL(source_workbook, '') = ? AND IFNULL(source_sheet, '') = ?
            """,
            (
                str(source_workbook or "").strip(),
                str(source_sheet or "").strip(),
            ),
        )
    return affected_rows


def get_latest_costing_import(db_path: Path) -> dict | None:
    with connect(db_path) as conn:
        import_row = conn.execute(
            """
            SELECT id, collection, season, source_pdf, debug_dir, raw_json, created_at, updated_at
            FROM costing_imports
            ORDER BY id DESC
            LIMIT 1
            """
        ).fetchone()
        if import_row is None:
            return None

        variant_rows = conn.execute(
            """
            SELECT style_code, color_label, description, image_path, page_image_path, page_number,
                   crop_meta_json, collection, season, created_at, updated_at
            FROM costing_variants
            WHERE costing_import_id = ?
            ORDER BY page_number, id
            """,
            (import_row["id"],),
        ).fetchall()

    items: list[dict] = []
    for row in variant_rows:
        try:
            crop_meta = json.loads(row["crop_meta_json"] or "{}")
        except Exception:
            crop_meta = {}
        items.append(
            {
                "style_code": row["style_code"],
                "color_label": row["color_label"],
                "description": row["description"],
                "image_path": remap_runtime_path(row["image_path"]),
                "page_image_path": remap_runtime_path(row["page_image_path"]),
                "page_number": row["page_number"],
                "crop_meta": crop_meta,
                "collection": row["collection"] or import_row["collection"],
                "season": row["season"] or import_row["season"],
                "source_pdf": import_row["source_pdf"],
                "created_at": row["created_at"] or import_row["created_at"],
                "updated_at": row["updated_at"] or import_row["updated_at"],
                "status": "draft",
                "is_active": True,
            }
        )

    return {
        "collection": import_row["collection"],
        "season": import_row["season"],
        "source_pdf": import_row["source_pdf"],
        "debug_dir": remap_runtime_path(import_row["debug_dir"]),
        "created_at": import_row["created_at"],
        "updated_at": import_row["updated_at"],
        "items": items,
    }


def get_all_costing_imports(db_path: Path) -> list[dict]:
    with connect(db_path) as conn:
        import_rows = conn.execute(
            """
            SELECT id, collection, season, source_pdf, debug_dir, raw_json, created_at, updated_at
            FROM costing_imports
            ORDER BY id
            """
        ).fetchall()

        imports: list[dict] = []
        for import_row in import_rows:
            variant_rows = conn.execute(
                """
                SELECT style_code, color_label, description, image_path, page_image_path, page_number,
                       crop_meta_json, collection, season, created_at, updated_at
                FROM costing_variants
                WHERE costing_import_id = ?
                ORDER BY page_number, id
                """,
                (import_row["id"],),
            ).fetchall()

            items: list[dict] = []
            for row in variant_rows:
                try:
                    crop_meta = json.loads(row["crop_meta_json"] or "{}")
                except Exception:
                    crop_meta = {}
                items.append(
                    {
                        "style_code": row["style_code"],
                        "color_label": row["color_label"],
                        "description": row["description"],
                        "image_path": remap_runtime_path(row["image_path"]),
                        "page_image_path": remap_runtime_path(row["page_image_path"]),
                        "page_number": row["page_number"],
                        "crop_meta": crop_meta,
                        "collection": row["collection"] or import_row["collection"],
                        "season": row["season"] or import_row["season"],
                        "source_pdf": import_row["source_pdf"],
                        "created_at": row["created_at"] or import_row["created_at"],
                        "updated_at": row["updated_at"] or import_row["updated_at"],
                        "status": "draft",
                        "is_active": True,
                    }
                )

            imports.append(
                {
                    "collection": import_row["collection"],
                    "season": import_row["season"],
                    "source_pdf": import_row["source_pdf"],
                    "debug_dir": remap_runtime_path(import_row["debug_dir"]),
                    "created_at": import_row["created_at"],
                    "updated_at": import_row["updated_at"],
                    "items": items,
                }
            )

        worksheet_import_rows = conn.execute(
            """
            SELECT id, collection, season, source_workbook, source_sheet, raw_json, created_at, updated_at
            FROM costing_ws_imports
            ORDER BY id
            """
        ).fetchall()

        for import_row in worksheet_import_rows:
            worksheet_rows = conn.execute(
                """
                SELECT
                    dept_number,
                    mj_class,
                    sub_class,
                    brand,
                    style_code,
                    description,
                    fabrication,
                    composition,
                    manf_color,
                    color_code,
                    product_type,
                    ticket_type,
                    packing_type,
                    total_units,
                    allocation_multiple,
                    min_qty,
                    max_qty,
                    inner_pack,
                    outer_pack,
                    size_range,
                    size_scale,
                    est_cbm,
                    hanger_supplier,
                    hanger_type,
                    item_type_for_freight,
                    item_retail,
                    po_number,
                    color_label,
                    collection,
                    season,
                    source_row_number,
                    status,
                    is_active,
                    created_at,
                    updated_at
                FROM costing_ws_rows
                WHERE costing_ws_import_id = ?
                ORDER BY id
                """,
                (import_row["id"],),
            ).fetchall()

            items: list[dict] = []
            for row in worksheet_rows:
                items.append(
                    {
                        "source_kind": "worksheet",
                        "style_code": row["style_code"],
                        "mfg_style": row["style_code"],
                        "color_label": row["color_label"],
                        "description": row["description"],
                        "image_path": "",
                        "page_image_path": "",
                        "page_number": None,
                        "crop_meta": {},
                        "collection": row["collection"] or import_row["collection"],
                        "season": row["season"] or import_row["season"],
                        "source_pdf": import_row["source_workbook"],
                        "source_workbook": import_row["source_workbook"],
                        "source_sheet": import_row["source_sheet"],
                        "source_row_number": row["source_row_number"],
                        "created_at": row["created_at"] or import_row["created_at"],
                        "updated_at": row["updated_at"] or import_row["updated_at"],
                        "dept_number": row["dept_number"],
                        "mj_class": row["mj_class"],
                        "sub_class": row["sub_class"],
                        "brand": row["brand"],
                        "fabrication": row["fabrication"],
                        "composition": row["composition"],
                        "manf_color": row["manf_color"],
                        "mfg_color": row["manf_color"],
                        "color_code": row["color_code"],
                        "product_type": row["product_type"],
                        "ticket_type": row["ticket_type"],
                        "packing_type": row["packing_type"],
                        "total_units": row["total_units"],
                        "allocation_multiple": row["allocation_multiple"],
                        "min_qty": row["min_qty"],
                        "max_qty": row["max_qty"],
                        "inner_pack": row["inner_pack"],
                        "outer_pack": row["outer_pack"],
                        "size_range": row["size_range"],
                        "size_scale": row["size_scale"],
                        "est_cbm": row["est_cbm"],
                        "hanger_supplier": row["hanger_supplier"],
                        "hanger_type": row["hanger_type"],
                        "item_type_for_freight": row["item_type_for_freight"],
                        "item_retail": row["item_retail"],
                        "po_number": row["po_number"],
                        "status": normalize_workflow_status(row["status"], "draft"),
                        "is_active": bool(row["is_active"]),
                    }
                )

            imports.append(
                {
                    "source_kind": "worksheet",
                    "collection": import_row["collection"],
                    "season": import_row["season"],
                    "source_pdf": import_row["source_workbook"],
                    "source_workbook": import_row["source_workbook"],
                    "source_sheet": import_row["source_sheet"],
                    "created_at": import_row["created_at"],
                    "updated_at": import_row["updated_at"],
                    "items": items,
                }
            )

    return imports


def list_costing_variants_for_web(db_path: Path, limit: int = 24) -> list[dict]:
    safe_limit = max(1, min(int(limit), 200))
    with connect(db_path) as conn:
        rows = conn.execute(
            """
            SELECT
                cv.id,
                cv.style_code,
                cv.color_label,
                cv.description,
                cv.collection,
                cv.season,
                cv.image_path,
                cv.page_number,
                ci.created_at
            FROM costing_variants cv
            JOIN costing_imports ci ON ci.id = cv.costing_import_id
            ORDER BY cv.id DESC
            LIMIT ?
            """,
            (safe_limit,),
        ).fetchall()

    variants: list[dict] = []
    for row in rows:
        row_dict = dict(row)
        row_dict["image_path"] = remap_runtime_path(row_dict.get("image_path"))
        variants.append(row_dict)
    return variants


def delete_costing_styles(db_path: Path, style_keys: list[object]) -> None:
    if not style_keys:
        return

    with connect(db_path) as conn:
        for item in style_keys:
            if isinstance(item, dict):
                source_kind = str(item.get("source_kind", "")).strip().lower()
                if source_kind == "worksheet":
                    conn.execute(
                        """
                        DELETE FROM costing_ws_rows
                        WHERE
                            IFNULL(collection, '') = ?
                            AND IFNULL(season, '') = ?
                            AND IFNULL(style_code, '') = ?
                            AND IFNULL(source_row_number, 0) = ?
                            AND costing_ws_import_id IN (
                                SELECT id
                                FROM costing_ws_imports
                                WHERE IFNULL(source_workbook, '') = ? AND IFNULL(source_sheet, '') = ?
                            )
                        """,
                        (
                            str(item.get("collection", "") or ""),
                            str(item.get("season", "") or ""),
                            str(item.get("style_code", "") or ""),
                            int(item.get("source_row_number") or 0),
                            str(item.get("source_workbook") or item.get("source_pdf") or ""),
                            str(item.get("source_sheet", "") or ""),
                        ),
                    )
                    continue

                conn.execute(
                    """
                    DELETE FROM costing_variants
                    WHERE IFNULL(collection, '') = ? AND IFNULL(season, '') = ? AND IFNULL(style_code, '') = ?
                    """,
                    (
                        str(item.get("collection", "") or ""),
                        str(item.get("season", "") or ""),
                        str(item.get("style_code", "") or ""),
                    ),
                )
                continue

            collection, season, style_code = item
            conn.execute(
                """
                DELETE FROM costing_variants
                WHERE IFNULL(collection, '') = ? AND IFNULL(season, '') = ? AND IFNULL(style_code, '') = ?
                """,
                (collection or "", season or "", style_code or ""),
            )
            conn.execute(
                """
                DELETE FROM costing_ws_rows
                WHERE IFNULL(collection, '') = ? AND IFNULL(season, '') = ? AND IFNULL(style_code, '') = ?
                """,
                (collection or "", season or "", style_code or ""),
            )

        conn.execute(
            """
            DELETE FROM costing_imports
            WHERE id IN (
                SELECT ci.id
                FROM costing_imports ci
                LEFT JOIN costing_variants cv ON cv.costing_import_id = ci.id
                GROUP BY ci.id
                HAVING COUNT(cv.id) = 0
            )
            """
        )
        conn.execute(
            """
            DELETE FROM costing_ws_imports
            WHERE id IN (
                SELECT cwi.id
                FROM costing_ws_imports cwi
                LEFT JOIN costing_ws_rows cwr ON cwr.costing_ws_import_id = cwi.id
                GROUP BY cwi.id
                HAVING COUNT(cwr.id) = 0
            )
            """
        )
