"""
Tareas Celery para importación CSV de Trailers y Cajas.
Flujo: scan_file (validación) → insert_valid_rows (commit).
Usa layouts_csv.common (storage, normalize, meta, responses); CSV con headers duplicados (dedupe).
"""
import json
import logging
import os
from typing import Dict, Any, Optional, List, Set

from core.celery_app import celery_app
from core.database import CoreSessionLocal

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
from .validators import validate_row_trailer, validate_row_trailer_desfase
from .common.mappers import row_to_trailer_data, row_to_trailer_data_for_update
from .common.fk_loader import load_trailers_fk_sets

logger = logging.getLogger(__name__)

JOB_TYPE = "trl"

# Para routes.py
TRL_IMPORT_FILE_PREFIX = "trl_import_file:"
TRL_IMPORT_META_PREFIX = "trl_import_meta:"
TRL_IMPORT_ERROR_LINES_PREFIX = "trl_import_error_lines:"
TRL_IMPORT_STATUS_PREFIX = "trl_import_status:"
TRL_IMPORT_REDIS_TTL = common_storage.IMPORT_REDIS_TTL


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 _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, "Trailers 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, "Trailers import")

    error_path = common_storage.error_path_for_job(JOB_TYPE, job_id)

    try:
        total_rows = common_csv_reader.count_csv_rows(file_path, has_header=True)
    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_trailer_numbers: Set[str] = set()
    if actualizar:
        try:
            from api.v1.modules.a76.transportation.trailers.models import Trailer
            with CoreSessionLocal() as session:
                for t in (
                    session.query(Trailer.trailer_number)
                    .filter(
                        Trailer.tenant_id == tenant_id,
                        Trailer.company_id == company_id,
                    )
                    .all()
                ):
                    if t[0] and (t[0] or "").strip():
                        existing_trailer_numbers.add((t[0] or "").strip())
        except Exception as e:
            logger.warning("Trailers import: could not load existing trailer_numbers for actualizar: %s", e)

    (
        valid_trailer_type_keys,
        valid_country_ame,
        state_descriptions_upper,
        state_country_set,
        state_ame_to_description,
    ) = load_trailers_fk_sets(tenant_id, company_id)

    error_count = 0
    processed_rows = 0
    errors_detail: List[Dict[str, Any]] = []
    error_lines_list: List[int] = []

    try:
        with open(error_path, "w", encoding="utf-8") as f_err:
            for i, row in common_csv_reader.iter_csv_rows_deduped_headers(file_path):
                if progress_callback:
                    progress_callback(i, total_rows, error_count)

                row_norm = row_from_template(row, common_normalize.normalize_header)
                # Desfase: advertencia no bloqueante (se muestra en UI pero no se agrega a error_lines)
                warn = validate_row_trailer_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_trailer(
                    row_norm,
                    i,
                    actualizar=actualizar,
                    existing_trailer_numbers=existing_trailer_numbers,
                    valid_trailer_type_keys=valid_trailer_type_keys,
                    valid_country_ame=valid_country_ame,
                    state_descriptions_upper=state_descriptions_upper,
                    state_country_set=state_country_set,
                    state_ame_to_description=state_ame_to_description,
                )
                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("Trailers 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"{TRL_IMPORT_STATUS_PREFIX}{job_id}",
            json.dumps(result).encode("utf-8"),
            ex=TRL_IMPORT_REDIS_TTL,
        )
    except Exception as e:
        logger.warning("Trailers 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("Trailers 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"{TRL_IMPORT_STATUS_PREFIX}{job_id}",
            json.dumps(result).encode("utf-8"),
            ex=TRL_IMPORT_REDIS_TTL,
        )
    except Exception as e:
        logger.warning("Trailers 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, "Trailers 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, "Trailers 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_trailer_numbers: Set[str] = set()
    if actualizar:
        try:
            from api.v1.modules.a76.transportation.trailers.models import Trailer
            with CoreSessionLocal() as session:
                for t in (
                    session.query(Trailer.trailer_number)
                    .filter(
                        Trailer.tenant_id == tenant_id,
                        Trailer.company_id == company_id,
                    )
                    .all()
                ):
                    if t[0] and (t[0] or "").strip():
                        existing_trailer_numbers.add((t[0] or "").strip())
        except Exception as e:
            logger.warning("Trailers import: could not load existing trailer_numbers for actualizar: %s", e)

    (
        valid_trailer_type_keys,
        valid_country_ame,
        state_descriptions_upper,
        state_country_set,
        state_ame_to_description,
    ) = load_trailers_fk_sets(tenant_id, company_id)

    from api.v1.modules.a76.transportation.trailers.services import TrailerService
    from api.v1.modules.a76.transportation.trailers.dto import TrailerCreateDTO, TrailerUpdateDTO

    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:
            for i, row in common_csv_reader.iter_csv_rows_deduped_headers(file_path):
                    if i in error_lines:
                        continue

                    row_norm = row_from_template(row, common_normalize.normalize_header)
                    row_errors = validate_row_trailer(
                        row_norm,
                        i,
                        actualizar=actualizar,
                        existing_trailer_numbers=existing_trailer_numbers,
                        valid_trailer_type_keys=valid_trailer_type_keys,
                        valid_country_ame=valid_country_ame,
                        state_descriptions_upper=state_descriptions_upper,
                        state_country_set=state_country_set,
                        state_ame_to_description=state_ame_to_description,
                    )
                    if row_errors and any(not e.get("warning", False) for e in row_errors):
                        skipped_invalid += 1
                        tn = (row_norm.get("NUMERO TRAILER") or row_norm.get("CLAVE TRAILER") or "").strip()[:20] 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,
                                "trailer_number": tn,
                                "invoice": tn,
                                "reason": f"{blocking.get('col', '')}: {blocking.get('msg', '')}",
                                "solution": blocking.get("solution", ""),
                            })
                        continue

                    tn = (row_norm.get("NUMERO TRAILER") or row_norm.get("CLAVE TRAILER") or "").strip()[:20] or ""
                    if not tn:
                        skipped_invalid += 1
                        continue

                    if tn in seen_keys_in_file:
                        skipped_duplicate += 1
                        skipped_details.append({
                            "line": i,
                            "trailer_number": tn,
                            "invoice": tn,
                            "reason": "Clave duplicada en el archivo (se usa la primera)",
                            "solution": "El sistema conserva la primera ocurrencia; no dupliques el NUMERO TRAILER (CLAVE TRAILER) en el archivo.",
                        })
                        continue
                    seen_keys_in_file[tn] = i

                    existing = TrailerService.get_by_id(session, tn, tenant_id, company_id)
                    try:
                        if existing:
                            if actualizar:
                                data = row_to_trailer_data_for_update(row_norm, existing)
                            else:
                                data = row_to_trailer_data(row_norm)
                            if not data or not data.get("trailer_number"):
                                skipped_invalid += 1
                                continue
                            update_data = TrailerUpdateDTO(**{k: v for k, v in data.items() if k != "trailer_number"})
                            TrailerService.update(session, tn, tenant_id, update_data, company_id)
                            updated_count += 1
                        else:
                            data = row_to_trailer_data(row_norm)
                            if not data or not data.get("trailer_number"):
                                skipped_invalid += 1
                                continue
                            create_data = TrailerCreateDTO(**data)
                            TrailerService.create(session, create_data, tenant_id, company_id)
                            inserted_count += 1
                    except Exception as db_err:
                        session.rollback()
                        skipped_invalid += 1
                        skipped_details.append({
                            "line": i,
                            "trailer_number": tn,
                            "invoice": tn,
                            "reason": str(db_err),
                            "solution": "Revisar el CSV para el NUMERO TRAILER indicado y corregir los datos/catálogos para resolver el error.",
                        })
                        continue

            try:
                session.commit()
            except Exception as db_err:
                session.rollback()
                logger.error("Trailers import DB error: %s", db_err)
                return {"status": "failed", "error": str(db_err)}

    except Exception as e:
        logger.exception("Trailers 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"{TRL_IMPORT_STATUS_PREFIX}{job_id}")
    except Exception as e:
        logger.warning("Trailers 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 válidos 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"{TRL_IMPORT_STATUS_PREFIX}{job_id}",
            json.dumps(result).encode("utf-8"),
            ex=TRL_IMPORT_REDIS_TTL,
        )
    except Exception as e:
        logger.warning("Trailers 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("Trailers import: starting commit for job %s", job_id)
    result = _do_commit(job_id)
    try:
        r = _get_redis()
        r.set(
            f"{TRL_IMPORT_STATUS_PREFIX}{job_id}",
            json.dumps(result).encode("utf-8"),
            ex=TRL_IMPORT_REDIS_TTL,
        )
    except Exception as e:
        logger.warning("Trailers import: failed to store commit status in Redis: %s", e)
    return result