Spaces:
Running
Running
File size: 16,105 Bytes
ebb9029 cd6f706 ebb9029 cd6f706 ebb9029 b7dddbe ebb9029 cd6f706 ebb9029 bd469c1 cd6f706 ebb9029 ebf3f60 ebb9029 bd469c1 ebb9029 cd6f706 ebf3f60 2ac4af1 ebf3f60 2ac4af1 ebf3f60 cd6f706 ebf3f60 f81d9cb ebf3f60 2ac4af1 f81d9cb cd6f706 ebf3f60 cd6f706 2ac4af1 cd6f706 ebf3f60 2ac4af1 ebf3f60 cd6f706 ebf3f60 cd6f706 ebf3f60 cd6f706 | 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 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 | from __future__ import annotations
import asyncio
import time
from typing import Annotated, Any, Dict, List, Optional
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, Field
from app.config import get_settings
from app.core.logger import get_logger
from app.core.thread_pool import run_in_executor
from app.services.gliner_service import gliner_service
from app.services.json_service import extract_json
from app.services.monetary_field_service import MonetaryFieldStatus, apply_monetary_fields
logger = get_logger(__name__)
router = APIRouter()
MAX_CONTENT_LENGTH = 10_000_000
class ExtractJsonRequest(BaseModel):
content: str = Field(
...,
description="Dirty string content potentially containing JSON wrapped in markdown, conversational text, etc.",
min_length=1,
)
limit: Optional[int] = Field(
default=None,
ge=1,
le=100,
description="Maximum number of JSON objects to extract. Omit for all.",
)
mode: str = Field(
default="all",
pattern=r"^(first|all)$",
description="'first' returns only the first JSON object; 'all' returns all extracted objects.",
)
class ExtractJsonResponse(BaseModel):
success: bool
time_ms: float
data: Any = None
count: int = 0
error_message: Optional[str] = None
@router.post(
"/json/extract",
response_model=ExtractJsonResponse,
summary="Extract JSON from dirty/markdown content",
description=(
"Accepts string content that may contain JSON embedded in markdown code fences "
"(```json), XML-style <json> tags, or mixed with conversational text. "
"Returns cleaned, parsed JSON objects. Handles malformed JSON via a repair pipeline "
"that fixes trailing commas, unquoted keys, single-quote strings, JS comments, etc."
),
)
async def extract_json_endpoint(
body: ExtractJsonRequest,
) -> ExtractJsonResponse:
start = time.perf_counter()
content_length = len(body.content)
if content_length > MAX_CONTENT_LENGTH:
elapsed = round((time.perf_counter() - start) * 1000, 3)
raise HTTPException(
status_code=413,
detail=ExtractJsonResponse(
success=False,
time_ms=elapsed,
data=None,
count=0,
error_message=f"Content exceeds maximum length of {MAX_CONTENT_LENGTH:,} characters.",
).model_dump(),
)
effective_limit = 1 if body.mode == "first" else body.limit
result = await run_in_executor(extract_json, body.content, limit=effective_limit)
elapsed = round((time.perf_counter() - start) * 1000, 3)
if not result.success:
logger.warning(
"JSON extraction returned no results",
extra={
"input_length": content_length,
"mode": body.mode,
"time_ms": elapsed,
},
)
raise HTTPException(
status_code=422,
detail=ExtractJsonResponse(
success=False,
time_ms=elapsed,
data=None,
count=0,
error_message=result.error_message or "No JSON content could be extracted from the provided input.",
).model_dump(),
)
response_data = result.data[0] if body.mode == "first" else result.data
logger.info(
"JSON extraction successful",
extra={
"count": result.total_extracted,
"method": result.extraction_method,
"input_length": content_length,
"time_ms": elapsed,
},
)
return ExtractJsonResponse(
success=True,
time_ms=elapsed,
data=response_data,
count=result.total_extracted,
error_message=None,
)
class NoAiExtractRequest(BaseModel):
content: str = Field(
...,
description=(
"Raw text to extract from (invoice text, OCR output, emails, etc.). "
"No external AI/LLM API is called."
),
min_length=1,
)
mode: str = Field(
default="json",
pattern=r"^(json|entities)$",
description=(
"'json' extracts structured fields using a GLiNER2 `structure` schema; "
"'entities' extracts zero-shot entities using a `labels` list."
),
)
structure: Optional[Dict[str, Any]] = Field(
default=None,
description=(
"Required for mode='json'. GLiNER2 structure schema mapping a parent "
"key to field specs, e.g. "
'{"invoice": ["number::str::Invoice number", "total::str::Total amount"]}. '
"Field spec: name::dtype::choices::description."
),
)
labels: Optional[List[str]] = Field(
default=None,
description="Required for mode='entities'. Entity types to detect, e.g. ['person', 'company', 'location'].",
)
threshold: float = Field(
default=0.5,
ge=0.0,
le=1.0,
description="Confidence threshold (0.0-1.0). Lower includes more candidates.",
)
monetary_fields: Optional["MonetaryFieldsConfig"] = Field(
default=None,
description=(
"OPTIONAL. Declare which extracted fields hold monetary values so they "
"are price-parsed in place: `field_names` lists the monetary fields "
"inside each object (e.g. ['total']) and `object_name` selects the "
"container key in the extracted `data` -- it is inferred from the "
"`structure` parent key when omitted (mode='json') and validated "
"against the structure when provided. Found values are replaced with "
"their parsed numeric amount; missing/blank/unparseable values are "
"reported per field. Omit to leave the extracted data untouched."
),
)
class MonetaryFieldsConfig(BaseModel):
"""Declares which extracted fields are monetary so they are price-parsed."""
object_name: Optional[str] = Field(
default=None,
description=(
"Container key in the extracted `data` whose value holds the objects to "
"process, e.g. 'invoice'. OPTIONAL for mode='json': inferred from the "
"single parent key of `structure`. When provided it must match a "
"structure object name. Required for mode='entities'."
),
)
field_names: List[str] = Field(
...,
min_length=1,
description="Monetary field names inside each object, e.g. ['total', 'cgst'].",
)
class NoAiExtractResponse(BaseModel):
success: bool
time_ms: float
mode: str
data: Any = None
count: int = 0
error_message: Optional[str] = None
monetary_fields: List[MonetaryFieldStatus] = Field(
default_factory=list,
description=(
"Per-field outcome of the optional `monetary_fields` config: status "
"is 'parsed', 'not_found', 'not_parsable' or 'skipped', with the "
"raw `value`, the `parsed_value` and a human-readable error. Empty "
"when no config was sent."
),
)
def _count_extracted(result: Any) -> int:
if not isinstance(result, dict):
return 0
total = 0
for parent, items in result.items():
if isinstance(items, list):
total += len(items)
elif isinstance(items, dict):
total += 1
return total
@router.post(
"/json/feature-extract",
response_model=List[NoAiExtractResponse],
summary="Batch-extract structured JSON / entities with a local on-device model (no external AI)",
description=(
"Send up to 5 requests as a JSON array (1-5 items). All items are "
"processed concurrently on the shared thread pool, bounded by the local "
"model's concurrency limit. No external AI/LLM API is contacted -- ideal "
"for private or invoice/OCR data. mode='json' uses a structure schema to "
"pull named fields; mode='entities' detects a flat list of entity types. "
"Each item reports its own success/error. "
"Returns HTTP 503 if the model failed to load or is disabled. "
"OPTIONAL per-item `monetary_fields` ({\"object_name\", \"field_names\"}) "
"price-parses the declared fields in place inside the extracted data."
),
)
async def no_ai_extract_endpoint(
body: Annotated[
List[NoAiExtractRequest],
Field(
min_length=1,
max_length=5,
description="Array of up to 5 extraction requests, processed concurrently.",
),
],
) -> List[NoAiExtractResponse]:
settings = get_settings()
total_start = time.perf_counter()
if not settings.gliner_enabled:
raise HTTPException(
status_code=503,
detail=NoAiExtractResponse(
success=False,
time_ms=0.0,
mode="",
data=None,
count=0,
error_message="The local extraction service is disabled.",
).model_dump(),
)
if not gliner_service.is_loaded():
try:
await run_in_executor(gliner_service.load_model)
except Exception:
logger.exception("Lazy GLiNER2 model load failed on request")
raise HTTPException(
status_code=503,
detail=NoAiExtractResponse(
success=False,
time_ms=round((time.perf_counter() - total_start) * 1000, 3),
mode="",
data=None,
count=0,
error_message="The local extraction model is unavailable. Please try again later.",
).model_dump(),
)
async def _process_one(index: int, item: NoAiExtractRequest) -> NoAiExtractResponse:
start = time.perf_counter()
if item.mode == "json" and not item.structure:
return NoAiExtractResponse(
success=False,
time_ms=round((time.perf_counter() - start) * 1000, 3),
mode=item.mode,
data=None,
count=0,
error_message="mode='json' requires a non-empty 'structure' schema.",
)
if item.mode == "entities" and not item.labels:
return NoAiExtractResponse(
success=False,
time_ms=round((time.perf_counter() - start) * 1000, 3),
mode=item.mode,
data=None,
count=0,
error_message="mode='entities' requires a non-empty 'labels' list.",
)
if len(item.content) > gliner_service.max_content_length:
return NoAiExtractResponse(
success=False,
time_ms=round((time.perf_counter() - start) * 1000, 3),
mode=item.mode,
data=None,
count=0,
error_message=(
f"Content exceeds maximum length of {gliner_service.max_content_length:,} characters."
),
)
# Resolve the monetary-fields target object name. It is optional for
# mode='json' (inferred from the single `structure` parent key) and
# validated against the structure when provided.
monetary_config = item.monetary_fields
object_name: Optional[str] = None
if monetary_config is not None:
object_name = monetary_config.object_name
if item.mode == "json" and isinstance(item.structure, dict):
structure_names = [k for k in item.structure if isinstance(k, str)]
if object_name is None:
if len(structure_names) == 1:
object_name = structure_names[0]
else:
return NoAiExtractResponse(
success=False,
time_ms=round((time.perf_counter() - start) * 1000, 3),
mode=item.mode,
data=None,
count=0,
error_message=(
"monetary_fields.object_name is required when the structure "
"has multiple object names: " + ", ".join(structure_names)
),
)
elif object_name not in structure_names:
return NoAiExtractResponse(
success=False,
time_ms=round((time.perf_counter() - start) * 1000, 3),
mode=item.mode,
data=None,
count=0,
error_message=(
f"monetary_fields.object_name '{object_name}' does not match "
"the structure object name(s): " + ", ".join(structure_names)
),
)
elif item.mode == "entities" and object_name is None:
return NoAiExtractResponse(
success=False,
time_ms=round((time.perf_counter() - start) * 1000, 3),
mode=item.mode,
data=None,
count=0,
error_message=(
"monetary_fields.object_name is required for mode='entities' "
"(no structure to infer it from)"
),
)
try:
if item.mode == "json":
result = await run_in_executor(
gliner_service.extract_json, item.content, item.structure, item.threshold
)
else:
result = await run_in_executor(
gliner_service.extract_entities, item.content, item.labels, item.threshold
)
except Exception:
logger.exception("GLiNER2 inference failed for item %s", index)
return NoAiExtractResponse(
success=False,
time_ms=round((time.perf_counter() - start) * 1000, 3),
mode=item.mode,
data=None,
count=0,
error_message="Extraction failed. Please try again later.",
)
elapsed = round((time.perf_counter() - start) * 1000, 3)
# OPTIONAL post-processing: price-parse the declared monetary fields
# in place inside the extracted data (e.g. "1250.75" -> 1250.75) and
# report the per-field outcome so callers can understand any failures.
monetary_report: List[MonetaryFieldStatus] = []
parsed_monetary = 0
if item.monetary_fields is not None:
monetary_report = apply_monetary_fields(
result,
object_name or "",
item.monetary_fields.field_names,
)
parsed_monetary = sum(1 for s in monetary_report if s.status == "parsed")
logger.info(
"Feature-extract item extracted",
extra={
"index": index,
"mode": item.mode,
"count": _count_extracted(result),
"monetary_fields_parsed": parsed_monetary,
"time_ms": elapsed,
},
)
return NoAiExtractResponse(
success=True,
time_ms=elapsed,
mode=item.mode,
data=result,
count=_count_extracted(result),
error_message=None,
monetary_fields=monetary_report,
)
results = await asyncio.gather(
*[_process_one(index, item) for index, item in enumerate(body)]
)
logger.info(
"Feature-extract batch processed",
extra={
"items": len(body),
"time_ms": round((time.perf_counter() - total_start) * 1000, 3),
},
)
return list(results)
|