app/api/v1/price_parser.py

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