"""The candidate fingerprint: what VISUAL says about a packet, hashed per part, to decide update, CO or nothing. Parts (each the sha256 hex of its canonical JSON): - major: `legs` (the sorted leg keys), `qty` (per leg), `rev`, `drawing` (upper-cased, spaces removed), `bom` (sorted `[item, part_id, qty]`, the old matSig); - minor: `tgt` (target hours per leg), `want` (the want date). Never in it: rel, folded, partspecs, title, statuses, issued, whs, loc, orders, labor, dept, op. A candidate whose materials are unknown (`bom is None`) has no `bom` part; `classify` skips a part the new side lacks and `merge` carries the old one forward, so a failed materials read never looks like a BOM change. """ from __future__ import annotations import hashlib import json import unicodedata from collections.abc import Mapping from dataclasses import dataclass from datetime import date, datetime from decimal import Decimal from typing import TYPE_CHECKING, Any if TYPE_CHECKING: from .candidates import Candidate MAJOR = ("legs", "qty", "rev", "drawing", "bom") MINOR = ("tgt", "want") PARTS = MAJOR + MINOR # ---------------------------------------------------------------- normalisation def norm_text(value: Any) -> str | None: """NFC, CRLF and CR to LF, trimmed; '' and None both None.""" if value is None: return None s = value if isinstance(value, str) else str(value) s = unicodedata.normalize("NFC", s.replace("\r\n", "\n").replace("\r", "\n")).strip() return s or None def norm_id(value: Any) -> str | None: """An ID (base, lot, split, sub, part, drawing): normalised text, upper-cased. Stays text.""" s = norm_text(value) return s.upper() if s is not None else None def norm_decimal(value: Any) -> str | None: """A number as text with no exponent and no trailing zeros: 2, 2.0 and Decimal('2.00') are all '2'.""" if value is None or value == "": return None d = value if isinstance(value, Decimal) else Decimal(str(value)) if d.is_nan() or d.is_infinite(): return None if d == 0: return "0" s = format(d.normalize(), "f") if "." in s: s = s.rstrip("0").rstrip(".") return s def normalise(value: Any) -> Any: """The canonical form of a JSON-like value.""" if value is None: return None if isinstance(value, bool): return value if isinstance(value, str): return norm_text(value) if isinstance(value, int | float | Decimal): return norm_decimal(value) if isinstance(value, datetime): return value.isoformat() if isinstance(value, date): return value.isoformat() if isinstance(value, Mapping): return {str(k): normalise(v) for k, v in value.items()} if isinstance(value, list | tuple | set | frozenset): return [normalise(v) for v in value] return norm_text(str(value)) def canonical(value: Any) -> bytes: """json.dumps(sort_keys, compact separators, ensure_ascii False) of the normalised value, UTF-8.""" return json.dumps(normalise(value), sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") def sha256_hex(value: Any) -> str: return hashlib.sha256(canonical(value)).hexdigest() def drawing_key(dwg: Any) -> str | None: """A drawing number for comparison: upper-cased, every whitespace removed.""" s = norm_text(dwg) if s is None: return None s = "".join(s.split()).upper() return s or None def leg_key(base: Any, lot: Any, split: Any, sub: Any, seq: Any) -> str: """'BASE|LOT|SPLIT|SUB|SEQ' with the ID parts upper-cased (as text) and seq as an integer.""" parts = [norm_id(x) or "" for x in (base, lot, split, sub)] return "|".join([*parts, str(int(seq)) if seq is not None and seq != "" else ""]) # ---------------------------------------------------------------- the fingerprint @dataclass(frozen=True) class Fingerprint: full: str parts: dict[str, str] @dataclass(frozen=True) class Change: major: frozenset[str] minor: frozenset[str] @property def changed(self) -> bool: return bool(self.major or self.minor) @property def is_minor(self) -> bool: return bool(self.minor) and not self.major def part_values(c: Candidate) -> dict[str, Any]: """The normalised value of every part, before hashing (tests and the diff text read these).""" legs = sorted( ( [ norm_id(lg.key.base) or "", norm_id(lg.key.lot) or "", norm_id(lg.key.split) or "", norm_id(lg.key.sub) or "", int(lg.seq), ] for lg in c.legs ), key=lambda x: (x[0], x[1], x[2], x[3], x[4]), ) qty: dict[str, Any] = {} tgt: dict[str, Any] = {} for lg in c.legs: k = leg_key(lg.key.base, lg.key.lot, lg.key.split, lg.key.sub, lg.seq) qty[k] = norm_decimal(lg.qty) tgt[k] = norm_decimal(lg.tgt) values: dict[str, Any] = { "legs": legs, "qty": qty, "rev": norm_text(c.rev), "drawing": drawing_key(c.drawing), "tgt": tgt, "want": c.want.isoformat() if c.want else None, } if c.bom is not None: values["bom"] = bom_sig(c.bom) return values def bom_sig(bom: Any) -> list[list[Any]]: """The old matSig: sorted [item, part_id, qty] of every line (BomLine objects or dicts).""" out: list[list[Any]] = [] for b in bom or []: get = b.get if isinstance(b, Mapping) else lambda k, _b=b: getattr(_b, k, None) out.append([norm_text(get("item")) or "", norm_id(get("part_id")) or "", norm_decimal(get("qty"))]) out.sort(key=lambda x: (x[0], x[1], x[2] or "")) return out def _full(parts: Mapping[str, str]) -> str: return sha256_hex(dict(parts)) def compute(c: Candidate) -> Fingerprint: parts = {name: sha256_hex(v) for name, v in part_values(c).items()} return Fingerprint(full=_full(parts), parts=parts) def merge(before: Mapping[str, str] | None, after: Fingerprint) -> Fingerprint: """Carry forward parts the new side does not know (the BOM when materials are unknown this run).""" if not before: return after parts = dict(after.parts) for name in PARTS: if name not in parts and name in before: parts[name] = str(before[name]) if parts == after.parts: return after return Fingerprint(full=_full(parts), parts=parts) def classify(before: Mapping[str, str] | None, after: Mapping[str, str]) -> Change: """Which parts changed. A part the new side lacks (unknown this run) is skipped. No `before`: all changed.""" major: set[str] = set() minor: set[str] = set() for name in PARTS: if name not in after: continue if before is not None and before.get(name) == after[name]: continue (major if name in MAJOR else minor).add(name) return Change(frozenset(major), frozenset(minor))