465 lines
15 KiB
Python
465 lines
15 KiB
Python
#!/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"
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Mapping from source component category to template target rows.
|
|
# Template columns (0-based):
|
|
# A = 0, B = 1, C = 2, D = 3, E = 4, ...
|
|
# Serial numbers are written into column E as specified by the task.
|
|
# ---------------------------------------------------------------------------
|
|
CATEGORY_TARGETS: dict[str, dict] = {
|
|
"All / Components / Barebone": {
|
|
"name_start": 10,
|
|
"name_rows": 1,
|
|
"serial_start": 10,
|
|
"serial_rows": 1,
|
|
},
|
|
"All / Components / CPU": {
|
|
"name_start": 11,
|
|
"name_rows": 1,
|
|
"serial_start": 11,
|
|
"serial_rows": 1,
|
|
},
|
|
"All / Components / RAM": {
|
|
"name_start": 12,
|
|
"name_rows": 8,
|
|
"serial_start": 12,
|
|
"serial_rows": 8,
|
|
},
|
|
"All / Components / GPU": {
|
|
"name_start": 20,
|
|
"name_rows": 4,
|
|
"serial_start": 20,
|
|
"serial_rows": 4,
|
|
},
|
|
"All / Components / SSD": {
|
|
"name_start": 24,
|
|
"name_rows": 3,
|
|
"serial_start": 24,
|
|
"serial_rows": 3,
|
|
},
|
|
"All / Components / Network": {
|
|
"name_start": 30,
|
|
"name_rows": 2,
|
|
"serial_start": 30,
|
|
"serial_rows": 2,
|
|
},
|
|
"All / Components / RAID": {
|
|
"name_start": 32,
|
|
"name_rows": 1,
|
|
"serial_start": 32,
|
|
"serial_rows": 1,
|
|
},
|
|
}
|
|
|
|
# Order in which categories are processed. For SSD/RAID the source file may
|
|
# contain multiple distinct blocks for the same category; this order defines
|
|
# which block goes into which template area.
|
|
CATEGORY_ORDER = [
|
|
"All / Components / Barebone",
|
|
"All / Components / CPU",
|
|
"All / Components / RAM",
|
|
"All / Components / GPU",
|
|
"All / Components / SSD",
|
|
"All / Components / Network",
|
|
"All / Components / RAID",
|
|
]
|
|
|
|
# Source column indices (0-based).
|
|
COL_CAT = 3 # D
|
|
COL_NAME = 4 # E
|
|
COL_SN = 6 # G
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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 _cell_style_name(cell: TableCell) -> str | None:
|
|
return cell.getAttribute("stylename")
|
|
|
|
|
|
def _set_cell_value(table: Table, row: int, col: int, value: str) -> None:
|
|
"""Set the text value of a table cell, preserving style and spans.
|
|
|
|
Because the template contains merged cells (C:D, etc.), the *logical*
|
|
column the user sees may span several physical cells. We interpret ``col``
|
|
as the 0-based visual/logical column: if a previous cell in the row is
|
|
merged across N columns, the physical index is adjusted by N-1.
|
|
"""
|
|
_ensure_rows(table, row + 1)
|
|
rows = _get_rows(table)
|
|
target_row = rows[row]
|
|
physical_cells = _get_cells(target_row)
|
|
|
|
# Map logical column -> physical cell index, accounting for merged cells.
|
|
physical_index = 0
|
|
span_carry = 0
|
|
for logical_col in range(col + 1):
|
|
if span_carry > 0:
|
|
span_carry -= 1
|
|
else:
|
|
if physical_index >= len(physical_cells):
|
|
raise IndexError(
|
|
f"Row {row + 1} does not have enough cells for logical column {col + 1}"
|
|
)
|
|
cell = physical_cells[physical_index]
|
|
physical_index += 1
|
|
span = cell.getAttribute("numbercolumnsspanned")
|
|
if span:
|
|
span_carry = int(span) - 1
|
|
# physical_index now points one past the target physical cell.
|
|
target_physical_index = physical_index - 1
|
|
old_cell = physical_cells[target_physical_index]
|
|
|
|
style = _cell_style_name(old_cell)
|
|
|
|
# Reuse the existing cell so ODF caches stay consistent.
|
|
new_cell = old_cell
|
|
# Clear any existing paragraphs/text.
|
|
for child in list(new_cell.childNodes):
|
|
new_cell.removeChild(child)
|
|
|
|
if style:
|
|
new_cell.setAttribute("stylename", style)
|
|
for attr in ("numbercolumnsspanned", "numberrowsspanned"):
|
|
val = new_cell.getAttribute(attr)
|
|
if val is not None:
|
|
new_cell.setAttribute(attr, val)
|
|
|
|
new_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 _is_component_block_empty(ws, start_row: int) -> tuple[int, list[str]]:
|
|
"""Return how many extra rows with serial numbers belong to a component.
|
|
|
|
Starting at ``start_row`` we walk downwards while the category (column D)
|
|
and the component name (column E) are empty. For every row that has a
|
|
serial number in column G we increase the corresponding m* counter and
|
|
collect the serial. We stop as soon as a non-empty category or component
|
|
name appears.
|
|
"""
|
|
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 len(extra_serials), extra_serials
|
|
|
|
|
|
def _find_source_rows(ws) -> dict[str, list[tuple[int, str, list[str]]]]:
|
|
"""Find all component blocks in the source file.
|
|
|
|
Returns a dict mapping source category -> list of blocks. Each block is
|
|
(row, name, serials_list). A new block starts when a row contains a
|
|
non-empty value in column E even if the category cell in column D is empty.
|
|
"""
|
|
blocks: dict[str, list[tuple[int, str, list[str]]]] = {
|
|
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)
|
|
|
|
# A component block must either belong to a known category or start with
|
|
# a component name directly below a known category (empty category cell
|
|
# but non-empty name cell).
|
|
if category not in CATEGORY_TARGETS:
|
|
r += 1
|
|
continue
|
|
|
|
if name is None and serial is None:
|
|
r += 1
|
|
continue
|
|
|
|
# Collect serials for this block.
|
|
serials: list[str] = []
|
|
if serial:
|
|
serials.append(serial)
|
|
_, extra_serials = _is_component_block_empty(ws, r)
|
|
serials.extend(extra_serials)
|
|
|
|
# Use the component name from this row; fall back to the name of the
|
|
# first block of the same category if the source has split blocks.
|
|
block_name = name if name else None
|
|
|
|
blocks[category].append((r, block_name, serials))
|
|
|
|
# Skip over the rows we already consumed.
|
|
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
|
|
|
|
fa: str | None = None
|
|
model: str | None = None
|
|
|
|
# B2 -> FA, C2 -> MODEL
|
|
b2 = _clean(ws.cell(row=2, column=2).value)
|
|
c2 = _clean(ws.cell(row=2, column=3).value)
|
|
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 = _find_source_rows(ws)
|
|
wb.close()
|
|
|
|
full_model = f"{model}-{fa}"
|
|
|
|
counters = {
|
|
"mGPU": 0,
|
|
"mRAM": 0,
|
|
"mSSD": 0,
|
|
"mNIC": 0,
|
|
}
|
|
components: dict[str, dict] = {cat: {"name": None, "serials": []} for cat in CATEGORY_TARGETS}
|
|
|
|
# Process each category according to the fixed template order.
|
|
for cat in CATEGORY_ORDER:
|
|
cat_blocks = blocks.get(cat, [])
|
|
if not cat_blocks:
|
|
continue
|
|
|
|
# First block always fills the primary template area.
|
|
_, name, serials = cat_blocks[0]
|
|
components[cat]["name"] = name
|
|
components[cat]["serials"] = serials
|
|
|
|
# Counters count the *additional* serial rows beyond the first one.
|
|
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(serials) - 1)
|
|
|
|
# Second block for SSD goes into the secondary NVME U.2 area (C27:E29).
|
|
if cat == "All / Components / SSD" and len(cat_blocks) > 1:
|
|
components["__SSD_SECOND__"] = {
|
|
"name": cat_blocks[1][1],
|
|
"serials": cat_blocks[1][2],
|
|
}
|
|
|
|
result = {
|
|
"fa": fa,
|
|
"model": model,
|
|
"full_model": full_model,
|
|
"components": components,
|
|
**counters,
|
|
}
|
|
if "__SSD_SECOND__" in components:
|
|
result["ssd_second"] = components.pop("__SSD_SECOND__")
|
|
return result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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']}"
|
|
)
|
|
|
|
# Write combined model-serial into B1 (merged B1:C2).
|
|
_set_cell_value(table, 0, 1, data["full_model"])
|
|
|
|
# Component names and serial numbers.
|
|
name_col = 2 # C (template uses merged C:D for names)
|
|
serial_col = 4 # E (serial numbers go into column E)
|
|
|
|
for category in CATEGORY_ORDER:
|
|
comp = data["components"][category]
|
|
targets = CATEGORY_TARGETS[category]
|
|
name = comp.get("name")
|
|
serials = comp.get("serials", [])
|
|
|
|
if name:
|
|
_set_cell_value(table, targets["name_start"] - 1, name_col, name)
|
|
print(f" {category} name -> C{targets['name_start']}: {name}")
|
|
|
|
for i, sn in enumerate(serials[: targets["serial_rows"]]):
|
|
target_row = targets["serial_start"] - 1 + i
|
|
_set_cell_value(table, target_row, serial_col, sn)
|
|
print(f" {category} S/N {i + 1} -> E{target_row + 1}: {sn}")
|
|
|
|
if len(serials) > targets["serial_rows"]:
|
|
print(
|
|
f" Warning: {category} has {len(serials)} serials, "
|
|
f"only {targets['serial_rows']} fit in the template",
|
|
file=sys.stderr,
|
|
)
|
|
|
|
# Optional second SSD block into rows 27-29 (C27:D29 / E27:E29).
|
|
ssd_second = data.get("ssd_second")
|
|
if ssd_second:
|
|
name = ssd_second.get("name")
|
|
serials = ssd_second.get("serials", [])
|
|
if name:
|
|
_set_cell_value(table, 26, name_col, name)
|
|
print(f" Second SSD name -> C27: {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())
|