dotfiles/VSCodium/User/History/114d05ac/V0N4.py

#!/usr/bin/env python3
"""Fill AIME production order template from Odoo MRP export.

Usage:
    python fill_production_order.py <source_xlsx>
    python fill_production_order.py               # uses newest matching source file

The script loads the specified Excel source file, extracts relevant fields,
fills the ODS template, and saves it as a new ODS named after cell B1:C2
of the resulting template (MODEL-FA without the prefixes).
"""

from __future__ import annotations

import argparse
import re
import sys
from pathlib import Path

import openpyxl
from odf import opendocument
from odf.table import Table, TableCell, TableRow
from odf.text import P

# ---------------------------------------------------------------------------
# Paths / globs
# ---------------------------------------------------------------------------
TEMPLATE_GLOB = "A4004_*.ods"
SOURCE_GLOB = "Production Order (mrp.production)*.xlsx"

# ---------------------------------------------------------------------------
# Source column indices (0-based).  Column D = index 3, E = 4, G = 6.
# ---------------------------------------------------------------------------
COL_CAT = 3   # D
COL_NAME = 4  # E
COL_SN = 6    # G

# ---------------------------------------------------------------------------
# Mapping from source component category to target template rows.
# Template columns (1-based as the user sees them):
#   Category = B (col 1), Name = C:D (cols 2-3), Serial = E (col 4).
# The rows are fixed by the template layout.
# ---------------------------------------------------------------------------
CATEGORY_TARGETS: dict[str, dict] = {
    "All / Components / Barebone": {
        "name_row": 10,
        "serial_row": 10,
        "max_serials": 1,
    },
    "All / Components / CPU": {
        "name_row": 11,
        "serial_row": 11,
        "max_serials": 1,
    },
    "All / Components / RAM": {
        "name_row": 12,
        "serial_row": 12,
        "max_serials": 8,
    },
    "All / Components / GPU": {
        "name_row": 20,
        "serial_row": 20,
        "max_serials": 4,
    },
    "All / Components / SSD": {
        "name_row": 24,
        "serial_row": 24,
        "max_serials": 3,
    },
    "All / Components / Network": {
        "name_row": 30,
        "serial_row": 30,  # NOTE: user's spec says E30 even though table row is 30
        "max_serials": 2,
    },
    "All / Components / RAID": {
        "name_row": 32,
        "serial_row": 32,  # NOTE: user's spec says E30; see fill logic below
        "max_serials": 1,
    },
}

# Order in which categories are processed.
CATEGORY_ORDER = [
    "All / Components / Barebone",
    "All / Components / CPU",
    "All / Components / RAM",
    "All / Components / GPU",
    "All / Components / SSD",
    "All / Components / Network",
    "All / Components / RAID",
]

# Source categories that may appear without an explicit category cell but belong
# to the preceding known category (e.g. continuation rows for RAM / SSD).

# ---------------------------------------------------------------------------
# ODS helpers
# ---------------------------------------------------------------------------
def _get_rows(table: Table) -> list[TableRow]:
    return list(table.getElementsByType(TableRow))


def _get_cells(row: TableRow) -> list[TableCell]:
    return list(row.getElementsByType(TableCell))


def _ensure_rows(table: Table, min_row_count: int) -> None:
    """Append empty rows until the table has at least ``min_row_count`` rows."""
    rows = _get_rows(table)
    if not rows:
        return
    first_row_cells = _get_cells(rows[0])
    while len(rows) < min_row_count:
        new_row = TableRow()
        for _ in first_row_cells:
            cell = TableCell()
            cell.addElement(P(text=""))
            new_row.addElement(cell)
        table.addElement(new_row)
        rows = _get_rows(table)


def _set_cell_value(table: Table, row: int, col: int, value: str) -> None:
    """Set the text value of the cell at physical ``row``/``col`` (0-based).

    Preserves existing horizontal/vertical merge attributes so that
    multi-row or multi-column merged cells stay merged after the update.
    """
    _ensure_rows(table, row + 1)
    rows = _get_rows(table)
    target_row = rows[row]
    cells = _get_cells(target_row)

    if col >= len(cells):
        raise IndexError(
            f"Row {row + 1} only has {len(cells)} cells, cannot write to physical column {col + 1}"
        )

    cell = cells[col]
    style = cell.getAttribute("stylename")

    # Preserve merge attributes.
    preserved_spans = {
        attr: cell.getAttribute(attr)
        for attr in ("numbercolumnsspanned", "numberrowsspanned")
        if cell.getAttribute(attr) is not None
    }

    # Clear existing paragraphs.
    for child in list(cell.childNodes):
        cell.removeChild(child)

    if style:
        cell.setAttribute("stylename", style)

    for attr, val in preserved_spans.items():
        cell.setAttribute(attr, val)

    cell.addElement(P(text=str(value) if value is not None else ""))


def _find_template(directory: Path) -> Path:
    candidates = sorted(directory.glob(TEMPLATE_GLOB))
    if not candidates:
        raise FileNotFoundError(
            f"No ODS template matching '{TEMPLATE_GLOB}' found in {directory}"
        )
    return candidates[0]


# ---------------------------------------------------------------------------
# Source parsing
# ---------------------------------------------------------------------------
def _clean(value) -> str | None:
    if value is None:
        return None
    text = str(value).strip()
    return text if text else None


def _count_empty_name_rows(ws, start_row: int) -> tuple[int, list[str]]:
    """Count consecutive continuation rows and collect their serial numbers.

    Starting on the row *below* ``start_row`` we walk downwards while both
    the category cell (column D) and the name cell (column E) are empty.
    For every such row that has a serial number in column G we collect it.
    We stop as soon as a non-empty category or name appears.

    Returns (number_of_continuation_rows, list_of_extra_serials_found).
    """
    extra_serials: list[str] = []
    max_row = ws.max_row
    r = start_row + 1
    while r <= max_row:
        cat_val = _clean(ws.cell(row=r, column=COL_CAT + 1).value)
        name_val = _clean(ws.cell(row=r, column=COL_NAME + 1).value)
        if cat_val is not None or name_val is not None:
            break
        sn = _clean(ws.cell(row=r, column=COL_SN + 1).value)
        if sn:
            extra_serials.append(sn)
        r += 1
    return (r - start_row - 1), extra_serials


def _extract_blocks(ws) -> dict[str, list[dict]]:
    """Extract every component block from the source workbook.

    Returns a dict mapping source category -> list of blocks.
    Each block has:
        - row:       first source row of the block
        - name:      component name (column E)
        - serials:   list of serial numbers belonging to this block

    A new block starts when a known category cell appears or when a non-empty
    name cell appears directly below a known category that did not yet have a
    name in its own row.
    """
    blocks: dict[str, list[dict]] = {cat: [] for cat in CATEGORY_TARGETS}

    max_row = ws.max_row
    r = 1
    while r <= max_row:
        category = _clean(ws.cell(row=r, column=COL_CAT + 1).value)
        name = _clean(ws.cell(row=r, column=COL_NAME + 1).value)
        serial = _clean(ws.cell(row=r, column=COL_SN + 1).value)

        if category not in CATEGORY_TARGETS:
            r += 1
            continue

        # Skip truly empty category rows.
        if name is None and serial is None:
            r += 1
            continue

        serials: list[str] = []
        if serial:
            serials.append(serial)

        _, extra_serials = _count_empty_name_rows(ws, r)
        serials.extend(extra_serials)

        blocks[category].append({
            "row": r,
            "name": name,
            "serials": serials,
        })

        r += 1 + len(extra_serials)

    return blocks


def extract_source_data(source_path: Path) -> dict:
    """Read the source workbook and return extracted values."""
    wb = openpyxl.load_workbook(source_path, data_only=True)
    ws = wb.active

    # B2 -> FA, C2 -> MODEL
    b2 = _clean(ws.cell(row=2, column=2).value)
    c2 = _clean(ws.cell(row=2, column=3).value)

    fa: str | None = None
    model: str | None = None
    if isinstance(b2, str):
        m = re.search(r"FA-([A-Za-z0-9]+)", b2)
        if m:
            fa = m.group(1)
    if isinstance(c2, str):
        m = re.search(r"AIME-([A-Za-z0-9]+)", c2)
        if m:
            model = m.group(1)

    if fa is None:
        raise ValueError(f"Could not find 'FA-' serial number in {source_path}")
    if model is None:
        raise ValueError(f"Could not find 'AIME-' model in {source_path}")

    blocks = _extract_blocks(ws)
    wb.close()

    full_model = f"{model}-{fa}"

    components: dict[str, dict] = {cat: {"name": None, "serials": []} for cat in CATEGORY_TARGETS}
    ssd_second: dict | None = None

    counters = {
        "mGPU": 0,
        "mRAM": 0,
        "mSSD": 0,
        "mNIC": 0,
    }

    # Per the user's variable definition, the counters count how many *empty*
    # name rows exist below the component before the next non-empty name.
    # Those empty rows are the rows that contain extra serial numbers.

    for cat in CATEGORY_ORDER:
        cat_blocks = blocks.get(cat, [])
        if not cat_blocks:
            continue

        first = cat_blocks[0]
        components[cat]["name"] = first["name"]
        components[cat]["serials"] = first["serials"]

        # The m* variables count how many *extra* serial rows exist beyond the first.
        counter_key = {
            "All / Components / GPU": "mGPU",
            "All / Components / RAM": "mRAM",
            "All / Components / SSD": "mSSD",
            "All / Components / Network": "mNIC",
        }.get(cat)
        if counter_key:
            counters[counter_key] = max(0, len(first["serials"]) - 1)

        # If there are additional SSD blocks beyond the primary one, the second
        # distinct block is placed in the secondary SSD area (rows 27-29).
        if cat == "All / Components / SSD" and len(cat_blocks) > 1:
            ssd_second = cat_blocks[1]

    return {
        "fa": fa,
        "model": model,
        "full_model": full_model,
        "components": components,
        "ssd_second": ssd_second,
        **counters,
    }


# ---------------------------------------------------------------------------
# Template filling
# ---------------------------------------------------------------------------
def fill_template(source_path: Path, output_path: Path | None = None) -> Path:
    directory = source_path.parent
    data = extract_source_data(source_path)

    template_path = _find_template(directory)
    doc = opendocument.load(template_path)

    tables = list(doc.spreadsheet.getElementsByType(Table))
    if not tables:
        raise ValueError("No table found in the template ODS file")
    table = tables[0]

    print(f"Source: {source_path.name}")
    print(f"FA: {data['fa']}, MODEL: {data['model']}, Combined: {data['full_model']}")
    print(
        f"Counters -> mGPU={data['mGPU']}, mRAM={data['mRAM']}, "
        f"mSSD={data['mSSD']}, mNIC={data['mNIC']}"
    )

    # Combined model-serial into B1 (merged B1:C2).
    # B = physical column 1, C = physical column 2; merged cells share the first physical cell.
    _set_cell_value(table, 0, 1, data["full_model"])

    # Template columns (0-based physical as stored in the ODS).
    # The example file stores the component name at index 2 (with a C:D
    # merge) and the serial numbers at index 3.  odfpy writes the value into
    # the physical cell we target, so we use the same physical indices.
    name_col = 2   # C (merge owner for the C:D name cell)
    serial_col = 3 # D / S/N column in the stored ODS

    for category in CATEGORY_ORDER:
        comp = data["components"][category]
        targets = CATEGORY_TARGETS[category]
        name = comp.get("name")
        serials = comp.get("serials", [])

        if name:
            row_idx = targets["name_row"] - 1
            _set_cell_value(table, row_idx, name_col, name)
            print(f"  {category} name -> C{targets['name_row']}:D{targets['name_row']}: {name}")

        for i, sn in enumerate(serials[: targets["max_serials"]]):
            target_row = targets["serial_row"] - 1 + i
            _set_cell_value(table, target_row, serial_col, sn)
            print(f"  {category} S/N {i + 1} -> E{target_row + 1}: {sn}")


    # Secondary SSD block into rows 27-29 (name C27:D27, serial E27:E29).
    ssd_second = data.get("ssd_second")
    if ssd_second:
        name = ssd_second.get("name")
        serials = ssd_second.get("serials", [])
        print(f"DEBUG ssd_second name={name!r} serials={serials}")
        if name:
            _set_cell_value(table, 26, name_col, name)
            print(f"  Second SSD name -> C27:D27: {name}")
        for i, sn in enumerate(serials[:3]):
            _set_cell_value(table, 26 + i, serial_col, sn)
            print(f"  Second SSD S/N {i + 1} -> E{27 + i}: {sn}")



    if output_path is None:
        output_path = directory / f"{data['full_model']}.ods"
    else:
        output_path = Path(output_path)

    doc.save(output_path)
    print(f"Saved: {output_path}")
    return output_path


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Fill production order template from an Odoo export."
    )
    parser.add_argument(
        "source",
        nargs="?",
        help="Path to the source Excel file. If omitted, the alphabetically last "
        "source file matching the production order glob is used.",
    )
    parser.add_argument(
        "-o", "--output", help="Optional explicit output ODS file path"
    )
    args = parser.parse_args(argv)

    directory = Path(__file__).resolve().parent

    if args.source:
        source_path = Path(args.source).expanduser()
        if not source_path.is_absolute():
            source_path = directory / source_path
    else:
        sources = sorted(directory.glob(SOURCE_GLOB))
        if not sources:
            print(f"No source file matching '{SOURCE_GLOB}' found.", file=sys.stderr)
            return 1
        source_path = sources[-1]
        print(f"Using source file: {source_path.name}")

    if not source_path.exists():
        print(f"Source file not found: {source_path}", file=sys.stderr)
        return 1

    fill_template(source_path, args.output)
    return 0


if __name__ == "__main__":
    sys.exit(main())