llm/column_units.py
Rama main · réplica de lectura del repositorio original.
# SPDX-License-Identifier: BUSL-1.1
# Copyright (c) 2024-2026 Ali Sadeghi Aghili
"""Which numbers of a result are money, for the wording of the answer summary.
The summary prompt used to say that every monetary value is in Rials and that
amounts are always written with the unit Rial/ریال. A model told that attaches
the unit to every number it sees: a total traded *volume* came back as
"33,493,503,232 Rial". The rule is now "Rial only for money", and this module
tells the model, per result column, which side of the rule each column is on,
decided here and not left to the model whenever the schema or the column name
settles it.
How a column is classified
--------------------------
In this order, the first that applies wins:
1. ``column_units:`` of ``schema.yaml`` (:func:`schema_data.registry.get_column_units`):
``rial`` is money, ``none`` is not. The operator's word.
2. The column name: a currency word (``Rial``, ``ریال``) is money; then a word
for a count, quantity, volume, weight, percentage, identifier, code or date
is not money; then a word for a price, value, amount, cost, fee or the like
(``قیمت``, ``ارزش``, ``مبلغ``, ``کارمزد``) is money. A count word beats a
money word on purpose: ``FeeCount`` is a number of fees.
3. Otherwise the column is *unknown* and the model decides: Rial only if the
numbers are clearly an amount of money.
Only columns that hold numbers are listed; a text or date column never gets a
currency unit.
Nothing here reads a row value except to see whether it is a number, and
nothing here can fail the summary: a ``schema.yaml`` that cannot be read means
the name rule alone.
"""
from __future__ import annotations
import logging
import numbers
import re
from collections.abc import Mapping, Sequence
logger = logging.getLogger(__name__)
__all__ = [
"MONEY",
"NOT_MONEY",
"UNKNOWN",
"classify_column",
"column_hints",
"declared_units",
"numeric_columns",
]
#: The three answers :func:`classify_column` gives.
MONEY = "money"
NOT_MONEY = "not-money"
UNKNOWN = "unknown"
#: Words of a column name that name the currency itself.
_CURRENCY_WORDS = frozenset({"rial", "rials", "irr", "ریال"})
#: Words for a number that is not an amount of money. Latin words are compared
#: in lower case; ``s`` of a plural is dropped before the comparison.
_NOT_MONEY_WORDS = frozenset({
"count", "cnt", "qty", "quantity", "number", "num", "volume", "vol",
"weight", "wt", "percent", "pct", "percentage", "ratio",
"id", "uid", "code", "date", "year", "month", "day", "rank",
"حجم", "تعداد", "مقدار", "وزن", "درصد", "نسبت",
"کد", "شناسه", "شماره", "تاریخ", "سال", "ماه", "روز", "رتبه",
})
#: Words for an amount of money.
_MONEY_WORDS = frozenset({
"price", "value", "amount", "amt", "cost", "fee", "payment", "salary",
"wage", "revenue", "income", "profit", "balance", "debt", "turnover",
"commission", "tax", "budget", "worth",
"ارزش", "قیمت", "مبلغ", "کارمزد", "هزینه", "بها", "درآمد", "سود", "بدهی", "مانده",
})
_CAMEL_BREAK = re.compile(r"(?<=[a-z0-9])(?=[A-Z])|(?<=[A-Z])(?=[A-Z][a-z])")
_DIGITS = re.compile(r"\d+")
_SEPARATORS = re.compile(r"[\s_\-./]+")
#: Persian endings that leave the word itself: plural ``ها``/``های`` and the
#: adjective ``ی`` (``ریالی`` is "in Rials").
_PERSIAN_ENDINGS = ("ها", "های", "ی")
def _words(name: str) -> list[str]:
"""The words of a column name, lower case: split at ``_``, spaces, ZWNJ and camel-case humps.
>>> _words("TotalTradeVolume"), _words("trade_value"), _words("RingID")
(['total', 'trade', 'volume'], ['trade', 'value'], ['ring', 'id'])
>>> _words("حجم معاملات"), _words("ارزش‌ها")
(['حجم', 'معاملات'], ['ارزش', 'ها'])
"""
spaced = _DIGITS.sub(" ", _CAMEL_BREAK.sub(" ", name))
return [word.lower() for word in _SEPARATORS.split(spaced) if word]
def _is_one_of(word: str, vocabulary: frozenset[str]) -> bool:
"""Whether *word* is in *vocabulary*, allowing an English plural or a Persian ending.
>>> _is_one_of("prices", _MONEY_WORDS), _is_one_of("ریالی", _CURRENCY_WORDS)
(True, True)
>>> _is_one_of("valid", _MONEY_WORDS), _is_one_of("paid", _NOT_MONEY_WORDS)
(False, False)
"""
if word in vocabulary:
return True
if word.endswith("s") and word[:-1] in vocabulary:
return True
return any(word.endswith(end) and word[: -len(end)] in vocabulary for end in _PERSIAN_ENDINGS)
def classify_column(name: str, declared: Mapping[str, str] | None = None) -> str:
"""Whether the numbers of result column *name* are money.
Parameters
----------
name:
The column's name in the result (an alias, as the query wrote it).
declared:
``{lower-case column name: "rial" | "none"}`` from ``schema.yaml``
(:func:`declared_units`); ``None`` or empty for none.
Returns
-------
str
:data:`MONEY`, :data:`NOT_MONEY` or :data:`UNKNOWN`.
Examples
--------
>>> classify_column("TotalVolume"), classify_column("TotalValue"), classify_column("Total")
('not-money', 'money', 'unknown')
>>> classify_column("حجم کل معاملات"), classify_column("ارزش معاملات")
('not-money', 'money')
>>> classify_column("VolumeInRial"), classify_column("FeeCount"), classify_column("RingID")
('money', 'not-money', 'not-money')
>>> classify_column("TotalValue", {"totalvalue": "none"})
'not-money'
>>> classify_column("Total", {"total": "rial"})
'money'
"""
unit = (declared or {}).get(name.lower())
if unit == "rial":
return MONEY
if unit == "none":
return NOT_MONEY
words = _words(name)
if any(_is_one_of(word, _CURRENCY_WORDS) for word in words):
return MONEY
if any(_is_one_of(word, _NOT_MONEY_WORDS) for word in words):
return NOT_MONEY
if any(_is_one_of(word, _MONEY_WORDS) for word in words):
return MONEY
return UNKNOWN
def declared_units() -> dict[str, str]:
"""``column_units:`` of the loaded ``schema.yaml``; ``{}`` when there is none to read.
Never raises: a deployment without a readable ``schema.yaml`` still gets a
summary, worded from the column names alone.
"""
try:
from schema_data.registry import get_column_units
return dict(get_column_units())
except Exception as exc: # noqa: BLE001 - see the docstring
logger.debug("No column units from schema.yaml (%s); using column names only", exc)
return {}
def numeric_columns(rows: Sequence[Mapping[str, object]]) -> list[str]:
"""The columns, in the order the rows have them, that hold a number in at least one row.
A boolean is not a number here.
>>> numeric_columns([{"a": "x", "b": 1, "c": True}, {"a": "y", "b": None, "d": 2.5}])
['b', 'd']
"""
found: dict[str, None] = {}
for row in rows:
for key, value in row.items():
if isinstance(value, numbers.Number) and not isinstance(value, bool):
found.setdefault(str(key), None)
return list(found)
def column_hints(rows: Sequence[Mapping[str, object]], declared: Mapping[str, str] | None = None) -> str:
"""The prompt lines that say which numeric columns of *rows* are money.
Parameters
----------
rows:
The rows shown to the model.
declared:
See :func:`classify_column`.
Returns
-------
str
A short block ending in a newline, or ``""`` when no column holds a number.
Examples
--------
>>> print(column_hints([{"Name": "A", "TotalVolume": 5, "TotalValue": 9, "Total": 3}]), end="")
Which numeric columns of the result are money (decided from the schema or the column name):
- Money, in Rials (write Rial/ریال): TotalValue
- Not money, plain numbers (no Rial/ریال, no other currency): TotalVolume
- Not known (write Rial/ریال only if the numbers are clearly an amount of money): Total
>>> column_hints([{"Name": "A"}])
''
"""
groups: dict[str, list[str]] = {MONEY: [], NOT_MONEY: [], UNKNOWN: []}
for name in numeric_columns(rows):
groups[classify_column(name, declared)].append(name)
lines = [
(MONEY, "- Money, in Rials (write Rial/ریال): "),
(NOT_MONEY, "- Not money, plain numbers (no Rial/ریال, no other currency): "),
(UNKNOWN, "- Not known (write Rial/ریال only if the numbers are clearly an amount of money): "),
]
body = [label + ", ".join(groups[kind]) for kind, label in lines if groups[kind]]
if not body:
return ""
header = "Which numeric columns of the result are money (decided from the schema or the column name):"
return "\n".join([header, *body]) + "\n"