526 lines
22 KiB
Python
526 lines
22 KiB
Python
"""
|
|
Tareas Celery para importación CSV de Conductores.
|
|
Flujo: scan_file (validación) → insert_valid_rows (commit).
|
|
Usa layouts_csv.common (storage, normalize, meta, responses); CSV con headers duplicados (dedupe) y clave de estado en Redis.
|
|
Paridad Clarion: actualizar, existing_driver_keys, valid_transporter_keys, valid_country_ame.
|
|
"""
|
|
import csv
|
|
import json
|
|
import logging
|
|
import os
|
|
from typing import Dict, Any, Optional, List, Set, Tuple
|
|
|
|
from core.celery_app import celery_app
|
|
from core.database import CoreSessionLocal
|
|
from sqlalchemy import func
|
|
|
|
from ..common import storage as common_storage
|
|
from ..common import normalize as common_normalize
|
|
from ..common import meta as common_meta
|
|
from ..common import responses as common_responses
|
|
from .template_config import row_from_template
|
|
from .validators import validate_row_driver, validate_row_driver_desfase
|
|
from .common.mappers import row_to_driver_data
|
|
from .common.fk_loader import load_drivers_fk_sets
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
JOB_TYPE = "drv"
|
|
|
|
# Para routes.py
|
|
DRV_IMPORT_FILE_PREFIX = "drv_import_file:"
|
|
DRV_IMPORT_META_PREFIX = "drv_import_meta:"
|
|
DRV_IMPORT_ERROR_LINES_PREFIX = "drv_import_error_lines:"
|
|
DRV_IMPORT_STATUS_PREFIX = "drv_import_status:"
|
|
DRV_IMPORT_REDIS_TTL = common_storage.IMPORT_REDIS_TTL
|
|
DRV_IMPORT_TRANSPORTER_MAP_PREFIX = "drv_import_transporter_map:"
|
|
|
|
|
|
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 _dedupe_headers(headers: List[str]) -> List[str]:
|
|
counts: Dict[str, int] = {}
|
|
unique: List[str] = []
|
|
for header in headers:
|
|
name = str(header or "").strip() or "COL"
|
|
count = counts.get(name, 0) + 1
|
|
counts[name] = count
|
|
unique.append(name if count == 1 else f"{name} {count}")
|
|
return unique
|
|
|
|
|
|
def _do_scan(job_id: str, progress_callback: Optional[Any] = None) -> Dict[str, Any]:
|
|
file_path = common_storage.ensure_file_from_redis(JOB_TYPE, job_id, "Drivers import")
|
|
if not file_path:
|
|
return {"status": "failed", "error": "Archivo no encontrado (expirado o no subido). Sube de nuevo."}
|
|
common_storage.ensure_meta_from_redis(JOB_TYPE, job_id, file_path, "Drivers import")
|
|
|
|
error_path = common_storage.error_path_for_job(JOB_TYPE, job_id)
|
|
|
|
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)}
|
|
|
|
try:
|
|
tenant_id, company_id = common_meta.require_tenant_context(file_path)
|
|
except ValueError as e:
|
|
return {"status": "failed", "error": str(e)}
|
|
|
|
meta = common_meta.load_meta(file_path) or {}
|
|
actualizar = meta.get("actualizar", False)
|
|
existing_driver_keys: Set[Tuple[str, int]] = set()
|
|
if actualizar:
|
|
try:
|
|
from api.v1.modules.a76.transportation.drivers.models import Driver
|
|
with CoreSessionLocal() as session:
|
|
for row in (
|
|
session.query(Driver.transporter_key, Driver.line)
|
|
.filter(
|
|
Driver.tenant_id == tenant_id,
|
|
Driver.company_id == company_id,
|
|
)
|
|
.all()
|
|
):
|
|
if row[0] is not None and row[1] is not None:
|
|
existing_driver_keys.add(
|
|
((row[0] or "").strip().upper(), int(row[1]))
|
|
)
|
|
except Exception as e:
|
|
logger.warning("Drivers import: could not load existing_driver_keys for actualizar: %s", e)
|
|
|
|
valid_transporter_keys, valid_country_ame, transporter_key_actual = load_drivers_fk_sets(tenant_id, company_id)
|
|
|
|
try:
|
|
r = _get_redis()
|
|
r.set(
|
|
f"{DRV_IMPORT_TRANSPORTER_MAP_PREFIX}{job_id}",
|
|
json.dumps(transporter_key_actual).encode("utf-8"),
|
|
ex=DRV_IMPORT_REDIS_TTL,
|
|
)
|
|
except Exception as e:
|
|
logger.warning("Drivers import: failed to store transporter map in Redis: %s", e)
|
|
|
|
logger.info(
|
|
"Drivers import scan: job_id=%s tenant_id=%s company_id=%s actualizar=%s transportistas=%d",
|
|
job_id, tenant_id, company_id, actualizar, len(valid_transporter_keys),
|
|
)
|
|
|
|
error_count = 0
|
|
processed_rows = 0
|
|
errors_detail: List[Dict[str, Any]] = []
|
|
error_lines_list: List[int] = []
|
|
|
|
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.reader(f_in, dialect=dialect)
|
|
try:
|
|
headers = next(reader)
|
|
except StopIteration:
|
|
headers = []
|
|
headers = _dedupe_headers(headers)
|
|
dict_reader = csv.DictReader(f_in, fieldnames=headers, dialect=dialect)
|
|
|
|
for i, row in enumerate(dict_reader, start=1):
|
|
if progress_callback and i % 500 == 0:
|
|
progress_callback(i, total_rows, error_count)
|
|
|
|
row_norm = row_from_template(row, common_normalize.normalize_header)
|
|
warn = validate_row_driver_desfase(row_norm, i)
|
|
if warn and len(errors_detail) < 500:
|
|
errors_detail.append({
|
|
"line": warn.get("line", i),
|
|
"col": warn.get("col", ""),
|
|
"msg": warn.get("msg", ""),
|
|
"solution": warn.get("solution", ""),
|
|
"warning": bool(warn.get("warning", False)),
|
|
})
|
|
|
|
row_errors = validate_row_driver(
|
|
row_norm,
|
|
i,
|
|
actualizar=actualizar,
|
|
existing_driver_keys=existing_driver_keys,
|
|
valid_transporter_keys=valid_transporter_keys,
|
|
valid_country_ame=valid_country_ame,
|
|
)
|
|
if row_errors:
|
|
has_blocking_error = any(not e.get("warning", False) for e in row_errors)
|
|
if has_blocking_error:
|
|
error_count += 1
|
|
error_lines_list.append(i)
|
|
for e in row_errors:
|
|
if not e.get("warning", False):
|
|
f_err.write(json.dumps(e) + "\n")
|
|
if len(errors_detail) < 500:
|
|
for e in row_errors:
|
|
if len(errors_detail) >= 500:
|
|
break
|
|
errors_detail.append({
|
|
"line": e.get("line", i),
|
|
"col": e.get("col", ""),
|
|
"msg": e.get("msg", ""),
|
|
"solution": e.get("solution", ""),
|
|
"warning": bool(e.get("warning", False)),
|
|
})
|
|
processed_rows += 1
|
|
|
|
if error_lines_list:
|
|
common_storage.store_error_lines(JOB_TYPE, job_id, error_lines_list)
|
|
except Exception as e:
|
|
logger.error("Drivers import scan failed: %s", e)
|
|
return {"status": "failed", "error": str(e)}
|
|
|
|
return common_responses.scan_result(job_id, processed_rows, error_count, errors_detail)
|
|
|
|
|
|
def run_scan_sync(job_id: str) -> Dict[str, Any]:
|
|
result = _do_scan(job_id, progress_callback=None)
|
|
try:
|
|
r = _get_redis()
|
|
r.set(
|
|
f"{DRV_IMPORT_STATUS_PREFIX}{job_id}",
|
|
json.dumps(result).encode("utf-8"),
|
|
ex=DRV_IMPORT_REDIS_TTL,
|
|
)
|
|
except Exception as e:
|
|
logger.warning("Drivers import: failed to store scan status in Redis: %s", e)
|
|
return result
|
|
|
|
|
|
@celery_app.task(bind=True)
|
|
def scan_file(self, job_id: str, config: str = None):
|
|
logger.info("Drivers import: starting scan for job %s", job_id)
|
|
|
|
def on_progress(current: int, total: int, errors: int) -> None:
|
|
self.update_state(state="PROGRESS", meta={"current": current, "total": total, "errors": errors})
|
|
|
|
result = _do_scan(job_id, progress_callback=on_progress)
|
|
try:
|
|
r = _get_redis()
|
|
r.set(
|
|
f"{DRV_IMPORT_STATUS_PREFIX}{job_id}",
|
|
json.dumps(result).encode("utf-8"),
|
|
ex=DRV_IMPORT_REDIS_TTL,
|
|
)
|
|
except Exception as e:
|
|
logger.warning("Drivers import: failed to store scan status in Redis: %s", e)
|
|
return result
|
|
|
|
|
|
def _do_commit(job_id: str) -> Dict[str, Any]:
|
|
file_path = common_storage.ensure_file_from_redis(JOB_TYPE, job_id, "Drivers import")
|
|
if not file_path:
|
|
alt_path = common_storage.file_path_for_job(JOB_TYPE, job_id)
|
|
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:
|
|
common_storage.ensure_meta_from_redis(JOB_TYPE, job_id, file_path, "Drivers import")
|
|
|
|
error_path = common_storage.error_path_for_job(JOB_TYPE, job_id)
|
|
error_lines = common_storage.get_error_lines(JOB_TYPE, job_id, error_path)
|
|
|
|
try:
|
|
tenant_id, company_id = common_meta.require_tenant_context(file_path)
|
|
except ValueError as e:
|
|
return {"status": "failed", "error": str(e)}
|
|
|
|
meta = common_meta.load_meta(file_path) or {}
|
|
actualizar = meta.get("actualizar", False)
|
|
existing_driver_keys: Set[Tuple[str, int]] = set()
|
|
if actualizar:
|
|
try:
|
|
from api.v1.modules.a76.transportation.drivers.models import Driver
|
|
with CoreSessionLocal() as session:
|
|
for row in (
|
|
session.query(Driver.transporter_key, Driver.line)
|
|
.filter(
|
|
Driver.tenant_id == tenant_id,
|
|
Driver.company_id == company_id,
|
|
)
|
|
.all()
|
|
):
|
|
if row[0] is not None and row[1] is not None:
|
|
existing_driver_keys.add(
|
|
((row[0] or "").strip().upper(), int(row[1]))
|
|
)
|
|
except Exception as e:
|
|
logger.warning("Drivers import: could not load existing_driver_keys for actualizar: %s", e)
|
|
|
|
valid_transporter_keys, valid_country_ame, transporter_key_actual = load_drivers_fk_sets(tenant_id, company_id)
|
|
|
|
transporter_map_from_redis: Optional[Dict[str, str]] = None
|
|
try:
|
|
r = _get_redis()
|
|
map_key = f"{DRV_IMPORT_TRANSPORTER_MAP_PREFIX}{job_id}"
|
|
raw = r.get(map_key)
|
|
if raw:
|
|
transporter_map_from_redis = json.loads(raw.decode("utf-8"))
|
|
logger.info(
|
|
"Drivers import commit: using transporter map from Redis (job_id=%s, keys=%d)",
|
|
job_id, len(transporter_map_from_redis),
|
|
)
|
|
else:
|
|
logger.warning(
|
|
"Drivers import commit: no transporter map in Redis for job_id=%s, using DB fallback",
|
|
job_id,
|
|
)
|
|
except Exception as e:
|
|
logger.warning(
|
|
"Drivers import: could not load transporter map from Redis (job_id=%s): %s",
|
|
job_id, e,
|
|
)
|
|
|
|
logger.info(
|
|
"Drivers import commit: job_id=%s tenant_id=%s company_id=%s transportistas=%d",
|
|
job_id, tenant_id, company_id, len(valid_transporter_keys),
|
|
)
|
|
|
|
from api.v1.modules.a76.transportation.drivers.services import DriverService
|
|
from api.v1.modules.a76.transportation.drivers.dto import DriverCreateDTO
|
|
|
|
inserted_count = 0
|
|
updated_count = 0
|
|
skipped_invalid = 0
|
|
skipped_duplicate = 0
|
|
skipped_details: List[Dict[str, Any]] = []
|
|
seen_keys_in_file: Dict[str, int] = {}
|
|
meta_path = common_meta.get_meta_path(file_path)
|
|
|
|
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.reader(f, dialect=dialect)
|
|
try:
|
|
headers = next(reader)
|
|
except StopIteration:
|
|
headers = []
|
|
headers = _dedupe_headers(headers)
|
|
dict_reader = csv.DictReader(f, fieldnames=headers, dialect=dialect)
|
|
|
|
for i, row in enumerate(dict_reader, start=1):
|
|
if i in error_lines:
|
|
continue
|
|
|
|
row_norm = row_from_template(row, common_normalize.normalize_header)
|
|
row_errors = validate_row_driver(
|
|
row_norm,
|
|
i,
|
|
actualizar=actualizar,
|
|
existing_driver_keys=existing_driver_keys,
|
|
valid_transporter_keys=valid_transporter_keys,
|
|
valid_country_ame=valid_country_ame,
|
|
)
|
|
if row_errors and any(not e.get("warning", False) for e in row_errors):
|
|
skipped_invalid += 1
|
|
driver_key = (row_norm.get("CLAVE CONDUCTOR") or "").strip()[:80] or "-"
|
|
blocking_errors = [e for e in row_errors if not e.get("warning", False)]
|
|
for blocking in blocking_errors:
|
|
skipped_details.append({
|
|
"line": i,
|
|
"driver_key": driver_key,
|
|
"invoice": driver_key,
|
|
"reason": f"{blocking.get('col', '')}: {blocking.get('msg', '')}",
|
|
"solution": blocking.get("solution", ""),
|
|
})
|
|
continue
|
|
|
|
data = row_to_driver_data(row_norm, tenant_id, company_id)
|
|
if not data or not data.get("transporter_key") or data.get("line") is None:
|
|
skipped_invalid += 1
|
|
continue
|
|
|
|
tk = (data["transporter_key"] or "").strip()
|
|
# Usar mapa del scan (Redis) si existe; si no, resolver en la sesión del commit (fallback)
|
|
if transporter_map_from_redis is not None:
|
|
tk_upper = tk.upper()
|
|
if tk_upper not in transporter_map_from_redis:
|
|
skipped_invalid += 1
|
|
skipped_details.append({
|
|
"line": i,
|
|
"driver_key": f"{tk}:{data.get('line')}",
|
|
"invoice": f"{tk}:{data.get('line')}",
|
|
"reason": f"El transportista {tk} no existe en el catálogo.",
|
|
"solution": "Capturar en el CSV un TRANSPORTISTA existente (columna TRANSPORTISTA CLAVE / CLAVE TRANSPORTISTA).",
|
|
})
|
|
continue
|
|
data["transporter_key"] = transporter_map_from_redis[tk_upper]
|
|
else:
|
|
from api.v1.modules.a76.transportation.transporters.models import Transporter
|
|
transporter_row = (
|
|
session.query(Transporter.transporter_key)
|
|
.filter(
|
|
Transporter.tenant_id == tenant_id,
|
|
Transporter.company_id == company_id,
|
|
func.upper(Transporter.transporter_key) == tk.upper(),
|
|
)
|
|
.first()
|
|
)
|
|
if not transporter_row or not transporter_row[0]:
|
|
skipped_invalid += 1
|
|
skipped_details.append({
|
|
"line": i,
|
|
"driver_key": f"{tk}:{data.get('line')}",
|
|
"invoice": f"{tk}:{data.get('line')}",
|
|
"reason": f"El transportista {tk} no existe en el catálogo.",
|
|
"solution": "Capturar en el CSV un TRANSPORTISTA existente (columna TRANSPORTISTA CLAVE / CLAVE TRANSPORTISTA).",
|
|
})
|
|
continue
|
|
data["transporter_key"] = (transporter_row[0] or "").strip()
|
|
|
|
key = f"{data['transporter_key']}:{data['line']}"
|
|
if key in seen_keys_in_file:
|
|
skipped_duplicate += 1
|
|
skipped_details.append({
|
|
"line": i,
|
|
"driver_key": key,
|
|
"invoice": key,
|
|
"reason": "Clave duplicada en el archivo (se usa la primera)",
|
|
"solution": "Revisa TRANSPORTISTA y LINEA para no duplicar la combinación en el archivo (se conserva la primera).",
|
|
})
|
|
continue
|
|
seen_keys_in_file[key] = i
|
|
|
|
existing = DriverService.get_driver_by_key_and_line(
|
|
session, data["transporter_key"], data["line"], str(company_id), tenant_id
|
|
)
|
|
try:
|
|
if existing:
|
|
update_fields = {
|
|
k: v for k, v in data.items()
|
|
if k not in ("transporter_key", "line", "company_id", "tenant_id")
|
|
}
|
|
for field, value in update_fields.items():
|
|
setattr(existing, field, value)
|
|
session.add(existing)
|
|
updated_count += 1
|
|
else:
|
|
create_data = DriverCreateDTO(**data)
|
|
DriverService.create_driver(session, create_data)
|
|
inserted_count += 1
|
|
except Exception as db_err:
|
|
session.rollback()
|
|
skipped_invalid += 1
|
|
err_msg = str(db_err)
|
|
if "ForeignKeyViolation" in err_msg or "foreign key constraint" in err_msg.lower() or "driver_transporter_key_fkey" in err_msg:
|
|
err_msg = f"El transportista {data.get('transporter_key', '')} no existe en el catálogo."
|
|
logger.warning(
|
|
"Drivers import: FK violation linea %d transporter_key=%r (valid_transporter_keys tiene %d claves)",
|
|
i, data.get("transporter_key"), len(valid_transporter_keys),
|
|
)
|
|
skipped_details.append({
|
|
"line": i,
|
|
"driver_key": key,
|
|
"invoice": key,
|
|
"reason": err_msg,
|
|
"solution": "Revisar la línea del CSV y los catálogos relacionados para corregir el error e intentar de nuevo.",
|
|
})
|
|
continue
|
|
|
|
try:
|
|
session.commit()
|
|
except Exception as db_err:
|
|
session.rollback()
|
|
logger.error("Drivers import DB error: %s", db_err)
|
|
return {"status": "failed", "error": str(db_err)}
|
|
|
|
except Exception as e:
|
|
logger.exception("Drivers import task failed")
|
|
return {"status": "failed", "error": str(e)}
|
|
|
|
common_storage.cleanup_import_job(
|
|
JOB_TYPE, job_id,
|
|
file_path=file_path,
|
|
error_path=error_path,
|
|
meta_path=meta_path,
|
|
)
|
|
try:
|
|
r = _get_redis()
|
|
r.delete(f"{DRV_IMPORT_STATUS_PREFIX}{job_id}")
|
|
r.delete(f"{DRV_IMPORT_TRANSPORTER_MAP_PREFIX}{job_id}")
|
|
except Exception as e:
|
|
logger.warning("Drivers import: failed to delete status key: %s", e)
|
|
|
|
total_ok = inserted_count + updated_count
|
|
if total_ok == 0 and (skipped_invalid + skipped_duplicate) > 0:
|
|
return {
|
|
"status": "warning",
|
|
"inserted": inserted_count,
|
|
"updated": updated_count,
|
|
"skipped_invalid": skipped_invalid,
|
|
"skipped_duplicate": skipped_duplicate,
|
|
"skipped_missing_fk": 0,
|
|
"skipped_details": skipped_details,
|
|
"message": f"No se insertaron registros. {skipped_invalid + skipped_duplicate} rechazados.",
|
|
}
|
|
if total_ok == 0:
|
|
return {
|
|
"status": "failed",
|
|
"error": "No hay registros validos en el archivo CSV",
|
|
"inserted": 0,
|
|
"updated": 0,
|
|
"skipped_invalid": skipped_invalid,
|
|
"skipped_duplicate": skipped_duplicate,
|
|
"skipped_missing_fk": 0,
|
|
"skipped_details": skipped_details,
|
|
}
|
|
return {
|
|
"status": "finished",
|
|
"inserted": inserted_count,
|
|
"updated": updated_count,
|
|
"skipped_invalid": skipped_invalid,
|
|
"skipped_duplicate": skipped_duplicate,
|
|
"skipped_missing_fk": 0,
|
|
"skipped_details": skipped_details,
|
|
}
|
|
|
|
|
|
def run_commit_sync(job_id: str) -> Dict[str, Any]:
|
|
result = _do_commit(job_id)
|
|
try:
|
|
r = _get_redis()
|
|
r.set(
|
|
f"{DRV_IMPORT_STATUS_PREFIX}{job_id}",
|
|
json.dumps(result).encode("utf-8"),
|
|
ex=DRV_IMPORT_REDIS_TTL,
|
|
)
|
|
except Exception as e:
|
|
logger.warning("Drivers import: failed to store commit status in Redis: %s", e)
|
|
return result
|
|
|
|
|
|
@celery_app.task(bind=True)
|
|
def insert_valid_rows(self, job_id: str):
|
|
logger.info("Drivers import: starting commit for job %s", job_id)
|
|
result = _do_commit(job_id)
|
|
try:
|
|
r = _get_redis()
|
|
r.set(
|
|
f"{DRV_IMPORT_STATUS_PREFIX}{job_id}",
|
|
json.dumps(result).encode("utf-8"),
|
|
ex=DRV_IMPORT_REDIS_TTL,
|
|
)
|
|
except Exception as e:
|
|
logger.warning("Drivers import: failed to store commit status in Redis: %s", e)
|
|
return result
|