import os import base64 from datetime import datetime from decimal import Decimal import csv import json import logging import re import unicodedata from typing import Dict, Any, Optional, List from core.celery_app import celery_app from core.database import CoreSessionLocal from .template_config import row_from_template # Models are imported inside tasks to avoid circular dependencies and mapper initialization issues in the API process # We'll need schemas for validation # from api.v1.modules.a76.invoices.schemas import InvoiceHeaderCreate # But for Phase 1 we use a lighter check logger = logging.getLogger(__name__) # Redis keys and TTL for import file/meta (shared between API and worker when no shared filesystem) IMPORT_FILE_KEY_PREFIX = "import_file:" IMPORT_META_KEY_PREFIX = "import_meta:" IMPORT_ERROR_LINES_KEY_PREFIX = "import_error_lines:" IMPORT_REDIS_TTL = 3600 # 1 hour def _get_redis(): """Redis client using same URL as Celery broker (worker and API can share data).""" 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: """Directory on the worker for temp CSV and meta (same structure as API, but local to worker).""" return os.path.join(os.getcwd(), "uploads", "temp") def _ensure_worker_has_file_from_redis(job_id: str) -> Optional[str]: """ Load file content from Redis and write to worker's upload dir. Returns local file_path if successful, None otherwise. """ redis_client = _get_redis() key = f"{IMPORT_FILE_KEY_PREFIX}{job_id}" data = redis_client.get(key) if not data: return None try: raw = base64.b64decode(data) except Exception as e: logger.warning(f"Failed to decode import 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"{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: """Load meta from Redis and write to worker's meta file. Returns True if meta was found and written.""" redis_client = _get_redis() key = f"{IMPORT_META_KEY_PREFIX}{job_id}" data = redis_client.get(key) if not data: return False try: meta = json.loads(data.decode("utf-8")) except Exception as e: logger.warning(f"Failed to decode import 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: """Remove file, meta and error lines from Redis after commit (cleanup).""" try: r = _get_redis() r.delete( f"{IMPORT_FILE_KEY_PREFIX}{job_id}", f"{IMPORT_META_KEY_PREFIX}{job_id}", f"{IMPORT_ERROR_LINES_KEY_PREFIX}{job_id}", ) except Exception as e: logger.warning(f"Failed to delete import keys from Redis: {e}") class ForeignKeyValidator: def __init__(self, session, tenant_id, company_id): self.session = session self.tenant_id = tenant_id self.company_id = company_id self.cache = {} # {(model_name, value): bool} def check_exists(self, model, value, field_name="id", is_public=False): if value is None: return True # Assume optional if None, or let DB handle not-null key = (model.__name__, value) if key in self.cache: return self.cache[key] query = self.session.query(getattr(model, field_name)).filter(getattr(model, field_name) == value) if not is_public: query = query.filter(model.tenant_id == self.tenant_id, model.company_id == self.company_id) exists = query.first() is not None self.cache[key] = exists return exists TRANSPORT_TYPE_VALUES = { "none", "transport", "box", "licence plates", "truck", "vessel", "rail_barge", "container", "airplane", "gondola", "flatbed", } def normalize_public_code(value: Optional[str]) -> Optional[str]: if value is None: return None text = str(value).strip().upper() return text or None def validate_public_code( validator: ForeignKeyValidator, model, value: Optional[str], line_num: int, col_name: str, field_name: str = "code", required: bool = False, ) -> Optional[Dict[str, Any]]: code = normalize_public_code(value) if not code: if required: return {"line": line_num, "col": col_name, "msg": "Requerido"} return None if not validator.check_exists(model, code, field_name=field_name, is_public=True): return {"line": line_num, "col": col_name, "msg": "No existe en el catalogo"} return None def validate_tenant_fk_id( validator: ForeignKeyValidator, model, value: Optional[int], line_num: int, col_name: str, required: bool = False, ) -> Optional[Dict[str, Any]]: if value is None: if required: return {"line": line_num, "col": col_name, "msg": "Requerido"} return None if not validator.check_exists(model, value): return {"line": line_num, "col": col_name, "msg": "No existe en el catalogo"} return None @celery_app.task(bind=True) def scan_file(self, job_id: str, model_target: str, config: str = None): """ Pass 1: Read CSV, Validate types, Write Errors to JSONL. File content is loaded from Redis (written by API on upload) so worker does not need shared filesystem. """ logger.info(f"Starting scan for job {job_id} target {model_target}") # 1. Get file from Redis and write to worker local disk file_path = _ensure_worker_has_file_from_redis(job_id) if not file_path: return {"status": "failed", "error": "File not found (missing or expired in queue). Please upload again."} _ensure_worker_has_meta_from_redis(job_id, file_path) # 2. Setup Error Log error_path = file_path.replace("temp", "errors").replace(".csv", ".jsonl") os.makedirs(os.path.dirname(error_path), exist_ok=True) total_rows = 0 error_count = 0 processed_rows = 0 # 3. Count Total (Quick Pass) or just estimate try: with open(file_path, 'r', encoding='utf-8-sig') as f: total_rows = sum(1 for _ in f) - 1 # Minus header except Exception as e: return {"status": "failed", "error": f"Cannot read file: {e}"} footer_config = parse_footer_config(config) date_format = footer_config.get("dateFormat") # Validate and set default date_format if not provided if not date_format: date_format = "yyyy-mm-dd" # Default to ISO format logger.info(f"No date_format specified in config, using default: {date_format}") meta_path = file_path.replace(".csv", ".meta.json") meta = {} tenant_id = None company_id = None if os.path.exists(meta_path): try: with open(meta_path, "r", encoding="utf-8") as f_meta: meta = json.load(f_meta) or {} tenant_id = meta.get("tenant_id") company_id = meta.get("company_id") except Exception as e: logger.warning(f"Failed to read meta for job {job_id}: {e}") if not tenant_id or not company_id: return {"status": "failed", "error": "Missing context (tenant/company)"} template_id = meta.get("template_id") or ( "imp_temp_header" if model_target == "invoice_header" else "imp_temp_details" ) inv_type_value = normalize_public_code(footer_config.get("invoice_type") or meta.get("invoice_type") or "TEM") if not inv_type_value: inv_type_value = "TEM" try: from api.v1.modules.a76.invoices.models import InvoiceHeader from api.v1.modules.a76.clients_and_providers.models import ClientProvider from api.v1.modules.a76.customs_brokers.models import CustomsBroker from api.v1.modules.public.reference_data.currency_types.models import CurrencyType from api.v1.modules.public.reference_data.pedimento_codes.models import PedimentoCode from api.v1.modules.public.reference_data.code_pedimento_regimens.models import ( CodePedimentoRegimen, ) from api.v1.modules.public.reference_data.pedimento_regimens.models import RegimenPedimento from api.v1.modules.public.reference_data.invoice_types.models import InvoiceType from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection from api.v1.modules.public.reference_data.incoterms.models import Incoterm from api.v1.modules.a76.parts.models import Part models = { "InvoiceHeader": InvoiceHeader, "InvoiceType": InvoiceType, "ClientProvider": ClientProvider, "CustomsBroker": CustomsBroker, "RegimenPedimento": RegimenPedimento, "CodePedimentoRegimen": CodePedimentoRegimen, "PedimentoCode": PedimentoCode, "CurrencyType": CurrencyType, "CustomsSection": CustomsSection, "Incoterm": Incoterm, "Part": Part, } with CoreSessionLocal() as session, \ open(file_path, 'r', encoding='utf-8-sig') as f_in, \ open(error_path, 'w', encoding='utf-8') as f_err: validator = ForeignKeyValidator(session, tenant_id, company_id) invoice_id_cache: Dict[str, Optional[int]] = {} # Detect Delimiter sample = f_in.read(2048) f_in.seek(0) try: dialect = csv.Sniffer().sniff(sample, delimiters=",;\t") except: dialect = 'excel' reader = csv.DictReader(f_in, dialect=dialect) for i, row in enumerate(reader, start=1): # Check for Progress Update if i % 1000 == 0: self.update_state(state='PROGRESS', meta={ 'current': i, 'total': total_rows, 'errors': error_count }) # Solo columnas de la plantilla (respetar plantilla tal cual) row_norm = row_from_template(row, template_id, normalize_header) errors = validate_row_strict( row_norm, model_target, i, date_format, validator, inv_type_value, invoice_id_cache, models, ) if errors: error_count += 1 # Write simple JSON error f_err.write(json.dumps(errors) + "\n") processed_rows += 1 except Exception as e: logger.error(f"Scan failed: {e}") return {"status": "failed", "error": str(e)} # 4. Store error line numbers in Redis so insert_valid_rows can skip them (any worker) error_lines_list = [] errors_detail: List[Dict[str, Any]] = [] 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"]) if len(errors_detail) < 500: errors_detail.append( { "line": err["line"], "col": err.get("col", ""), "msg": err.get("msg", ""), } ) except Exception: pass if error_lines_list: r = _get_redis() r.set( f"{IMPORT_ERROR_LINES_KEY_PREFIX}{job_id}", json.dumps(error_lines_list).encode("utf-8"), ex=IMPORT_REDIS_TTL, ) except Exception as e: logger.warning(f"Failed to store error lines in Redis: {e}") # 5. Result (incluye lista de errores para que el usuario pueda corregir el CSV) 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 validate_row_phase_1( row: Dict[str, Any], target: str, line_num: int, date_format: Optional[str], ) -> Optional[Dict[str, Any]]: """ Validation: Unique IDs, Dates, and Numeric constraint checks. Target: 'invoice_header' or 'invoice_details' """ def check_decimal(col_name): val = row.get(col_name) if val and str(val).strip(): if parse_decimal(val) is None: return {"line": line_num, "col": col_name, "msg": "Debe ser un número decimal válido"} return None def check_int(col_name): val = row.get(col_name) if val and str(val).strip(): if parse_int(val) is None: return {"line": line_num, "col": col_name, "msg": "Debe ser un número entero válido"} return None def check_date(col_name): date_str = row.get(col_name) if date_str and str(date_str).strip(): if not is_valid_date(date_str, date_format): expected = display_date_format(date_format) return { "line": line_num, "col": col_name, "msg": f"Formato de fecha inválido ({expected})", } return None def check_weight(col_name): val = row.get(col_name) if val and str(val).strip(): if parse_weight_unit(val) is None: return {"line": line_num, "col": col_name, "msg": "Unidad de peso inválida (ej. KGS, LBS)"} return None def check_currency(col_name): val = row.get(col_name) if val and str(val).strip(): parsed_currency = parse_currency(val, None) val_norm = normalize_header(val) # parse_currency returns MANUAL if unknown, so if it wasn't explicitly MANUAL, it's invalid if parsed_currency.value == "manual" and "MANUAL" not in val_norm: return {"line": line_num, "col": col_name, "msg": "Moneda inválida (ej. MN, ME, USD, PESOS)"} return None def check_transport_type(col_name): val = row.get(col_name) if val and str(val).strip(): if str(val).strip().lower() not in TRANSPORT_TYPE_VALUES: return {"line": line_num, "col": col_name, "msg": "Tipo de transporte inválido (ej. box, truck, container)"} return None # A. Invoice Header if target == 'invoice_header': # 1. Unique ID if not row.get('NUMERO FACTURA') and not row.get('NUM FACTURA') and not row.get('ID'): return {"line": line_num, "col": "NUMERO FACTURA", "msg": "Requerido"} # 2. Date Format date_str = row.get('FECHA FACTURA') or row.get('FECHA') if not date_str or not str(date_str).strip(): return {"line": line_num, "col": "FECHA FACTURA", "msg": "Requerido"} err = check_date('FECHA FACTURA') or check_date('FECHA') if err: return err err = check_date('FECHA EMISION') if err: return err # 3. Numeric Fields for col in ['TIPO DE CAMBIO', 'FLETES', 'VALOR SEGUROS', 'SEGUROS', 'EMBALAJES', 'OTROS INCREMENTABLES']: err = check_decimal(col) if err: return err # 4. Integer FKs for col in ['CLAVE PROVEEDOR', 'CLAVE VENDIDO A', 'CLAVE ENVIADO A', 'AGENTE ADUANAL', 'REMESA']: err = check_int(col) if err: return err # 5. Enums for col in ['TIPO PESO']: err = check_weight(col) if err: return err for col in ['TIPO MONEDA']: err = check_currency(col) if err: return err for col in ['TIPO TRANSPORTE']: err = check_transport_type(col) if err: return err # B. Invoice Details (Parts) elif target == 'invoice_details': # 1. Line Number if not row.get('LINEA') and not row.get('RENGLON') and not row.get('PARTIDA'): return {"line": line_num, "col": "LINEA", "msg": "Requerido"} # 2. Parent Link (Invoice Number) if not (row.get('NUMERO FACTURA') or row.get('NUM FACTURA') or row.get('FACTURA')): return {"line": line_num, "col": "NUMERO FACTURA", "msg": "Requerido"} # 3. Numeric Fields for col in ['PRECIO UNITARIO', 'PRECIOUNITARIO', 'VALOR COMERCIAL', 'VALORCOMERCIAL', 'CANTIDAD']: err = check_decimal(col) if err: return err for col in ['CANTIDAD BULTOS', 'CANTIDADBULTOS', 'LINEA', 'RENGLON', 'PARTIDA']: err = check_int(col) if err: return err return None def validate_row_strict( row: Dict[str, Any], target: str, line_num: int, date_format: Optional[str], validator: ForeignKeyValidator, inv_type_value: str, invoice_id_cache: Dict[str, Optional[int]], models: Dict[str, Any], ) -> Optional[Dict[str, Any]]: err = validate_row_phase_1(row, target, line_num, date_format) if err: return err InvoiceHeader = models["InvoiceHeader"] InvoiceType = models["InvoiceType"] ClientProvider = models["ClientProvider"] CustomsBroker = models["CustomsBroker"] RegimenPedimento = models["RegimenPedimento"] CurrencyType = models["CurrencyType"] CustomsSection = models["CustomsSection"] Incoterm = models["Incoterm"] Part = models["Part"] if target == "invoice_header": if not validator.check_exists(InvoiceType, inv_type_value, field_name="key", is_public=True): return {"line": line_num, "col": "TIPO FACTURA", "msg": "No existe en el catalogo"} provider_id = parse_int(row.get("CLAVE PROVEEDOR")) err = validate_tenant_fk_id(validator, ClientProvider, provider_id, line_num, "CLAVE PROVEEDOR", required=True) if err: return err sold_to_id = parse_int(row.get("CLAVE VENDIDO A")) err = validate_tenant_fk_id(validator, ClientProvider, sold_to_id, line_num, "CLAVE VENDIDO A", required=True) if err: return err shipped_to_id = parse_int(row.get("CLAVE ENVIADO A")) err = validate_tenant_fk_id(validator, ClientProvider, shipped_to_id, line_num, "CLAVE ENVIADO A", required=True) if err: return err broker_id = parse_int(row.get("AGENTE ADUANAL")) err = validate_tenant_fk_id(validator, CustomsBroker, broker_id, line_num, "AGENTE ADUANAL") if err: return err err = validate_public_code( validator, RegimenPedimento, row.get("REGIMEN") or row.get("CLAVEDOCUMENTO"), line_num, "CLAVEDOCUMENTO", ) if err: return err err = validate_public_code( validator, CustomsSection, row.get("ADUANA DE CRUCE"), line_num, "ADUANA DE CRUCE", field_name="customs_code", ) if err: return err err = validate_public_code( validator, CurrencyType, row.get("CLAVE MONEDA"), line_num, "CLAVE MONEDA", ) if err: return err err = validate_public_code( validator, Incoterm, row.get("CLAVE INCOTERM"), line_num, "CLAVE INCOTERM", ) if err: return err elif target == "invoice_details": invoice_number = (row.get("NUMERO FACTURA") or row.get("NUM FACTURA") or row.get("FACTURA") or "").strip() if not invoice_number: return {"line": line_num, "col": "NUMERO FACTURA", "msg": "Requerido"} cache_key = f"{invoice_number}|{inv_type_value}" if cache_key in invoice_id_cache: invoice_id = invoice_id_cache[cache_key] else: invoice_id = ( validator.session.query(InvoiceHeader.id) .filter( InvoiceHeader.tenant_id == validator.tenant_id, InvoiceHeader.company_id == validator.company_id, InvoiceHeader.invoice_number == invoice_number, InvoiceHeader.invoice_type == inv_type_value, ) .scalar() ) invoice_id_cache[cache_key] = invoice_id if not invoice_id: return {"line": line_num, "col": "NUMERO FACTURA", "msg": "Factura no existe"} part_num = (row.get("NUMPARTE") or row.get("NUMERO PARTE") or "").strip() if not part_num: return {"line": line_num, "col": "NUMPARTE", "msg": "Requerido"} if not validator.check_exists(Part, part_num, field_name="part_number"): return {"line": line_num, "col": "NUMPARTE", "msg": "No existe en el catalogo"} return None def parse_footer_config(config: Optional[str]) -> Dict[str, Any]: if not config: return {} try: if isinstance(config, str): return json.loads(config) if isinstance(config, dict): return config except Exception: return {} return {} def display_date_format(date_format: Optional[str]) -> str: if not date_format: return "YYYY-MM-DD" return date_format.upper() def parse_date(date_text: Optional[str], date_format: Optional[str]) -> Optional[datetime.date]: if not date_text: return None candidates = [] fmt_map = { "dd/mm/yyyy": "%d/%m/%Y", "mm/dd/yyyy": "%m/%d/%Y", "yyyy-mm-dd": "%Y-%m-%d", } if date_format and date_format in fmt_map: candidates.append(fmt_map[date_format]) candidates.extend(["%Y-%m-%d", "%d/%m/%Y", "%m/%d/%Y"]) for fmt in candidates: try: return datetime.strptime(str(date_text).strip(), fmt).date() except ValueError: continue return None def is_valid_date(date_text: Optional[str], date_format: Optional[str]) -> bool: return parse_date(date_text, date_format) is not None def normalize_header(name: Optional[str]) -> str: if not name: return "" name = unicodedata.normalize("NFKD", str(name)).upper() name = "".join(ch for ch in name if not unicodedata.combining(ch)) name = re.sub(r"[^A-Z0-9]+", " ", name) return re.sub(r"\s+", " ", name).strip() def normalize_row(row: Dict[str, Any]) -> Dict[str, Any]: return {normalize_header(k): v for k, v in row.items()} def parse_int(value: Any) -> Optional[int]: if value is None: return None text = str(value).strip() if not text: return None try: return int(text) except ValueError: return None def parse_decimal(value: Any) -> Optional[Decimal]: if value is None: return None text = str(value).strip() if not text: return None text = text.replace(",", "") try: return Decimal(text) except Exception: return None def decimal_or_zero(value: Any) -> Decimal: """Return parsed decimal or Decimal('0') for CSV nulls/empty (vanilla default).""" return parse_decimal(value) or Decimal("0") def int_or_zero(value: Any) -> int: """Return parsed int or 0 for CSV nulls/empty (vanilla default).""" return parse_int(value) if parse_int(value) is not None else 0 def parse_currency(value: Optional[str], currency_type: Optional[str]): from api.v1.modules.a76.invoices.models import Currency if value: normalized = normalize_header(value) if normalized in {"MN", "M N", "NACIONAL", "LOCAL", "PESOS", "PESO"}: return Currency.LOCAL if normalized in {"ME", "M E", "EXTRANJERA", "EXTRANJERO", "FOREIGN", "USD", "DOLAR", "DOLARES"}: return Currency.FOREIGN if "MANUAL" in normalized: return Currency.MANUAL if currency_type and str(currency_type).strip().upper() == "MXN": return Currency.LOCAL if currency_type: return Currency.FOREIGN return Currency.MANUAL def parse_weight_unit(value: Optional[str]): from api.v1.modules.a76.invoices.models import WeightUnit if not value: return None normalized = normalize_header(value) if normalized in {"KG", "KGS", "KILOS", "KILOGRAMOS"}: return WeightUnit.KGS if normalized in {"LB", "LBS", "LIBRAS"}: return WeightUnit.LBS return None def resolve_tenant_fk_id( session: CoreSessionLocal, model, value: Optional[int], tenant_id: int, company_id: int, cache: Dict[int, Optional[int]], ) -> Optional[int]: if value is None: return None if value in cache: return cache[value] exists = ( session.query(model.id) .filter( model.id == value, model.tenant_id == tenant_id, model.company_id == company_id, ) .scalar() ) cache[value] = value if exists is not None else None return cache[value] def resolve_public_code( session: CoreSessionLocal, model, column, value: Optional[str], cache: Dict[str, Optional[str]], ) -> Optional[str]: if not value: return None normalized = str(value).strip().upper() if not normalized: return None if normalized in cache: return cache[normalized] exists = session.query(column).filter(column == normalized).scalar() cache[normalized] = normalized if exists is not None else None return cache[normalized] @celery_app.task(bind=True) def insert_valid_rows(self, job_id: str, model_target: str): """ Pass 2: Re-read CSV, Skip Errors, Bulk Insert. File and meta are loaded from Redis if present (same as scan_file), so worker does not need shared filesystem. """ logger.info(f"Starting Commit for {job_id} target {model_target}") # Ensure we have the file on this worker: prefer Redis (so any worker can run commit) file_path = _ensure_worker_has_file_from_redis(job_id) if not file_path: upload_dir = _worker_upload_dir() file_path = os.path.join(upload_dir, f"{job_id}.csv") if not os.path.exists(file_path): return {"status": "failed", "error": "File not found (missing or expired). Please upload and confirm again."} else: _ensure_worker_has_meta_from_redis(job_id, file_path) try: from api.v1.modules.a76.invoices.models import ( InvoiceHeader, InvoiceComplianceMx, InvoiceFinancials, InvoiceLogistics, InvoiceSalesDetails, OperationType, TransportType, WeightUnit, ) from api.v1.modules.a76.clients_and_providers.models import ClientProvider from api.v1.modules.a76.customs_brokers.models import CustomsBroker from api.v1.modules.public.reference_data.currency_types.models import CurrencyType from api.v1.modules.public.reference_data.pedimento_codes.models import PedimentoCode from api.v1.modules.public.reference_data.code_pedimento_regimens.models import CodePedimentoRegimen from api.v1.modules.public.reference_data.pedimento_regimens.models import RegimenPedimento from api.v1.modules.public.reference_data.invoice_types.models import InvoiceType from api.v1.modules.public.reference_data.customs_sections.models import CustomsSection from api.v1.modules.public.reference_data.incoterms.models import Incoterm from api.v1.modules.a76.items.models import LineItem from api.v1.modules.a76.items.line_financials.models import LineFinancial from api.v1.modules.a76.items.line_quantities.models import LineQuantity from api.v1.modules.a76.items.line_customs.models import LineCustom from api.v1.modules.a76.items.line_descriptions.models import LineDescription from api.v1.modules.a76.parts.models import Part error_path = file_path.replace("temp", "errors").replace(".csv", ".jsonl") # 1. Load Error Line Numbers (from Redis if scan ran on another worker, else from file) error_lines = set() try: r = _get_redis() raw = r.get(f"{IMPORT_ERROR_LINES_KEY_PREFIX}{job_id}") if raw: error_lines = set(json.loads(raw.decode("utf-8"))) except Exception as e: logger.debug(f"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 # Load Metadata (Context) meta_path = file_path.replace(".csv", ".meta.json") tenant_id = None company_id = None footer_config = {} meta = {} if os.path.exists(meta_path): try: with open(meta_path, 'r') as f: meta = json.load(f) or {} tenant_id = meta.get('tenant_id') company_id = meta.get('company_id') operation_type_raw = meta.get('operation_type', 'imp') footer_config = parse_footer_config(meta.get('footer_config')) except: pass if not tenant_id or not company_id: return {"status": "failed", "error": "Missing context (tenant/company)"} # 2. Re-read and Map # Initialize counters outside the session block so they're accessible later headers_to_insert = [] details_to_insert = [] skipped_invalid = 0 skipped_missing_invoice = 0 skipped_missing_fk = 0 skipped_fk_details = [] inserted_count = 0 response = None # Will be set inside the session block date_format = footer_config.get("dateFormat") # Validate and set default date_format if not provided if not date_format: date_format = "yyyy-mm-dd" # Default to ISO format logger.info(f"No date_format specified in config, using default: {date_format}") else: logger.info(f"Using date_format from config: {date_format}") # Default types from config or fallback op_type_value = OperationType(meta.get('operation_type', 'imp').lower()) inv_type_value = normalize_public_code(footer_config.get('invoice_type') or 'TEM') or 'TEM' logger.info(f"Processing CSV with operation_type={op_type_value}, invoice_type={inv_type_value}, date_format={date_format}") with CoreSessionLocal() as session: invoice_id_cache = {} cleared_invoices = set() # Track invoices where we've already cleared items in this job provider_cache: Dict[int, Optional[int]] = {} sold_to_cache: Dict[int, Optional[int]] = {} shipped_to_cache: Dict[int, Optional[int]] = {} broker_cache: Dict[int, Optional[int]] = {} regimen_cache: Dict[str, Optional[str]] = {} currency_type_cache: Dict[str, Optional[str]] = {} customs_section_cache: Dict[str, Optional[str]] = {} part_cache: Dict[str, Optional[int]] = {} validator = ForeignKeyValidator(session, tenant_id, company_id) with open(file_path, 'r', encoding='utf-8-sig') as f: # Detect Delimiter sample = f.read(2048) f.seek(0) try: dialect = csv.Sniffer().sniff(sample, delimiters=",;\t") except: dialect = 'excel' reader = csv.DictReader(f, dialect=dialect) template_id = meta.get("template_id") or ( "imp_temp_header" if model_target == "invoice_header" else "imp_temp_details" ) for i, row in enumerate(reader, start=1): if i in error_lines: continue row_norm = row_from_template(row, template_id, normalize_header) # Mapping Logic (solo campos que acepta el modelo de facturas) if model_target == 'invoice_header': invoice_number = (row_norm.get('NUMERO FACTURA') or row_norm.get('NUM FACTURA') or row_norm.get('FACTURA') or '').strip() invoice_date = parse_date(row_norm.get('FECHA FACTURA') or row_norm.get('FECHA'), date_format) if not invoice_number or not invoice_date: skipped_invalid += 1 logger.debug(f"Row {i}: Skipped - missing invoice_number or invalid invoice_date. " f"Invoice: {invoice_number}, Date: {row_norm.get('FECHA FACTURA') or row_norm.get('FECHA')}") continue # --- NEW: Foreign Key Validations --- # 1. Invoice Type (Public) if not validator.check_exists(InvoiceType, inv_type_value, field_name="key", is_public=True): skipped_missing_fk += 1 reason = f"Tipo de factura '{inv_type_value}' no existe" skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") continue provider_id = parse_int(row_norm.get('CLAVE PROVEEDOR')) err = validate_tenant_fk_id( validator, ClientProvider, provider_id, i, "CLAVE PROVEEDOR", required=True, ) if err: skipped_invalid += 1 reason = f"{err['col']}: {err['msg']}" skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") continue sold_to_id = parse_int(row_norm.get('CLAVE VENDIDO A')) err = validate_tenant_fk_id( validator, ClientProvider, sold_to_id, i, "CLAVE VENDIDO A", required=True, ) if err: skipped_invalid += 1 reason = f"{err['col']}: {err['msg']}" skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") continue shipped_to_id = parse_int(row_norm.get('CLAVE ENVIADO A')) err = validate_tenant_fk_id( validator, ClientProvider, shipped_to_id, i, "CLAVE ENVIADO A", required=True, ) if err: skipped_invalid += 1 reason = f"{err['col']}: {err['msg']}" skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") continue broker_id = parse_int(row_norm.get('AGENTE ADUANAL')) err = validate_tenant_fk_id( validator, CustomsBroker, broker_id, i, "AGENTE ADUANAL", ) if err: skipped_missing_fk += 1 reason = f"{err['col']}: {err['msg']}" skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") continue err = validate_public_code( validator, RegimenPedimento, row_norm.get('REGIMEN') or row_norm.get('CLAVEDOCUMENTO'), i, "CLAVEDOCUMENTO", ) if err: skipped_missing_fk += 1 reason = f"{err['col']}: {err['msg']}" skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") continue err = validate_public_code( validator, CustomsSection, row_norm.get('ADUANA DE CRUCE'), i, "ADUANA DE CRUCE", field_name="customs_code", ) if err: skipped_missing_fk += 1 reason = f"{err['col']}: {err['msg']}" skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") continue err = validate_public_code( validator, CurrencyType, row_norm.get('CLAVE MONEDA'), i, "CLAVE MONEDA", ) if err: skipped_missing_fk += 1 reason = f"{err['col']}: {err['msg']}" skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") continue err = validate_public_code( validator, Incoterm, row_norm.get('CLAVE INCOTERM'), i, "CLAVE INCOTERM", ) if err: skipped_missing_fk += 1 reason = f"{err['col']}: {err['msg']}" skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") continue transport_type_val = row_norm.get('TIPO TRANSPORTE') if transport_type_val and str(transport_type_val).strip().lower() not in TRANSPORT_TYPE_VALUES: skipped_invalid += 1 reason = "TIPO TRANSPORTE: Tipo de transporte invalido" skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") continue currency_val = row_norm.get('TIPO MONEDA') if currency_val and str(currency_val).strip(): parsed_currency = parse_currency(currency_val, None) val_norm = normalize_header(currency_val) if parsed_currency.value == "manual" and "MANUAL" not in val_norm: skipped_invalid += 1 reason = "TIPO MONEDA: Moneda invalida" skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") continue # 2. Client/Provider and broker checks are handled above # --- 4. Check for Existing Invoice (Upsert Logic) --- existing_header = None if invoice_number: existing_header = ( session.query(InvoiceHeader) .filter( InvoiceHeader.tenant_id == tenant_id, InvoiceHeader.company_id == company_id, InvoiceHeader.invoice_number == invoice_number, InvoiceHeader.invoice_type == inv_type_value ) .first() ) if existing_header: # UPDATE existing header header = existing_header header.invoice_date = invoice_date header.operation_type = op_type_value header.is_updated = True # Mark as updated header.updated_date = datetime.utcnow() header.document_type = resolve_public_code( session, RegimenPedimento, RegimenPedimento.code, (row_norm.get('REGIMEN') or row_norm.get('CLAVEDOCUMENTO')), regimen_cache, ) header.project_number = (row_norm.get('NUM PROYECTO') or row_norm.get('NUMPROYECTO') or None) header.purchase_order = (row_norm.get('ORDEN COMPRA') or row_norm.get('ORDENCOMPRA') or None) header.alternate_invoice = (row_norm.get('FACTURA ALTERNA') or None) header.invoice_ref = (row_norm.get('FACTURA EXPO REF') or row_norm.get('FACTURAEXPOREF') or None) header.emission_date = parse_date(row_norm.get('FECHA EMISION'), date_format) header.observation_es = (row_norm.get('OBSERVACIONES E') or None) header.observation_en = (row_norm.get('OBSERVACIONES I') or None) logger.info(f"Row {i}: Updating existing invoice {invoice_number}") # Clean up related data that will be re-inserted/updated # Note: compliance, financials, logistics are 1-to-1 relationships and will be updated by assignment below # but we might want to be explicit if ORM doesn't handle replace well. # SQLAlchemy relationship assignment usually handles 1-to-1 updates correctly. else: # CREATE new header header = InvoiceHeader( invoice_number=invoice_number, invoice_date=invoice_date, operation_type=op_type_value, is_updated=False, system="CSV", capture_date=datetime.utcnow(), invoice_type=inv_type_value, document_type=resolve_public_code( session, RegimenPedimento, RegimenPedimento.code, (row_norm.get('REGIMEN') or row_norm.get('CLAVEDOCUMENTO')), regimen_cache, ), project_number=(row_norm.get('NUM PROYECTO') or row_norm.get('NUMPROYECTO') or None), purchase_order=(row_norm.get('ORDEN COMPRA') or row_norm.get('ORDENCOMPRA') or None), alternate_invoice=(row_norm.get('FACTURA ALTERNA') or None), invoice_ref=(row_norm.get('FACTURA EXPO REF') or row_norm.get('FACTURAEXPOREF') or None), emission_date=parse_date(row_norm.get('FECHA EMISION'), date_format), observation_es=(row_norm.get('OBSERVACIONES E') or None), observation_en=(row_norm.get('OBSERVACIONES I') or None), tenant_id=tenant_id, company_id=company_id, ) compliance = InvoiceComplianceMx( remesa=parse_int(row_norm.get('REMESA')), aduana=resolve_public_code( session, CustomsSection, CustomsSection.customs_code, row_norm.get('ADUANA DE CRUCE'), customs_section_cache, ), provider_id=resolve_tenant_fk_id( session, ClientProvider, parse_int(row_norm.get('CLAVE PROVEEDOR')), tenant_id, company_id, provider_cache, ), sold_to_id=resolve_tenant_fk_id( session, ClientProvider, parse_int(row_norm.get('CLAVE VENDIDO A')), tenant_id, company_id, sold_to_cache, ), shipped_to_id=resolve_tenant_fk_id( session, ClientProvider, parse_int(row_norm.get('CLAVE ENVIADO A')), tenant_id, company_id, shipped_to_cache, ), customs_broker_id=resolve_tenant_fk_id( session, CustomsBroker, parse_int(row_norm.get('AGENTE ADUANAL')), tenant_id, company_id, broker_cache, ), edocument=(row_norm.get('E DOCUMENT') or None), vucem_operation_num=(row_norm.get('NUM OPERACION') or None), tenant_id=tenant_id, company_id=company_id, ) financials_currency_type = resolve_public_code( session, CurrencyType, CurrencyType.code, row_norm.get('CLAVE MONEDA'), currency_type_cache, ) financials = InvoiceFinancials( currency=parse_currency(row_norm.get('TIPO MONEDA'), financials_currency_type), currency_type=financials_currency_type, exchange_rate=decimal_or_zero(row_norm.get('TIPO DE CAMBIO')), freight=decimal_or_zero(row_norm.get('FLETES')), insurance_value=decimal_or_zero(row_norm.get('VALOR SEGUROS')), insurance=decimal_or_zero(row_norm.get('SEGUROS')), packaging=decimal_or_zero(row_norm.get('EMBALAJES')), other_increments=decimal_or_zero(row_norm.get('OTROS INCREMENTABLES')), tenant_id=tenant_id, company_id=company_id, ) weight_type = parse_weight_unit(row_norm.get('TIPO PESO')) logistics = None if weight_type or row_norm.get('TIPO TRANSPORTE') or row_norm.get('NUMERO TRANSPORTE'): raw_transport = (row_norm.get('TIPO TRANSPORTE') or "none") transport_str = str(raw_transport).strip().lower() or "none" try: transport_type = TransportType(transport_str) except ValueError: transport_type = TransportType.NONE logistics = InvoiceLogistics( carrier_id=(row_norm.get('CLAVE TRANSPORTISTA') or None), driver_name=(row_norm.get('NOMBRE CONDUCTOR') or None), transport_type=transport_type, transport_num=(row_norm.get('NUMERO TRANSPORTE') or None), weight_type=weight_type or WeightUnit.KGS, seal_number=(row_norm.get('PRECINTO') or None), incoterm=(row_norm.get('CLAVE INCOTERM') or None), entry_exit_date=parse_date(row_norm.get('FECHA EMISION'), date_format), tenant_id=tenant_id, company_id=company_id, ) header.compliance_mx = compliance header.financials = financials if logistics: header.logistics = logistics headers_to_insert.append(header) elif model_target == 'invoice_details': invoice_number = (row_norm.get('NUMERO FACTURA') or row_norm.get('NUM FACTURA') or '').strip() if not invoice_number: skipped_invalid += 1 continue cache_key = f"{invoice_number}|{inv_type_value}" if cache_key in invoice_id_cache: invoice_id = invoice_id_cache[cache_key] else: invoice_id = ( session.query(InvoiceHeader.id) .filter( InvoiceHeader.tenant_id == tenant_id, InvoiceHeader.company_id == company_id, InvoiceHeader.invoice_number == invoice_number, InvoiceHeader.invoice_type == inv_type_value, ) .scalar() ) invoice_id_cache[cache_key] = invoice_id if not invoice_id: logger.warning( "Invoice not found for details row %s (invoice_number=%s)", i, invoice_number, ) skipped_missing_invoice += 1 continue part_num = (row_norm.get('NUMPARTE') or row_norm.get('NUMERO PARTE') or '').strip() if not part_num: skipped_invalid += 1 reason = "NUMPARTE: Requerido" skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") continue if not validator.check_exists(Part, part_num, field_name="part_number"): skipped_missing_fk += 1 reason = f"NUMPARTE '{part_num}' no existe" skipped_fk_details.append({"line": i, "invoice": invoice_number, "reason": reason}) logger.warning(f"Row {i} (Invoice {invoice_number}): {reason}") continue # --- Prevent Duplicates: Clear existing items for this invoice (Once per job) --- if invoice_id not in cleared_invoices: logger.info(f"Clearing existing details for Invoice {invoice_number} (ID: {invoice_id}) to prevent duplicates") # 1. Delete Items (Cascades to LineItem, LineFinancial, etc. if DB configured, check models) # Checking Item model, we usually need to be careful. # Assuming Cascade delete is set up on FKs or we rely on ORM cascade if using relationships. # Here we use bulk delete. session.query(Item).filter(Item.invoice_id == invoice_id).delete(synchronize_session=False) # 2. Delete InvoiceSalesDetails session.query(InvoiceSalesDetails).filter(InvoiceSalesDetails.invoice_id == invoice_id).delete(synchronize_session=False) cleared_invoices.add(invoice_id) # --- NEW LOGIC: Expanded Anexo 76 Structure --- # A. Find/Cache Part part_id = None part_id = part_cache.get(part_num) if part_id is None: p = session.query(Part.id).filter( Part.part_number == part_num, Part.tenant_id == tenant_id, Part.company_id == company_id ).first() if p: part_id = p.id part_cache[part_num] = part_id line_num_val = (row_norm.get('LINEA') or row_norm.get('RENGLON') or row_norm.get('PARTIDA')) line_num = parse_int(line_num_val) or (len(details_to_insert) + 1) # 1. Parent Item item = Item( invoice_id=invoice_id, tenant_id=tenant_id, company_id=company_id, item_type="N", # Default to Normal system_origin="CSV" ) session.add(item) session.flush() # Need item.id # 2. Main Line line = LineItem( item_id=item.id, line_number=line_num, part_number=part_id, tenant_id=tenant_id, company_id=company_id ) session.add(line) session.flush() # Need line.id # 3. Financial Data (vanilla: nulls from CSV -> 0) price = parse_decimal(row_norm.get('PRECIO UNITARIO') or row_norm.get('PRECIOUNITARIO')) val_com = parse_decimal(row_norm.get('VALOR COMERCIAL') or row_norm.get('VALORCOMERCIAL')) qty = parse_decimal(row_norm.get('CANTIDAD')) commercial_total = val_com or (price * qty if price and qty else None) session.add(LineFinancial( item_line_id=line.id, unit_cost_capture=decimal_or_zero(price), total_commercial_value=decimal_or_zero(commercial_total), )) # 4. Quantities (vanilla: nulls -> 0 so we always have a quantity row) session.add(LineQuantity( item_line_id=line.id, quantity=decimal_or_zero(qty), )) # 5. Customs/Fraction origin = row_norm.get('PAIS ORIGEN') or row_norm.get('PAISORIGEN') fraction = row_norm.get('FRACCION') if origin or fraction: session.add(LineCustom( item_line_id=line.id, fraction=fraction, origin_country=origin, )) # 6. Description desc = row_norm.get('DESCRIPCION') if desc: session.add(LineDescription( item_line_id=line.id, description_spanish=desc, )) # 7. Legacy Sales Details (For specific audit/UI fields; vanilla: nulls -> 0) detail = InvoiceSalesDetails( invoice_id=invoice_id, line_number=line_num, sales_order=(row_norm.get('ORDEN DE COMPRA') or row_norm.get('ORDENCOMPRA') or None), line_bundles=int_or_zero(row_norm.get('CANTIDAD BULTOS') or row_norm.get('CANTIDADBULTOS')), tenant_id=tenant_id, company_id=company_id, ) session.add(detail) details_to_insert.append(item) # Use as counter/ref # 3. Bulk Insert (ORM Transaction) try: if model_target == 'invoice_header': if headers_to_insert: logger.info(f"Attempting to commit {len(headers_to_insert)} headers") session.add_all(headers_to_insert) session.commit() inserted_count = len(headers_to_insert) logger.info(f"Headers commit successful. Inserted: {inserted_count}") else: logger.warning(f"No headers to insert for job {job_id}") else: if details_to_insert: logger.info(f"Attempting to commit {len(details_to_insert)} items and related data") session.commit() # Everything was already added with session.add() inserted_count = len(details_to_insert) logger.info(f"Details commit successful. Inserted: {inserted_count}") else: logger.warning(f"No details to insert for job {job_id}") except Exception as db_err: session.rollback() logger.error(f"DB Error during {model_target} commit: {db_err}") import traceback logger.error(traceback.format_exc()) return {"status": "failed", "error": str(db_err)} # 4. Determine final status and prepare response (inside session block to access variables) total_skipped = skipped_invalid + skipped_missing_fk + skipped_missing_invoice # Log summary logger.info(f"Job {job_id} completed. Inserted: {inserted_count}, Skipped: {total_skipped} " f"(invalid: {skipped_invalid}, missing_fk: {skipped_missing_fk}, missing_invoice: {skipped_missing_invoice})") # Prepare response based on results if inserted_count == 0: if total_skipped > 0: logger.warning(f"No valid records to insert for job {job_id}. All {total_skipped} records were rejected.") response = { "status": "warning", "inserted": 0, "skipped_invalid": skipped_invalid, "skipped_missing_invoice": skipped_missing_invoice, "skipped_missing_fk": skipped_missing_fk, "skipped_details": skipped_fk_details, "message": f"No se insertaron registros. {total_skipped} fueron rechazados." } else: logger.error(f"No valid records found in CSV for job {job_id}") response = { "status": "failed", "error": "No hay registros válidos en el archivo CSV", "inserted": 0, "skipped_invalid": skipped_invalid, "skipped_missing_invoice": skipped_missing_invoice, "skipped_missing_fk": skipped_missing_fk, "skipped_details": skipped_fk_details } else: # Success case - at least some records were inserted response = { "status": "finished", "inserted": inserted_count, "skipped_invalid": skipped_invalid, "skipped_missing_invoice": skipped_missing_invoice, "skipped_missing_fk": skipped_missing_fk, "skipped_details": skipped_fk_details } except Exception as e: logger.error(f"Task failed: {e}") import traceback logger.error(traceback.format_exc()) return {"status": "failed", "error": str(e)} # 5. Cleanup: remove temp files and Redis keys so data is not kept indefinitely try: if file_path and os.path.exists(file_path): os.remove(file_path) if os.path.exists(error_path): os.remove(error_path) _delete_import_from_redis(job_id) except Exception as cleanup_err: logger.warning("Failed to cleanup temp files or Redis: %s", cleanup_err) # Ensure response is defined (fallback in case of unexpected errors) if response is None: logger.error(f"Unexpected error: response not set for job {job_id}") response = { "status": "failed", "error": "Error inesperado durante el procesamiento", "inserted": 0, "skipped_invalid": skipped_invalid, "skipped_missing_invoice": skipped_missing_invoice, "skipped_missing_fk": skipped_missing_fk, "skipped_details": skipped_fk_details } return response