feature/BOMs-csv-WIP

This commit is contained in:
hreyes
2026-03-03 09:54:12 -07:00
parent 343054d274
commit 9da84990a8
9 changed files with 669 additions and 0 deletions

View File

@@ -0,0 +1,7 @@
"""
Módulo de importación CSV para BOMs (Bills of Materials).
"""
from .routes import router
__all__ = ["router"]

View File

@@ -0,0 +1 @@
# boms.imports

View File

@@ -0,0 +1,151 @@
"""
Rutas de importación CSV para BOMs.
Flujo: 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,
BOM_IMPORT_FILE_PREFIX,
BOM_IMPORT_META_PREFIX,
BOM_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),
):
try:
tenant_id = validate_access_to_resource(db, company_id, current_user)
except Exception as e:
logger.error(f"BOMs 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": "boms",
}
try:
r = _get_redis()
r.set(
f"{BOM_IMPORT_FILE_PREFIX}{job_id}",
base64.b64encode(contents),
ex=BOM_IMPORT_REDIS_TTL,
)
r.set(
f"{BOM_IMPORT_META_PREFIX}{job_id}",
json.dumps(meta_data).encode("utf-8"),
ex=BOM_IMPORT_REDIS_TTL,
)
except Exception as e:
logger.error(f"BOMs 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"bom_{job_id}.csv"), "wb") as f:
f.write(contents)
with open(os.path.join(upload_dir, f"bom_{job_id}.meta.json"), "w") as f:
json.dump(meta_data, f)
except Exception as e:
logger.warning(f"BOMs 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):
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("BOMs 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):
task = insert_valid_rows.delay(job_id)
return {
"status": "committing",
"message": "Inserción iniciada.",
"commit_job_id": task.id,
}

View File

@@ -0,0 +1,22 @@
from pydantic import BaseModel
from typing import Optional
class ImportJobResponse(BaseModel):
job_id: str
status: str
message: str
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_duplicate: Optional[int] = 0
skipped_details: Optional[list] = None

View File

@@ -0,0 +1,430 @@
"""
Tareas Celery para importación CSV de BOMs.
Flujo: scan_file (validación) → insert_valid_rows (commit).
Sin tabla BOM dedicada aún: insert_valid_rows solo valida y devuelve resultado; el mapeo a tabla se añadirá cuando exista.
"""
import os
import base64
import csv
import json
import logging
import re
import unicodedata
from decimal import Decimal, InvalidOperation
from typing import Dict, Any, Optional, List, Set
from core.celery_app import celery_app
from core.database import CoreSessionLocal
from .template_config import row_from_template
logger = logging.getLogger(__name__)
BOM_IMPORT_FILE_PREFIX = "bom_import_file:"
BOM_IMPORT_META_PREFIX = "bom_import_meta:"
BOM_IMPORT_ERROR_LINES_PREFIX = "bom_import_error_lines:"
BOM_IMPORT_REDIS_TTL = 3600
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"{BOM_IMPORT_FILE_PREFIX}{job_id}")
if not data:
return None
try:
raw = base64.b64decode(data)
except Exception as e:
logger.warning(f"BOMs 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"bom_{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"{BOM_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"BOMs 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"{BOM_IMPORT_FILE_PREFIX}{job_id}",
f"{BOM_IMPORT_META_PREFIX}{job_id}",
f"{BOM_IMPORT_ERROR_LINES_PREFIX}{job_id}",
)
except Exception as e:
logger.warning(f"BOMs 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 _validate_row_bom(
row: Dict[str, Any],
line_num: int,
valid_part_numbers: Optional[Set[str]] = None,
) -> Optional[Dict[str, Any]]:
"""Valida una fila BOM según columnas del template. FK opcional a a76.parts."""
parent = (row.get("NUMPARTE_PADRE") or "").strip()
if not parent:
return {"line": line_num, "col": "NUMPARTE_PADRE", "msg": "Requerido"}
if len(parent) > 70:
return {"line": line_num, "col": "NUMPARTE_PADRE", "msg": "Máximo 70 caracteres"}
component = (row.get("NUMPARTE_COMPONENTE") or "").strip()
if not component:
return {"line": line_num, "col": "NUMPARTE_COMPONENTE", "msg": "Requerido"}
if len(component) > 70:
return {"line": line_num, "col": "NUMPARTE_COMPONENTE", "msg": "Máximo 70 caracteres"}
qty = row.get("CANTIDAD")
if qty is None or qty == "":
return {"line": line_num, "col": "CANTIDAD", "msg": "Requerido"}
try:
val = Decimal(str(qty))
if val < 0:
return {"line": line_num, "col": "CANTIDAD", "msg": "Debe ser mayor o igual a cero"}
except (InvalidOperation, ValueError, TypeError):
return {"line": line_num, "col": "CANTIDAD", "msg": "Debe ser número"}
uom = (row.get("UNIMED") or "").strip()
if uom and len(uom) > 10:
return {"line": line_num, "col": "UNIMED", "msg": "Máximo 10 caracteres"}
def _optional_number(val: Any) -> bool:
if val is None:
return True
s = re.sub(r"\s+", "", str(val).strip())
if not s:
return True
try:
float(s.replace(",", "."))
return True
except (ValueError, TypeError):
return False
version_bom = row.get("VERSION_BOM")
if not _optional_number(version_bom):
return {"line": line_num, "col": "VERSION_BOM", "msg": "Debe ser número"}
version_bill = row.get("VERSION_BILL")
if not _optional_number(version_bill):
return {"line": line_num, "col": "VERSION_BILL", "msg": "Debe ser número"}
# Solo exigir que padre/componente existan en catálogo si hay partes cargadas (evita rechazar todo cuando el catálogo está vacío o en pruebas)
if valid_part_numbers is not None and len(valid_part_numbers) > 0:
if parent not in valid_part_numbers:
return {"line": line_num, "col": "NUMPARTE_PADRE", "msg": "Parte padre no existe en catálogo"}
if component not in valid_part_numbers:
return {"line": line_num, "col": "NUMPARTE_COMPONENTE", "msg": "Parte componente no existe en catálogo"}
return None
@celery_app.task(bind=True)
def scan_file(self, job_id: str, config: str = None):
logger.info(f"BOMs 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"bom_{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"BOMs 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)"}
valid_part_numbers: Set[str] = set()
try:
with CoreSessionLocal() as session:
from api.v1.modules.a76.parts.models import Part
for p in (
session.query(Part.part_number)
.filter(
Part.tenant_id == tenant_id,
Part.company_id == company_id,
)
.all()
):
valid_part_numbers.add(p[0])
except Exception as e:
logger.warning(f"BOMs import: could not load parts for FK validation: {e}")
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)
err = _validate_row_bom(row_norm, i, valid_part_numbers=valid_part_numbers)
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"BOMs 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"{BOM_IMPORT_ERROR_LINES_PREFIX}{job_id}",
json.dumps(error_lines_list).encode("utf-8"),
ex=BOM_IMPORT_REDIS_TTL,
)
except Exception as e:
logger.warning(f"BOMs 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 _decimal_or_none(val: Any) -> Optional[Decimal]:
if val is None or val == "":
return None
try:
return Decimal(str(val))
except (InvalidOperation, ValueError, TypeError):
return None
def _int_or_none(val: Any) -> Optional[int]:
if val is None or val == "":
return None
try:
return int(val)
except (ValueError, TypeError):
return None
@celery_app.task(bind=True)
def insert_valid_rows(self, job_id: str):
logger.info(f"BOMs 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"bom_{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"bom_{job_id}.jsonl")
error_lines = set()
try:
r = _get_redis()
raw = r.get(f"{BOM_IMPORT_ERROR_LINES_PREFIX}{job_id}")
if raw:
error_lines = set(json.loads(raw.decode("utf-8")))
except Exception as e:
logger.debug(f"BOMs 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)"}
valid_part_numbers: Set[str] = set()
try:
with CoreSessionLocal() as session:
from api.v1.modules.a76.parts.models import Part
for p in (
session.query(Part.part_number)
.filter(
Part.tenant_id == tenant_id,
Part.company_id == company_id,
)
.all()
):
valid_part_numbers.add(p[0])
except Exception as e:
logger.warning(f"BOMs import: could not load parts: {e}")
inserted_count = 0
skipped_invalid = 0
skipped_details: List[Dict[str, Any]] = []
valid_count = 0
try:
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)
err = _validate_row_bom(row_norm, i, valid_part_numbers=valid_part_numbers)
if err:
skipped_invalid += 1
skipped_details.append(
{"line": i, "reason": f"{err.get('col', '')}: {err.get('msg', '')}"}
)
continue
valid_count += 1
# Sin tabla BOM dedicada: no se escribe en DB; solo se cuentan filas válidas.
# Cuando exista la tabla de destino, aquí se hará insert/update.
response = {
"status": "finished",
"inserted": inserted_count,
"skipped_invalid": skipped_invalid,
"skipped_missing_fk": 0,
"skipped_duplicate": 0,
"skipped_details": skipped_details,
}
if valid_count > 0 and inserted_count == 0:
response["message"] = f"WIP: {valid_count} filas válidas. La tabla BOM aún no existe en el sistema; no se insertó nada."
except Exception as e:
logger.error(f"BOMs import task failed: {e}")
import traceback
logger.error(traceback.format_exc())
response = {
"status": "failed",
"error": str(e),
"inserted": 0,
"skipped_invalid": skipped_invalid,
"skipped_missing_fk": 0,
"skipped_duplicate": 0,
"skipped_details": skipped_details,
}
try:
if file_path and os.path.exists(file_path):
os.remove(file_path)
if os.path.exists(error_path):
os.remove(error_path)
if os.path.exists(meta_path):
os.remove(meta_path)
_delete_import_from_redis(job_id)
except Exception as cleanup_err:
logger.warning(f"BOMs import cleanup failed: {cleanup_err}")
return response

View File

@@ -0,0 +1,42 @@
"""
Configuración de plantilla CSV para BOMs (EstructuraBOMS.xlsx).
Placeholders hasta tener el XLS definitivo; ajustar canónicos y aliases según el archivo.
"""
from typing import Dict, List, Any
TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = {
"boms": [
{"canonical": "NUMPARTE_PADRE", "aliases": ["PARTE PADRE", "PART NUMBER", "PARENT PART", "NUM PARTE PADRE"]},
{"canonical": "NUMPARTE_COMPONENTE", "aliases": ["PARTE COMPONENTE", "COMPONENT PART", "NUM PARTE COMPONENTE"]},
{"canonical": "CANTIDAD", "aliases": ["QTY", "QUANTITY", "CANT"]},
{"canonical": "UNIMED", "aliases": ["UNIDAD MEDIDA", "UNIT", "UOM"]},
{"canonical": "VERSION_BOM", "aliases": ["VERSION BOM", "BOM VERSION", "VERSIONBOM"]},
{"canonical": "VERSION_BILL", "aliases": ["VERSION BILL", "BILL VERSION", "VERSIONBILL"]},
],
}
def build_normalized_lookup(normalize_header_fn) -> Dict[str, str]:
cols = TEMPLATE_COLUMNS.get("boms")
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) -> Dict[str, Any]:
lookup = build_normalized_lookup(normalize_header_fn)
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

View File

@@ -0,0 +1,13 @@
"""
Endpoints para importación CSV de BOMs.
Mismo patrón que parts y classes: upload → scan → status → commit.
"""
from fastapi import APIRouter
from .imports.routes import router as imports_router
router = APIRouter()
# CSV import (upload → scan → status → commit)
router.include_router(imports_router, prefix="/imports", tags=["a76 / boms / csv_import"])

View File

@@ -23,6 +23,7 @@ from .transportation.drivers.routes import router as drivers_router
from .doc_types_dig.routes import router as doc_types_dig_router
from .fraction_rule_octave.routes import router as fraction_rule_octave_router
from .parts import router as parts_router
from .boms import router as boms_router
from .pedmientos.router import router as pedimentos_router
from .permission_rule_oct.routes import router as permission_rule_oct_router
from .transportation.trailers.routes import router as trailers_router
@@ -59,6 +60,7 @@ router.include_router(pedimentos_router, prefix="/a76")
router.include_router(client_and_provider_router, prefix="/a76", tags=["a76 / clients_and_providers"])
router.include_router(classes_router, prefix="/a76/classes", tags=["a76 / classes"])
router.include_router(parts_router, prefix="/a76", tags=["a76 / parts"])
router.include_router(boms_router, prefix="/a76/boms", tags=["a76 / boms"])
router.include_router(permission_rule_oct_router, prefix="/a76", tags=["a76 / permission_rule_oct"])
router.include_router(fraction_rule_octave_router, prefix="/a76", tags=["a76 / fraction_rule_octave"])
router.include_router(country_rule_oct_router, prefix="/a76", tags=["a76 / country_rule_oct"])

View File

@@ -40,6 +40,7 @@ celery_app.conf.update(
"api.v1.modules.a76.general_catalogs.fractions.us_tariff_fractions.imports.tasks",
"api.v1.modules.a76.classes.imports.tasks",
"api.v1.modules.a76.parts.imports.tasks",
"api.v1.modules.a76.boms.imports.tasks",
"api.v1.modules.a76.transportation.vehicles.imports.tasks",
"api.v1.modules.a76.reports.exportacion.transmission.MAINX30.task",
"api.v1.modules.a76.reports.importacion.transmission.temporal.MAINX30.task",