feature/csv-fracc-ame
This commit is contained in:
@@ -0,0 +1 @@
|
||||
# CSV import for US Tariff Fractions (Fracción Americana): upload → scan → commit
|
||||
@@ -0,0 +1,160 @@
|
||||
"""
|
||||
Rutas de importación CSV para Fracción Americana (US Tariff Fractions).
|
||||
Mismo flujo que exchange_rate/imports: upload → scan → status (polling) → commit.
|
||||
"""
|
||||
import base64
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
from uuid import uuid4
|
||||
|
||||
from fastapi import APIRouter, File, HTTPException, Query, UploadFile, Depends
|
||||
from sqlalchemy.orm import Session
|
||||
from typing import Dict, Any
|
||||
|
||||
from core.celery_app import celery_app
|
||||
from core.database import get_core_db
|
||||
from core.security import get_current_user, validate_access_to_resource
|
||||
|
||||
from .schemas import ImportJobResponse
|
||||
from .tasks import (
|
||||
scan_file,
|
||||
insert_valid_rows,
|
||||
FA_IMPORT_FILE_PREFIX,
|
||||
FA_IMPORT_META_PREFIX,
|
||||
FA_IMPORT_REDIS_TTL,
|
||||
)
|
||||
|
||||
router = APIRouter()
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _get_redis():
|
||||
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", response_model=ImportJobResponse)
|
||||
async def upload_import_file(
|
||||
file: UploadFile = File(...),
|
||||
company_id: int = Query(..., description="Company ID"),
|
||||
db: Session = Depends(get_core_db),
|
||||
current_user: Dict[str, Any] = Depends(get_current_user),
|
||||
):
|
||||
"""
|
||||
Fase 1: Subir CSV, guardar en Redis, encolar tarea de escaneo.
|
||||
"""
|
||||
try:
|
||||
tenant_id = validate_access_to_resource(db, company_id, current_user)
|
||||
except Exception as e:
|
||||
logger.error(f"FA import: access validation failed: {e}")
|
||||
raise HTTPException(status_code=403, detail="Invalid company access")
|
||||
|
||||
if not file.filename or not file.filename.lower().endswith(".csv"):
|
||||
raise HTTPException(status_code=400, detail="Solo se permiten archivos .csv")
|
||||
|
||||
job_id = str(uuid4())
|
||||
contents = await file.read()
|
||||
|
||||
meta_data = {
|
||||
"tenant_id": tenant_id,
|
||||
"company_id": company_id,
|
||||
"user_id": current_user.get("id"),
|
||||
"template_id": "us_tariff_fractions",
|
||||
}
|
||||
|
||||
try:
|
||||
r = _get_redis()
|
||||
r.set(
|
||||
f"{FA_IMPORT_FILE_PREFIX}{job_id}",
|
||||
base64.b64encode(contents),
|
||||
ex=FA_IMPORT_REDIS_TTL,
|
||||
)
|
||||
r.set(
|
||||
f"{FA_IMPORT_META_PREFIX}{job_id}",
|
||||
json.dumps(meta_data).encode("utf-8"),
|
||||
ex=FA_IMPORT_REDIS_TTL,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"FA import: Redis store error: {e}")
|
||||
raise HTTPException(status_code=500, detail="No se pudo encolar el archivo.")
|
||||
|
||||
try:
|
||||
upload_dir = os.path.join(os.getcwd(), "uploads", "temp")
|
||||
os.makedirs(upload_dir, exist_ok=True)
|
||||
with open(os.path.join(upload_dir, f"fa_{job_id}.csv"), "wb") as f:
|
||||
f.write(contents)
|
||||
with open(os.path.join(upload_dir, f"fa_{job_id}.meta.json"), "w") as f:
|
||||
json.dump(meta_data, f)
|
||||
except Exception as e:
|
||||
logger.warning(f"FA import: local file save failed: {e}")
|
||||
|
||||
scan_file.apply_async(args=[job_id], task_id=job_id)
|
||||
|
||||
return ImportJobResponse(
|
||||
job_id=job_id,
|
||||
status="queued",
|
||||
message="Archivo subido. Escaneo iniciado.",
|
||||
)
|
||||
|
||||
|
||||
@router.get("/{job_id}/status")
|
||||
async def get_import_status(job_id: str):
|
||||
"""
|
||||
Polling: estado del escaneo o del commit.
|
||||
"""
|
||||
task_result = celery_app.AsyncResult(job_id)
|
||||
|
||||
if task_result.state == "PENDING":
|
||||
return {"status": "processing", "progress": 0}
|
||||
if task_result.state == "PROGRESS":
|
||||
info = task_result.info or {}
|
||||
return {
|
||||
"status": "processing",
|
||||
"progress": info.get("current", 0),
|
||||
"total": info.get("total", 0),
|
||||
}
|
||||
if task_result.state == "SUCCESS":
|
||||
result = task_result.result
|
||||
if isinstance(result, dict) and "status" in result:
|
||||
return result
|
||||
return {"status": "finished", "result": result}
|
||||
|
||||
result = getattr(task_result, "result", None)
|
||||
if isinstance(result, dict) and result.get("status") in ("finished", "warning"):
|
||||
return result
|
||||
|
||||
logger.warning("FA import task %s failed: state=%s", job_id, task_result.state)
|
||||
err_msg = None
|
||||
tb = getattr(task_result, "traceback", None)
|
||||
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:
|
||||
try:
|
||||
exc = task_result.get(propagate=False)
|
||||
if exc is not None:
|
||||
err_msg = str(exc)
|
||||
except Exception:
|
||||
pass
|
||||
if not err_msg and result is not None:
|
||||
if not isinstance(result, dict):
|
||||
err_msg = str(result)
|
||||
elif result.get("error") or result.get("message"):
|
||||
err_msg = result.get("error") or result.get("message")
|
||||
return {"status": "failed", "error": err_msg or "Task failed"}
|
||||
|
||||
|
||||
@router.post("/{job_id}/commit")
|
||||
async def commit_import_job(job_id: str):
|
||||
"""
|
||||
Fase 2: Usuario confirma; se encola la inserción de filas válidas.
|
||||
"""
|
||||
task = insert_valid_rows.delay(job_id)
|
||||
return {
|
||||
"status": "committing",
|
||||
"message": "Inserción iniciada.",
|
||||
"commit_job_id": task.id,
|
||||
}
|
||||
@@ -0,0 +1,25 @@
|
||||
from pydantic import BaseModel
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class ImportJobResponse(BaseModel):
|
||||
job_id: str
|
||||
status: str
|
||||
message: str
|
||||
|
||||
|
||||
class CommitRequest(BaseModel):
|
||||
pass # no body needed for single model
|
||||
|
||||
|
||||
class ImportJobStatus(BaseModel):
|
||||
status: str
|
||||
job_id: str
|
||||
total_rows: Optional[int] = 0
|
||||
error_count: Optional[int] = 0
|
||||
valid_rows: Optional[int] = 0
|
||||
error: Optional[str] = None
|
||||
inserted: Optional[int] = 0
|
||||
skipped_invalid: Optional[int] = 0
|
||||
skipped_missing_fk: Optional[int] = 0
|
||||
skipped_details: Optional[list] = None
|
||||
@@ -0,0 +1,474 @@
|
||||
"""
|
||||
Tareas Celery para importación CSV de Fracción Americana (US Tariff Fractions).
|
||||
Flujo en dos fases: scan_file (validación) → insert_valid_rows (commit).
|
||||
"""
|
||||
import os
|
||||
import base64
|
||||
import csv
|
||||
import json
|
||||
import logging
|
||||
import re
|
||||
import unicodedata
|
||||
from decimal import Decimal
|
||||
from typing import Dict, Any, Optional, List
|
||||
|
||||
from core.celery_app import celery_app
|
||||
from core.database import CoreSessionLocal
|
||||
|
||||
from .template_config import row_from_template
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
FA_IMPORT_FILE_PREFIX = "fa_import_file:"
|
||||
FA_IMPORT_META_PREFIX = "fa_import_meta:"
|
||||
FA_IMPORT_ERROR_LINES_PREFIX = "fa_import_error_lines:"
|
||||
FA_IMPORT_REDIS_TTL = 3600 # 1 hour
|
||||
|
||||
TEMPLATE_ID = "us_tariff_fractions"
|
||||
|
||||
|
||||
def _get_redis():
|
||||
import redis
|
||||
url = os.getenv("VALKEY_URL", os.getenv("REDIS_URL", "redis://valkey:6379/0"))
|
||||
return redis.Redis.from_url(url, decode_responses=False)
|
||||
|
||||
|
||||
def _worker_upload_dir() -> str:
|
||||
return os.path.join(os.getcwd(), "uploads", "temp")
|
||||
|
||||
|
||||
def _ensure_worker_has_file_from_redis(job_id: str) -> Optional[str]:
|
||||
r = _get_redis()
|
||||
data = r.get(f"{FA_IMPORT_FILE_PREFIX}{job_id}")
|
||||
if not data:
|
||||
return None
|
||||
try:
|
||||
raw = base64.b64decode(data)
|
||||
except Exception as e:
|
||||
logger.warning(f"FA import: failed to decode file from Redis: {e}")
|
||||
return None
|
||||
upload_dir = _worker_upload_dir()
|
||||
os.makedirs(upload_dir, exist_ok=True)
|
||||
file_path = os.path.join(upload_dir, f"fa_{job_id}.csv")
|
||||
with open(file_path, "wb") as f:
|
||||
f.write(raw)
|
||||
return file_path
|
||||
|
||||
|
||||
def _ensure_worker_has_meta_from_redis(job_id: str, file_path: str) -> bool:
|
||||
r = _get_redis()
|
||||
data = r.get(f"{FA_IMPORT_META_PREFIX}{job_id}")
|
||||
if not data:
|
||||
return False
|
||||
try:
|
||||
meta = json.loads(data.decode("utf-8"))
|
||||
except Exception as e:
|
||||
logger.warning(f"FA import: failed to decode meta from Redis: {e}")
|
||||
return False
|
||||
meta_path = file_path.replace(".csv", ".meta.json")
|
||||
with open(meta_path, "w", encoding="utf-8") as f:
|
||||
json.dump(meta, f)
|
||||
return True
|
||||
|
||||
|
||||
def _delete_import_from_redis(job_id: str) -> None:
|
||||
try:
|
||||
r = _get_redis()
|
||||
r.delete(
|
||||
f"{FA_IMPORT_FILE_PREFIX}{job_id}",
|
||||
f"{FA_IMPORT_META_PREFIX}{job_id}",
|
||||
f"{FA_IMPORT_ERROR_LINES_PREFIX}{job_id}",
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(f"FA import: failed to delete Redis keys: {e}")
|
||||
|
||||
|
||||
def normalize_header(name: Optional[str]) -> str:
|
||||
if not name:
|
||||
return ""
|
||||
name = unicodedata.normalize("NFKD", str(name)).upper()
|
||||
name = "".join(ch for ch in name if not unicodedata.combining(ch))
|
||||
name = re.sub(r"[^A-Z0-9]+", " ", name)
|
||||
return re.sub(r"\s+", " ", name).strip()
|
||||
|
||||
|
||||
def _normalize_code(raw: Optional[str]) -> str:
|
||||
"""Normalize fraction code: strip and remove dots/dashes, max 16 chars."""
|
||||
if not raw:
|
||||
return ""
|
||||
s = str(raw).strip().replace(".", "").replace("-", "")
|
||||
return s[:16] if len(s) > 16 else s
|
||||
|
||||
|
||||
def _validate_row_us_tariff_fraction(row: Dict[str, Any], line_num: int) -> Optional[Dict[str, Any]]:
|
||||
"""Valida una fila para Fracción Americana. Retorna error dict o None."""
|
||||
code_raw = (row.get("FRACCION_ARANCELARIA") or "").strip()
|
||||
if not code_raw:
|
||||
return {"line": line_num, "col": "FRACCION_ARANCELARIA", "msg": "Requerido"}
|
||||
code_norm = _normalize_code(code_raw)
|
||||
if not code_norm:
|
||||
return {"line": line_num, "col": "FRACCION_ARANCELARIA", "msg": "Requerido"}
|
||||
if len(code_norm) > 16:
|
||||
return {"line": line_num, "col": "FRACCION_ARANCELARIA", "msg": "Máximo 16 caracteres"}
|
||||
|
||||
prefix_raw = (row.get("PREFIJO") or "").strip()
|
||||
if prefix_raw and len(prefix_raw) > 10:
|
||||
return {"line": line_num, "col": "PREFIJO", "msg": "Máximo 10 caracteres"}
|
||||
|
||||
um_raw = (row.get("UNIDAD_DE_MEDIDA") or "").strip()
|
||||
if um_raw and len(um_raw) > 10:
|
||||
return {"line": line_num, "col": "UNIDAD_DE_MEDIDA", "msg": "Máximo 10 caracteres"}
|
||||
|
||||
tipo_raw = (row.get("TIPO_DE_ADVALOREM") or "").strip()
|
||||
if tipo_raw and len(tipo_raw) > 10:
|
||||
return {"line": line_num, "col": "TIPO_DE_ADVALOREM", "msg": "Máximo 10 caracteres"}
|
||||
|
||||
adv_pct = row.get("ADVALOREM_PCT")
|
||||
if adv_pct is not None and str(adv_pct).strip():
|
||||
try:
|
||||
v = float(str(adv_pct).strip().replace(",", "."))
|
||||
if v < 0:
|
||||
return {"line": line_num, "col": "ADVALOREM_PCT", "msg": "Debe ser >= 0"}
|
||||
except ValueError:
|
||||
return {"line": line_num, "col": "ADVALOREM_PCT", "msg": "Debe ser un número"}
|
||||
|
||||
adv_dlls = row.get("ADVALOREM_DLLS")
|
||||
if adv_dlls is not None and str(adv_dlls).strip():
|
||||
try:
|
||||
v = float(str(adv_dlls).strip().replace(",", "."))
|
||||
if v < 0:
|
||||
return {"line": line_num, "col": "ADVALOREM_DLLS", "msg": "Debe ser >= 0"}
|
||||
except ValueError:
|
||||
return {"line": line_num, "col": "ADVALOREM_DLLS", "msg": "Debe ser un número"}
|
||||
|
||||
return None
|
||||
|
||||
|
||||
@celery_app.task(bind=True)
|
||||
def scan_file(self, job_id: str, config: str = None):
|
||||
"""
|
||||
Fase 1: Leer CSV, validar filas, escribir errores en JSONL.
|
||||
Devuelve waiting_confirmation con total_rows, error_count, valid_rows, errors.
|
||||
"""
|
||||
logger.info(f"FA import: starting scan for job {job_id}")
|
||||
|
||||
file_path = _ensure_worker_has_file_from_redis(job_id)
|
||||
if not file_path:
|
||||
return {"status": "failed", "error": "Archivo no encontrado (expirado o no subido). Sube de nuevo."}
|
||||
_ensure_worker_has_meta_from_redis(job_id, file_path)
|
||||
|
||||
error_dir = os.path.join(os.path.dirname(file_path).replace("temp", "errors"), "")
|
||||
os.makedirs(error_dir, exist_ok=True)
|
||||
error_path = os.path.join(error_dir, f"fa_{job_id}.jsonl")
|
||||
|
||||
total_rows = 0
|
||||
try:
|
||||
with open(file_path, "r", encoding="utf-8-sig") as f:
|
||||
total_rows = sum(1 for _ in f) - 1
|
||||
except Exception as e:
|
||||
return {"status": "failed", "error": str(e)}
|
||||
|
||||
meta_path = file_path.replace(".csv", ".meta.json")
|
||||
meta = {}
|
||||
if os.path.exists(meta_path):
|
||||
try:
|
||||
with open(meta_path, "r", encoding="utf-8") as f:
|
||||
meta = json.load(f) or {}
|
||||
except Exception as e:
|
||||
logger.warning(f"FA import: failed to read meta: {e}")
|
||||
|
||||
tenant_id = meta.get("tenant_id")
|
||||
company_id = meta.get("company_id")
|
||||
if not tenant_id or not company_id:
|
||||
return {"status": "failed", "error": "Falta contexto (tenant/company)"}
|
||||
|
||||
error_count = 0
|
||||
processed_rows = 0
|
||||
errors_detail: List[Dict[str, Any]] = []
|
||||
|
||||
try:
|
||||
with open(file_path, "r", encoding="utf-8-sig") as f_in, open(
|
||||
error_path, "w", encoding="utf-8"
|
||||
) as f_err:
|
||||
sample = f_in.read(2048)
|
||||
f_in.seek(0)
|
||||
try:
|
||||
dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
|
||||
except Exception:
|
||||
dialect = "excel"
|
||||
reader = csv.DictReader(f_in, dialect=dialect)
|
||||
|
||||
for i, row in enumerate(reader, start=1):
|
||||
if i % 500 == 0:
|
||||
self.update_state(
|
||||
state="PROGRESS",
|
||||
meta={"current": i, "total": total_rows, "errors": error_count},
|
||||
)
|
||||
|
||||
row_norm = row_from_template(row, normalize_header, TEMPLATE_ID)
|
||||
err = _validate_row_us_tariff_fraction(row_norm, i)
|
||||
if err:
|
||||
error_count += 1
|
||||
f_err.write(json.dumps(err) + "\n")
|
||||
if len(errors_detail) < 500:
|
||||
errors_detail.append(
|
||||
{"line": err["line"], "col": err.get("col", ""), "msg": err.get("msg", "")}
|
||||
)
|
||||
processed_rows += 1
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"FA import scan failed: {e}")
|
||||
return {"status": "failed", "error": str(e)}
|
||||
|
||||
error_lines_list = []
|
||||
try:
|
||||
if os.path.exists(error_path):
|
||||
with open(error_path, "r", encoding="utf-8") as f:
|
||||
for line in f:
|
||||
try:
|
||||
err = json.loads(line)
|
||||
if "line" in err:
|
||||
error_lines_list.append(err["line"])
|
||||
except Exception:
|
||||
pass
|
||||
if error_lines_list:
|
||||
r = _get_redis()
|
||||
r.set(
|
||||
f"{FA_IMPORT_ERROR_LINES_PREFIX}{job_id}",
|
||||
json.dumps(error_lines_list).encode("utf-8"),
|
||||
ex=FA_IMPORT_REDIS_TTL,
|
||||
)
|
||||
except Exception as e:
|
||||
logger.warning(f"FA import: failed to store error lines in Redis: {e}")
|
||||
|
||||
return {
|
||||
"status": "waiting_confirmation",
|
||||
"job_id": job_id,
|
||||
"total_rows": processed_rows,
|
||||
"error_count": error_count,
|
||||
"valid_rows": processed_rows - error_count,
|
||||
"errors": errors_detail,
|
||||
}
|
||||
|
||||
|
||||
def _str_or_none(val: Any, max_len: Optional[int] = None) -> Optional[str]:
|
||||
if val is None:
|
||||
return None
|
||||
s = str(val).strip()
|
||||
if not s:
|
||||
return None
|
||||
if max_len and len(s) > max_len:
|
||||
return s[:max_len]
|
||||
return s
|
||||
|
||||
|
||||
def _parse_float(val: Any) -> Optional[float]:
|
||||
if val is None or str(val).strip() == "":
|
||||
return None
|
||||
try:
|
||||
return float(str(val).strip().replace(",", "."))
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
|
||||
|
||||
@celery_app.task(bind=True)
|
||||
def insert_valid_rows(self, job_id: str):
|
||||
"""
|
||||
Fase 2: Re-leer CSV, omitir filas con error, upsert USTariffFraction por (tenant_id, company_id, code).
|
||||
"""
|
||||
logger.info(f"FA import: starting commit for job {job_id}")
|
||||
|
||||
file_path = _ensure_worker_has_file_from_redis(job_id)
|
||||
if not file_path:
|
||||
alt_path = os.path.join(_worker_upload_dir(), f"fa_{job_id}.csv")
|
||||
if not os.path.exists(alt_path):
|
||||
return {
|
||||
"status": "failed",
|
||||
"error": "Archivo no encontrado (expirado). Sube y confirma de nuevo.",
|
||||
}
|
||||
file_path = alt_path
|
||||
else:
|
||||
_ensure_worker_has_meta_from_redis(job_id, file_path)
|
||||
|
||||
base_dir = os.path.dirname(file_path)
|
||||
error_dir = base_dir.replace("temp", "errors")
|
||||
error_path = os.path.join(error_dir, f"fa_{job_id}.jsonl")
|
||||
|
||||
error_lines = set()
|
||||
try:
|
||||
r = _get_redis()
|
||||
raw = r.get(f"{FA_IMPORT_ERROR_LINES_PREFIX}{job_id}")
|
||||
if raw:
|
||||
error_lines = set(json.loads(raw.decode("utf-8")))
|
||||
except Exception as e:
|
||||
logger.debug(f"FA import: could not load error lines from Redis: {e}")
|
||||
if not error_lines and os.path.exists(error_path):
|
||||
with open(error_path, "r", encoding="utf-8") as f:
|
||||
for line in f:
|
||||
try:
|
||||
err = json.loads(line)
|
||||
error_lines.add(err["line"])
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
meta_path = file_path.replace(".csv", ".meta.json")
|
||||
tenant_id = None
|
||||
company_id = None
|
||||
meta = {}
|
||||
if os.path.exists(meta_path):
|
||||
try:
|
||||
with open(meta_path, "r", encoding="utf-8") as f:
|
||||
meta = json.load(f) or {}
|
||||
tenant_id = meta.get("tenant_id")
|
||||
company_id = meta.get("company_id")
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if not tenant_id or not company_id:
|
||||
return {"status": "failed", "error": "Falta contexto (tenant/company)"}
|
||||
|
||||
from api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.models import USTariffFraction
|
||||
|
||||
inserted_count = 0
|
||||
skipped_invalid = 0
|
||||
skipped_details: List[Dict[str, Any]] = []
|
||||
response = None
|
||||
|
||||
try:
|
||||
with CoreSessionLocal() as session:
|
||||
with open(file_path, "r", encoding="utf-8-sig") as f:
|
||||
sample = f.read(2048)
|
||||
f.seek(0)
|
||||
try:
|
||||
dialect = csv.Sniffer().sniff(sample, delimiters=",;\t")
|
||||
except Exception:
|
||||
dialect = "excel"
|
||||
reader = csv.DictReader(f, dialect=dialect)
|
||||
|
||||
for i, row in enumerate(reader, start=1):
|
||||
if i in error_lines:
|
||||
continue
|
||||
|
||||
row_norm = row_from_template(row, normalize_header, TEMPLATE_ID)
|
||||
err = _validate_row_us_tariff_fraction(row_norm, i)
|
||||
if err:
|
||||
skipped_invalid += 1
|
||||
skipped_details.append(
|
||||
{
|
||||
"line": i,
|
||||
"reason": f"{err.get('col', '')}: {err.get('msg', '')}",
|
||||
}
|
||||
)
|
||||
continue
|
||||
|
||||
code = _normalize_code(row_norm.get("FRACCION_ARANCELARIA"))
|
||||
if not code:
|
||||
skipped_invalid += 1
|
||||
skipped_details.append({"line": i, "reason": "FRACCION_ARANCELARIA: vacío"})
|
||||
continue
|
||||
|
||||
prefix = _str_or_none(row_norm.get("PREFIJO"), 10)
|
||||
unit_of_measure = _str_or_none(row_norm.get("UNIDAD_DE_MEDIDA"), 10)
|
||||
description = _str_or_none(row_norm.get("DESCRIPCION"))
|
||||
type_code = _str_or_none(row_norm.get("TIPO_DE_ADVALOREM"), 10)
|
||||
ad_valorem = _parse_float(row_norm.get("ADVALOREM_PCT"))
|
||||
fixed_cost_raw = _parse_float(row_norm.get("ADVALOREM_DLLS"))
|
||||
fixed_cost = Decimal(str(round(fixed_cost_raw, 8))) if fixed_cost_raw is not None else None
|
||||
|
||||
existing = (
|
||||
session.query(USTariffFraction)
|
||||
.filter(
|
||||
USTariffFraction.tenant_id == tenant_id,
|
||||
USTariffFraction.company_id == company_id,
|
||||
USTariffFraction.code == code,
|
||||
)
|
||||
.first()
|
||||
)
|
||||
|
||||
if existing:
|
||||
existing.prefix = prefix
|
||||
existing.type_code = type_code
|
||||
existing.ad_valorem = ad_valorem
|
||||
existing.fixed_cost = fixed_cost
|
||||
existing.unit_of_measure = unit_of_measure
|
||||
existing.description = description
|
||||
session.add(existing)
|
||||
inserted_count += 1
|
||||
else:
|
||||
new_row = USTariffFraction(
|
||||
tenant_id=tenant_id,
|
||||
company_id=company_id,
|
||||
code=code,
|
||||
prefix=prefix,
|
||||
type_code=type_code,
|
||||
ad_valorem=ad_valorem,
|
||||
fixed_cost=fixed_cost,
|
||||
unit_of_measure=unit_of_measure,
|
||||
description=description,
|
||||
)
|
||||
session.add(new_row)
|
||||
inserted_count += 1
|
||||
|
||||
try:
|
||||
session.commit()
|
||||
except Exception as db_err:
|
||||
session.rollback()
|
||||
logger.error(f"FA import DB error: {db_err}")
|
||||
return {"status": "failed", "error": str(db_err)}
|
||||
|
||||
total_skipped = skipped_invalid
|
||||
if inserted_count == 0 and total_skipped > 0:
|
||||
response = {
|
||||
"status": "warning",
|
||||
"inserted": 0,
|
||||
"skipped_invalid": skipped_invalid,
|
||||
"skipped_missing_fk": 0,
|
||||
"skipped_details": skipped_details,
|
||||
"message": f"No se insertaron registros. {total_skipped} rechazados.",
|
||||
}
|
||||
elif inserted_count == 0:
|
||||
response = {
|
||||
"status": "failed",
|
||||
"error": "No hay registros válidos en el archivo CSV",
|
||||
"inserted": 0,
|
||||
"skipped_invalid": skipped_invalid,
|
||||
"skipped_missing_fk": 0,
|
||||
"skipped_details": skipped_details,
|
||||
}
|
||||
else:
|
||||
response = {
|
||||
"status": "finished",
|
||||
"inserted": inserted_count,
|
||||
"skipped_invalid": skipped_invalid,
|
||||
"skipped_missing_fk": 0,
|
||||
"skipped_details": skipped_details,
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"FA import task failed: {e}")
|
||||
import traceback
|
||||
logger.error(traceback.format_exc())
|
||||
return {"status": "failed", "error": str(e)}
|
||||
|
||||
try:
|
||||
if file_path and os.path.exists(file_path):
|
||||
os.remove(file_path)
|
||||
if os.path.exists(error_path):
|
||||
os.remove(error_path)
|
||||
meta_path_clean = file_path.replace(".csv", ".meta.json")
|
||||
if os.path.exists(meta_path_clean):
|
||||
os.remove(meta_path_clean)
|
||||
_delete_import_from_redis(job_id)
|
||||
except Exception as cleanup_err:
|
||||
logger.warning(f"FA import cleanup failed: {cleanup_err}")
|
||||
|
||||
if response is None:
|
||||
response = {
|
||||
"status": "failed",
|
||||
"error": "Error inesperado",
|
||||
"inserted": 0,
|
||||
"skipped_invalid": skipped_invalid,
|
||||
"skipped_missing_fk": 0,
|
||||
"skipped_details": skipped_details,
|
||||
}
|
||||
return response
|
||||
@@ -0,0 +1,47 @@
|
||||
"""
|
||||
Configuración de plantilla CSV para Fracción Americana (EstructuraCatFraccAme.xls).
|
||||
Solo se leen columnas definidas aquí; el resto se ignora.
|
||||
"""
|
||||
|
||||
from typing import Dict, List, Any, Optional
|
||||
|
||||
TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
|
||||
"us_tariff_fractions": [
|
||||
{"canonical": "FRACCION_ARANCELARIA", "aliases": ["FRACCION ARANCELARIA", "CODE", "FRACCION"]},
|
||||
{"canonical": "PREFIJO"},
|
||||
{"canonical": "UNIDAD_DE_MEDIDA", "aliases": ["UNIDAD DE MEDIDA", "UMT"]},
|
||||
{"canonical": "DESCRIPCION"},
|
||||
{"canonical": "TIPO_DE_ADVALOREM", "aliases": ["TIPO DE ADVALOREM", "TIPO"]},
|
||||
{"canonical": "ADVALOREM_PCT", "aliases": ["ADVALOREM %", "ADVALOREM"]},
|
||||
{"canonical": "ADVALOREM_DLLS", "aliases": ["ADVALOREM DLLS", "ADVALOREM DLL"]},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def build_normalized_lookup(normalize_header_fn, template_id: str = "us_tariff_fractions") -> Dict[str, str]:
|
||||
"""normalized_header -> canonical_name para plantilla us_tariff_fractions."""
|
||||
cols = TEMPLATE_COLUMNS.get(template_id)
|
||||
if not cols:
|
||||
return {}
|
||||
lookup: Dict[str, str] = {}
|
||||
for item in cols:
|
||||
canonical = item["canonical"]
|
||||
lookup[normalize_header_fn(canonical)] = canonical
|
||||
for alias in item.get("aliases") or []:
|
||||
lookup[normalize_header_fn(alias)] = canonical
|
||||
return lookup
|
||||
|
||||
|
||||
def row_from_template(
|
||||
row: Dict[str, Any], normalize_header_fn, template_id: str = "us_tariff_fractions"
|
||||
) -> Dict[str, Any]:
|
||||
"""Fila CSV con solo columnas de la plantilla, en nombres canónicos."""
|
||||
lookup = build_normalized_lookup(normalize_header_fn, template_id)
|
||||
if not lookup:
|
||||
return {normalize_header_fn(k): v for k, v in row.items()}
|
||||
out: Dict[str, Any] = {}
|
||||
for csv_header, value in row.items():
|
||||
key_norm = normalize_header_fn(csv_header)
|
||||
if key_norm in lookup:
|
||||
out[lookup[key_norm]] = value
|
||||
return out
|
||||
@@ -17,23 +17,28 @@ from .dto import (
|
||||
)
|
||||
from .service import USTariffFractionService
|
||||
|
||||
# Create router using TenantCRUDRoutes factory for basic CRUD operations
|
||||
# Create router using TenantCRUDRoutes factory for basic CRUD operations (prefix="" so we mount under main_router)
|
||||
crud_router = TenantCRUDRoutes(
|
||||
service=USTariffFractionService,
|
||||
create_schema=USTariffFractionCreateDTO,
|
||||
update_schema=USTariffFractionUpdateDTO,
|
||||
response_schema=USTariffFractionResponseDTO,
|
||||
prefix="/us-tariff-fractions",
|
||||
prefix="",
|
||||
tags=["a76 / general catalogs / us tariff fractions"],
|
||||
resource_name="US Tariff Fraction",
|
||||
id_name="id",
|
||||
enable_list=False, # We implement our custom list endpoint
|
||||
)
|
||||
|
||||
router = crud_router.router
|
||||
# Master router with prefix so all routes live under /us-tariff-fractions
|
||||
from .imports.routes import router as imports_router
|
||||
main_router = APIRouter(prefix="/us-tariff-fractions", tags=["a76 / general catalogs / us tariff fractions"])
|
||||
main_router.include_router(imports_router, prefix="/imports", tags=["us_tariff_fractions / csv_import"])
|
||||
main_router.include_router(crud_router.router)
|
||||
|
||||
# Custom list endpoint with search filter
|
||||
@router.get(
|
||||
|
||||
# Custom list endpoint with search filter (under /us-tariff-fractions/)
|
||||
@main_router.get(
|
||||
"/",
|
||||
response_model=Dict[str, Any],
|
||||
summary="List US Tariff Fractions",
|
||||
@@ -65,3 +70,6 @@ async def list_us_tariff_fractions(
|
||||
"page_size": page_size,
|
||||
"pages": (total + page_size - 1) // page_size,
|
||||
}
|
||||
|
||||
|
||||
router = main_router
|
||||
|
||||
@@ -36,6 +36,7 @@ celery_app.conf.update(
|
||||
"api.v1.modules.a76.customs_brokers.imports.tasks",
|
||||
"api.v1.modules.a76.clients_and_providers.imports.tasks",
|
||||
"api.v1.modules.a76.general_catalogs.exchange_rate.imports.tasks",
|
||||
"api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.imports.tasks",
|
||||
"api.v1.modules.a76.reports.exportacion.transmission.MAINX30.task",
|
||||
"api.v1.modules.a76.reports.importacion.transmission.temporal.MAINX30.task",
|
||||
"api.v1.modules.a76.reports.importacion.transmission.definitive.MAINX30.task",
|
||||
|
||||
@@ -419,6 +419,21 @@ export const api = {
|
||||
api.post(`/v1/a76/exchange-rate/imports/${jobId}/commit`, {})
|
||||
},
|
||||
|
||||
// CSV import for Fracción Americana (us_tariff_fractions/imports)
|
||||
americanFractionImports: {
|
||||
upload: (file: File, companyId: number) => {
|
||||
const formData = new FormData();
|
||||
formData.append('file', file);
|
||||
return fetchApi(
|
||||
`/v1/a76/us-tariff-fractions/imports/upload?company_id=${companyId}`,
|
||||
{ method: 'POST', body: formData }
|
||||
);
|
||||
},
|
||||
status: (jobId: string) => api.get(`/v1/a76/us-tariff-fractions/imports/${jobId}/status`),
|
||||
commit: (jobId: string) =>
|
||||
api.post(`/v1/a76/us-tariff-fractions/imports/${jobId}/commit`, {})
|
||||
},
|
||||
|
||||
// Generic request for custom needs (like file uploads)
|
||||
request: <T = any>(endpoint: string, options: RequestInit = {}) => fetchApi<T>(endpoint, options)
|
||||
};
|
||||
|
||||
@@ -30,6 +30,8 @@
|
||||
let useClientProviderImport = $state(false);
|
||||
// Cuando es true, usamos API de importación de Tipos de Cambio (exchange_rate/imports)
|
||||
let useExchangeRateImport = $state(false);
|
||||
// Cuando es true, usamos API de importación de Fracción Americana (us_tariff_fractions/imports)
|
||||
let useAmericanFractionImport = $state(false);
|
||||
|
||||
// Initialize settings for all tabs upfront to avoid reactivity loops
|
||||
let allSettings = $state<Record<string, any>>(() => {
|
||||
@@ -51,6 +53,7 @@
|
||||
useCustomsBrokerImport = config.id === 'customs_brokers';
|
||||
useClientProviderImport = config.id === 'clients_providers';
|
||||
useExchangeRateImport = config.id === 'exchange_rates';
|
||||
useAmericanFractionImport = config.id === 'american_fractions';
|
||||
|
||||
const companyId = companyStore.activeCompany?.id || 1;
|
||||
|
||||
@@ -108,6 +111,24 @@
|
||||
return;
|
||||
}
|
||||
|
||||
if (useAmericanFractionImport) {
|
||||
try {
|
||||
const res = await api.americanFractionImports.upload(file, companyId);
|
||||
if (res.data?.job_id) {
|
||||
currentJobId = res.data.job_id;
|
||||
pollStatus();
|
||||
} else {
|
||||
toast.error(res.error || 'Error al subir el archivo');
|
||||
isUploading = false;
|
||||
}
|
||||
} catch (e) {
|
||||
console.error('Upload exception', e);
|
||||
toast.error('Error inesperado al subir el archivo');
|
||||
isUploading = false;
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
const currentSettings = allSettings[activeTab] || {};
|
||||
const footerConfig = { ...currentSettings };
|
||||
if (activeTab === 'importacion') {
|
||||
@@ -148,7 +169,9 @@
|
||||
? await api.clientProviderImports.status(currentJobId)
|
||||
: useExchangeRateImport
|
||||
? await api.exchangeRateImports.status(currentJobId)
|
||||
: await api.imports.status(currentJobId);
|
||||
: useAmericanFractionImport
|
||||
? await api.americanFractionImports.status(currentJobId)
|
||||
: await api.imports.status(currentJobId);
|
||||
console.log('Poll response', res);
|
||||
if (res.error && !res.data) {
|
||||
toast.error(res.error || 'Error al consultar el estado');
|
||||
@@ -296,7 +319,9 @@
|
||||
? await api.clientProviderImports.commit(currentJobId)
|
||||
: useExchangeRateImport
|
||||
? await api.exchangeRateImports.commit(currentJobId)
|
||||
: await api.imports.commit(currentJobId, activeModelTarget || '');
|
||||
: useAmericanFractionImport
|
||||
? await api.americanFractionImports.commit(currentJobId)
|
||||
: await api.imports.commit(currentJobId, activeModelTarget || '');
|
||||
if (res.data?.commit_job_id) {
|
||||
currentJobId = res.data.commit_job_id;
|
||||
pollStatus();
|
||||
|
||||
Reference in New Issue
Block a user