"""API REST del microservicio de extraccion."""
from __future__ import annotations

import os
import uuid
from contextlib import asynccontextmanager
from datetime import datetime, timezone
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, Response, UploadFile
from pydantic import BaseModel, Field, model_validator
from sqlalchemy import delete, select

from api.desktop_ingest import (
  mark_task_completed_if_done,
  persist_desktop_ingest_result,
  record_desktop_ingest_error,
)

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.openai_balance import fetch_openai_balance
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="Flujo IPS 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
  job_id: str | None = None
  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,
    "job_id": task.job_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.get("/openai/balance")
def openai_balance() -> dict[str, Any]:
  """Crédito disponible en la cuenta OpenAI (o proveedor compatible)."""
  return fetch_openai_balance()


@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 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():
      with session_scope() as session:
        task = session.get(Task, task_id, with_for_update=True)
        if task is None:
          return {"error": "task_not_found"}
        payload = persist_desktop_ingest_result(
          session,
          task=task,
          batch_id=batch_id,
          radicado=radicado,
          filename=effective_name,
          doc_type=doc_type,
          request=request,
          normalized_obj=normalized_obj,
        )
        mark_task_completed_if_done(session, task, batch_id)
        session.flush()  # persiste status=completed / finished_at si aplica
        processed_now = int((task.result or {}).get("processed") or 0)
        _log.info(
          "✓ guardado radicado=%s processed=%d/%d status=%s",
          radicado, processed_now, int((task.result or {}).get("total") or 0), task.status,
        )
        session.refresh(task)
        return {**payload, "task": serialize_task(task)}

    return await _asyncio.to_thread(_persist)

  except Exception as exc:
    # La llamada HTTP conserva el error, pero el progreso debe quedar persistido
    # para que un documento fallido no deje la tarea eternamente en running.
    def _record_error(error: Exception) -> None:
      with session_scope() as session:
        task = session.get(Task, task_id, with_for_update=True)
        if task is None:
          return
        record_desktop_ingest_error(
          session,
          task=task,
          radicado=radicado,
          filename=effective_name,
          doc_type=doc_type,
          error=error,
        )
        mark_task_completed_if_done(session, task, batch_id)
        session.flush()

    try:
      await _asyncio.to_thread(_record_error, exc)
    except Exception:
      _log.exception(
        "No se pudo registrar el error de radicado=%s filename=%s",
        radicado,
        effective_name,
      )
    raise

  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, response_class=Response)
def delete_job(job_id: str) -> Response:
  """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()
  return Response(status_code=204)


# ═══════════════════════════════════════════════════════════════════════════════
# 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
  nit: str | None = None
  object_key: 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]
  download_job_id: str | None = None
  source: str | None = None
  started_at: str | None = None  # ISO del cliente (inicio real del conteo)


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
  nit: str | None = None
  object_key: str | None = None


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(),
    "nit": getattr(f, "nit", None),
    "object_key": getattr(f, "object_key", None),
  }


@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)"
  finished = utcnow()
  started = finished
  if request.started_at:
    try:
      parsed = datetime.fromisoformat(request.started_at.replace("Z", "+00:00"))
      if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
      started = parsed
    except ValueError:
      started = finished

  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={
        "source": request.source or ("from-download" if request.download_job_id else "desktop"),
        "download_job_id": request.download_job_id,
      },
      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"),
        "nits": sorted({r.nit for r in request.results if r.nit}),
      },
      started_at=started,
      finished_at=finished,
      updated_at=finished,
    )
    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,
          nit=r.nit,
          object_key=r.object_key,
          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