"""
Limpieza periódica de archivos huérfanos de import CSV (layouts/imports/temp y errors).
"""
from __future__ import annotations

import logging
import os
import re
import time
from typing import Optional, Tuple

from sqlalchemy import or_

from api.v1.modules.core.tasks_tracking.models import TaskRun, TaskStatus
from core.celery_app import celery_app
from core.database import CoreSessionLocal

from . import storage as common_storage
from .import_redis_keys import (
    FILE_PREFIX_TO_JOB_TYPE,
    any_layout_import_key_exists,
    delete_extra_layout_import_keys_from_redis,
)

logger = logging.getLogger(__name__)

# Evita carreras con uploads/commits recién creados (segundos)
MIN_ORPHAN_FILE_AGE_SEC = int(os.getenv("LAYOUT_ORPHAN_MIN_AGE_SEC", "600"))

_UUID_RE = r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}"


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 parse_temp_csv_filename(name: str) -> Optional[Tuple[str, str]]:
    """Devuelve (job_type, job_id) o None si no es un CSV de import conocido."""
    if not name.endswith(".csv"):
        return None
    base = name[:-4]
    m = re.match(rf"^({_UUID_RE})$", base, re.I)
    if m:
        return "", m.group(1)
    m = re.match(rf"^([a-z0-9]+)_({_UUID_RE})$", base, re.I)
    if not m:
        return None
    prefix, jid = m.group(1).lower(), m.group(2)
    jt = FILE_PREFIX_TO_JOB_TYPE.get(prefix)
    if jt is None:
        return None
    return jt, jid


def parse_errors_jsonl_filename(name: str) -> Optional[Tuple[str, str]]:
    if not name.endswith(".jsonl"):
        return None
    base = name[:-6]
    m = re.match(rf"^({_UUID_RE})$", base, re.I)
    if m:
        return "", m.group(1)
    m = re.match(rf"^([a-z0-9]+)_({_UUID_RE})$", base, re.I)
    if not m:
        return None
    prefix, jid = m.group(1).lower(), m.group(2)
    jt = FILE_PREFIX_TO_JOB_TYPE.get(prefix)
    if jt is None:
        return None
    return jt, jid


def _layout_import_protected_by_task_run(db, job_id: str) -> bool:
    row = (
        db.query(TaskRun)
        .filter(
            TaskRun.task_group == "layouts_csv",
            TaskRun.status.in_([TaskStatus.PENDING.value, TaskStatus.ACTIVE.value]),
            or_(
                TaskRun.task_id == job_id,
                TaskRun.meta_payload.contains({"layout_import_job_id": job_id}),
            ),
        )
        .first()
    )
    return row is not None


def _reference_mtime(job_type: str, job_id: str) -> Optional[float]:
    csv_path = common_storage.file_path_for_job(job_type, job_id)
    err_path = common_storage.error_path_for_job(job_type, job_id)
    mtimes = []
    if os.path.isfile(csv_path):
        mtimes.append(os.path.getmtime(csv_path))
    if os.path.isfile(err_path):
        mtimes.append(os.path.getmtime(err_path))
    if not mtimes:
        return None
    return max(mtimes)


def _try_cleanup_layout_import_job(r, db, job_type: str, job_id: str) -> bool:
    if any_layout_import_key_exists(r, job_type, job_id):
        return False
    if _layout_import_protected_by_task_run(db, job_id):
        return False
    mtime = _reference_mtime(job_type, job_id)
    if mtime is None:
        return False
    if time.time() - mtime < MIN_ORPHAN_FILE_AGE_SEC:
        return False

    file_path = common_storage.file_path_for_job(job_type, job_id)
    error_path = common_storage.error_path_for_job(job_type, job_id)
    meta_path = None
    if os.path.isfile(file_path):
        meta_path = file_path.replace(".csv", ".meta.json")

    common_storage.cleanup_import_job(
        job_type,
        job_id,
        file_path=file_path if os.path.isfile(file_path) else None,
        error_path=error_path if os.path.isfile(error_path) else None,
        meta_path=meta_path if meta_path and os.path.isfile(meta_path) else None,
    )
    delete_extra_layout_import_keys_from_redis(r, job_type, job_id)
    return True


def run_orphan_layout_import_cleanup() -> dict:
    """Barrido síncrono; devuelve contadores para logs/resultado Celery."""
    removed_jobs: list[str] = []
    r = _get_redis()
    db = CoreSessionLocal()
    try:
        temp_dir = common_storage.upload_dir()
        os.makedirs(temp_dir, exist_ok=True)
        for name in os.listdir(temp_dir):
            parsed = parse_temp_csv_filename(name)
            if not parsed:
                continue
            job_type, job_id = parsed
            if _try_cleanup_layout_import_job(r, db, job_type, job_id):
                removed_jobs.append(f"{job_type or 'invoice'}:{job_id}")

        err_dir = common_storage.error_dir()
        if os.path.isdir(err_dir):
            for name in os.listdir(err_dir):
                parsed = parse_errors_jsonl_filename(name)
                if not parsed:
                    continue
                job_type, job_id = parsed
                csv_path = common_storage.file_path_for_job(job_type, job_id)
                if os.path.isfile(csv_path):
                    continue
                if _try_cleanup_layout_import_job(r, db, job_type, job_id):
                    rid = f"{job_type or 'invoice'}:{job_id}:errors_only"
                    if rid not in removed_jobs:
                        removed_jobs.append(rid)
    finally:
        db.close()

    if removed_jobs:
        logger.info("layout import cleanup removed %s job(s): %s", len(removed_jobs), removed_jobs[:20])
    return {"removed_count": len(removed_jobs), "removed": removed_jobs}


@celery_app.task(name="cleanup_orphan_layout_imports")
def cleanup_orphan_layout_imports():
    return run_orphan_layout_import_cleanup()
