"""Batch amount / price parsing API. Exposes the pure amount parser (:mod:`app.services.price_parser`) as a batch endpoint: send up to 100 raw text fields and get every candidate amount plus the single best one back per item, processed concurrently. POST /api/v1/price-parser [{"text": "I want to go home with ssja^67.00 and $896,009.0 0"}] -> [{"success": true, "count": 2, "amounts": [{"amount": 896009.0, "currency": "$", ...}, ...], "best": {"amount": 896009.0, ...}, ...}] """ from __future__ import annotations import asyncio import time from typing import Annotated, List, Optional from fastapi import APIRouter from pydantic import BaseModel, Field from app.core.logger import get_logger from app.core.thread_pool import thread_pool from app.services.price_parser import CONFIDENCE_FLOOR, extract_amounts, parse_amount logger = get_logger(__name__) router = APIRouter() MAX_BATCH_SIZE = 100 MAX_AMOUNTS_PER_ITEM = 100 class PriceParseRequest(BaseModel): text: str = Field( ..., min_length=1, max_length=1_000_000, description="Raw text field to parse amounts from (OCR output, invoice lines, emails, ...).", ) currency_hint: Optional[str] = Field( default=None, description=( "Currency to assume for values with no currency marker of their own, " "e.g. 'USD', 'руб'. Ignored when a value already carries a currency." ), ) default_currency: Optional[str] = Field( default=None, description="Currency label applied to the final result when no currency was found.", ) limit: Optional[int] = Field( default=None, ge=1, le=MAX_AMOUNTS_PER_ITEM, description="Maximum number of candidate amounts to return for this item. Omit for all.", ) class ParsedAmount(BaseModel): """One candidate amount found in a text field.""" amount: float amount_text: str raw: str currency: Optional[str] = None currency_code: Optional[str] = None confidence: float is_negative: bool class PriceParseItemResponse(BaseModel): success: bool time_ms: float text: str count: int amounts: List[ParsedAmount] = Field( default_factory=list, description="All candidate amounts found, ranked by confidence (best first).", ) best: Optional[ParsedAmount] = Field( default=None, description="The single most likely amount, or null when none was found.", ) error_message: Optional[str] = None def _to_parsed(amount) -> ParsedAmount: return ParsedAmount( amount=amount.amount_float, amount_text=amount.amount_text, raw=amount.raw, currency=amount.currency, currency_code=amount.currency_code, confidence=amount.confidence, is_negative=amount.is_negative, ) def _run_parse(text: str, limit, hint, default_currency): """CPU-bound parse work, executed on the shared thread pool.""" amounts = extract_amounts(text, limit=limit, currency_hint=hint) # Only candidates that clear the confidence floor are real amounts; # reject-word/date/quantity leftovers (confidence 0) must not count. amounts = [a for a in amounts if a.confidence >= CONFIDENCE_FLOOR] best = parse_amount(text, currency_hint=hint, default_currency=default_currency) return amounts, best def _error_response(start: float, text: str, message: str) -> PriceParseItemResponse: return PriceParseItemResponse( success=False, time_ms=round((time.perf_counter() - start) * 1000, 3), text=text, count=0, amounts=[], best=None, error_message=message, ) async def _parse_one(index: int, body: PriceParseRequest) -> PriceParseItemResponse: start = time.perf_counter() try: loop = asyncio.get_running_loop() amounts, best = await loop.run_in_executor( thread_pool, _run_parse, body.text, body.limit, body.currency_hint, body.default_currency, ) except ValueError as exc: return _error_response(start, body.text, str(exc)) except Exception as exc: # pragma: no cover - defensive logger.exception("price_parser item %s failed", index) return _error_response(start, body.text, f"parsing failed: {exc}") elapsed = round((time.perf_counter() - start) * 1000, 3) logger.info( "price_parser item parsed", extra={"index": index, "count": len(amounts), "time_ms": elapsed}, ) return PriceParseItemResponse( success=True, time_ms=elapsed, text=body.text, count=len(amounts), amounts=[_to_parsed(a) for a in amounts], best=_to_parsed(best) if best is not None else None, error_message=None, ) @router.post( "/price-parser", response_model=List[PriceParseItemResponse], summary="Parse monetary amounts from dirty text fields (batch of up to 100, processed concurrently)", description=( "Send up to 100 raw text fields as a JSON array (1-100 items). Each item " "returns every candidate amount found (ranked by confidence) plus the " "single best one, with currency, confidence and the matched substrings. " "Supports US/EU/Indian number formats, OCR distortion ('$iom89.00', " "'896,009.0 0'), magnitudes ('₹5 Cr', '1.5M'), negatives and " "currency hints. Fields with only dates/quantities/references return " "count=0 with an empty amounts array. Parsing is exact (Decimal) and " "dependency-free." ), ) async def price_parser_batch( body: Annotated[ List[PriceParseRequest], Field( min_length=1, max_length=MAX_BATCH_SIZE, description="Array of 1-100 text fields to parse, processed concurrently.", ), ], ) -> List[PriceParseItemResponse]: start = time.perf_counter() logger.info("price_parser_batch | start count=%d", len(body)) results = await asyncio.gather(*[_parse_one(index, item) for index, item in enumerate(body)]) elapsed = round((time.perf_counter() - start) * 1000, 3) succeeded = sum(1 for r in results if r.success) logger.info( "price_parser_batch | done total=%d succeeded=%d time_ms=%s", len(results), succeeded, elapsed, ) return results