diff --git a/backend/api/v1/modules/a76/pedmientos/imports/__init__.py b/backend/api/v1/modules/a76/pedmientos/imports/__init__.py new file mode 100644 index 00000000..8378bf88 --- /dev/null +++ b/backend/api/v1/modules/a76/pedmientos/imports/__init__.py @@ -0,0 +1 @@ +# CSV import for Pedimentos (upload → scan → commit) diff --git a/backend/api/v1/modules/a76/pedmientos/imports/routes.py b/backend/api/v1/modules/a76/pedmientos/imports/routes.py new file mode 100644 index 00000000..f624e5dc --- /dev/null +++ b/backend/api/v1/modules/a76/pedmientos/imports/routes.py @@ -0,0 +1,160 @@ +""" +Rutas de importación CSV para Pedimentos. +Mismo flujo que customs_brokers/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, + PED_IMPORT_FILE_PREFIX, + PED_IMPORT_META_PREFIX, + PED_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"Pedimentos 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": "pedimentos", + } + + try: + r = _get_redis() + r.set( + f"{PED_IMPORT_FILE_PREFIX}{job_id}", + base64.b64encode(contents), + ex=PED_IMPORT_REDIS_TTL, + ) + r.set( + f"{PED_IMPORT_META_PREFIX}{job_id}", + json.dumps(meta_data).encode("utf-8"), + ex=PED_IMPORT_REDIS_TTL, + ) + except Exception as e: + logger.error(f"Pedimentos 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"ped_{job_id}.csv"), "wb") as f: + f.write(contents) + with open(os.path.join(upload_dir, f"ped_{job_id}.meta.json"), "w") as f: + json.dump(meta_data, f) + except Exception as e: + logger.warning(f"Pedimentos 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("Pedimentos 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, + } diff --git a/backend/api/v1/modules/a76/pedmientos/imports/schemas.py b/backend/api/v1/modules/a76/pedmientos/imports/schemas.py new file mode 100644 index 00000000..6fe69385 --- /dev/null +++ b/backend/api/v1/modules/a76/pedmientos/imports/schemas.py @@ -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: Optional[str] = None + 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 diff --git a/backend/api/v1/modules/a76/pedmientos/imports/tasks.py b/backend/api/v1/modules/a76/pedmientos/imports/tasks.py new file mode 100644 index 00000000..82b630d2 --- /dev/null +++ b/backend/api/v1/modules/a76/pedmientos/imports/tasks.py @@ -0,0 +1,563 @@ +""" +Tareas Celery para importación CSV de Pedimentos. +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, InvalidOperation +from datetime import datetime +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__) + +# Redis keys (prefijo propio para no colisionar con otros imports) +PED_IMPORT_FILE_PREFIX = "ped_import_file:" +PED_IMPORT_META_PREFIX = "ped_import_meta:" +PED_IMPORT_ERROR_LINES_PREFIX = "ped_import_error_lines:" +PED_IMPORT_REDIS_TTL = 3600 # 1 hour + + +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"{PED_IMPORT_FILE_PREFIX}{job_id}") + if not data: + return None + try: + raw = base64.b64decode(data) + except Exception as e: + logger.warning(f"Pedimentos 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"ped_{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"{PED_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"Pedimentos 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"{PED_IMPORT_FILE_PREFIX}{job_id}", + f"{PED_IMPORT_META_PREFIX}{job_id}", + f"{PED_IMPORT_ERROR_LINES_PREFIX}{job_id}", + ) + except Exception as e: + logger.warning(f"Pedimentos 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_pedimento( + row: Dict[str, Any], + line_num: int, + valid_client_ids: Set[int], + valid_regimes: Set[str], + valid_pedimento_codes: Set[str], +) -> Optional[Dict[str, Any]]: + """Valida una fila para Pedimento. Retorna error dict o None.""" + # Required: AÑO, ADUANA, PATENTE, NUMERO, CLIENTE_ID, CODIGO_PEDIMENTO, REGIMEN + year = (row.get("AÑO") or "").strip() + if not year: + return {"line": line_num, "col": "AÑO", "msg": "Requerido"} + if len(year) > 2: + return {"line": line_num, "col": "AÑO", "msg": "Máximo 2 caracteres"} + + customs_office = (row.get("ADUANA") or "").strip() + if not customs_office: + return {"line": line_num, "col": "ADUANA", "msg": "Requerido"} + if len(customs_office) > 3: + return {"line": line_num, "col": "ADUANA", "msg": "Máximo 3 caracteres"} + + license_val = (row.get("PATENTE") or "").strip() + if not license_val: + return {"line": line_num, "col": "PATENTE", "msg": "Requerido"} + if len(license_val) > 4: + return {"line": line_num, "col": "PATENTE", "msg": "Máximo 4 caracteres"} + + pedimento_number = (row.get("NUMERO") or "").strip() + if not pedimento_number: + return {"line": line_num, "col": "NUMERO", "msg": "Requerido"} + if len(pedimento_number) > 7: + return {"line": line_num, "col": "NUMERO", "msg": "Máximo 7 caracteres"} + + client_id_str = (row.get("CLIENTE_ID") or "").strip() + if not client_id_str: + return {"line": line_num, "col": "CLIENTE_ID", "msg": "Requerido"} + try: + client_id = int(client_id_str) + except ValueError: + return {"line": line_num, "col": "CLIENTE_ID", "msg": "Debe ser número entero"} + if client_id not in valid_client_ids: + return {"line": line_num, "col": "CLIENTE_ID", "msg": "Cliente no existe en catálogo"} + + pedimento_code = (row.get("CODIGO_PEDIMENTO") or "").strip() + if not pedimento_code: + return {"line": line_num, "col": "CODIGO_PEDIMENTO", "msg": "Requerido"} + if len(pedimento_code) > 2: + return {"line": line_num, "col": "CODIGO_PEDIMENTO", "msg": "Máximo 2 caracteres"} + if pedimento_code not in valid_pedimento_codes: + return {"line": line_num, "col": "CODIGO_PEDIMENTO", "msg": "Código no existe en catálogo"} + + regime = (row.get("REGIMEN") or "").strip() + if not regime: + return {"line": line_num, "col": "REGIMEN", "msg": "Requerido"} + if len(regime) > 3: + return {"line": line_num, "col": "REGIMEN", "msg": "Máximo 3 caracteres"} + if regime not in valid_regimes: + return {"line": line_num, "col": "REGIMEN", "msg": "Régimen no existe en catálogo"} + + # Optional numeric/string fields - validate format if present + status = (row.get("ESTATUS") or "").strip() + if status and len(status) > 30: + return {"line": line_num, "col": "ESTATUS", "msg": "Máximo 30 caracteres"} + + for col, max_len in [("VALOR_USD", 17), ("PRECIO_PAGADO", 17), ("PESO_BRUTO", 19), ("TIPO_CAMBIO", 9)]: + val = (row.get(col) or "").strip() + if not val: + continue + try: + Decimal(val.replace(",", ".")) + except (InvalidOperation, ValueError): + return {"line": line_num, "col": col, "msg": "Valor numérico inválido"} + + operation_type = (row.get("TIPO_OPERACION") or "").strip().lower() + if operation_type and operation_type not in ("imp", "exp", ""): + return {"line": line_num, "col": "TIPO_OPERACION", "msg": "Debe ser imp o exp"} + + pedimento_type = (row.get("TIPO_PEDIMENTO") or "").strip().lower() + if pedimento_type and pedimento_type not in ("normal", "consolidated", "complementary", "automobile", ""): + return {"line": line_num, "col": "TIPO_PEDIMENTO", "msg": "Tipo no válido (normal, consolidated, complementary, automobile)"} + + return None + + +def _parse_decimal(val: Any) -> Optional[Decimal]: + if val is None or (isinstance(val, str) and not val.strip()): + return None + try: + return Decimal(str(val).strip().replace(",", ".")) + except (InvalidOperation, ValueError): + return None + + +def _row_to_pedimentos_create(row: Dict[str, Any]) -> Dict[str, Any]: + """Build dict for PedimentosCreate from normalized CSV row (canonical names).""" + year = (row.get("AÑO") or "").strip()[:2] + customs_office = (row.get("ADUANA") or "").strip()[:3] + license_val = (row.get("PATENTE") or "").strip()[:4] + pedimento_number = (row.get("NUMERO") or "").strip()[:7] + client_id_str = (row.get("CLIENTE_ID") or "").strip() + client_id = int(client_id_str) if client_id_str else None + pedimento_code = (row.get("CODIGO_PEDIMENTO") or "").strip()[:2] + regime = (row.get("REGIMEN") or "").strip()[:3] + + data = { + "year": year, + "customs_office": customs_office, + "license": license_val, + "pedimento_number": pedimento_number, + "client_id": client_id, + "pedimento_code": pedimento_code, + "regime": regime, + } + + op = (row.get("TIPO_OPERACION") or "").strip().lower() + if op in ("imp", "exp"): + data["operation_type"] = op + + ptype = (row.get("TIPO_PEDIMENTO") or "").strip().lower() + if ptype in ("normal", "consolidated", "complementary", "automobile"): + data["pedimento_type"] = ptype + + status = (row.get("ESTATUS") or "").strip() + if status: + data["status"] = status[:30] + + data["usd_value"] = _parse_decimal(row.get("VALOR_USD")) + data["paid_price"] = _parse_decimal(row.get("PRECIO_PAGADO")) + data["gross_weight"] = _parse_decimal(row.get("PESO_BRUTO")) + data["exchange_rate"] = _parse_decimal(row.get("TIPO_CAMBIO")) + + obs = (row.get("OBSERVACIONES") or "").strip() + if obs: + data["observations"] = obs + + return data + + +@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"Pedimentos 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"ped_{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"Pedimentos 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)"} + + # Load valid FK sets for validation + valid_client_ids: Set[int] = set() + valid_regimes: Set[str] = set() + valid_pedimento_codes: Set[str] = set() + try: + with CoreSessionLocal() as session: + from api.v1.modules.a76.clients_and_providers.models import ClientProvider + from api.v1.modules.public.reference_data.pedimento_regimens.models import RegimenPedimento + from api.v1.modules.public.reference_data.pedimento_codes.models import PedimentoCode + + for cp in session.query(ClientProvider).filter( + ClientProvider.tenant_id == tenant_id, + ClientProvider.company_id == company_id, + ).all(): + valid_client_ids.add(cp.id) + for r in session.query(RegimenPedimento).all(): + valid_regimes.add(r.code) + for pc in session.query(PedimentoCode).all(): + valid_pedimento_codes.add(pc.code) + except Exception as e: + logger.error(f"Pedimentos import: failed to load FK sets: {e}") + return {"status": "failed", "error": f"No se pudo cargar catálogos: {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, "pedimentos") + err = _validate_row_pedimento( + row_norm, i, valid_client_ids, valid_regimes, valid_pedimento_codes + ) + 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"Pedimentos 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"{PED_IMPORT_ERROR_LINES_PREFIX}{job_id}", + json.dumps(error_lines_list).encode("utf-8"), + ex=PED_IMPORT_REDIS_TTL, + ) + except Exception as e: + logger.warning(f"Pedimentos 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, + } + + +@celery_app.task(bind=True) +def insert_valid_rows(self, job_id: str): + """ + Fase 2: Re-leer CSV, omitir filas con error, insertar Pedimentos vía PedimentosService.create. + """ + logger.info(f"Pedimentos 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"ped_{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"ped_{job_id}.jsonl") + + error_lines = set() + try: + r = _get_redis() + raw = r.get(f"{PED_IMPORT_ERROR_LINES_PREFIX}{job_id}") + if raw: + error_lines = set(json.loads(raw.decode("utf-8"))) + except Exception as e: + logger.debug(f"Pedimentos 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)"} + + # Reload FK sets for commit-time validation + valid_client_ids: Set[int] = set() + valid_regimes: Set[str] = set() + valid_pedimento_codes: Set[str] = set() + try: + with CoreSessionLocal() as session: + from api.v1.modules.a76.clients_and_providers.models import ClientProvider + from api.v1.modules.public.reference_data.pedimento_regimens.models import RegimenPedimento + from api.v1.modules.public.reference_data.pedimento_codes.models import PedimentoCode + + for cp in session.query(ClientProvider).filter( + ClientProvider.tenant_id == tenant_id, + ClientProvider.company_id == company_id, + ).all(): + valid_client_ids.add(cp.id) + for r in session.query(RegimenPedimento).all(): + valid_regimes.add(r.code) + for pc in session.query(PedimentoCode).all(): + valid_pedimento_codes.add(pc.code) + except Exception as e: + logger.error(f"Pedimentos import: failed to load FK sets: {e}") + return {"status": "failed", "error": str(e)} + + from api.v1.modules.a76.pedmientos.dtos.pedimentos import PedimentosCreate + from api.v1.modules.a76.pedmientos.dtos.pedimento_dates import PedimentoDatesCreate + from api.v1.modules.a76.pedmientos.services.pedimentos import PedimentosService + + inserted_count = 0 + skipped_invalid = 0 + skipped_missing_fk = 0 + skipped_duplicate = 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, "pedimentos") + err = _validate_row_pedimento( + row_norm, i, valid_client_ids, valid_regimes, valid_pedimento_codes + ) + if err: + skipped_invalid += 1 + skipped_details.append( + {"line": i, "reason": f"{err.get('col', '')}: {err.get('msg', '')}"} + ) + continue + + try: + data = _row_to_pedimentos_create(row_norm) + # Service expects at least pedimento_dates with entry_date/end_date for DB NOT NULL + if "pedimento_dates" not in data or data.get("pedimento_dates") is None: + data["pedimento_dates"] = PedimentoDatesCreate( + entry_date=datetime.now(), + end_date=datetime.now(), + ) + create_data = PedimentosCreate(**data) + PedimentosService.create(session, create_data, tenant_id, company_id) + inserted_count += 1 + except ValueError as ve: + if "Ya existe" in str(ve) or "duplicate" in str(ve).lower(): + skipped_duplicate += 1 + skipped_details.append({"line": i, "reason": str(ve)}) + else: + skipped_invalid += 1 + skipped_details.append({"line": i, "reason": str(ve)}) + except Exception as e: + logger.warning(f"Pedimentos import line {i}: {e}") + skipped_invalid += 1 + skipped_details.append({"line": i, "reason": str(e)}) + + total_skipped = skipped_invalid + skipped_missing_fk + skipped_duplicate + if inserted_count == 0 and total_skipped > 0: + response = { + "status": "warning", + "inserted": 0, + "skipped_invalid": skipped_invalid, + "skipped_missing_fk": skipped_missing_fk, + "skipped_duplicate": skipped_duplicate, + "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": skipped_missing_fk, + "skipped_duplicate": skipped_duplicate, + "skipped_details": skipped_details, + } + else: + response = { + "status": "finished", + "inserted": inserted_count, + "skipped_invalid": skipped_invalid, + "skipped_missing_fk": skipped_missing_fk, + "skipped_duplicate": skipped_duplicate, + "skipped_details": skipped_details, + } + + except Exception as e: + logger.error(f"Pedimentos 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 = file_path.replace(".csv", ".meta.json") + if os.path.exists(meta_path): + os.remove(meta_path) + _delete_import_from_redis(job_id) + except Exception as cleanup_err: + logger.warning(f"Pedimentos import cleanup failed: {cleanup_err}") + + if response is None: + response = { + "status": "failed", + "error": "Error inesperado", + "inserted": 0, + "skipped_invalid": skipped_invalid, + "skipped_missing_fk": skipped_missing_fk, + "skipped_duplicate": skipped_duplicate, + "skipped_details": skipped_details, + } + return response diff --git a/backend/api/v1/modules/a76/pedmientos/imports/template_config.py b/backend/api/v1/modules/a76/pedmientos/imports/template_config.py new file mode 100644 index 00000000..06fea167 --- /dev/null +++ b/backend/api/v1/modules/a76/pedmientos/imports/template_config.py @@ -0,0 +1,53 @@ +""" +Configuración de plantilla CSV para Pedimentos (EstructuraCatPedimentos.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]]] = { + "pedimentos": [ + {"canonical": "AÑO", "aliases": ["YEAR", "ANIO"]}, + {"canonical": "ADUANA", "aliases": ["CUSTOMS_OFFICE", "CUSTOMS OFFICE"]}, + {"canonical": "PATENTE", "aliases": ["LICENCIA", "LICENSE", "LIC"]}, + {"canonical": "NUMERO", "aliases": ["PEDIMENTO_NUMBER", "PEDIMENTO NUMBER", "NUMERO PEDIMENTO"]}, + {"canonical": "CLIENTE_ID", "aliases": ["CLIENT_ID", "CLIENTE", "ID CLIENTE"]}, + {"canonical": "TIPO_OPERACION", "aliases": ["OPERATION_TYPE", "OPERACION"]}, + {"canonical": "TIPO_PEDIMENTO", "aliases": ["PEDIMENTO_TYPE", "TIPO"]}, + {"canonical": "CODIGO_PEDIMENTO", "aliases": ["PEDIMENTO_CODE", "CODIGO", "CLAVE PEDIMENTO"]}, + {"canonical": "REGIMEN", "aliases": ["REGIME"]}, + {"canonical": "ESTATUS", "aliases": ["STATUS", "ESTADO"]}, + {"canonical": "VALOR_USD", "aliases": ["USD_VALUE", "VALOR USD", "USD"]}, + {"canonical": "PRECIO_PAGADO", "aliases": ["PAID_PRICE", "PRECIO PAGADO"]}, + {"canonical": "PESO_BRUTO", "aliases": ["GROSS_WEIGHT", "PESO BRUTO", "PESO"]}, + {"canonical": "TIPO_CAMBIO", "aliases": ["EXCHANGE_RATE", "TIPO CAMBIO", "CAMBIO"]}, + {"canonical": "OBSERVACIONES", "aliases": ["OBSERVATIONS", "OBS", "NOTAS"]}, + ], +} + + +def build_normalized_lookup(normalize_header_fn, template_id: str = "pedimentos") -> Dict[str, str]: + """normalized_header -> canonical_name para plantilla pedimentos.""" + 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 = "pedimentos") -> 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 diff --git a/backend/api/v1/modules/a76/pedmientos/router.py b/backend/api/v1/modules/a76/pedmientos/router.py index f294cba4..2b987b46 100644 --- a/backend/api/v1/modules/a76/pedmientos/router.py +++ b/backend/api/v1/modules/a76/pedmientos/router.py @@ -31,6 +31,7 @@ from .routes.pedimento_rectification_origin import ( from .routes.pedimento_transport_means import router as pedimento_transport_means_router from .routes.pedimento_validation import router as pedimento_validation_router from .routes.pedimentos import router as pedimentos_router +from .imports.routes import router as pedimentos_imports_router router = APIRouter() @@ -117,3 +118,8 @@ router.include_router( router.include_router( pedimentos_router, prefix="/pedimentos", tags=["a76 / pedimentos"] ) +router.include_router( + pedimentos_imports_router, + prefix="/pedimentos/imports", + tags=["a76 / pedimentos / csv_import"], +) diff --git a/backend/core/celery_app.py b/backend/core/celery_app.py index 41a98b99..f0b91d0a 100644 --- a/backend/core/celery_app.py +++ b/backend/core/celery_app.py @@ -35,6 +35,7 @@ celery_app.conf.update( "api.v1.modules.a76.imports.tasks", "api.v1.modules.a76.customs_brokers.imports.tasks", "api.v1.modules.a76.clients_and_providers.imports.tasks", + "api.v1.modules.a76.pedmientos.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", diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 19d4d82c..55964f53 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -434,6 +434,21 @@ export const api = { api.post(`/v1/a76/us-tariff-fractions/imports/${jobId}/commit`, {}) }, + // CSV import for Pedimentos (pedimentos/imports) + pedimentosImports: { + upload: (file: File, companyId: number) => { + const formData = new FormData(); + formData.append('file', file); + return fetchApi( + `/v1/a76/pedimentos/imports/upload?company_id=${companyId}`, + { method: 'POST', body: formData } + ); + }, + status: (jobId: string) => api.get(`/v1/a76/pedimentos/imports/${jobId}/status`), + commit: (jobId: string) => + api.post(`/v1/a76/pedimentos/imports/${jobId}/commit`, {}) + }, + // Generic request for custom needs (like file uploads) request: (endpoint: string, options: RequestInit = {}) => fetchApi(endpoint, options) }; diff --git a/frontend/src/routes/dashboard/csv-upload/+page.svelte b/frontend/src/routes/dashboard/csv-upload/+page.svelte index 05471225..7b2f5aa0 100644 --- a/frontend/src/routes/dashboard/csv-upload/+page.svelte +++ b/frontend/src/routes/dashboard/csv-upload/+page.svelte @@ -32,6 +32,8 @@ let useExchangeRateImport = $state(false); // Cuando es true, usamos API de importación de Fracción Americana (us_tariff_fractions/imports) let useAmericanFractionImport = $state(false); + // Cuando es true, usamos API de importación de Pedimentos (pedimentos/imports) + let usePedimentosImport = $state(false); // Initialize settings for all tabs upfront to avoid reactivity loops let allSettings = $state>(() => { @@ -54,6 +56,7 @@ useClientProviderImport = config.id === 'clients_providers'; useExchangeRateImport = config.id === 'exchange_rates'; useAmericanFractionImport = config.id === 'american_fractions'; + usePedimentosImport = config.id === 'pedimentos'; const companyId = companyStore.activeCompany?.id || 1; @@ -129,6 +132,24 @@ return; } + if (usePedimentosImport) { + try { + const res = await api.pedimentosImports.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') { @@ -171,7 +192,9 @@ ? await api.exchangeRateImports.status(currentJobId) : useAmericanFractionImport ? await api.americanFractionImports.status(currentJobId) - : await api.imports.status(currentJobId); + : usePedimentosImport + ? await api.pedimentosImports.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'); @@ -205,7 +228,8 @@ const inserted = res.data?.inserted || 0; const skippedInvalid = res.data?.skipped_invalid || 0; const skippedFk = res.data?.skipped_missing_fk || 0; - const totalSkipped = skippedInvalid + skippedFk; + const skippedDup = res.data?.skipped_duplicate || 0; + const totalSkipped = skippedInvalid + skippedFk + skippedDup; if (inserted === 0) { toast.error(`No se insertaron registros. ${totalSkipped} fueron rechazados.`); @@ -219,12 +243,13 @@ const inserted = res.data?.inserted || 0; const skippedInvalid = res.data?.skipped_invalid || 0; const skippedFk = res.data?.skipped_missing_fk || 0; + const skippedDup = res.data?.skipped_duplicate || 0; const skippedDetails = res.data?.skipped_details || []; if (inserted > 0) { toast.success(`Importación completada: ${inserted} registros insertados`); - if (skippedInvalid > 0 || skippedFk > 0) { - const totalSkipped = skippedInvalid + skippedFk; + if (skippedInvalid > 0 || skippedFk > 0 || skippedDup > 0) { + const totalSkipped = skippedInvalid + skippedFk + skippedDup; toast.warning(`${totalSkipped} registros fueron rechazados`); } } else { @@ -321,7 +346,9 @@ ? await api.exchangeRateImports.commit(currentJobId) : useAmericanFractionImport ? await api.americanFractionImports.commit(currentJobId) - : await api.imports.commit(currentJobId, activeModelTarget || ''); + : usePedimentosImport + ? await api.pedimentosImports.commit(currentJobId) + : await api.imports.commit(currentJobId, activeModelTarget || ''); if (res.data?.commit_job_id) { currentJobId = res.data.commit_job_id; pollStatus(); diff --git a/scripts/backend-entrypoint.sh b/scripts/backend-entrypoint.sh index 20c5f89f..85cb5432 100755 --- a/scripts/backend-entrypoint.sh +++ b/scripts/backend-entrypoint.sh @@ -19,7 +19,7 @@ wait_for_tcp() { echo "Esperando a que $service esté disponible en ${host}:${port}..." while [ $attempt -le $max_attempts ]; do - if python -c "import socket; s = socket.socket(); s.settimeout(3); s.connect(('$host', $port)); s.close()" 2>/dev/null; then + if python3 -c "import socket; s = socket.socket(); s.settimeout(3); s.connect(('$host', $port)); s.close()" 2>/dev/null; then echo "✓ $service está listo y accesible" return 0 fi