lean101/data-entry-app
1
0
Fork 0
Code Issues 7 Pull Requests Actions Packages Projects Releases Wiki Activity

v0.1.19 - Throughput overview & responsive header

Browse Source 
- Throughput: new "Throughput Overview" header with Gauge icon and brand-green
  badge icons on each card (Today, This week, 4-week average, Horse Mix, Grain Mix)
- Throughput: inline rolling-range selector (7d / 4w / 6w / 12w, default 4 weeks)
  driving the customer-mix cards; stats window widened to 12 weeks so switching
  range is a pure client-side re-filter
- Throughput: cards collapse to a single even 5-across row on laptop and up,
  with container-query value text that scales to each card's width
- Throughput: date logic pinned to Australian Eastern time (fixes the day-early
  date); This week subtitle shows the Mon-Sun date range
- Throughput: subtler tinted add-form; removed the inline-entry kicker and the
  "Open full form" link
- Topbar: fix cramped laptop header - action toggles no longer wrap above the
  user button; search drops to its own row earlier

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
admin
2026-06-13 10:01:10 +12:00
co-authored by Claude Opus 4.8
parent 4ff372d307
commit 2de82776cb
64 changed files with 6034 additions and 1134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/mix_calculator_pdf.py
+4 -1
View File
@@ -319,7 +319,10 @@ def build_mix_calculator_pdf(session_record: MixCalculatorSession | dict) -> byt
fit_text(line.raw_material_name, "Helvetica-Bold", table_font_size, content_width - 210),
)
pdf.setFont("Helvetica", table_font_size)
pdf.drawString(right_col_x, text_y, f"{_fmt_number(line.required_kg)}kg")
# Each ingredient carries its own rounding (set in the Ingredients Editor)
# so the printed sheet matches the on-screen calculated output.
line_decimals = getattr(line, "rounding_decimals", 2)
pdf.drawString(right_col_x, text_y, f"{_fmt_number(line.required_kg, line_decimals)}kg")
strip_y = table_bottom - 6
if note_lines:
backend/app/services/mix_calculator_service.py
+5
View File
@@ -43,6 +43,7 @@ def _resolved_formula_rows(product: Product) -> tuple[list[dict], float]:
"raw_material_name": ingredient.raw_material.name,
"quantity_kg": ingredient.quantity_kg,
"unit": ingredient.raw_material.unit_of_measure,
"rounding_decimals": ingredient.raw_material.rounding_decimals,
"sort_order": ingredient.sort_order,
}
for ingredient in product.ingredients
@@ -55,6 +56,7 @@ def _resolved_formula_rows(product: Product) -> tuple[list[dict], float]:
"raw_material_name": ingredient.raw_material.name if ingredient.raw_material is not None else f"Raw material {ingredient.raw_material_id}",
"quantity_kg": ingredient.quantity_kg,
"unit": ingredient.raw_material.unit_of_measure if ingredient.raw_material is not None else "kg",
"rounding_decimals": ingredient.raw_material.rounding_decimals if ingredient.raw_material is not None else 2,
"sort_order": index,
}
for index, ingredient in enumerate(product.mix.ingredients, start=1)
@@ -128,6 +130,7 @@ def calculate_mix_calculator_preview(
"required_kg": required_kg,
"mix_percentage": mix_percentage,
"unit": ingredient["unit"],
"rounding_decimals": ingredient.get("rounding_decimals", 2),
"sort_order": ingredient["sort_order"] or index,
}
)
@@ -260,6 +263,7 @@ def serialize_mix_calculator_session(session_record: MixCalculatorSession, auth_
"required_kg": round(line.required_kg, 4),
"mix_percentage": round(line.mix_percentage, 4),
"unit": line.unit,
"rounding_decimals": line.rounding_decimals,
"sort_order": line.sort_order,
}
for line in session_record.lines
@@ -331,6 +335,7 @@ def create_mix_calculator_session(db: Session, *, auth_session: AuthSession, pay
required_kg=line["required_kg"],
mix_percentage=line["mix_percentage"],
unit=line["unit"],
rounding_decimals=line.get("rounding_decimals", 2),
sort_order=line["sort_order"],
)
for line in preview["lines"]
backend/app/services/throughput_service.py
+309
View File
@@ -1,5 +1,7 @@
from __future__ import annotations
import csv
import io
import logging
import os
from datetime import date, datetime
@@ -369,3 +371,310 @@ def resolve_workbook_path() -> Path | None:
if candidate.exists():
return candidate
return None
# ── Ad-hoc CSV / spreadsheet upload import ──────────────────────────────────
# Lets an operator upload their own CSV or .xlsx of packing runs (from Settings
# → Import) and have every row saved as a throughput entry. Unlike the bundled
# workbook seed above, this is column-header driven so the file can be a simple
# hand-built sheet rather than the exact "Operations Throughput.xlsx" layout.
# Maps the column headers we accept (normalised: lower-cased, spaces/dashes →
# single spaces) onto the canonical field used internally. Several aliases per
# field so a human-built sheet "just works".
_HEADER_ALIASES: dict[str, str] = {
"date": "date",
"production date": "date",
"production_date": "date",
"product": "product",
"product name": "product",
"product_name": "product",
"product name snapshot": "product",
"name": "product",
"item id": "item_id",
"item_id": "item_id",
"itemid": "item_id",
"sku": "item_id",
"quantity": "quantity",
"qty": "quantity",
"packed": "quantity",
"quantity packed": "quantity",
"amount": "quantity",
"quantity type": "quantity_type",
"type": "quantity_type",
"unit": "quantity_type",
"packed as": "quantity_type",
"bag size": "bag_size",
"bag_size": "bag_size",
"kg per bag": "bag_size",
"kg/bag": "bag_size",
"bagsize": "bag_size",
"staff": "staff_name",
"staff name": "staff_name",
"packed by": "staff_name",
"operator": "staff_name",
"for order": "for_order",
"order": "for_order",
"for stock": "for_stock",
"stock": "for_stock",
"job number": "job_number",
"job": "job_number",
"job no": "job_number",
"order number": "job_number",
"stock quantity": "stock_quantity",
"stock qty": "stock_quantity",
"sample box no": "sample_box_no",
"sample box": "sample_box_no",
"scales checked": "scales_checked",
"scales": "scales_checked",
"label correct": "label_correct",
"label": "label_correct",
"bag sealed": "bag_sealed",
"sealed": "bag_sealed",
"pallet good condition": "pallet_good_condition",
"pallet": "pallet_good_condition",
"notes": "notes",
"note": "notes",
"comment": "notes",
"comments": "notes",
}
# How many row-level errors we collect before truncating, to keep the response
# (and the toast) sane on a badly-formed file.
_MAX_REPORTED_ERRORS = 50
def _normalise_header(raw: object) -> str | None:
if raw is None:
return None
key = " ".join(str(raw).strip().lower().replace("-", " ").replace("_", " ").split())
if not key:
return None
if key in _HEADER_ALIASES:
return _HEADER_ALIASES[key]
# Test weights: "test weight 1".."test weight 5" (and "tw1" style).
for n in range(1, 6):
if key in {f"test weight {n}", f"tw{n}", f"test {n}"}:
return f"test_weight_{n}"
return None
def _coerce_quantity_type(value: object) -> str | None:
if value is None:
return None
text = str(value).strip().lower()
if not text:
return None
if text in {"bag", "bags", "b"}:
return "bags"
if text in {"kg", "kgs", "kilogram", "kilograms", "bulka", "bulk"}:
return "kg"
return None
def _read_tabular_file(filename: str, content: bytes) -> tuple[list[str | None], list[tuple]]:
"""Return (headers, data_rows). Detects CSV vs .xlsx by extension/content."""
lowered = (filename or "").lower()
is_excel = lowered.endswith((".xlsx", ".xlsm", ".xls"))
if is_excel:
workbook = load_workbook(io.BytesIO(content), data_only=True, read_only=True)
ws = workbook.active
rows = [tuple(r) for r in ws.iter_rows(values_only=True)]
workbook.close()
else:
text = None
for encoding in ("utf-8-sig", "utf-8", "latin-1"):
try:
text = content.decode(encoding)
break
except UnicodeDecodeError:
continue
if text is None:
raise ValueError("Could not decode the file as text. Save it as UTF-8 CSV or .xlsx.")
# Sniff the delimiter (comma/semicolon/tab) but fall back to comma.
sample = text[:4096]
try:
dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
except csv.Error:
dialect = csv.excel
rows = [tuple(r) for r in csv.reader(io.StringIO(text), dialect)]
# Find the first row that has at least one recognised header; treat it as
# the header row and everything after as data.
for index, row in enumerate(rows):
if any(_normalise_header(cell) is not None for cell in row):
return list(row), rows[index + 1 :]
return [], []
def import_entries_from_file(
db: Session,
*,
filename: str,
content: bytes,
tenant_id: str,
created_by: str | None,
) -> dict:
"""Parse an uploaded CSV/spreadsheet and persist each row as a throughput
entry. Products are matched by item_id then name, and auto-created when not
found so every entry stays linked. Returns a summary with row-level errors.
"""
headers, data_rows = _read_tabular_file(filename, content)
if not headers:
raise ValueError(
"No recognised columns found. The file needs a header row with at "
"least Date, Product and Quantity columns."
)
# Map canonical field name → column index. First occurrence wins.
field_index: dict[str, int] = {}
for col, raw in enumerate(headers):
field = _normalise_header(raw)
if field and field not in field_index:
field_index[field] = col
for required in ("date", "product", "quantity"):
if required not in field_index:
raise ValueError(
f"Missing required '{required}' column. Required columns are "
"Date, Product and Quantity."
)
def cell(row: tuple, field: str) -> object:
idx = field_index.get(field)
if idx is None or idx >= len(row):
return None
return row[idx]
# Index existing products for matching (by item_id and by lower-cased name).
by_item: dict[str, ThroughputProduct] = {}
by_name: dict[str, ThroughputProduct] = {}
for product in db.scalars(
select(ThroughputProduct).where(ThroughputProduct.tenant_id == tenant_id)
).all():
if product.item_id:
by_item[str(product.item_id)] = product
by_name[product.name.lower()] = product
imported = 0
skipped = 0
products_created = 0
errors: list[str] = []
def note_error(message: str) -> None:
if len(errors) < _MAX_REPORTED_ERRORS:
errors.append(message)
for offset, row in enumerate(data_rows):
# Sheet/file row number for human-friendly error messages (header = 1).
line_no = offset + 2
if not row or all(value is None or str(value).strip() == "" for value in row):
continue
production_date = _coerce_date(cell(row, "date"))
product_name = _coerce_text(cell(row, "product"))
quantity = _coerce_float(cell(row, "quantity"))
if production_date is None:
skipped += 1
note_error(f"Row {line_no}: missing or invalid date.")
continue
if not product_name:
skipped += 1
note_error(f"Row {line_no}: missing product name.")
continue
if quantity is None or quantity < 0:
skipped += 1
note_error(f"Row {line_no}: missing or invalid quantity.")
continue
bag_size = _coerce_float(cell(row, "bag_size"))
quantity_type = _coerce_quantity_type(cell(row, "quantity_type"))
if quantity_type is None:
# Infer: bulka-style rows have a blank or very large bag size.
if bag_size is None or bag_size >= _BULKA_BAG_SIZE_THRESHOLD or "bulka" in product_name.lower():
quantity_type = "kg"
else:
quantity_type = "bags"
if quantity_type == "bags" and (bag_size is None or bag_size <= 0):
skipped += 1
note_error(f"Row {line_no}: bag size is required when packed as bags.")
continue
item_id_raw = cell(row, "item_id")
item_id = None
if item_id_raw is not None:
if isinstance(item_id_raw, float) and item_id_raw.is_integer():
item_id = str(int(item_id_raw))
else:
item_id = _coerce_text(item_id_raw)
product = (by_item.get(item_id) if item_id else None) or by_name.get(product_name.lower())
if product is None:
product = ThroughputProduct(
tenant_id=tenant_id,
item_id=item_id,
name=product_name,
default_bag_size=bag_size,
is_bulka_default=_infer_bulka_default(product_name, bag_size),
active=True,
notes="Auto-created during throughput import",
)
db.add(product)
db.flush()
products_created += 1
if item_id:
by_item[item_id] = product
by_name[product_name.lower()] = product
for_order = _coerce_bool(cell(row, "for_order")) if field_index.get("for_order") is not None else False
for_stock = _coerce_bool(cell(row, "for_stock")) if field_index.get("for_stock") is not None else False
stock_quantity = _coerce_float(cell(row, "stock_quantity")) if for_stock else None
calculated = calculate_kg(quantity, quantity_type, bag_size)
entry = ProductionThroughput(
tenant_id=tenant_id,
production_date=production_date,
product_id=product.id,
product_name_snapshot=product_name,
bag_size=bag_size,
scales_checked=_coerce_bool(cell(row, "scales_checked")),
label_correct=_coerce_bool(cell(row, "label_correct")),
bag_sealed=_coerce_bool(cell(row, "bag_sealed")),
pallet_good_condition=_coerce_bool(cell(row, "pallet_good_condition")),
for_order=for_order,
for_stock=for_stock,
job_number=_coerce_text(cell(row, "job_number")) if for_order else None,
stock_quantity=stock_quantity,
sample_box_no=_coerce_text(cell(row, "sample_box_no")),
test_weight_1=_coerce_float(cell(row, "test_weight_1")),
test_weight_2=_coerce_float(cell(row, "test_weight_2")),
test_weight_3=_coerce_float(cell(row, "test_weight_3")),
test_weight_4=_coerce_float(cell(row, "test_weight_4")),
test_weight_5=_coerce_float(cell(row, "test_weight_5")),
quantity=quantity,
quantity_type=quantity_type,
calculated_kg=calculated,
staff_name=normalise_staff_name(cell(row, "staff_name")),
notes=_coerce_text(cell(row, "notes")),
created_by=created_by or "csv-import",
)
db.add(entry)
imported += 1
if imported == 0 and products_created == 0:
# Nothing landed — don't leave a half-open transaction.
db.rollback()
else:
db.commit()
return {
"entries_imported": imported,
"entries_skipped": skipped,
"products_created": products_created,
"errors": errors,
}
backend/app/services/xero_service.py
+80 -11
View File
@@ -23,7 +23,7 @@ from dataclasses import dataclass, field
from datetime import datetime
from app.models.client_access import ClientAccount
from app.models.ordering import Order
from app.models.ordering import Order, XeroContactLink
@dataclass
@@ -57,11 +57,75 @@ class XeroSubmissionResult:
line_items: list[dict] = field(default_factory=list)
def map_customer_to_contact(customer: ClientAccount) -> dict:
"""Map a customer account onto a Xero contact payload."""
@dataclass
class XeroContact:
"""A Xero contact available to link a customer against."""
contact_id: str
name: str
email: str | None = None
status: str = "ACTIVE"
def as_dict(self) -> dict:
return {
"contact_id": self.contact_id,
"name": self.name,
"email": self.email,
"status": self.status,
}
# Deterministic sample contacts used while running in stub mode (no Xero
# credentials). They stand in for "what's in Xero" so the customer→contact
# mapping UI is usable before the live API is wired. Ids mimic Xero GUIDs.
_STUB_CONTACTS: tuple[XeroContact, ...] = (
XeroContact("STUB-CON-0001", "Hunter Premium Produce", "accounts@hunterpremium.example", "ACTIVE"),
XeroContact("STUB-CON-0002", "Mayreef Pty Ltd", "ap@mayreef.example", "ACTIVE"),
XeroContact("STUB-CON-0003", "Ian McKay Stock Feeds", "ian@mckayfeeds.example", "ACTIVE"),
XeroContact("STUB-CON-0004", "Peckish Bird Foods", "orders@peckish.example", "ACTIVE"),
XeroContact("STUB-CON-0005", "Hay & Straw Co", "info@hayandstraw.example", "ACTIVE"),
XeroContact("STUB-CON-0006", "PHF Horse Mixes", "accounts@phfhorse.example", "ACTIVE"),
)
def _fetch_contacts_from_api(config: XeroConfig) -> list[XeroContact]:
"""Live contact fetch. Stubbed until credentials/endpoints are wired.
TODO (go-live): GET ``{config.base_url}/Contacts`` with the
``Xero-tenant-id`` header, page through ``Contacts[]`` and map each onto a
:class:`XeroContact` (``ContactID``/``Name``/``EmailAddress``/``ContactStatus``).
"""
raise NotImplementedError("Live Xero contact fetch is not implemented yet.")
def list_xero_contacts(config: XeroConfig | None = None) -> tuple[list[XeroContact], bool]:
"""Return the Xero contacts available for linking and whether they're stubbed.
Never raises — on a live-mode error it returns an empty list so the mapping
console still renders.
"""
config = config or XeroConfig.from_env()
if not config.configured:
return list(_STUB_CONTACTS), True
try:
return _fetch_contacts_from_api(config), False
except Exception: # pragma: no cover - defensive: never break the request path
return [], False
def map_customer_to_contact(customer: ClientAccount, link: XeroContactLink | None = None) -> dict:
"""Map a customer account onto a Xero contact payload.
When the customer has been linked to a Xero contact we send the real
``ContactID`` so Xero attaches the invoice to the existing contact. Without a
link we fall back to keying on the client code (Xero will match-or-create).
"""
if link is not None and link.xero_contact_id:
return {
"ContactID": link.xero_contact_id,
"Name": link.xero_contact_name or customer.name,
}
return {
# TODO: persist and reuse a real Xero ContactID once the contact has
# been created/matched in Xero. For now we key on the client code.
"ContactNumber": customer.client_code,
"Name": customer.name,
}
@@ -76,7 +140,9 @@ def map_product_to_item_code(product_sku: str) -> str:
return product_sku
def build_invoice_payload(order: Order, customer: ClientAccount) -> dict:
def build_invoice_payload(
order: Order, customer: ClientAccount, link: XeroContactLink | None = None
) -> dict:
"""Build the Xero draft-invoice payload for a confirmed order."""
line_items = []
for line in order.lines:
@@ -98,7 +164,7 @@ def build_invoice_payload(order: Order, customer: ClientAccount) -> dict:
return {
"Type": "ACCREC",
"Status": "DRAFT",
"Contact": map_customer_to_contact(customer),
"Contact": map_customer_to_contact(customer, link),
"Reference": order.purchase_order_number or order.order_number or f"Order {order.id}",
"LineAmountTypes": "Exclusive",
"LineItems": line_items,
@@ -127,14 +193,17 @@ def _submit_to_xero_api(config: XeroConfig, payload: dict) -> XeroSubmissionResu
)
def submit_order_to_xero(order: Order, customer: ClientAccount) -> XeroSubmissionResult:
def submit_order_to_xero(
order: Order, customer: ClientAccount, link: XeroContactLink | None = None
) -> XeroSubmissionResult:
"""Submit a confirmed order to Xero, or stub it when unconfigured.
Never raises — failures are returned as ``status="failed"`` results so the
order lifecycle can record the attempt and continue.
Pass ``link`` to invoice against the customer's mapped Xero contact. Never
raises — failures are returned as ``status="failed"`` results so the order
lifecycle can record the attempt and continue.
"""
config = XeroConfig.from_env()
payload = build_invoice_payload(order, customer)
payload = build_invoice_payload(order, customer, link)
summary = f"{len(payload['LineItems'])} line(s) for {payload['Contact']['Name']}"
if not config.configured: