"""Extract seed CSVs from Mike's real spreadsheets in ../notes/. Run on the host (openpyxl is a dev-time dependency, not a runtime one): python3 main/scripts/extract_seed.py Writes main/seed/assets.csv and main/seed/parts.csv. Those CSVs are the committed seed fixtures; the container never reads .xlsx. Sources (see brief.md): assets <- Britco Equipment List COGZ.xlsx (377 real assets) parts <- BP Parts Report.xlsx (1,015 transactions, deduplicated to the distinct parts they reference — the MaintainX parts-import file itself holds only one example row) """ from __future__ import annotations import csv import sys from pathlib import Path import openpyxl PROJECT = Path(__file__).resolve().parents[2] NOTES = PROJECT / "notes" SEED = PROJECT / "main" / "seed" EQUIPMENT_XLSX = NOTES / "Maintenance Planning" / "Britco Equipment List COGZ.xlsx" PARTS_REPORT_XLSX = ( NOTES / "Maintenance Planning" / "MAINTAINX" / "BP Parts Report.xlsx" ) _repairs = 0 def _demojibake(text: str) -> str: """Repair UTF-8 bytes that were decoded as cp1252 in the source export. Mike's MaintainX export contains values like '17.26a EUR (tm) wide' where a quote or apostrophe should be - the classic mojibake signature (U+00E2 U+20AC ...). Round-tripping through cp1252 restores the original character. Encoding is per character with a latin-1 fallback: the mangled text mixes cp1252-mappable characters (U+00E2, U+20AC) with C1 control codepoints (U+009D) that cp1252 cannot encode at all. A whole-string cp1252 encode raises on those and silently leaves the value mangled - which is exactly what the first version of this function did, and what tests/test_contracts.py now guards against. Deliberately conservative: the repair is applied only when the round-trip succeeds AND changes something, so clean values are never touched. The source .xlsx is not modified. """ global _repairs if "â" not in text and "Ã" not in text: return text raw = bytearray() for char in text: for encoding in ("cp1252", "latin-1"): try: raw += char.encode(encoding) break except UnicodeEncodeError: continue else: return text # not representable - leave the value untouched try: repaired = raw.decode("utf-8") except UnicodeDecodeError: return text if repaired != text: _repairs += 1 return repaired def _clean(value: object) -> str: if value is None: return "" return _demojibake(str(value).strip()) def extract_assets() -> int: """Equipment No / Description / Line / Manufacturer -> assets.csv.""" book = openpyxl.load_workbook(EQUIPMENT_XLSX, read_only=True, data_only=True) sheet = book.active rows = sheet.iter_rows(values_only=True) header = [_clean(c).lstrip(">").strip().rstrip("*") for c in next(rows)] try: idx = { "equipment_no": header.index("Equipment No"), "description": header.index("Description"), "line": header.index("Line"), "manufacturer": header.index("Manufacturer"), } except ValueError as exc: raise SystemExit(f"Unexpected equipment header {header!r}: {exc}") from exc seen: set[str] = set() out: list[dict[str, str]] = [] for row in rows: equipment_no = _clean( row[idx["equipment_no"]] if idx["equipment_no"] < len(row) else "" ) description = _clean( row[idx["description"]] if idx["description"] < len(row) else "" ) if not equipment_no or not description: continue if equipment_no in seen: continue seen.add(equipment_no) out.append( { "equipment_no": equipment_no, "description": description, "line": _clean(row[idx["line"]] if idx["line"] < len(row) else ""), "manufacturer": _clean( row[idx["manufacturer"]] if idx["manufacturer"] < len(row) else "" ), } ) book.close() path = SEED / "assets.csv" with path.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter( handle, fieldnames=["equipment_no", "description", "line", "manufacturer"] ) writer.writeheader() writer.writerows(out) return len(out) def extract_parts() -> int: """Distinct parts referenced by the transaction report -> parts.csv. The report is movement data, so the same part appears many times. The last transaction for a part carries the most recent quantity/cost, which is the closest thing to a current inventory snapshot the source offers. """ book = openpyxl.load_workbook(PARTS_REPORT_XLSX, read_only=True, data_only=True) sheet = book.active rows = sheet.iter_rows(values_only=True) header = [_clean(c) for c in next(rows)] def col(name: str) -> int | None: return header.index(name) if name in header else None idx = { "part_id": col("Part ID"), "part_name": col("Part Name"), "area": col("Part Area"), "location": col("Part Location"), "barcode": col("QR/Bar code"), "unit_cost": col("Unit Cost"), "quantity_after": col("Quantity After"), "uom": col("U/M"), "part_number": col("Part #"), "part_type": col("Part Types"), } if idx["part_name"] is None: raise SystemExit(f"Unexpected parts header {header!r}") def cell(row: tuple, key: str) -> str: position = idx[key] if position is None or position >= len(row): return "" return _clean(row[position]) parts: dict[str, dict[str, str]] = {} for row in rows: name = cell(row, "part_name") if not name: continue key = cell(row, "part_id") or name record = { "part_number": cell(row, "part_number") or cell(row, "part_id"), "part_name": name, "description": "", "uom": cell(row, "uom"), "location": cell(row, "location"), "area": cell(row, "area"), "part_type": cell(row, "part_type"), "barcode": cell(row, "barcode"), "unit_cost": cell(row, "unit_cost"), "quantity": cell(row, "quantity_after"), } parts[key] = record # later rows win: most recent movement book.close() path = SEED / "parts.csv" fields = [ "part_number", "part_name", "description", "uom", "location", "area", "part_type", "barcode", "unit_cost", "quantity", ] with path.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=fields) writer.writeheader() writer.writerows(parts.values()) return len(parts) def main() -> int: SEED.mkdir(parents=True, exist_ok=True) for source in (EQUIPMENT_XLSX, PARTS_REPORT_XLSX): if not source.exists(): print(f"missing source: {source}", file=sys.stderr) return 1 assets = extract_assets() parts = extract_parts() print(f"assets.csv: {assets} rows") print(f"parts.csv: {parts} rows") print(f"mojibake repaired in {_repairs} source values") return 0 if __name__ == "__main__": raise SystemExit(main())