dodil/erp-brick-manufacturing

DODIL ERP brick: manufacturing

scripts/gen_schema.py

"""Regenerate schema.sql and migrations/00001_init/up.sql from infrastructure/tables.py (the one column model).
Run: python scripts/gen_schema.py  (from the repo root, with src on the path). A NEW column is a NEW migration, never an edit here."""
import pathlib
import sys

ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))

from sqlalchemy.dialects import postgresql  # noqa: E402

from brick_manufacturing.infrastructure.tables import BOM, ORDER, md  # noqa: E402

SRC = ROOT / "src" / "brick_manufacturing"
HEADER = ("-- Every table prefixed `mfg_`, a natural primary key each (never SERIAL), money NUMERIC(18,2), rates and quantities NUMERIC(19,4),\n"
          "-- creation columns last so a later column is an append. Additive only.\n")


def _plain(sql: str) -> str:
    return sql.replace(" WITHOUT TIME ZONE", "")


def ddl() -> str:
    d, out = postgresql.dialect(), []
    for t in md.sorted_tables:
        if t.name in (BOM.versions.name, ORDER.versions.name):
            continue
        cols = [f"    {c.name:<24} " + c.type.compile(dialect=d).replace(" WITHOUT TIME ZONE", "").replace(", ", ",")
                + (" NOT NULL" if c.primary_key else "") + "," for c in t.columns]
        out.append(f"CREATE TABLE IF NOT EXISTS {t.name} (\n" + "\n".join(cols)
                   + f"\n    PRIMARY KEY ({', '.join(c.name for c in t.primary_key.columns)})\n);\n")
        if t.name == "mfg_bom":
            out.append(_plain(BOM.version_ddl()))
        if t.name == "mfg_production_order":
            out.append(_plain(ORDER.version_ddl()))
    return "\n".join(out)


def _cols(sql: str) -> dict:
    import re
    out = {}
    for name, body in re.findall(r"CREATE TABLE IF NOT EXISTS (\w+) \(\n(.*?)\n\);", sql, re.S):
        out[name] = [ln.strip().split()[0] for ln in body.splitlines() if ln.strip() and not ln.strip().startswith(("PRIMARY KEY", "--"))]
    return out


def slice_b(full: str) -> tuple:
    """(up, down) for 00002_slice_b: every table slice A's FROZEN first migration does not have, and every column it lacks."""
    import re
    first = _cols((SRC / "migrations" / "00001_init" / "up.sql").read_text())
    d, up, down = postgresql.dialect(), [], []
    for t in md.sorted_tables:
        if t.name not in first:
            block = re.search(rf"CREATE TABLE IF NOT EXISTS {t.name} \(\n.*?\n\);\n", full, re.S)
            up.append(block.group(0))
            down.append(f"DROP TABLE IF EXISTS {t.name};\n")
        else:
            for c in t.columns:
                if c.name not in first[t.name]:
                    up.append(f"ALTER TABLE {t.name} ADD COLUMN {c.name} {c.type.compile(dialect=d).replace(' WITHOUT TIME ZONE', '').replace(', ', ',')};\n")
    return ("-- 00002_slice_b: work centres and routings, confirmations and scrap, MRP-lite, cost roll-ups, the AI audit, the Odoo slice;\n"
            "-- and four columns on the order (routing, plan reference, stale note). Additive only.\n\n" + "\n".join(up),
            "-- 00002_slice_b down: the tables go; the columns stay (DROP COLUMN is disruptive on DataK3).\n" + "".join(reversed(down)))


if __name__ == "__main__":
    body = ddl()
    (SRC / "schema.sql").write_text(HEADER + "\n" + body)
    first = SRC / "migrations" / "00001_init"
    if not (first / "up.sql").exists():           # slice A's migration is FROZEN once written: a later column is a later migration
        first.mkdir(parents=True, exist_ok=True)
        (first / "up.sql").write_text(
            "-- 00001_init: slice A (bills of materials, production orders, movements, outbox, settlements, controls).\n"
            "-- Versioned layout migrations/NNNNN_<name>/{up,down}.sql with a per-brick journal, mfg_schema_migrations.\n\n"
            "CREATE TABLE IF NOT EXISTS mfg_schema_migrations (\n    version     VARCHAR NOT NULL,\n    applied_at  TIMESTAMP,\n"
            "    PRIMARY KEY (version)\n);\n\n" + body)
        names = [t.name for t in md.sorted_tables]
        (first / "down.sql").write_text("-- 00001_init down: development only; a released migration is never dropped (additive only).\n"
                                        + "".join(f"DROP TABLE IF EXISTS {n};\n" for n in reversed(names)))
    up, down = slice_b(HEADER + "\n" + body)
    second = SRC / "migrations" / "00002_slice_b"
    second.mkdir(parents=True, exist_ok=True)
    (second / "up.sql").write_text(up)
    (second / "down.sql").write_text(down)
    print("wrote schema.sql and 00002_slice_b")