from __future__ import annotations
from dataclasses import dataclass, asdict
from pathlib import Path
import os
import shutil
import pandas as pd
import json
import tempfile
from app.db import get_engine
from app.queries import QUERY_FOLIOS_AGG
from app.dwh.paths import FOLIOS_DIR, ensure_dirs
# Librerias
from google.oauth2 import service_account
from googleapiclient.discovery import build
SCOPES = ["https://www.googleapis.com/auth/drive"]

from googleapiclient.http import MediaFileUpload
def _resolve_sa_path(path_str: str) -> str:
    p = Path(path_str)

    if not p.is_absolute():
        # etl_folios.py está en: CABANNA-API/app/dwh/etl_folios.py
        # repo root es: parents[2] -> CABANNA-API/
        ROOT = Path(__file__).resolve().parents[2]
        p = (ROOT / p).resolve()

    return str(p)

def _looks_like_json(s: str) -> bool:
    s = (s or "").strip()
    return s.startswith("{") and s.endswith("}")

def _normalize_private_key(info: dict) -> dict:
    """
    Asegura que private_key tenga saltos de línea reales.
    Sirve tanto si viene con \\n escapados como si viene normal.
    """
    info = dict(info)
    pk = info.get("private_key")
    if isinstance(pk, str):
        if "\\n" in pk:
            pk = pk.replace("\\n", "\n")
        if not pk.endswith("\n"):
            pk += "\n"
        info["private_key"] = pk
    return info

def _load_sa_credentials_from_env():
    """
    ÚNICA función válida para local + producción.

    Prioridad:
    1) Streamlit Cloud secrets: [gcp_service_account] (dict)
    2) ENV GDRIVE_SA_JSON como JSON (texto)
    3) ENV GDRIVE_SA_JSON como ruta a archivo .json (local)
    """

    # 2/3) ENV var
    raw = os.getenv("GDRIVE_SA_JSON")
    if not raw:
        raise FileNotFoundError(
            "Falta credencial. Define:\n"
            "- Streamlit Secrets: [gcp_service_account]\n"
            "o\n"
            "- ENV: GDRIVE_SA_JSON (JSON completo o ruta al .json)"
        )

    raw = raw.strip().replace("\r\n", "\n")

    # 2) Si es JSON -> parsea como texto
    if _looks_like_json(raw):
        try:
            info = json.loads(raw)
        except json.JSONDecodeError as e:
            raise RuntimeError(
                "GDRIVE_SA_JSON parece JSON pero no es válido.\n"
                "Si lo pegaste en secrets como string, asegúrate de que el private_key use \\n escapados.\n"
                "Recomendado: usa [gcp_service_account] en Streamlit secrets.\n"
                f"Detalle: {e}"
            ) from e

        info = _normalize_private_key(info)
        return service_account.Credentials.from_service_account_info(info, scopes=SCOPES)

    # 3) Si no es JSON -> se trata como ruta (local)
    json_path = _resolve_sa_path(raw)
    if not os.path.exists(json_path):
        raise FileNotFoundError(
            f"No encuentro el JSON del Service Account en: {json_path}\n"
            "En local: define GDRIVE_SA_JSON como ruta válida.\n"
            "En Streamlit Cloud: usa [gcp_service_account] en Secrets."
        )

    return service_account.Credentials.from_service_account_file(json_path, scopes=SCOPES)

def get_drive_service():
    creds = _load_sa_credentials_from_env()
    return build("drive", "v3", credentials=creds)

def get_setting(key: str, default: str | None = None) -> str | None:

    # 2) env
    v = os.getenv(key)
    if v is not None and str(v).strip() != "":
        return str(v).strip()
    return default

FOLDER_ID_FOLIOS = get_setting("FOLDER_ID_FOLIOS")
if not FOLDER_ID_FOLIOS:
    raise RuntimeError("Falta FOLDER_ID_FOLIOS en .env o variables de entorno.")


def month_start(dt: pd.Timestamp) -> pd.Timestamp:
    return dt.normalize().replace(day=1)

def next_month(dt: pd.Timestamp) -> pd.Timestamp:
    return (dt + pd.offsets.MonthBegin(1)).normalize()

def month_ranges(start_dt: pd.Timestamp, end_dt: pd.Timestamp):
    cur = month_start(start_dt)
    end_dt = pd.to_datetime(end_dt)
    while cur < end_dt:
        nxt = next_month(cur)
        ini = cur
        fin = min(nxt, end_dt)
        yield ini, fin
        cur = nxt

def _dir_size_bytes(path: Path) -> int:
    if not path.exists():
        return 0
    total = 0
    for p in path.rglob("*"):
        if p.is_file():
            total += p.stat().st_size
    return total

def _disk_free_bytes(path: Path) -> int:
    # usa el filesystem donde vive la carpeta
    usage = shutil.disk_usage(str(path))
    return usage.free

from googleapiclient.http import MediaFileUpload

def upload_or_replace_in_drive(folder_id: str, filename: str, local_path: str):
    service = get_drive_service()

    q = f"'{folder_id}' in parents and name='{filename}' and trashed=false"
    res = service.files().list(q=q, fields="files(id)").execute()
    files = res.get("files", [])

    media = MediaFileUpload(local_path, resumable=True)

    if files:
        # update
        service.files().update(
            fileId=files[0]["id"],
            media_body=media
        ).execute()
    else:
        # create
        service.files().create(
            body={"name": filename, "parents": [folder_id]},
            media_body=media
        ).execute()


@dataclass
class EtlFoliosResult:
    status: str
    start: str
    end: str
    files_written: int
    rows_written: int
    months_total: int
    months_empty: int
    empty_months: list[str]
    last_nonempty_end: str | None
    last_file: str | None
    folios_dir_bytes: int
    disk_free_bytes: int
    warnings: list[str]


def build_folios_parquet(
    start: str,
    end: str,
    *,
    stop_on_consecutive_empty: int = 3,   # 🔧 “me detengo si tengo N meses seguidos vacíos”
    min_rows_to_write: int = 1            # 🔧 no guardo si df tiene menos de esto
) -> dict:
    """
    Construye parquets mensuales folios_YYYY_MM.parquet.
    - NO guarda archivos vacíos.
    - Opcional: se detiene tras N meses consecutivos vacíos.
    - Regresa un resumen (para que /etl/folios lo reporte).
    """

    ensure_dirs()
    engine = get_engine()

    start_dt = pd.to_datetime(start)
    end_dt = pd.to_datetime(end)

    files_written = 0
    rows_written = 0
    months_total = 0
    months_empty = 0
    empty_months: list[str] = []
    warnings: list[str] = []

    last_nonempty_end: pd.Timestamp | None = None
    last_file: Path | None = None

    consecutive_empty = 0

    for ini, fin in month_ranges(start_dt, end_dt):
        months_total += 1
        ym = ini.strftime("%Y-%m")
        out = FOLIOS_DIR / f"folios_{ini.strftime('%Y_%m')}.parquet"

        with engine.connect() as conn:
            # params=(ini, fin) OK si QUERY_FOLIOS_AGG usa 2 markers
            df = pd.read_sql(QUERY_FOLIOS_AGG, conn, params=(ini, fin))

        # Normaliza Fecha si existe
        if not df.empty and "Fecha" in df.columns:
            df["Fecha"] = pd.to_datetime(df["Fecha"], errors="coerce")

        # ✅ No guardar vacíos (ni casi vacíos)
        if df is None or len(df) < min_rows_to_write:
            months_empty += 1
            empty_months.append(ym)
            consecutive_empty += 1

            # 👇 estrategia: cortar si ya son muchos vacíos seguidos
            if stop_on_consecutive_empty and consecutive_empty >= stop_on_consecutive_empty:
                warnings.append(
                    f"Se detuvo por {consecutive_empty} meses consecutivos sin datos. "
                    f"Último mes procesado: {ym} (rango {ini.date()}→{fin.date()})."
                )
                break

            # si no cortamos, seguimos con el siguiente mes
            continue

        # si aquí llegó, hay data real
        consecutive_empty = 0

        # Guarda parquet
        tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".parquet")
        df.to_parquet(tmp.name, index=False)
        tmp.close()

        upload_or_replace_in_drive(
            folder_id=FOLDER_ID_FOLIOS,
            filename=out.name,        # folios_YYYY_MM.parquet
            local_path=tmp.name
        )

        os.unlink(tmp.name)


        files_written += 1
        rows_written += len(df)
        last_nonempty_end = fin
        last_file = out

    # Métricas de disco
    folios_dir_bytes = _dir_size_bytes(FOLIOS_DIR)
    disk_free_bytes = _disk_free_bytes(FOLIOS_DIR)

    # Alertas simples (no “exactas”, pero útiles)
    # Nota: Render tiene filesystem efímero por defecto; cambios se pierden en deploy/restart. :contentReference[oaicite:1]{index=1}
    if folios_dir_bytes > 1_500_000_000:
        warnings.append("Carpeta folios supera ~1.5GB; considera mover parquets a almacenamiento persistente (S3/GCS/Render Disk).")
    if disk_free_bytes < 1_000_000_000:
        warnings.append("Espacio libre en disco bajo (<~1GB). Podrías quedarte sin espacio al seguir generando parquets.")

    res = EtlFoliosResult(
        status="ok",
        start=start,
        end=end,
        files_written=files_written,
        rows_written=rows_written,
        months_total=months_total,
        months_empty=months_empty,
        empty_months=empty_months[-24:],  # limita tamaño del payload
        last_nonempty_end=last_nonempty_end.isoformat() if last_nonempty_end is not None else None,
        last_file=str(last_file) if last_file is not None else None,
        folios_dir_bytes=folios_dir_bytes,
        disk_free_bytes=disk_free_bytes,
        warnings=warnings
    )

    # Si no escribió nada, márcalo claro
    if files_written == 0:
        res.status = "no_data"
        res.warnings.append("No se escribió ningún parquet con datos. Revisa QUERY_FOLIOS_AGG y/o el rango de fechas.")

    return asdict(res)