254 lines
9.2 KiB
Python
254 lines
9.2 KiB
Python
from datetime import datetime
|
|
from uuid import uuid4
|
|
import os
|
|
import logging
|
|
from fastapi import APIRouter, UploadFile, File, Form, HTTPException, Depends, Query
|
|
from sqlalchemy.orm import Session
|
|
from typing import Optional, Literal, Dict, Any
|
|
|
|
from core.celery_app import celery_app
|
|
from core.config import settings
|
|
from core.database import get_core_db
|
|
from core.security import get_current_user, validate_access_to_resource
|
|
from api.v1.modules.core.tasks_tracking import track_and_dispatch
|
|
|
|
from .tasks import (
|
|
scan_file,
|
|
insert_valid_rows,
|
|
JOB_TYPE,
|
|
IMPORT_META_KEY_PREFIX,
|
|
IMPORT_REDIS_TTL,
|
|
)
|
|
from .schemas import ImportJobResponse, ImportJobStatus, CommitRequest
|
|
from ..common import storage as common_storage
|
|
from ..common.error_csv import download_scan_errors_csv_stream
|
|
from ..common.responses import normalize_commit_status_payload
|
|
from ..common.track_commit_dispatch import dispatch_tracked_layouts_csv_commit
|
|
|
|
router = APIRouter()
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _get_redis():
|
|
"""Redis client (same broker as Celery so worker can read)."""
|
|
import redis
|
|
url = os.getenv("VALKEY_URL", os.getenv("REDIS_URL", "redis://valkey:6379/0"))
|
|
return redis.Redis.from_url(url, decode_responses=False)
|
|
|
|
@router.post("/upload/{model_target}", response_model=ImportJobResponse)
|
|
async def upload_import_file(
|
|
model_target: Literal["invoice_header", "invoice_details", "invoice_series"],
|
|
file: UploadFile = File(...),
|
|
footer_config: Optional[str] = Form(None), # JSON string with settings
|
|
template_id: Optional[str] = Form(None), # id de la plantilla (ej. imp_temp_header) para respetar columnas
|
|
company_id: int = Query(..., description="Company ID"), # Required for context
|
|
operation_type: Optional[str] = Query("imp"),
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Step 1: Upload CSV, save to temp, trigger scan task.
|
|
Si se envía template_id, solo se leen las columnas de esa plantilla.
|
|
"""
|
|
header_company_id = None
|
|
try:
|
|
if db is not None and getattr(db, "info", None):
|
|
header_company_id = db.info.get("rls_company_id")
|
|
except Exception:
|
|
header_company_id = None
|
|
if header_company_id is not None and int(header_company_id) != int(company_id):
|
|
logger.warning(
|
|
"Facturas upload company mismatch request_company_id=%s db_rls_company_id=%s",
|
|
company_id,
|
|
header_company_id,
|
|
)
|
|
|
|
# 1. Validate Access & Get Tenant
|
|
try:
|
|
tenant_id = validate_access_to_resource(db, company_id, current_user, ["csv_upload.process"])
|
|
except Exception as e:
|
|
logger.error(f"Access validation failed: {e}")
|
|
raise HTTPException(status_code=403, detail="Invalid company access")
|
|
|
|
if not file.filename.endswith(".csv"):
|
|
raise HTTPException(status_code=400, detail="Only .csv files allowed")
|
|
|
|
job_id = str(uuid4())
|
|
contents = await file.read()
|
|
|
|
meta_data = {
|
|
"tenant_id": tenant_id,
|
|
"company_id": company_id,
|
|
"user_id": current_user.get("id"),
|
|
"capture_user": (
|
|
current_user.get("preferred_username")
|
|
or current_user.get("email")
|
|
or current_user.get("sub")
|
|
or "CSV"
|
|
),
|
|
"footer_config": footer_config,
|
|
"operation_type": operation_type,
|
|
"template_id": template_id,
|
|
}
|
|
|
|
try:
|
|
common_storage.store_import_file(
|
|
JOB_TYPE,
|
|
job_id,
|
|
contents,
|
|
meta_data,
|
|
tenant_id=int(tenant_id),
|
|
company_id=company_id,
|
|
ttl=IMPORT_REDIS_TTL,
|
|
log_label="Facturas import",
|
|
)
|
|
except common_storage.ImportStoreError as e:
|
|
logger.error(f"Import store error: {e}")
|
|
raise HTTPException(status_code=500, detail="Failed to queue file for processing.")
|
|
except Exception as e:
|
|
logger.error(f"Redis store error: {e}")
|
|
raise HTTPException(status_code=500, detail="Failed to queue file for processing.")
|
|
|
|
# Trigger Celery Task (Async). Worker loads file from Redis / MinIO.
|
|
logger.info(
|
|
"Queueing facturas scan job=%s tenant_id=%s company_id=%s template_id=%s operation_type=%s",
|
|
job_id,
|
|
tenant_id,
|
|
company_id,
|
|
template_id,
|
|
operation_type,
|
|
)
|
|
track_and_dispatch(
|
|
db=db,
|
|
task=scan_file,
|
|
tenant_id=int(tenant_id),
|
|
company_id=company_id,
|
|
requested_by_user=current_user.get("preferred_username") or current_user.get("email") or current_user.get("sub"),
|
|
task_name="facturas_scan_file",
|
|
task_group="layouts_csv",
|
|
task_origin="a76/layouts_csv/facturas/upload",
|
|
args=[job_id, model_target, footer_config],
|
|
task_id=job_id,
|
|
)
|
|
|
|
return ImportJobResponse(
|
|
job_id=job_id,
|
|
status="queued",
|
|
message="File uploaded. Scanning started."
|
|
)
|
|
|
|
@router.get("/{job_id}/status")
|
|
async def get_import_status(job_id: str):
|
|
"""
|
|
Poll to get progress or final report. Always returns an object with "status".
|
|
"""
|
|
task_result = celery_app.AsyncResult(job_id)
|
|
|
|
if task_result.state == "PENDING":
|
|
return {"status": "processing", "progress": 0}
|
|
# task_track_started=True: el worker pasa por STARTED (y a veces RECEIVED) antes de SUCCESS/PROGRESS.
|
|
# Sin esto, el polling cae en la rama final y devuelve status=failed aunque el scan vaya bien.
|
|
if task_result.state in ("STARTED", "RECEIVED"):
|
|
info = task_result.info if isinstance(task_result.info, dict) else {}
|
|
return {
|
|
"status": "processing",
|
|
"progress": info.get("current", 0),
|
|
"total": info.get("total", 0),
|
|
}
|
|
if task_result.state == "PROGRESS":
|
|
return {
|
|
"status": "processing",
|
|
"progress": (task_result.info or {}).get("current", 0),
|
|
"total": (task_result.info or {}).get("total", 0),
|
|
}
|
|
if task_result.state == "SUCCESS":
|
|
result = task_result.result
|
|
if isinstance(result, dict) and "status" in result:
|
|
return normalize_commit_status_payload(result)
|
|
return {"status": "finished", "result": result}
|
|
|
|
# Recuperación: raro pero posible con backend de resultados — el payload ya está pero el estado no es SUCCESS.
|
|
raw = getattr(task_result, "result", None)
|
|
if isinstance(raw, dict) and raw.get("status") in (
|
|
"waiting_confirmation",
|
|
"finished",
|
|
"warning",
|
|
"failed",
|
|
):
|
|
return normalize_commit_status_payload(raw)
|
|
|
|
# FAILURE: obtener mensaje real (traceback, result o get(propagate=False))
|
|
logger.warning("Import task %s failed: state=%s", job_id, task_result.state)
|
|
err_msg = None
|
|
tb = getattr(task_result, "traceback", None)
|
|
if tb:
|
|
logger.debug("Task traceback: %s", tb[:500] if isinstance(tb, str) else tb)
|
|
if tb and isinstance(tb, str):
|
|
lines = [l.strip() for l in tb.strip().split("\n") if l.strip()]
|
|
if lines:
|
|
err_msg = lines[-1]
|
|
if not err_msg and len(lines) > 1:
|
|
err_msg = lines[-2] + " " + (lines[-1] or "")
|
|
if not err_msg:
|
|
try:
|
|
exc = task_result.get(propagate=False)
|
|
if exc is not None:
|
|
err_msg = str(exc)
|
|
except Exception:
|
|
pass
|
|
if not err_msg:
|
|
result = getattr(task_result, "result", None)
|
|
info = getattr(task_result, "info", None)
|
|
if result is not None and not isinstance(result, dict):
|
|
err_msg = str(result)
|
|
elif isinstance(result, dict) and (result.get("error") or result.get("message")):
|
|
err_msg = result.get("error") or result.get("message")
|
|
if not err_msg and isinstance(info, str):
|
|
err_msg = info
|
|
elif not err_msg and isinstance(info, dict) and "error" in info:
|
|
err_msg = str(info["error"])
|
|
if not err_msg:
|
|
err_msg = "Task failed"
|
|
return {"status": "failed", "error": err_msg}
|
|
|
|
|
|
@router.get("/{job_id}/errors/scan-csv")
|
|
async def download_scan_errors_csv(job_id: str):
|
|
"""
|
|
Descarga CSV con TODO el detalle de errores del scan (sin límite),
|
|
leyendo el archivo JSONL generado por el worker (mismas columnas que otros layouts_csv).
|
|
"""
|
|
|
|
# Facturas (impo/exp delega a facturas) usan job_type vacío ("").
|
|
return download_scan_errors_csv_stream("", job_id, filename_prefix="errores")
|
|
|
|
|
|
@router.post("/{job_id}/commit")
|
|
async def commit_import_job(
|
|
job_id: str,
|
|
body: CommitRequest,
|
|
db: Session = Depends(get_core_db),
|
|
current_user: Dict[str, Any] = Depends(get_current_user),
|
|
):
|
|
"""
|
|
Step 2: User confirms import. Trigger bulk insert.
|
|
"""
|
|
r = _get_redis()
|
|
commit_id = dispatch_tracked_layouts_csv_commit(
|
|
db=db,
|
|
current_user=current_user,
|
|
redis_client=r,
|
|
layout_job_id=job_id,
|
|
meta_redis_key=f"{IMPORT_META_KEY_PREFIX}{job_id}",
|
|
celery_task=insert_valid_rows,
|
|
task_name="facturas_insert_valid_rows",
|
|
task_origin="a76/layouts_csv/facturas/commit",
|
|
args=[job_id, body.model_target],
|
|
)
|
|
|
|
return {
|
|
"status": "committing",
|
|
"message": "Bulk insert started.",
|
|
"commit_job_id": commit_id,
|
|
}
|