diff --git a/backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/__init__.py b/backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/__init__.py new file mode 100644 index 00000000..df6c436e --- /dev/null +++ b/backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/__init__.py @@ -0,0 +1 @@ +# layouts_csv.cambio_regimen_regularizacion — carga CSV Cambio de régimen y Regularización (encabezado y partidas) diff --git a/backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/routes.py b/backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/routes.py new file mode 100644 index 00000000..0bba9dce --- /dev/null +++ b/backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/routes.py @@ -0,0 +1,157 @@ +""" +Rutas de importación CSV para Cambio de régimen y Regularización (encabezado y partidas). +Flujo: upload → scan → status (polling) → commit. Sin validaciones ni inserción aún. +""" +import base64 +import json +import logging +import os +from uuid import uuid4 + +from fastapi import APIRouter, File, HTTPException, UploadFile, Depends, Form, Query +from sqlalchemy.orm import Session +from typing import Literal, Optional, Dict, Any + +from core.celery_app import celery_app +from core.database import get_core_db +from core.paths import layout_path +from core.security import get_current_user, validate_access_to_resource + +from .schemas import ImportJobResponse, CommitRequest +from .tasks import scan_file, insert_valid_rows, JOB_TYPE, CRREG_IMPORT_REDIS_TTL +from ..common import storage as common_storage + +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) + + +def _default_template_id(model_target: str, document_type: Optional[str]) -> str: + if document_type == "regulariz": + return "regulariz_header" if model_target == "invoice_header" else "regulariz_details" + return "cam_reg_header" if model_target == "invoice_header" else "cam_reg_details" + + +@router.post("/upload/{model_target}", response_model=ImportJobResponse) +async def upload_import_file( + model_target: Literal["invoice_header", "invoice_details"], + file: UploadFile = File(...), + footer_config: Optional[str] = Form(None), + template_id: Optional[str] = Form(None), + document_type: Optional[str] = Query(None, description="cam_reg | regulariz"), + company_id: int = Query(..., description="Company ID"), + db: Session = Depends(get_core_db), + current_user: Dict[str, Any] = Depends(get_current_user), +): + """Subir CSV, guardar en Redis, encolar scan. template_id/document_type distinguen Cambio de régimen vs Regularización.""" + try: + tenant_id = validate_access_to_resource(db, company_id, current_user) + except Exception as e: + logger.error("Cambio régimen/Regularización import: access validation failed: %s", 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() + + file_key, meta_key, _ = common_storage.storage_keys(JOB_TYPE, job_id) + meta_data = { + "tenant_id": tenant_id, + "company_id": company_id, + "user_id": current_user.get("id"), + "footer_config": footer_config, + "document_type": document_type or "cam_reg", + "template_id": template_id or _default_template_id(model_target, document_type), + } + + try: + r = _get_redis() + r.set(file_key, base64.b64encode(contents), ex=CRREG_IMPORT_REDIS_TTL) + r.set(meta_key, json.dumps(meta_data).encode("utf-8"), ex=CRREG_IMPORT_REDIS_TTL) + except Exception as e: + logger.error("Cambio régimen/Regularización import: Redis store error: %s", e) + raise HTTPException(status_code=500, detail="No se pudo encolar el archivo.") + + try: + upload_dir = layout_path("imports", "temp") + os.makedirs(upload_dir, exist_ok=True) + csv_path = common_storage.file_path_for_job(JOB_TYPE, job_id) + with open(csv_path, "wb") as f: + f.write(contents) + meta_path = csv_path.replace(".csv", ".meta.json") + with open(meta_path, "w") as f: + json.dump(meta_data, f) + except Exception as e: + logger.warning("Cambio régimen/Regularización import: local file save failed: %s", e) + + scan_file.apply_async(args=[job_id, model_target, footer_config], 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} + + if isinstance(getattr(task_result, "result", None), dict) and task_result.result.get("status") in ("finished", "warning"): + return task_result.result + + logger.warning("Cambio régimen/Regularización 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: + result = getattr(task_result, "result", None) + if result is not None and not isinstance(result, dict): + err_msg = str(result) + elif isinstance(result, dict) and (result.get("error") or result.get("message")): + err_msg = result.get("error") or result.get("message") + return {"status": "failed", "error": err_msg or "Task failed"} + + +@router.post("/{job_id}/commit") +async def commit_import_job(job_id: str, body: CommitRequest): + """Usuario confirma; se encola la tarea de commit (por ahora sin inserción real).""" + task = insert_valid_rows.delay(job_id, body.model_target) + return { + "status": "committing", + "message": "Proceso de commit iniciado.", + "commit_job_id": task.id, + } diff --git a/backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/schemas.py b/backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/schemas.py new file mode 100644 index 00000000..2a043f30 --- /dev/null +++ b/backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/schemas.py @@ -0,0 +1,23 @@ +from pydantic import BaseModel +from typing import Optional, Literal + + +class ImportJobResponse(BaseModel): + job_id: str + status: str + message: str + + +class CommitRequest(BaseModel): + model_target: Literal["invoice_header", "invoice_details"] + + +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 + error_file: Optional[str] = None diff --git a/backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/tasks.py b/backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/tasks.py new file mode 100644 index 00000000..c06ce2b9 --- /dev/null +++ b/backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/tasks.py @@ -0,0 +1,146 @@ +""" +Tareas Celery para importación CSV de Cambio de régimen y Regularización (encabezado y partidas). +Flujo: scan_file (sin validaciones) → insert_valid_rows (sin inserción en BD). +Usa layouts_csv.common (storage, normalize, meta, responses, csv_reader). +Validaciones independientes por document_type (cam_reg vs regulariz) se añadirán después. +""" +import logging +import os +from typing import Dict, Any, Optional + +from core.celery_app import celery_app + +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 ..common import csv_reader as common_csv_reader +from .template_config import row_from_template + +logger = logging.getLogger(__name__) + +JOB_TYPE = "crreg" +CRREG_IMPORT_REDIS_TTL = common_storage.IMPORT_REDIS_TTL + + +def _ensure_file(job_id: str) -> Optional[str]: + return common_storage.ensure_file_from_redis(JOB_TYPE, job_id, "Cambio régimen/Regularización import") + + +def _ensure_meta(job_id: str, file_path: str) -> bool: + return common_storage.ensure_meta_from_redis(JOB_TYPE, job_id, file_path, "Cambio régimen/Regularización import") + + +def _norm_row(row: Dict[str, Any], template_id: str) -> Dict[str, Any]: + return row_from_template(row, template_id, common_normalize.normalize_header) + + +@celery_app.task(bind=True, name="api.v1.modules.a76.layouts_csv.cambio_regimen_regularizacion.tasks.scan_file") +def scan_file(self, job_id: str, model_target: str, config: str = None): + """ + Scan CSV sin validaciones: leer, normalizar con plantilla, devolver total_rows y 0 errores. + """ + logger.info("Cambio régimen/Regularización import: starting scan for job %s target %s", job_id, model_target) + + file_path = _ensure_file(job_id) + if not file_path: + return {"status": "failed", "error": "Archivo no encontrado (expirado o no subido). Sube de nuevo."} + if os.path.getsize(file_path) == 0: + return {"status": "failed", "error": "El archivo está vacío."} + _ensure_meta(job_id, file_path) + + try: + 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 {} + document_type = meta.get("document_type") or "cam_reg" + template_id = meta.get("template_id") or ( + "cam_reg_header" if model_target == "invoice_header" else "cam_reg_details" + ) + if document_type == "regulariz" and not meta.get("template_id"): + template_id = "regulariz_header" if model_target == "invoice_header" else "regulariz_details" + + total_rows = 0 + processed_rows = 0 + + try: + total_rows = common_csv_reader.count_csv_rows(file_path, has_header=True) + except Exception as e: + return {"status": "failed", "error": str(e)} + + def on_progress(current: int, total: int) -> None: + self.update_state(state="PROGRESS", meta={"current": current, "total": total}) + + try: + for i, row in common_csv_reader.iter_csv_rows(file_path, fieldnames=None): + if i % 500 == 0: + on_progress(i, total_rows) + _norm_row(row, template_id) + processed_rows += 1 + except Exception as e: + logger.error("Cambio régimen/Regularización import scan failed: %s", e) + return {"status": "failed", "error": str(e)} + + return common_responses.scan_result(job_id, processed_rows, 0, []) + + +@celery_app.task(bind=True, name="api.v1.modules.a76.layouts_csv.cambio_regimen_regularizacion.tasks.insert_valid_rows") +def insert_valid_rows(self, job_id: str, model_target: str): + """ + Commit sin inserción en BD: leer CSV, omitir líneas de error (vacío por ahora), cleanup, devolver finished con inserted=0. + """ + logger.info("Cambio régimen/Regularización import: starting commit for job %s target %s", job_id, model_target) + + file_path = _ensure_file(job_id) + 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: + _ensure_meta(job_id, file_path) + + try: + 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 {} + meta_path = common_meta.get_meta_path(file_path) + 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) + + document_type = meta.get("document_type") or "cam_reg" + template_id = meta.get("template_id") or ( + "cam_reg_header" if model_target == "invoice_header" else "cam_reg_details" + ) + if document_type == "regulariz" and not meta.get("template_id"): + template_id = "regulariz_header" if model_target == "invoice_header" else "regulariz_details" + + try: + for i, row in common_csv_reader.iter_csv_rows(file_path, fieldnames=None): + if i in error_lines: + continue + _norm_row(row, template_id) + except Exception as e: + logger.error("Cambio régimen/Regularización import commit read failed: %s", e) + 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, + ) + + return { + "status": "finished", + "inserted": 0, + "skipped_invalid": 0, + "skipped_missing_fk": 0, + "skipped_duplicate": 0, + "skipped_details": [], + "message": "Proceso base listo; validaciones e inserción pendientes.", + } diff --git a/backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/template_config.py b/backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/template_config.py new file mode 100644 index 00000000..642e413e --- /dev/null +++ b/backend/api/v1/modules/a76/layouts_csv/cambio_regimen_regularizacion/template_config.py @@ -0,0 +1,138 @@ +""" +Plantillas CSV para Cambio de régimen y Regularización (encabezado y partidas). +Por ahora misma estructura que encabezado/partidas de exportación; luego se ajustan columnas si difieren. +""" + +from typing import Dict, List, Any, Optional + +# Cambio de régimen: cam_reg_header, cam_reg_details +# Regularización: regulariz_header, regulariz_details +TEMPLATE_COLUMNS: Dict[str, List[Dict[str, Any]]] = { + "cam_reg_header": [ + {"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA", "ID"]}, + {"canonical": "FECHA FACTURA", "aliases": ["FECHA"]}, + {"canonical": "FECHA EMISION"}, + {"canonical": "CLAVE PROVEEDOR"}, + {"canonical": "CLAVE VENDIDO A"}, + {"canonical": "CLAVE ENVIADO A"}, + {"canonical": "REGIMEN", "aliases": ["CLAVEDOCUMENTO"]}, + {"canonical": "ADUANA DE CRUCE"}, + {"canonical": "CLAVE MONEDA"}, + {"canonical": "CLAVE INCOTERM"}, + {"canonical": "TIPO MONEDA"}, + {"canonical": "TIPO DE CAMBIO"}, + {"canonical": "TIPO PESO"}, + {"canonical": "TIPO TRANSPORTE"}, + {"canonical": "REMESA"}, + {"canonical": "AGENTE ADUANAL"}, + {"canonical": "FLETES"}, + {"canonical": "VALOR SEGUROS"}, + {"canonical": "SEGUROS"}, + {"canonical": "EMBALAJES"}, + {"canonical": "OTROS INCREMENTABLES"}, + {"canonical": "NUM PROYECTO", "aliases": ["NUMPROYECTO"]}, + {"canonical": "ORDEN COMPRA", "aliases": ["ORDENCOMPRA"]}, + {"canonical": "FACTURA ALTERNA"}, + {"canonical": "FACTURA EXPO REF", "aliases": ["FACTURAEXPOREF"]}, + {"canonical": "OBSERVACIONES E"}, + {"canonical": "OBSERVACIONES I"}, + {"canonical": "E DOCUMENT"}, + {"canonical": "NUM OPERACION"}, + {"canonical": "CLAVE TRANSPORTISTA"}, + {"canonical": "NOMBRE CONDUCTOR"}, + {"canonical": "NUMERO TRANSPORTE"}, + {"canonical": "PRECINTO"}, + ], + "cam_reg_details": [ + {"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA"]}, + {"canonical": "LINEA", "aliases": ["RENGLON", "PARTIDA"]}, + {"canonical": "NUMPARTE", "aliases": ["NUMERO PARTE"]}, + {"canonical": "PRECIO UNITARIO", "aliases": ["PRECIOUNITARIO"]}, + {"canonical": "VALOR COMERCIAL", "aliases": ["VALORCOMERCIAL"]}, + {"canonical": "CANTIDAD"}, + {"canonical": "CANTIDAD BULTOS", "aliases": ["CANTIDADBULTOS"]}, + {"canonical": "DESCRIPCION"}, + {"canonical": "PAIS ORIGEN", "aliases": ["PAISORIGEN"]}, + {"canonical": "FRACCION"}, + {"canonical": "ORDEN DE COMPRA", "aliases": ["ORDENCOMPRA"]}, + ], + "regulariz_header": [ + {"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA", "ID"]}, + {"canonical": "FECHA FACTURA", "aliases": ["FECHA"]}, + {"canonical": "FECHA EMISION"}, + {"canonical": "CLAVE PROVEEDOR"}, + {"canonical": "CLAVE VENDIDO A"}, + {"canonical": "CLAVE ENVIADO A"}, + {"canonical": "REGIMEN", "aliases": ["CLAVEDOCUMENTO"]}, + {"canonical": "ADUANA DE CRUCE"}, + {"canonical": "CLAVE MONEDA"}, + {"canonical": "CLAVE INCOTERM"}, + {"canonical": "TIPO MONEDA"}, + {"canonical": "TIPO DE CAMBIO"}, + {"canonical": "TIPO PESO"}, + {"canonical": "TIPO TRANSPORTE"}, + {"canonical": "REMESA"}, + {"canonical": "AGENTE ADUANAL"}, + {"canonical": "FLETES"}, + {"canonical": "VALOR SEGUROS"}, + {"canonical": "SEGUROS"}, + {"canonical": "EMBALAJES"}, + {"canonical": "OTROS INCREMENTABLES"}, + {"canonical": "NUM PROYECTO", "aliases": ["NUMPROYECTO"]}, + {"canonical": "ORDEN COMPRA", "aliases": ["ORDENCOMPRA"]}, + {"canonical": "FACTURA ALTERNA"}, + {"canonical": "FACTURA EXPO REF", "aliases": ["FACTURAEXPOREF"]}, + {"canonical": "OBSERVACIONES E"}, + {"canonical": "OBSERVACIONES I"}, + {"canonical": "E DOCUMENT"}, + {"canonical": "NUM OPERACION"}, + {"canonical": "CLAVE TRANSPORTISTA"}, + {"canonical": "NOMBRE CONDUCTOR"}, + {"canonical": "NUMERO TRANSPORTE"}, + {"canonical": "PRECINTO"}, + ], + "regulariz_details": [ + {"canonical": "NUMERO FACTURA", "aliases": ["NUM FACTURA", "FACTURA"]}, + {"canonical": "LINEA", "aliases": ["RENGLON", "PARTIDA"]}, + {"canonical": "NUMPARTE", "aliases": ["NUMERO PARTE"]}, + {"canonical": "PRECIO UNITARIO", "aliases": ["PRECIOUNITARIO"]}, + {"canonical": "VALOR COMERCIAL", "aliases": ["VALORCOMERCIAL"]}, + {"canonical": "CANTIDAD"}, + {"canonical": "CANTIDAD BULTOS", "aliases": ["CANTIDADBULTOS"]}, + {"canonical": "DESCRIPCION"}, + {"canonical": "PAIS ORIGEN", "aliases": ["PAISORIGEN"]}, + {"canonical": "FRACCION"}, + {"canonical": "ORDEN DE COMPRA", "aliases": ["ORDENCOMPRA"]}, + ], +} + + +def _resolve_template_columns(template_id: str) -> Optional[List[Dict[str, Any]]]: + return TEMPLATE_COLUMNS.get(template_id) + + +def build_normalized_lookup(template_id: str, normalize_header_fn) -> Dict[str, str]: + """normalized_header -> canonical_name.""" + cols = _resolve_template_columns(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], template_id: str, normalize_header_fn) -> Dict[str, Any]: + """Fila CSV -> dict con nombres canónicos de la plantilla.""" + lookup = build_normalized_lookup(template_id, 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 diff --git a/backend/api/v1/modules/a76/router.py b/backend/api/v1/modules/a76/router.py index a8d36774..9d8781d8 100644 --- a/backend/api/v1/modules/a76/router.py +++ b/backend/api/v1/modules/a76/router.py @@ -16,6 +16,7 @@ from .classes import router as classes_router from .clients_and_providers import router as client_and_provider_router from .layouts_csv.facturas.routes import router as imports_router from .layouts_csv.exportacion.routes import router as exportacion_imports_router +from .layouts_csv.cambio_regimen_regularizacion.routes import router as cambio_regimen_regularizacion_imports_router from .csv_templates.routes import router as csv_templates_router from .invoice_settings.routes import router as invoice_settings_router from .item_presets.routes import router as item_presets_router @@ -58,6 +59,7 @@ router.include_router(invoices_router, prefix="/a76", tags=["a76 / invoices"]) router.include_router(items_router, prefix="/a76", tags=["a76 / items"]) router.include_router(imports_router, prefix="/a76/imports", tags=["a76 / imports"]) router.include_router(exportacion_imports_router, prefix="/a76/imports/exportacion", tags=["a76 / imports / exportacion"]) +router.include_router(cambio_regimen_regularizacion_imports_router, prefix="/a76/imports/cambio-regimen-regularizacion", tags=["a76 / imports / cambio_regimen_regularizacion"]) router.include_router(csv_templates_router, prefix="/a76/csv-templates", tags=["a76 / csv_templates"]) router.include_router(invoice_settings_router) router.include_router(item_presets_router, prefix="/a76/item-presets", tags=["a76 / item_presets"]) diff --git a/backend/core/celery_app.py b/backend/core/celery_app.py index 3ca84277..bf15ae89 100644 --- a/backend/core/celery_app.py +++ b/backend/core/celery_app.py @@ -34,6 +34,7 @@ celery_app.conf.update( "api.v1.modules.a76.reports.exportacion.descargo.task", "api.v1.modules.a76.layouts_csv.facturas.tasks", "api.v1.modules.a76.layouts_csv.exportacion.tasks", + "api.v1.modules.a76.layouts_csv.cambio_regimen_regularizacion.tasks", "api.v1.modules.a76.layouts_csv.customs_brokers.tasks", "api.v1.modules.a76.layouts_csv.clients_and_providers.tasks", "api.v1.modules.a76.layouts_csv.pedmientos.tasks",