PacketStation/server/packetstation/feed/fingerprint.py

"""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))