89 lines
2.7 KiB
Python
89 lines
2.7 KiB
Python
"""Metrics — quality signals without golden dataset.
|
|
|
|
Checks that work immediately:
|
|
- Arithmetic: sum == price * qty (free signal of LLM/data errors)
|
|
- JSON fix rate: how often _safe_json_parse has to repair LLM output
|
|
"""
|
|
from decimal import Decimal, InvalidOperation
|
|
|
|
|
|
def check_arithmetic(ops: list) -> list[dict]:
|
|
"""Check sum == price * qty for ADD/UPDATE operations.
|
|
Returns list of mismatches: [{action, name, price, qty, expected_sum, actual_sum, diff}]
|
|
"""
|
|
mismatches = []
|
|
for op in ops:
|
|
action = op.get("action", "")
|
|
if action not in ("ADD", "UPDATE"):
|
|
continue
|
|
|
|
row = op.get("new_row") or op.get("new_values") or {}
|
|
price = _to_decimal(row.get("price"))
|
|
qty = _to_decimal(row.get("qty"))
|
|
actual_sum = _to_decimal(row.get("sum"))
|
|
|
|
if price is None or qty is None or actual_sum is None:
|
|
continue # can't check without all three
|
|
|
|
expected = price * qty
|
|
if expected != actual_sum:
|
|
mismatches.append({
|
|
"action": action,
|
|
"name": row.get("name", "")[:100],
|
|
"price": float(price),
|
|
"qty": float(qty),
|
|
"expected_sum": float(expected),
|
|
"actual_sum": float(actual_sum),
|
|
"diff": float(actual_sum - expected),
|
|
})
|
|
return mismatches
|
|
|
|
|
|
class ClassifyMetrics:
|
|
"""Track _safe_json_parse fix rate per batch."""
|
|
def __init__(self):
|
|
self.total = 0
|
|
self.fixes = 0 # how many times JSON needed repair
|
|
|
|
def record(self, needed_fix: bool):
|
|
self.total += 1
|
|
if needed_fix:
|
|
self.fixes += 1
|
|
|
|
@property
|
|
def fix_rate(self) -> float:
|
|
return self.fixes / self.total if self.total else 0.0
|
|
|
|
def summary(self) -> dict:
|
|
return {
|
|
"total_classifications": self.total,
|
|
"json_fixes": self.fixes,
|
|
"json_fix_rate": round(self.fix_rate, 3),
|
|
}
|
|
|
|
|
|
def _to_decimal(val) -> Decimal | None:
|
|
"""Safe conversion to Decimal with number normalization."""
|
|
if val is None or val == "":
|
|
return None
|
|
try:
|
|
s = str(val).replace("\xa0", "").replace(" ", "").replace(",", ".")
|
|
return Decimal(s)
|
|
except (InvalidOperation, ValueError):
|
|
return None
|
|
|
|
|
|
def normalize_date(ds: str) -> str | None:
|
|
"""Normalize date to YYYY-MM-DD. Handles DD.MM.YYYY, YYYY-MM-DD, etc."""
|
|
if not ds:
|
|
return None
|
|
import re
|
|
# DD.MM.YYYY → YYYY-MM-DD
|
|
m = re.match(r"(\d{2})\.(\d{2})\.(\d{4})", ds)
|
|
if m:
|
|
return f"{m.group(3)}-{m.group(2)}-{m.group(1)}"
|
|
# YYYY-MM-DD — already canonical
|
|
if re.match(r"\d{4}-\d{2}-\d{2}", ds):
|
|
return ds
|
|
return ds # return as-is if unrecognized format
|