1486 lines
49 KiB
Python
1486 lines
49 KiB
Python
"""API REST del microservicio de extraccion."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import uuid
|
|
from contextlib import asynccontextmanager
|
|
from typing import Any
|
|
|
|
try:
|
|
from arq import create_pool as _arq_create_pool
|
|
from arq.connections import RedisSettings as _RedisSettings
|
|
_ARQ_AVAILABLE = True
|
|
except ImportError:
|
|
_arq_create_pool = None # type: ignore[assignment]
|
|
_RedisSettings = None # type: ignore[assignment]
|
|
_ARQ_AVAILABLE = False
|
|
|
|
from fastapi import BackgroundTasks, FastAPI, File, Form, HTTPException, UploadFile
|
|
from pydantic import BaseModel, Field, model_validator
|
|
from sqlalchemy import delete, select
|
|
|
|
from api.desktop_ingest import mark_task_completed_if_done, process_desktop_ingest
|
|
|
|
from domain.batch_processor import BatchProcessor
|
|
from domain.config import extractor_llm_config, normalizer_llm_config, settings
|
|
from domain.config import get_allowed_document_types
|
|
from domain.estimator import Estimator
|
|
from domain.service import InvoiceProcessingService
|
|
from domain.strategies import OpenAICompatibleExtractor
|
|
from infrastructure.database import (
|
|
CleanedRow,
|
|
CleaningJob,
|
|
CountJobFile,
|
|
ExtractionBatch,
|
|
ExtractionRow,
|
|
Job,
|
|
Task,
|
|
init_db,
|
|
session_scope,
|
|
utcnow,
|
|
)
|
|
from infrastructure.serialization import flatten_extraction_result
|
|
from infrastructure.uploads import temp_upload_path
|
|
|
|
|
|
def build_processing_service() -> InvoiceProcessingService:
|
|
extractor_config = extractor_llm_config()
|
|
extractor = OpenAICompatibleExtractor(
|
|
api_key=str(extractor_config["api_key"]),
|
|
base_url=str(extractor_config["base_url"]),
|
|
model=str(extractor_config["model"]),
|
|
)
|
|
return InvoiceProcessingService(
|
|
extractor,
|
|
batch_processor_factory=lambda service: BatchProcessor(service),
|
|
estimator_factory=lambda service: Estimator(service),
|
|
)
|
|
|
|
|
|
processing_service = build_processing_service()
|
|
|
|
|
|
def _redis_settings_from_url(url: str):
|
|
url = url.removeprefix("redis://")
|
|
host_port, *db_parts = url.split("/")
|
|
host, *port_parts = host_port.split(":")
|
|
port = int(port_parts[0]) if port_parts else 6379
|
|
db = int(db_parts[0]) if db_parts else 0
|
|
return _RedisSettings(host=host, port=port, database=db)
|
|
|
|
|
|
# Pool ARQ global — compartido por todos los requests
|
|
_arq_pool = None
|
|
|
|
|
|
async def get_arq_pool():
|
|
global _arq_pool
|
|
if not _ARQ_AVAILABLE:
|
|
raise RuntimeError("arq no instalado; ASYNC_QUEUE requiere 'pip install arq'.")
|
|
if _arq_pool is None:
|
|
_arq_pool = await _arq_create_pool(_redis_settings_from_url(settings.redis_url))
|
|
return _arq_pool
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
init_db()
|
|
if settings.processing.async_queue:
|
|
os.makedirs(settings.queue_dir, exist_ok=True)
|
|
try:
|
|
await get_arq_pool()
|
|
except Exception as exc:
|
|
import logging as _logging
|
|
_logging.getLogger(__name__).warning(
|
|
"No se pudo conectar a Redis (%s). Usando modo sincrono como fallback.", exc
|
|
)
|
|
yield
|
|
if _arq_pool is not None:
|
|
await _arq_pool.aclose()
|
|
|
|
|
|
app = FastAPI(
|
|
title="IPSExtractor Extraction Service",
|
|
version="1.0.0",
|
|
lifespan=lifespan,
|
|
)
|
|
|
|
|
|
class DocumentType(BaseModel):
|
|
name: str
|
|
keyword: str = ""
|
|
|
|
|
|
class SchemaRequest(BaseModel):
|
|
document_path: str | None = None
|
|
root_path: str | None = None
|
|
doc_types: list[DocumentType] | None = None
|
|
instruction: str | None = None
|
|
tiff_page_range: tuple[int, int] | None = None
|
|
structure_mode: str = "radicados"
|
|
|
|
|
|
class DesktopDocument(BaseModel):
|
|
radicado: str
|
|
filename: str
|
|
doc_type: str | None = None
|
|
|
|
|
|
class ExtractionRequest(BaseModel):
|
|
source: str = "legacy"
|
|
root_path: str = ""
|
|
keywords: list[str] | None = None
|
|
doc_types: list[DocumentType] | None = None
|
|
fields: list[str] = Field(default_factory=list)
|
|
alias_map: dict[str, str] = Field(default_factory=dict)
|
|
instruction: str | None = None
|
|
tiff_page_range: tuple[int, int] | None = None
|
|
structure_mode: str = "radicados"
|
|
retry_on_empty: bool = False
|
|
radicado_filter: list[str] | None = None
|
|
documents: list[DesktopDocument] | None = None
|
|
document_count: int | None = None
|
|
job_id: str | None = None # vincular al job existente
|
|
model: str | None = None
|
|
|
|
@model_validator(mode="after")
|
|
def validate_source(self) -> "ExtractionRequest":
|
|
if self.source == "desktop":
|
|
if not self.documents:
|
|
raise ValueError("Envia documents cuando source=desktop.")
|
|
return self
|
|
if not self.root_path:
|
|
raise ValueError("Envia root_path para extraccion legacy.")
|
|
return self
|
|
|
|
|
|
class EstimateRequest(BaseModel):
|
|
root_path: str
|
|
keywords: list[str] | None = None
|
|
doc_types: list[DocumentType] | None = None
|
|
schema_map: dict[str, str] | None = Field(default=None, alias="schema")
|
|
selected_fields: list[str] | None = None
|
|
schema_by_type: dict[str, dict[str, str]] | None = None
|
|
selected_paths: list[str] | None = None
|
|
structure_mode: str = "radicados"
|
|
|
|
|
|
class ExtractionBatchRead(BaseModel):
|
|
id: str
|
|
status: str
|
|
root_path: str | None
|
|
request: dict[str, Any]
|
|
error: str | None
|
|
created_at: str
|
|
updated_at: str
|
|
|
|
|
|
class TaskRead(BaseModel):
|
|
id: str
|
|
type: str
|
|
status: str
|
|
title: str | None
|
|
request: dict[str, Any]
|
|
result: dict[str, Any]
|
|
error: str | None
|
|
owner_id: str | None
|
|
created_at: str
|
|
updated_at: str
|
|
started_at: str | None
|
|
finished_at: str | None
|
|
|
|
|
|
def serialize_batch(batch: ExtractionBatch) -> dict[str, Any]:
|
|
return {
|
|
"id": batch.id,
|
|
"status": batch.status,
|
|
"root_path": batch.root_path,
|
|
"request": batch.request,
|
|
"error": batch.error,
|
|
"created_at": batch.created_at.isoformat(),
|
|
"updated_at": batch.updated_at.isoformat(),
|
|
}
|
|
|
|
|
|
def serialize_task(task: Task) -> dict[str, Any]:
|
|
return {
|
|
"id": task.id,
|
|
"type": task.type,
|
|
"status": task.status,
|
|
"title": task.title,
|
|
"request": task.request,
|
|
"result": task.result,
|
|
"error": task.error,
|
|
"owner_id": task.owner_id,
|
|
"created_at": task.created_at.isoformat(),
|
|
"updated_at": task.updated_at.isoformat(),
|
|
"started_at": task.started_at.isoformat() if task.started_at else None,
|
|
"finished_at": task.finished_at.isoformat() if task.finished_at else None,
|
|
}
|
|
|
|
|
|
def persist_extraction_request(request: ExtractionRequest) -> dict[str, Any]:
|
|
with session_scope() as session:
|
|
batch = ExtractionBatch(
|
|
status="running",
|
|
root_path=request.root_path,
|
|
request=request.model_dump(),
|
|
updated_at=utcnow(),
|
|
)
|
|
session.add(batch)
|
|
session.flush()
|
|
batch_id = batch.id
|
|
|
|
try:
|
|
results = processing_service.process_root_with_field_selection(
|
|
request.root_path,
|
|
request.keywords,
|
|
doc_types=[dt.model_dump() for dt in request.doc_types or []] or None,
|
|
fields=request.fields,
|
|
csv_output_path=None,
|
|
instruction_override=request.instruction,
|
|
alias_map=request.alias_map,
|
|
tiff_page_range=request.tiff_page_range,
|
|
structure_mode=request.structure_mode,
|
|
retry_on_empty=request.retry_on_empty,
|
|
radicado_filter=request.radicado_filter,
|
|
)
|
|
|
|
with session_scope() as session:
|
|
row_index = 0
|
|
for item in results:
|
|
raw = item if isinstance(item, dict) else {"data": item}
|
|
rows = flatten_extraction_result(raw, request.fields, request.alias_map)
|
|
for row in rows:
|
|
row_index += 1
|
|
session.add(
|
|
ExtractionRow(
|
|
batch_id=batch_id,
|
|
radicado=str(raw.get("radicado") or row.get("radicado") or ""),
|
|
source_pdf=raw.get("_source_pdf"),
|
|
row_index=row_index,
|
|
row_json=row,
|
|
raw_json=raw,
|
|
)
|
|
)
|
|
batch = session.get(ExtractionBatch, batch_id)
|
|
if batch is None:
|
|
raise RuntimeError("No se encontro el lote recien creado.")
|
|
batch.status = "completed"
|
|
batch.updated_at = utcnow()
|
|
session.flush()
|
|
serialized = serialize_batch(batch)
|
|
serialized["row_count"] = row_index
|
|
return serialized
|
|
except Exception as exc:
|
|
with session_scope() as session:
|
|
batch = session.get(ExtractionBatch, batch_id)
|
|
if batch is not None:
|
|
batch.status = "failed"
|
|
batch.error = str(exc)
|
|
batch.updated_at = utcnow()
|
|
raise
|
|
|
|
|
|
def run_extraction_task(task_id: str) -> None:
|
|
with session_scope() as session:
|
|
task = session.get(Task, task_id)
|
|
if task is None:
|
|
return
|
|
task.status = "running"
|
|
task.started_at = utcnow()
|
|
task.updated_at = utcnow()
|
|
request_payload = dict(task.request or {})
|
|
|
|
try:
|
|
request = ExtractionRequest.model_validate(request_payload)
|
|
batch = persist_extraction_request(request)
|
|
with session_scope() as session:
|
|
task = session.get(Task, task_id)
|
|
if task is None:
|
|
return
|
|
task.status = "completed"
|
|
task.result = {
|
|
"extraction_batch_id": batch["id"],
|
|
"row_count": batch.get("row_count", 0),
|
|
"batch": batch,
|
|
}
|
|
task.error = None
|
|
task.finished_at = utcnow()
|
|
task.updated_at = utcnow()
|
|
except Exception as exc:
|
|
with session_scope() as session:
|
|
task = session.get(Task, task_id)
|
|
if task is None:
|
|
return
|
|
task.status = "failed"
|
|
task.error = str(exc)
|
|
task.finished_at = utcnow()
|
|
task.updated_at = utcnow()
|
|
|
|
|
|
@app.get("/health")
|
|
def health() -> dict[str, str]:
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.get("/config")
|
|
def get_config() -> dict[str, Any]:
|
|
extractor_config = extractor_llm_config()
|
|
normalizer_config = normalizer_llm_config()
|
|
return {
|
|
"extractor": {
|
|
"provider": extractor_config["provider"],
|
|
"base_url": extractor_config["base_url"],
|
|
"model": extractor_config["model"],
|
|
"timeout_s": extractor_config["timeout_s"],
|
|
"api_key_configured": bool(str(extractor_config["api_key"]).strip()),
|
|
},
|
|
"normalizer": {
|
|
"provider": normalizer_config["provider"],
|
|
"base_url": normalizer_config["base_url"],
|
|
"model": normalizer_config["model"],
|
|
"timeout_s": normalizer_config["timeout_s"],
|
|
"api_key_configured": bool(str(normalizer_config["api_key"]).strip()),
|
|
},
|
|
"processing": {
|
|
"concurrency_limit": settings.processing.concurrency_limit,
|
|
"allowed_document_types": list(get_allowed_document_types()),
|
|
},
|
|
"imaging": {
|
|
"pdf_dpi": settings.imaging.pdf_dpi,
|
|
"pdf_thread_count": settings.imaging.pdf_thread_count,
|
|
"tiff_pages": settings.imaging.tiff_pages,
|
|
"poppler_path": settings.imaging.poppler_path,
|
|
},
|
|
}
|
|
|
|
|
|
@app.get("/models")
|
|
def list_models() -> dict[str, Any]:
|
|
extractor_config = extractor_llm_config()
|
|
normalizer_config = normalizer_llm_config()
|
|
try:
|
|
models = processing_service.extractor.llm.list_models()
|
|
return {
|
|
"available": models,
|
|
"configured_extractor_model": extractor_config["model"],
|
|
"configured_normalizer_model": normalizer_config["model"],
|
|
"source": extractor_config["base_url"],
|
|
}
|
|
except Exception as exc:
|
|
return {
|
|
"available": [],
|
|
"configured_extractor_model": extractor_config["model"],
|
|
"configured_normalizer_model": normalizer_config["model"],
|
|
"source": extractor_config["base_url"],
|
|
"warning": f"No se pudieron consultar modelos del backend: {exc}",
|
|
}
|
|
|
|
|
|
@app.post("/schemas")
|
|
def build_schema(request: SchemaRequest) -> dict[str, Any]:
|
|
"""Descubre campos disponibles para que el cliente API elija `fields`."""
|
|
try:
|
|
if request.document_path:
|
|
schema, sample = processing_service.build_schema_from_first(
|
|
request.document_path,
|
|
instruction=request.instruction,
|
|
tiff_page_range=request.tiff_page_range,
|
|
)
|
|
return {"schema": schema, "sample": sample}
|
|
|
|
if request.root_path and request.doc_types:
|
|
schema_by_type, sample_by_type, first_radicado = (
|
|
processing_service.build_schema_from_first_for_types(
|
|
request.root_path,
|
|
[dt.model_dump() for dt in request.doc_types],
|
|
instruction=request.instruction,
|
|
tiff_page_range=request.tiff_page_range,
|
|
structure_mode=request.structure_mode,
|
|
)
|
|
)
|
|
return {
|
|
"schema_by_type": schema_by_type,
|
|
"sample_by_type": sample_by_type,
|
|
"first_radicado": first_radicado,
|
|
}
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
except FileNotFoundError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Envia document_path o root_path con doc_types para descubrir campos.",
|
|
)
|
|
|
|
|
|
@app.post("/schemas/upload")
|
|
def build_schema_upload(
|
|
file: UploadFile = File(...),
|
|
structure_mode: str = Form("radicados"),
|
|
doc_type_name: str | None = Form(None),
|
|
instruction: str | None = Form(None),
|
|
tiff_start: int | None = Form(None),
|
|
tiff_end: int | None = Form(None),
|
|
) -> dict[str, Any]:
|
|
"""Descubre campos a partir de un documento subido desde la app desktop."""
|
|
tiff_page_range = None
|
|
if tiff_start is not None and tiff_end is not None and not (tiff_start == 0 and tiff_end == 0):
|
|
tiff_page_range = (tiff_start, tiff_end)
|
|
|
|
try:
|
|
with temp_upload_path(file) as temp_path:
|
|
schema, sample = processing_service.build_schema_from_first(
|
|
temp_path,
|
|
instruction=instruction,
|
|
tiff_page_range=tiff_page_range,
|
|
)
|
|
if doc_type_name:
|
|
return {
|
|
"schema_by_type": {doc_type_name: schema},
|
|
"sample_by_type": {doc_type_name: sample},
|
|
"structure_mode": structure_mode,
|
|
}
|
|
return {"schema": schema, "sample": sample, "structure_mode": structure_mode}
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
except FileNotFoundError as exc:
|
|
raise HTTPException(status_code=404, detail=str(exc)) from exc
|
|
|
|
|
|
@app.post("/estimate")
|
|
def estimate(request: EstimateRequest) -> dict[str, Any]:
|
|
try:
|
|
return processing_service.estimate_tokens_and_cost(
|
|
request.root_path,
|
|
request.keywords,
|
|
request.schema_map,
|
|
request.selected_fields,
|
|
doc_types=[dt.model_dump() for dt in request.doc_types or []] or None,
|
|
schema_by_type=request.schema_by_type,
|
|
selected_paths=request.selected_paths,
|
|
structure_mode=request.structure_mode,
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
|
|
|
|
@app.post("/estimate/upload")
|
|
def estimate_upload(
|
|
file: UploadFile = File(...),
|
|
doc_count: int = Form(1),
|
|
selected_paths: str | None = Form(None),
|
|
structure_mode: str = Form("radicados"),
|
|
tiff_start: int | None = Form(None),
|
|
tiff_end: int | None = Form(None),
|
|
model: str | None = Form(None),
|
|
) -> dict[str, Any]:
|
|
"""Estima tokens y costos a partir de un documento subido desde la app desktop."""
|
|
import json as _json
|
|
|
|
tiff_page_range = None
|
|
if tiff_start is not None and tiff_end is not None and not (tiff_start == 0 and tiff_end == 0):
|
|
tiff_page_range = (tiff_start, tiff_end)
|
|
|
|
paths_list: list[str] = []
|
|
if selected_paths:
|
|
try:
|
|
paths_list = _json.loads(selected_paths)
|
|
except Exception:
|
|
paths_list = []
|
|
|
|
try:
|
|
with temp_upload_path(file) as temp_path:
|
|
return processing_service.estimator.estimate_from_file(
|
|
temp_path,
|
|
doc_count=max(1, doc_count),
|
|
selected_paths=paths_list or None,
|
|
tiff_page_range=tiff_page_range,
|
|
model=model or None,
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
|
|
|
|
@app.post("/batches", response_model=ExtractionBatchRead)
|
|
def run_extraction(request: ExtractionRequest) -> dict[str, Any]:
|
|
if not request.fields:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="El API no tiene selector visual: envia fields. Usa POST /schemas para descubrirlos.",
|
|
)
|
|
|
|
try:
|
|
return persist_extraction_request(request)
|
|
except Exception as exc:
|
|
raise HTTPException(status_code=500, detail=str(exc)) from exc
|
|
|
|
|
|
@app.post("/tasks/extraction", response_model=TaskRead)
|
|
def create_extraction_task(request: ExtractionRequest, background_tasks: BackgroundTasks) -> dict[str, Any]:
|
|
if not request.fields:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Envia fields. Usa POST /schemas para descubrir campos antes de crear la tarea.",
|
|
)
|
|
|
|
if request.source == "desktop":
|
|
documents = [doc.model_dump() for doc in request.documents or []]
|
|
unique_radicados = {str(doc["radicado"]) for doc in documents}
|
|
total = len(unique_radicados) if unique_radicados else (request.document_count or len(documents))
|
|
title = f"Extraccion desktop ({total} documentos)"
|
|
with session_scope() as session:
|
|
batch = ExtractionBatch(
|
|
status="running",
|
|
root_path="desktop://upload",
|
|
request=request.model_dump(),
|
|
updated_at=utcnow(),
|
|
)
|
|
session.add(batch)
|
|
session.flush()
|
|
batch_id = batch.id
|
|
task = Task(
|
|
job_id=request.job_id,
|
|
type="extraction",
|
|
status="running",
|
|
title=title,
|
|
request=request.model_dump(),
|
|
result={
|
|
"source": "desktop",
|
|
"extraction_batch_id": batch_id,
|
|
"processed": 0,
|
|
"total": total,
|
|
"row_count": 0,
|
|
"partials": {},
|
|
"errors": [],
|
|
},
|
|
started_at=utcnow(),
|
|
updated_at=utcnow(),
|
|
)
|
|
session.add(task)
|
|
session.flush()
|
|
return serialize_task(task)
|
|
|
|
with session_scope() as session:
|
|
task = Task(
|
|
job_id=request.job_id,
|
|
type="extraction",
|
|
status="pending",
|
|
title=f"Extraccion {request.root_path}",
|
|
request=request.model_dump(),
|
|
result={},
|
|
updated_at=utcnow(),
|
|
)
|
|
session.add(task)
|
|
session.flush()
|
|
task_id = task.id
|
|
serialized = serialize_task(task)
|
|
|
|
background_tasks.add_task(run_extraction_task, task_id)
|
|
return serialized
|
|
|
|
|
|
async def _process_sync(
|
|
task_id: str,
|
|
batch_id: str,
|
|
content: bytes,
|
|
effective_name: str,
|
|
radicado: str,
|
|
doc_type: str | None,
|
|
request: dict[str, Any],
|
|
) -> dict[str, Any]:
|
|
"""Procesa un documento sincronicamente en un thread pool (modo sin Redis).
|
|
|
|
Estrategia de concurrencia:
|
|
- La llamada a la LLM (lenta) se ejecuta sin ningún lock, permitiendo
|
|
que todos los documentos llamen a OpenAI en paralelo.
|
|
- Solo la escritura del resultado en la BD usa SELECT FOR UPDATE para
|
|
evitar la race condition en el contador `processed`.
|
|
"""
|
|
import asyncio as _asyncio
|
|
import logging as _logging
|
|
import tempfile as _tempfile
|
|
from domain.service import InvoiceProcessingService
|
|
from domain.strategies import OpenAICompatibleExtractor
|
|
from domain.batch_processor import BatchProcessor
|
|
from domain.estimator import Estimator
|
|
from api.desktop_ingest import _service_for_model, _paths_for_type
|
|
from domain.profiles import get_profile
|
|
from domain.prompts import PromptBuilder
|
|
from domain.normalizer import normalize_to_schema
|
|
|
|
_log = _logging.getLogger("extraction.ingest")
|
|
|
|
ext = os.path.splitext(effective_name)[1] or ".pdf"
|
|
with _tempfile.NamedTemporaryFile(delete=False, suffix=ext) as tmp:
|
|
tmp.write(content)
|
|
temp_path = tmp.name
|
|
|
|
try:
|
|
# ── Fase 1: llamada a la LLM (sin lock, en paralelo) ──────────────
|
|
def _call_llm():
|
|
model = request.get("model") or None
|
|
svc = _service_for_model(processing_service, model)
|
|
req = dict(request)
|
|
fields = list(req.get("fields") or [])
|
|
doc_type_arg = doc_type
|
|
structure_mode = str(req.get("structure_mode") or "radicados")
|
|
doc_types = list(req.get("doc_types") or [])
|
|
prof = get_profile("invoice")
|
|
selected_paths = _paths_for_type(
|
|
fields,
|
|
doc_type_arg if structure_mode == "radicados" and doc_types else None,
|
|
)
|
|
if doc_type_arg and structure_mode == "radicados" and doc_types:
|
|
try:
|
|
prompt = PromptBuilder.build_for_type(
|
|
doc_type_arg, selected_paths, {}, profile_name=prof.get("name", "invoice")
|
|
)
|
|
except Exception:
|
|
prompt = PromptBuilder.build(selected_paths, {}, profile_name=prof.get("name", "invoice"))
|
|
else:
|
|
instr = req.get("instruction")
|
|
prompt = instr or PromptBuilder.build(selected_paths, {}, profile_name=prof.get("name", "invoice"))
|
|
tiff_range = req.get("tiff_page_range")
|
|
model_name = getattr(svc.extractor, "model", "?")
|
|
_log.info("→ OpenAI [%s] radicado=%s filename=%s", model_name, radicado, effective_name)
|
|
data, _usage = svc.extract_full_from_pdf(
|
|
temp_path,
|
|
instruction=prompt,
|
|
tiff_page_range=tiff_range,
|
|
)
|
|
tok_in = (_usage or {}).get("prompt_tokens", "?")
|
|
tok_out = (_usage or {}).get("completion_tokens", "?")
|
|
_log.info("← OpenAI [%s] radicado=%s tokens=%s→%s", model_name, radicado, tok_in, tok_out)
|
|
if isinstance(data, dict):
|
|
data["radicado"] = radicado
|
|
else:
|
|
data = {"radicado": radicado, "data": data}
|
|
norm = normalize_to_schema(data, paths=selected_paths, profile_name=prof.get("name", "invoice"))
|
|
normalized_obj = data
|
|
if isinstance(norm, dict):
|
|
normalized_obj = norm.get("normalized") or data
|
|
return normalized_obj, selected_paths
|
|
|
|
normalized_obj, selected_paths = await _asyncio.to_thread(_call_llm)
|
|
|
|
# ── Fase 2: persistir resultado (con FOR UPDATE para serializar) ───
|
|
def _persist():
|
|
from infrastructure.serialization import flatten_extraction_result
|
|
from infrastructure.database import ExtractionRow, ExtractionBatch, utcnow
|
|
req = dict(request)
|
|
fields = list(req.get("fields") or [])
|
|
alias_map = dict(req.get("alias_map") or {})
|
|
with session_scope() as session:
|
|
task = session.get(Task, task_id, with_for_update=True)
|
|
if task is None:
|
|
return {"error": "task_not_found"}
|
|
result = dict(task.result or {})
|
|
fused = {"radicado": radicado, "_source_pdf": effective_name}
|
|
if doc_type:
|
|
fused[doc_type] = normalized_obj or {}
|
|
else:
|
|
fused.update(
|
|
normalized_obj if isinstance(normalized_obj, dict)
|
|
else {"data": normalized_obj}
|
|
)
|
|
row_index = int(result.get("row_count") or 0)
|
|
for row in flatten_extraction_result(fused, fields, alias_map):
|
|
row_index += 1
|
|
session.add(ExtractionRow(
|
|
batch_id=batch_id,
|
|
radicado=radicado,
|
|
source_pdf=effective_name,
|
|
row_index=row_index,
|
|
row_json=row,
|
|
raw_json=fused,
|
|
))
|
|
result["row_count"] = row_index
|
|
processed_now = int(result.get("processed") or 0) + 1
|
|
result["processed"] = processed_now
|
|
task.result = result
|
|
task.updated_at = utcnow()
|
|
session.flush()
|
|
mark_task_completed_if_done(session, task, batch_id)
|
|
session.flush() # persiste status=completed / finished_at si aplica
|
|
_log.info(
|
|
"✓ guardado radicado=%s processed=%d/%d status=%s",
|
|
radicado, processed_now, int(result.get("total") or 0), task.status,
|
|
)
|
|
session.refresh(task)
|
|
return {
|
|
"radicado": radicado,
|
|
"filename": effective_name,
|
|
"processed": processed_now,
|
|
"total": int(result.get("total") or 0),
|
|
"task": serialize_task(task),
|
|
}
|
|
|
|
return await _asyncio.to_thread(_persist)
|
|
|
|
finally:
|
|
try:
|
|
os.unlink(temp_path)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
async def _enqueue_async(
|
|
task_id: str,
|
|
batch_id: str,
|
|
content: bytes,
|
|
effective_name: str,
|
|
radicado: str,
|
|
doc_type: str | None,
|
|
instruction: str | None,
|
|
tiff_page_range: Any,
|
|
) -> None:
|
|
"""Guarda el archivo en el volumen compartido y encola el job en ARQ."""
|
|
ext = os.path.splitext(effective_name)[1] or ".pdf"
|
|
job_file = os.path.join(settings.queue_dir, f"{uuid.uuid4().hex}{ext}")
|
|
with open(job_file, "wb") as fh:
|
|
fh.write(content)
|
|
pool = await get_arq_pool()
|
|
await pool.enqueue_job(
|
|
"process_document",
|
|
task_id=task_id,
|
|
batch_id=batch_id,
|
|
queue_path=job_file,
|
|
radicado=radicado,
|
|
filename=effective_name,
|
|
doc_type=doc_type,
|
|
instruction=instruction,
|
|
tiff_page_range=list(tiff_page_range) if tiff_page_range else None,
|
|
)
|
|
|
|
|
|
def _load_task_info(task_id: str) -> tuple[str, dict[str, Any], dict[str, Any]]:
|
|
"""Carga batch_id, request y task serializado desde BD."""
|
|
with session_scope() as session:
|
|
task = session.get(Task, task_id)
|
|
if task is None:
|
|
raise HTTPException(status_code=404, detail="Tarea no encontrada")
|
|
if task.status not in {"pending", "running"}:
|
|
raise HTTPException(status_code=400, detail="La tarea ya no acepta documentos.")
|
|
if str((task.request or {}).get("source") or "") != "desktop":
|
|
raise HTTPException(status_code=400, detail="La tarea no fue creada en modo desktop.")
|
|
batch_id = str((task.result or {}).get("extraction_batch_id") or "")
|
|
if not batch_id:
|
|
raise HTTPException(status_code=400, detail="La tarea no tiene lote asociado.")
|
|
return batch_id, dict(task.request or {}), serialize_task(task)
|
|
|
|
|
|
@app.post("/tasks/{task_id}/ingest")
|
|
async def ingest_task_document(
|
|
task_id: str,
|
|
file: UploadFile = File(...),
|
|
radicado: str = Form(...),
|
|
filename: str | None = Form(None),
|
|
doc_type: str | None = Form(None),
|
|
) -> dict[str, Any]:
|
|
"""Procesa o encola un documento.
|
|
|
|
- ASYNC_QUEUE=false (default): procesa sincronicamente, responde con la tarea actualizada.
|
|
- ASYNC_QUEUE=true: encola en Redis/ARQ, responde de inmediato (sondea GET /tasks/{id}).
|
|
"""
|
|
batch_id, request, serialized = _load_task_info(task_id)
|
|
effective_name = filename or file.filename or "documento.pdf"
|
|
content = await file.read()
|
|
|
|
if settings.processing.async_queue:
|
|
await _enqueue_async(
|
|
task_id, batch_id, content, effective_name,
|
|
radicado, doc_type, request.get("instruction"), request.get("tiff_page_range"),
|
|
)
|
|
return {"queued": True, "task": serialized}
|
|
|
|
# Modo sincrono: procesa ahora y devuelve resultado completo
|
|
return await _process_sync(task_id, batch_id, content, effective_name, radicado, doc_type, request)
|
|
|
|
|
|
@app.post("/tasks/{task_id}/ingest-many")
|
|
async def ingest_task_documents_bulk(
|
|
task_id: str,
|
|
files: list[UploadFile] = File(...),
|
|
radicados: str = Form(...),
|
|
filenames: str | None = Form(None),
|
|
doc_types: str | None = Form(None),
|
|
) -> dict[str, Any]:
|
|
"""Procesa o encola múltiples documentos en una sola petición HTTP.
|
|
|
|
Los parámetros `radicados`, `filenames` y `doc_types` son listas
|
|
separadas por `|` que se corresponden posicionalmente con `files`.
|
|
|
|
En modo sincrono (ASYNC_QUEUE=false) procesa en paralelo con asyncio.
|
|
En modo async (ASYNC_QUEUE=true) encola todos en ARQ.
|
|
"""
|
|
import asyncio as _asyncio
|
|
|
|
rad_list = radicados.split("|")
|
|
fname_list = (filenames or "").split("|") if filenames else [""] * len(files)
|
|
dtype_list = (doc_types or "").split("|") if doc_types else [None] * len(files)
|
|
|
|
if len(files) != len(rad_list):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Numero de archivos ({len(files)}) no coincide con radicados ({len(rad_list)})",
|
|
)
|
|
|
|
batch_id, request, serialized = _load_task_info(task_id)
|
|
contents = await _asyncio.gather(*[f.read() for f in files])
|
|
|
|
if settings.processing.async_queue:
|
|
await _asyncio.gather(*[
|
|
_enqueue_async(
|
|
task_id, batch_id, contents[i],
|
|
fname_list[i] or files[i].filename or f"doc_{i}.pdf",
|
|
rad_list[i], dtype_list[i] or None,
|
|
request.get("instruction"), request.get("tiff_page_range"),
|
|
)
|
|
for i in range(len(files))
|
|
])
|
|
return {"queued": len(files), "task": serialized}
|
|
|
|
# Modo sincrono: procesar todos en paralelo con semaforo para no saturar la API
|
|
sem = _asyncio.Semaphore(settings.processing.concurrency_limit)
|
|
|
|
async def _bounded(i: int) -> dict[str, Any]:
|
|
async with sem:
|
|
return await _process_sync(
|
|
task_id, batch_id, contents[i],
|
|
fname_list[i] or files[i].filename or f"doc_{i}.pdf",
|
|
rad_list[i], dtype_list[i] or None,
|
|
request,
|
|
)
|
|
|
|
results = await _asyncio.gather(*[_bounded(i) for i in range(len(files))], return_exceptions=True)
|
|
errors = [str(r) for r in results if isinstance(r, Exception)]
|
|
return {"processed": len(files) - len(errors), "errors": errors, "task": serialized}
|
|
|
|
|
|
@app.get("/tasks", response_model=list[TaskRead])
|
|
def list_tasks(type: str | None = None, status: str | None = None) -> list[dict[str, Any]]:
|
|
with session_scope() as session:
|
|
query = select(Task).order_by(Task.created_at.desc())
|
|
if type:
|
|
query = query.where(Task.type == type)
|
|
if status:
|
|
query = query.where(Task.status == status)
|
|
tasks = session.scalars(query).all()
|
|
return [serialize_task(task) for task in tasks]
|
|
|
|
|
|
@app.get("/tasks/{task_id}", response_model=TaskRead)
|
|
def get_task(task_id: str) -> dict[str, Any]:
|
|
with session_scope() as session:
|
|
task = session.get(Task, task_id)
|
|
if task is None:
|
|
raise HTTPException(status_code=404, detail="Tarea no encontrada")
|
|
return serialize_task(task)
|
|
|
|
|
|
@app.get("/tasks/{task_id}/rows")
|
|
def get_task_rows(task_id: str, limit: int = 500, offset: int = 0) -> dict[str, Any]:
|
|
with session_scope() as session:
|
|
task = session.get(Task, task_id)
|
|
if task is None:
|
|
raise HTTPException(status_code=404, detail="Tarea no encontrada")
|
|
batch_id = (task.result or {}).get("extraction_batch_id")
|
|
if not batch_id:
|
|
raise HTTPException(status_code=400, detail="La tarea aun no tiene lote de extraccion asociado")
|
|
batch = session.get(ExtractionBatch, batch_id)
|
|
if batch is None:
|
|
raise HTTPException(status_code=404, detail="Lote no encontrado")
|
|
rows = session.scalars(
|
|
select(ExtractionRow)
|
|
.where(ExtractionRow.batch_id == batch_id)
|
|
.order_by(ExtractionRow.row_index, ExtractionRow.id)
|
|
.offset(offset)
|
|
.limit(limit)
|
|
).all()
|
|
return {
|
|
"task_id": task_id,
|
|
"batch_id": batch_id,
|
|
"rows": [row.row_json for row in rows],
|
|
"limit": limit,
|
|
"offset": offset,
|
|
}
|
|
|
|
|
|
@app.get("/batches")
|
|
def list_batches() -> list[dict[str, Any]]:
|
|
with session_scope() as session:
|
|
batches = session.scalars(
|
|
select(ExtractionBatch).order_by(ExtractionBatch.created_at.desc())
|
|
).all()
|
|
return [serialize_batch(batch) for batch in batches]
|
|
|
|
|
|
@app.get("/batches/{batch_id}")
|
|
def get_batch(batch_id: str) -> dict[str, Any]:
|
|
with session_scope() as session:
|
|
batch = session.get(ExtractionBatch, batch_id)
|
|
if batch is None:
|
|
raise HTTPException(status_code=404, detail="Lote no encontrado")
|
|
return serialize_batch(batch)
|
|
|
|
|
|
@app.get("/batches/{batch_id}/rows")
|
|
def get_rows(batch_id: str, limit: int = 500, offset: int = 0) -> dict[str, Any]:
|
|
with session_scope() as session:
|
|
batch = session.get(ExtractionBatch, batch_id)
|
|
if batch is None:
|
|
raise HTTPException(status_code=404, detail="Lote no encontrado")
|
|
rows = session.scalars(
|
|
select(ExtractionRow)
|
|
.where(ExtractionRow.batch_id == batch_id)
|
|
.order_by(ExtractionRow.row_index, ExtractionRow.id)
|
|
.offset(offset)
|
|
.limit(limit)
|
|
).all()
|
|
return {
|
|
"batch_id": batch_id,
|
|
"rows": [row.row_json for row in rows],
|
|
"limit": limit,
|
|
"offset": offset,
|
|
}
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# Jobs (contenedores multimodulo)
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
class JobRead(BaseModel):
|
|
id: str
|
|
status: str
|
|
title: str | None
|
|
metadata: dict[str, Any]
|
|
created_at: str
|
|
updated_at: str
|
|
|
|
|
|
def serialize_job(job: Job) -> dict[str, Any]:
|
|
return {
|
|
"id": job.id,
|
|
"status": job.status,
|
|
"title": job.title,
|
|
"metadata": job.job_metadata or {},
|
|
"created_at": job.created_at.isoformat(),
|
|
"updated_at": job.updated_at.isoformat(),
|
|
}
|
|
|
|
|
|
class CreateJobRequest(BaseModel):
|
|
title: str | None = None
|
|
folder_path: str | None = None
|
|
|
|
|
|
class UpdateJobRequest(BaseModel):
|
|
title: str | None = None
|
|
folder_path: str | None = None
|
|
status: str | None = None
|
|
|
|
|
|
@app.post("/jobs", response_model=JobRead)
|
|
def create_job(request: CreateJobRequest) -> dict[str, Any]:
|
|
"""Crea un trabajo (job) independiente como contenedor para módulos."""
|
|
with session_scope() as session:
|
|
job = Job(
|
|
status="pending",
|
|
title=request.title,
|
|
job_metadata={"folder_path": request.folder_path} if request.folder_path else {},
|
|
updated_at=utcnow(),
|
|
)
|
|
session.add(job)
|
|
session.flush()
|
|
return serialize_job(job)
|
|
|
|
|
|
@app.patch("/jobs/{job_id}", response_model=JobRead)
|
|
def update_job(job_id: str, request: UpdateJobRequest) -> dict[str, Any]:
|
|
"""Actualiza título, metadata o estado de un job."""
|
|
with session_scope() as session:
|
|
job = session.get(Job, job_id)
|
|
if job is None:
|
|
raise HTTPException(status_code=404, detail="Job no encontrado")
|
|
if request.title is not None:
|
|
job.title = request.title
|
|
if request.status is not None:
|
|
job.status = request.status
|
|
if request.folder_path is not None:
|
|
meta = dict(job.job_metadata or {})
|
|
meta["folder_path"] = request.folder_path
|
|
job.job_metadata = meta
|
|
job.updated_at = utcnow()
|
|
session.flush()
|
|
return serialize_job(job)
|
|
|
|
|
|
@app.get("/jobs", response_model=list[JobRead])
|
|
def list_jobs() -> list[dict[str, Any]]:
|
|
with session_scope() as session:
|
|
jobs = session.scalars(select(Job).order_by(Job.created_at.desc())).all()
|
|
return [serialize_job(j) for j in jobs]
|
|
|
|
|
|
@app.get("/jobs/{job_id}", response_model=JobRead)
|
|
def get_job(job_id: str) -> dict[str, Any]:
|
|
with session_scope() as session:
|
|
job = session.get(Job, job_id)
|
|
if job is None:
|
|
raise HTTPException(status_code=404, detail="Job no encontrado")
|
|
return serialize_job(job)
|
|
|
|
|
|
@app.get("/jobs/{job_id}/tasks", response_model=list[TaskRead])
|
|
def get_job_tasks(job_id: str) -> list[dict[str, Any]]:
|
|
with session_scope() as session:
|
|
job = session.get(Job, job_id)
|
|
if job is None:
|
|
raise HTTPException(status_code=404, detail="Job no encontrado")
|
|
tasks = session.scalars(
|
|
select(Task).where(Task.job_id == job_id).order_by(Task.created_at)
|
|
).all()
|
|
return [serialize_task(t) for t in tasks]
|
|
|
|
|
|
@app.delete("/jobs/{job_id}", status_code=204)
|
|
def delete_job(job_id: str) -> None:
|
|
"""Elimina un job y todos sus resultados (tasks, conteos, extracciones, limpiezas)."""
|
|
with session_scope() as session:
|
|
job = session.get(Job, job_id)
|
|
if job is None:
|
|
raise HTTPException(status_code=404, detail="Job no encontrado")
|
|
|
|
tasks = session.scalars(select(Task).where(Task.job_id == job_id)).all()
|
|
task_ids = [t.id for t in tasks]
|
|
|
|
batch_ids: list[str] = []
|
|
if task_ids:
|
|
batch_ids = list(
|
|
session.scalars(
|
|
select(ExtractionBatch.id).where(ExtractionBatch.job_module_id.in_(task_ids))
|
|
).all()
|
|
)
|
|
|
|
cleaning_ids = list(
|
|
session.scalars(select(CleaningJob.id).where(CleaningJob.job_id == job_id)).all()
|
|
)
|
|
if batch_ids:
|
|
cleaning_ids.extend(
|
|
session.scalars(
|
|
select(CleaningJob.id).where(CleaningJob.extraction_batch_id.in_(batch_ids))
|
|
).all()
|
|
)
|
|
cleaning_ids = list(dict.fromkeys(cleaning_ids))
|
|
|
|
if cleaning_ids:
|
|
session.execute(delete(CleanedRow).where(CleanedRow.cleaning_job_id.in_(cleaning_ids)))
|
|
session.execute(delete(CleaningJob).where(CleaningJob.id.in_(cleaning_ids)))
|
|
|
|
if batch_ids:
|
|
session.execute(delete(CleanedRow).where(CleanedRow.extraction_batch_id.in_(batch_ids)))
|
|
session.execute(delete(ExtractionRow).where(ExtractionRow.batch_id.in_(batch_ids)))
|
|
session.execute(delete(ExtractionBatch).where(ExtractionBatch.id.in_(batch_ids)))
|
|
|
|
if task_ids:
|
|
session.execute(delete(CountJobFile).where(CountJobFile.job_module_id.in_(task_ids)))
|
|
session.execute(delete(Task).where(Task.id.in_(task_ids)))
|
|
|
|
session.delete(job)
|
|
session.flush()
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# Count Jobs — persistir y gestionar resultados del servicio de conteo
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
class CountJobFileCreate(BaseModel):
|
|
radicado: str | None = None
|
|
filename: str
|
|
file_path: str | None = None
|
|
doc_type: str | None = None
|
|
service_count: int | None = None
|
|
services: list[dict[str, Any]] = Field(default_factory=list)
|
|
status: str = "ok"
|
|
error: str | None = None
|
|
|
|
|
|
class CreateCountJobRequest(BaseModel):
|
|
title: str | None = None
|
|
job_id: str | None = None # si se pasa, se usa el job existente
|
|
results: list[CountJobFileCreate]
|
|
|
|
|
|
class CountJobFileRead(BaseModel):
|
|
id: str
|
|
job_module_id: str
|
|
radicado: str | None
|
|
filename: str
|
|
file_path: str | None
|
|
doc_type: str | None
|
|
service_count: int | None
|
|
services: list[dict[str, Any]]
|
|
status: str
|
|
is_valid: bool | None
|
|
error: str | None
|
|
created_at: str
|
|
|
|
|
|
class UpdateCountJobFileRequest(BaseModel):
|
|
is_valid: bool | None
|
|
|
|
|
|
class ExtractionFromCountRequest(BaseModel):
|
|
counting_task_id: str
|
|
min_service_count: int = 0
|
|
valid_counts: list[int] | None = None # conteos exactos válidos (vacío = todos)
|
|
job_id: str | None = None # vincular al job existente del conteo
|
|
fields: list[str] = Field(default_factory=list)
|
|
alias_map: dict[str, str] = Field(default_factory=dict)
|
|
doc_types: list[DocumentType] | None = None
|
|
structure_mode: str = "general"
|
|
instruction: str | None = None
|
|
tiff_page_range: tuple[int, int] | None = None
|
|
retry_on_empty: bool = False
|
|
model: str | None = None
|
|
|
|
|
|
def serialize_count_file(f: CountJobFile) -> dict[str, Any]:
|
|
return {
|
|
"id": f.id,
|
|
"job_module_id": f.job_module_id,
|
|
"radicado": f.radicado,
|
|
"filename": f.filename,
|
|
"file_path": f.file_path,
|
|
"doc_type": f.doc_type,
|
|
"service_count": f.service_count,
|
|
"services": f.services or [],
|
|
"status": f.status,
|
|
"is_valid": f.is_valid,
|
|
"error": f.error,
|
|
"created_at": f.created_at.isoformat(),
|
|
}
|
|
|
|
|
|
@app.post("/count-jobs", response_model=TaskRead)
|
|
def create_count_job(request: CreateCountJobRequest) -> dict[str, Any]:
|
|
"""Persiste los resultados del servicio de conteo de facturas en la BD compartida."""
|
|
total = len(request.results)
|
|
default_title = request.title or f"Conteo ({total} archivos)"
|
|
with session_scope() as session:
|
|
if request.job_id:
|
|
job = session.get(Job, request.job_id)
|
|
if job is None:
|
|
raise HTTPException(status_code=404, detail="Job no encontrado")
|
|
# No sobrescribir el nombre que el usuario dio al trabajo.
|
|
job.status = "running"
|
|
job.updated_at = utcnow()
|
|
else:
|
|
job = Job(
|
|
status="completed",
|
|
title=default_title,
|
|
updated_at=utcnow(),
|
|
)
|
|
session.add(job)
|
|
session.flush()
|
|
|
|
task = Task(
|
|
job_id=job.id,
|
|
type="counting",
|
|
status="completed",
|
|
title=default_title,
|
|
request={},
|
|
result={
|
|
"total": total,
|
|
"processed": total,
|
|
"ok": sum(1 for r in request.results if r.status == "ok"),
|
|
"failed": sum(1 for r in request.results if r.status != "ok"),
|
|
},
|
|
started_at=utcnow(),
|
|
finished_at=utcnow(),
|
|
updated_at=utcnow(),
|
|
)
|
|
session.add(task)
|
|
session.flush()
|
|
|
|
for r in request.results:
|
|
session.add(
|
|
CountJobFile(
|
|
job_module_id=task.id,
|
|
radicado=r.radicado,
|
|
filename=r.filename,
|
|
file_path=r.file_path,
|
|
doc_type=r.doc_type,
|
|
service_count=r.service_count,
|
|
services=r.services,
|
|
status=r.status,
|
|
error=r.error,
|
|
)
|
|
)
|
|
session.flush()
|
|
return serialize_task(task)
|
|
|
|
|
|
@app.get("/count-jobs", response_model=list[TaskRead])
|
|
def list_count_jobs() -> list[dict[str, Any]]:
|
|
with session_scope() as session:
|
|
tasks = session.scalars(
|
|
select(Task)
|
|
.where(Task.type == "counting")
|
|
.order_by(Task.created_at.desc())
|
|
).all()
|
|
return [serialize_task(t) for t in tasks]
|
|
|
|
|
|
@app.get("/count-jobs/{task_id}", response_model=TaskRead)
|
|
def get_count_job(task_id: str) -> dict[str, Any]:
|
|
with session_scope() as session:
|
|
task = session.get(Task, task_id)
|
|
if task is None or task.type != "counting":
|
|
raise HTTPException(status_code=404, detail="Tarea de conteo no encontrada")
|
|
return serialize_task(task)
|
|
|
|
|
|
@app.get("/count-jobs/{task_id}/files", response_model=list[CountJobFileRead])
|
|
def list_count_job_files(task_id: str) -> list[dict[str, Any]]:
|
|
with session_scope() as session:
|
|
task = session.get(Task, task_id)
|
|
if task is None or task.type != "counting":
|
|
raise HTTPException(status_code=404, detail="Tarea de conteo no encontrada")
|
|
files = session.scalars(
|
|
select(CountJobFile)
|
|
.where(CountJobFile.job_module_id == task_id)
|
|
.order_by(CountJobFile.created_at)
|
|
).all()
|
|
return [serialize_count_file(f) for f in files]
|
|
|
|
|
|
@app.patch("/count-jobs/{task_id}/files/{file_id}", response_model=CountJobFileRead)
|
|
def update_count_job_file(
|
|
task_id: str, file_id: str, request: UpdateCountJobFileRequest
|
|
) -> dict[str, Any]:
|
|
with session_scope() as session:
|
|
f = session.get(CountJobFile, file_id)
|
|
if f is None or f.job_module_id != task_id:
|
|
raise HTTPException(status_code=404, detail="Archivo no encontrado")
|
|
f.is_valid = request.is_valid
|
|
session.flush()
|
|
return serialize_count_file(f)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
# Extraccion desde resultados de conteo (sin re-subir archivos)
|
|
# ═══════════════════════════════════════════════════════════════════════════════
|
|
|
|
def run_from_count_task(task_id: str) -> None:
|
|
"""Procesa archivos validos de un job de conteo en background (lee desde el volumen /data)."""
|
|
import logging as _logging
|
|
_log = _logging.getLogger("extraction.from_count")
|
|
|
|
from api.desktop_ingest import _service_for_model, mark_task_completed_if_done
|
|
|
|
with session_scope() as session:
|
|
task = session.get(Task, task_id)
|
|
if task is None:
|
|
return
|
|
files_data: list[dict[str, Any]] = list((task.result or {}).get("files", []))
|
|
req = dict(task.request or {})
|
|
batch_id = str((task.result or {}).get("extraction_batch_id") or "")
|
|
|
|
model_name: str | None = req.get("model")
|
|
instruction: str | None = req.get("instruction")
|
|
tiff_page_range = req.get("tiff_page_range")
|
|
|
|
for file_data in files_data:
|
|
file_path: str | None = file_data.get("file_path")
|
|
filename = str(file_data.get("filename") or "documento.pdf")
|
|
radicado = str(file_data.get("radicado") or file_data.get("filename") or "")
|
|
doc_type: str | None = file_data.get("doc_type")
|
|
|
|
if not file_path or not __import__("os").path.exists(file_path):
|
|
_log.warning("Archivo no disponible: %s", file_path or filename)
|
|
with session_scope() as session:
|
|
task_obj = session.get(Task, task_id)
|
|
if task_obj is None:
|
|
return
|
|
result = dict(task_obj.result or {})
|
|
errors = list(result.get("errors") or [])
|
|
errors.append(f"Archivo no encontrado: {filename}")
|
|
result["errors"] = errors
|
|
result["processed"] = int(result.get("processed") or 0) + 1
|
|
task_obj.result = result
|
|
task_obj.updated_at = utcnow()
|
|
mark_task_completed_if_done(session, task_obj, batch_id)
|
|
continue
|
|
|
|
try:
|
|
svc = _service_for_model(processing_service, model_name)
|
|
with session_scope() as session:
|
|
task_obj = session.get(Task, task_id, with_for_update=True)
|
|
if task_obj is None:
|
|
return
|
|
from api.desktop_ingest import process_desktop_ingest
|
|
process_desktop_ingest(
|
|
svc, session,
|
|
task=task_obj,
|
|
batch_id=batch_id,
|
|
temp_path=file_path,
|
|
radicado=radicado,
|
|
filename=filename,
|
|
doc_type=doc_type,
|
|
instruction=instruction,
|
|
tiff_page_range=tiff_page_range,
|
|
model=model_name,
|
|
)
|
|
mark_task_completed_if_done(session, task_obj, batch_id)
|
|
except Exception as exc:
|
|
_log.exception("Error procesando %s: %s", filename, exc)
|
|
with session_scope() as session:
|
|
task_obj = session.get(Task, task_id)
|
|
if task_obj is None:
|
|
return
|
|
result = dict(task_obj.result or {})
|
|
errors = list(result.get("errors") or [])
|
|
errors.append(f"{filename}: {exc}")
|
|
result["errors"] = errors
|
|
result["processed"] = int(result.get("processed") or 0) + 1
|
|
task_obj.result = result
|
|
task_obj.updated_at = utcnow()
|
|
mark_task_completed_if_done(session, task_obj, batch_id)
|
|
|
|
|
|
@app.post("/tasks/extraction/from-count", response_model=TaskRead)
|
|
def create_extraction_from_count(
|
|
request: ExtractionFromCountRequest, background_tasks: BackgroundTasks
|
|
) -> dict[str, Any]:
|
|
"""Crea una tarea de extraccion usando archivos validos de un job de conteo previo.
|
|
|
|
No requiere subir archivos: usa las rutas almacenadas en count_job_files.
|
|
"""
|
|
if not request.fields:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="Envia fields. Usa POST /extraction/schemas/upload para descubrir campos.",
|
|
)
|
|
|
|
with session_scope() as session:
|
|
counting_task = session.get(Task, request.counting_task_id)
|
|
if counting_task is None or counting_task.type != "counting":
|
|
raise HTTPException(status_code=404, detail="Tarea de conteo no encontrada")
|
|
|
|
all_files = session.scalars(
|
|
select(CountJobFile).where(
|
|
CountJobFile.job_module_id == request.counting_task_id
|
|
)
|
|
).all()
|
|
|
|
use_exact_counts = request.valid_counts is not None and len(request.valid_counts) > 0
|
|
valid_files = [
|
|
f for f in all_files
|
|
if f.status == "ok"
|
|
and f.is_valid is not False
|
|
and (
|
|
# Si el usuario marcó manualmente como válido, siempre se incluye
|
|
f.is_valid is True
|
|
# Si hay conteos exactos definidos, el service_count debe estar en la lista
|
|
or (use_exact_counts and f.service_count in request.valid_counts)
|
|
# Sin filtro exacto, usar umbral mínimo (compatibilidad hacia atrás)
|
|
or (not use_exact_counts and (f.service_count is None or f.service_count >= request.min_service_count))
|
|
)
|
|
]
|
|
|
|
if not valid_files:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail="No hay archivos validos para extraer con los criterios indicados.",
|
|
)
|
|
|
|
files_data = [
|
|
{
|
|
"id": f.id,
|
|
"radicado": f.radicado or f.filename,
|
|
"filename": f.filename,
|
|
"file_path": f.file_path,
|
|
"doc_type": f.doc_type,
|
|
"service_count": f.service_count,
|
|
}
|
|
for f in valid_files
|
|
]
|
|
total = len(files_data)
|
|
|
|
if request.job_id:
|
|
job = session.get(Job, request.job_id)
|
|
if job is None:
|
|
raise HTTPException(status_code=404, detail="Job no encontrado")
|
|
job.status = "running"
|
|
job.updated_at = utcnow()
|
|
else:
|
|
job = Job(
|
|
status="running",
|
|
title=f"Extraccion desde conteo ({total} archivos)",
|
|
updated_at=utcnow(),
|
|
)
|
|
session.add(job)
|
|
session.flush()
|
|
|
|
batch = ExtractionBatch(
|
|
status="running",
|
|
root_path="count://",
|
|
request={
|
|
"source": "from-count",
|
|
"counting_task_id": request.counting_task_id,
|
|
"fields": request.fields,
|
|
"alias_map": request.alias_map,
|
|
"min_service_count": request.min_service_count,
|
|
"structure_mode": request.structure_mode,
|
|
},
|
|
updated_at=utcnow(),
|
|
)
|
|
session.add(batch)
|
|
session.flush()
|
|
|
|
task = Task(
|
|
job_id=job.id,
|
|
type="extraction",
|
|
status="running",
|
|
title=job.title,
|
|
request={
|
|
"source": "from-count",
|
|
"counting_task_id": request.counting_task_id,
|
|
"fields": request.fields,
|
|
"alias_map": request.alias_map,
|
|
"doc_types": [dt.model_dump() for dt in request.doc_types or []],
|
|
"structure_mode": request.structure_mode,
|
|
"instruction": request.instruction,
|
|
"tiff_page_range": list(request.tiff_page_range) if request.tiff_page_range else None,
|
|
"min_service_count": request.min_service_count,
|
|
"model": request.model,
|
|
},
|
|
result={
|
|
"source": "from-count",
|
|
"extraction_batch_id": batch.id,
|
|
"processed": 0,
|
|
"total": total,
|
|
"row_count": 0,
|
|
"errors": [],
|
|
"files": files_data,
|
|
"partials": {},
|
|
},
|
|
started_at=utcnow(),
|
|
updated_at=utcnow(),
|
|
)
|
|
session.add(task)
|
|
session.flush()
|
|
|
|
batch.job_module_id = task.id
|
|
session.flush()
|
|
|
|
serialized = serialize_task(task)
|
|
task_id = task.id
|
|
|
|
background_tasks.add_task(run_from_count_task, task_id)
|
|
return serialized
|