Spaces:
Running
Running
File size: 5,911 Bytes
ebf3f60 2ac4af1 ebf3f60 f81d9cb ebf3f60 2ac4af1 ebf3f60 f81d9cb ebf3f60 f81d9cb ebf3f60 f81d9cb ebf3f60 f81d9cb ebf3f60 f81d9cb ebf3f60 f81d9cb ebf3f60 2ac4af1 f81d9cb ebf3f60 f81d9cb ebf3f60 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 | """In-place monetary-field parsing for the feature-extraction pipeline.
When a feature-extract request declares ``monetary_fields`` — an object name
plus a list of field names — the extracted ``data`` payload is post-processed
so those fields become real numbers instead of raw strings, and a per-field
status report is returned so callers can see exactly what happened to each
declared field.
Example
-------
Request config::
{"object_name": "invoice", "field_names": ["total", "invoice_no"]}
Extracted data::
{"invoice": [{"number": "INV-2026-00125", "total": "1250.75"}]}
After :func:`apply_monetary_fields` the same dict has ``"total": 1250.75``
(price-parsed) and the returned report explains each field:
* ``total`` -> status ``parsed``, parsed_value 1250.75
* ``invoice_no`` -> status ``not_found`` (no such field in the object)
"""
from __future__ import annotations
from typing import Any, List, Optional
from pydantic import BaseModel, Field
from app.services.price_parser import parse_amount
class MonetaryFieldStatus(BaseModel):
"""Outcome of one declared monetary field, per extracted object item."""
object_name: str
"""Container key the field was looked up in (from the request config)."""
field: str
"""Monetary field name (from the request config)."""
item_index: int = 0
"""Index of the object inside the container this report refers to."""
status: str = Field(
default="not_found",
description="One of: parsed, not_found, not_parsable, skipped.",
)
value: Optional[str] = None
"""The raw value as extracted (string form), when present."""
parsed_value: Optional[float] = None
"""Parsed numeric amount when status == 'parsed'."""
error: Optional[str] = None
"""Human-readable reason when the field was not parsed."""
def _status(object_name: str, field: str, item_index: int, **kw) -> MonetaryFieldStatus:
return MonetaryFieldStatus(object_name=object_name, field=field, item_index=item_index, **kw)
def apply_monetary_fields(data: Any, object_name: str, field_names: List[str]) -> List[MonetaryFieldStatus]:
"""Price-parse the configured monetary fields inside ``data``, in place.
``object_name`` selects the container key in ``data`` (e.g. ``"invoice"``);
its value may be a list of objects or a single object. For every listed
``field_names`` entry inside each object:
* value missing -> status ``not_found`` (error ``field 'x' not found``)
* value ``None`` / blank -> status ``skipped`` (error ``value is empty``)
* value that does not parse -> status ``not_parsable`` with the raw value
* value that parses -> replaced with the numeric amount, status ``parsed``
Returns one :class:`MonetaryFieldStatus` per (object item, field).
"""
report: List[MonetaryFieldStatus] = []
if not isinstance(data, dict) or not object_name:
for field in field_names:
report.append(
_status(
object_name,
field,
0,
status="not_found",
error=f"object '{object_name}' not found in extracted data",
)
)
return report
container = data.get(object_name)
if isinstance(container, list):
items: List[Any] = container
elif isinstance(container, dict):
items = [container]
else:
for field in field_names:
report.append(
_status(
object_name,
field,
0,
status="not_found",
error=f"object '{object_name}' not found in extracted data",
)
)
return report
for idx, item in enumerate(items):
if not isinstance(item, dict):
for field in field_names:
report.append(
_status(
object_name,
field,
idx,
status="not_found",
error="item is not an object",
)
)
continue
for field in field_names:
if field not in item:
report.append(
_status(
object_name,
field,
idx,
status="not_found",
error=f"field '{field}' not found",
)
)
continue
value = item[field]
if value is None or (isinstance(value, str) and not value.strip()):
report.append(
_status(
object_name,
field,
idx,
status="skipped",
error="value is empty",
)
)
continue
parsed = parse_amount(value)
if parsed is None:
report.append(
_status(
object_name,
field,
idx,
status="not_parsable",
value=str(value),
error=f"value is non-parsable: {value}",
)
)
continue
item[field] = parsed.amount_float
report.append(
_status(
object_name,
field,
idx,
status="parsed",
value=str(value),
parsed_value=parsed.amount_float,
)
)
return report
__all__ = ["MonetaryFieldStatus", "apply_monetary_fields"]
|