"""
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:
                    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