"""Isolated material contracts. No imports from the original SHS runtime. _to_pence is copied from the supplied SHS materials_catalogue.py (lines 70–86). See docs/PROVENANCE.md. All surrounding contracts and quantity rules are new. """ from datetime import date from decimal import Decimal, InvalidOperation, ROUND_HALF_UP import re from typing import Annotated, Literal from urllib.parse import urlsplit from fastapi import HTTPException from pydantic import Field, StrictInt, field_validator, model_validator from .models import Strict MAX_PENCE = 100_000_000 def _to_pence(value, field: str) -> int: """Parse a money amount into exact integer pence. Rejects negatives and anything beyond 2 decimal places.""" if isinstance(value, bool): raise HTTPException(status_code=400, detail=f"{field} must be a valid amount") try: d = Decimal(str(value).strip()) except (InvalidOperation, ValueError, AttributeError): raise HTTPException(status_code=400, detail=f"{field} must be a valid amount") if not d.is_finite() or d < 0: raise HTTPException(status_code=400, detail=f"{field} must be a non-negative amount") pence = d * 100 if pence != pence.to_integral_value(): raise HTTPException(status_code=400, detail=f"{field} can have at most 2 decimal places") if pence > MAX_PENCE: raise HTTPException(status_code=400, detail=f"{field} is unreasonably large") return int(pence) def price(value, field='Cost price'): # Bound input before Decimal arithmetic; exclude exponents, bools and NaN. if not isinstance(value, str) or not re.fullmatch(r'\d{1,10}(\.\d{1,2})?', value.strip()): raise ValueError(f'{field}: use pounds and up to two decimal places.') return _to_pence(value, field) def quantity(value): if not isinstance(value, str) or not re.fullmatch(r'\d{1,7}(\.\d{1,3})?', value.strip()): raise ValueError('Quantity: use a positive number with up to three decimal places.') q = Decimal(value.strip()) if not 0 < q <= 1_000_000: raise ValueError('Quantity must be greater than zero and no more than 1,000,000.') return format(q.normalize(), 'f') def total_pence(qty, cost): total = int((Decimal(qty) * cost).quantize(Decimal('1'), rounding=ROUND_HALF_UP)) if total > MAX_PENCE: raise HTTPException(422, 'One material line cannot exceed £1,000,000.') return total Identity = Annotated[str, Field(pattern=r'^[a-f0-9]{32}$')] Stage = Literal['to_order', 'ordered', 'received'] VersionNumber = Annotated[StrictInt, Field(ge=1)] class Reference(Strict): label: str = Field(min_length=1, max_length=100) url: str = Field(min_length=1, max_length=2000) last_price: str = Field(default='', max_length=30) last_checked: str = '' @field_validator('url') @classmethod def valid_url(cls, value): try: p = urlsplit(value) if p.scheme not in ('http', 'https') or not p.hostname or p.username or p.password: raise ValueError() p.port except ValueError: raise ValueError('Supplier references need an http(s) address without login details.') if re.search(r'[\s\x00-\x1f\\]', value): raise ValueError('Invalid supplier reference address.') return value @field_validator('last_price') @classmethod def valid_price(cls, value): if value: price(value, 'Last price') return value @field_validator('last_checked') @classmethod def valid_date(cls, value): if value: date.fromisoformat(value) if len(value) != 10: raise ValueError('Use YYYY-MM-DD.') return value class CatalogueFields(Strict): name: str = Field(min_length=1, max_length=180) supplier: str = Field(default='', max_length=180) supplier_sku: str = Field(default='', max_length=100) unit: str = Field(default='each', min_length=1, max_length=30) cost_price: str = Field(max_length=30) charge_price: str = Field(default='', max_length=30) @field_validator('cost_price', 'charge_price') @classmethod def valid_price(cls, value, info): if value or info.field_name == 'cost_price': price(value, info.field_name.replace('_', ' ')) return value class CatalogueEdit(CatalogueFields): version: VersionNumber class Version(Strict): version: VersionNumber class MaterialFields(Strict): name: str = Field(min_length=1, max_length=180) quantity: str = Field(default='1', max_length=20) unit: str = Field(default='each', min_length=1, max_length=30) cost_price: str = Field(max_length=30) supplier: str = Field(default='', max_length=180) supplier_sku: str = Field(default='', max_length=100) notes: str = Field(default='', max_length=2000) status: Stage = 'to_order' ordered_date: str = '' received_date: str = '' supplier_links: list[Reference] = Field(default_factory=list, max_length=8) @field_validator('quantity') @classmethod def valid_qty(cls, value): return quantity(value) @field_validator('cost_price') @classmethod def valid_price(cls, value): price(value) return value @field_validator('ordered_date', 'received_date') @classmethod def valid_date(cls, value): return Reference.valid_date(value) @model_validator(mode='after') def valid_total(self): total_pence(self.quantity, price(self.cost_price)) if self.ordered_date and self.received_date and self.ordered_date > self.received_date: raise ValueError('Received date cannot be before ordered date.') return self class MaterialCreate(MaterialFields): job_id: Identity class MaterialEdit(MaterialFields): version: VersionNumber class CatalogueAdd(Strict): job_id: Identity version: VersionNumber quantity: str = Field(default='1', max_length=20) cost_override: str = Field(default='', max_length=30) notes: str = Field(default='', max_length=2000) @field_validator('quantity') @classmethod def valid_qty(cls, value): return quantity(value) @field_validator('cost_override') @classmethod def valid_price(cls, value): if value: price(value) return value class Selection(Version): id: Identity class BulkStage(Strict): items: list[Selection] = Field(min_length=1, max_length=200) status: Stage @field_validator('items') @classmethod def unique_items(cls, values): if len({v.id for v in values}) != len(values): raise ValueError('Choose each material once.') return values class CsvImport(Strict): job_id: Identity rows: list[MaterialFields] = Field(min_length=1, max_length=200)