excel-planner/export/rtc_tsv_exporter.py

"""
Convierte proyectos de Planner al TSV que consume Tablero_proyectos_v7.html.

Formato (10 columnas, separadas por tabulador):
  id | nombre | estadoGlobal | progreso | fechaObjetivo | comentarioRetraso |
  tarea | responsable | fechaFin | estado
"""

from __future__ import annotations

from datetime import datetime
from pathlib import Path

from config import OUTPUT_DIR

TSV_HEADER = [
    "id",
    "nombre",
    "estadoGlobal",
    "progreso",
    "fechaObjetivo",
    "comentarioRetraso",
    "tarea",
    "responsable",
    "fechaFin",
    "estado",
]


def _sanitize_tsv_field(value: object) -> str:
    """Evita romper el TSV con tabuladores o saltos de linea."""
    text = "" if value is None else str(value)
    return text.replace("\t", " ").replace("\r", " ").replace("\n", " ").strip()


def _parse_date(value: object) -> datetime | None:
    if not value:
        return None
    text = str(value).strip()
    if not text:
        return None
    if text.endswith("Z"):
        text = text[:-1] + "+00:00"
    try:
        return datetime.fromisoformat(text)
    except ValueError:
        pass
    for fmt in ("%Y-%m-%d", "%d/%m/%Y"):
        try:
            return datetime.strptime(text[:10], fmt)
        except ValueError:
            continue
    return None


def _format_date(value: object) -> str:
    parsed = _parse_date(value)
    return parsed.strftime("%Y-%m-%d") if parsed else ""


def _task_due_date(task: dict) -> str:
    for key in (
        "fecha_vencimiento",
        "dueDateTime",
        "due_date",
        "fechaFin",
        "fecha_limite",
    ):
        formatted = _format_date(task.get(key))
        if formatted:
            return formatted
    return ""


def _project_estado_global(project: dict) -> str:
    progreso = int(round(float(project.get("porcentaje_avance") or 0)))
    total = int(project.get("total_tareas") or 0)
    completadas = int(project.get("completadas") or 0)
    if progreso >= 100 or (total > 0 and completadas >= total):
        return "Completado"
    return "En curso"


def _project_comentario_retraso(project: dict) -> str:
    vencidas = int(project.get("vencidas") or 0)
    if vencidas <= 0:
        return "Ninguno"
    if vencidas == 1:
        return "1 tarea vencida"
    return f"{vencidas} tareas vencidas"


def _project_fecha_objetivo(project: dict) -> str:
    fechas: list[datetime] = []
    for task in project.get("tareas") or []:
        parsed = _parse_date(_task_due_date(task) or None)
        if parsed:
            fechas.append(parsed)
    if not fechas:
        return ""
    return max(fechas).strftime("%Y-%m-%d")


def _task_estado(task: dict) -> str:
    porcentaje = int(round(float(task.get("porcentaje") or 0)))
    estado = _sanitize_tsv_field(task.get("estado")).lower()
    if porcentaje >= 100 or estado in {"completado", "completada", "done", "completed"}:
        return "100%"
    return f"{porcentaje}%"


def _project_row_values(project: dict) -> tuple[str, str, str, int, str, str]:
    progreso = int(round(float(project.get("porcentaje_avance") or 0)))
    return (
        _sanitize_tsv_field(project.get("plan_id")),
        _sanitize_tsv_field(project.get("proyecto")),
        _project_estado_global(project),
        progreso,
        _project_fecha_objetivo(project),
        _project_comentario_retraso(project),
    )


def projects_to_rows(projects: list[dict]) -> list[list[str]]:
    """Devuelve filas listas para Google Sheets (incluye encabezado)."""
    rows: list[list[str]] = [TSV_HEADER.copy()]

    for project in projects:
        id_proy, nombre, estado_global, progreso, fecha_objetivo, comentario = _project_row_values(project)
        tareas = project.get("tareas") or []

        if not tareas:
            rows.append([
                id_proy,
                nombre,
                estado_global,
                str(progreso),
                fecha_objetivo,
                comentario,
                "Sin tareas",
                "Sin asignar",
                fecha_objetivo,
                "0%",
            ])
            continue

        for task in tareas:
            responsable = _sanitize_tsv_field(task.get("asignados")) or "Sin asignar"
            rows.append([
                id_proy,
                nombre,
                estado_global,
                str(progreso),
                fecha_objetivo,
                comentario,
                _sanitize_tsv_field(task.get("tarea")) or "Sin nombre",
                responsable,
                _format_date(_task_due_date(task) or None),
                _task_estado(task),
            ])

    return rows


def projects_to_tsv(projects: list[dict]) -> str:
    rows = projects_to_rows(projects)
    return "\n".join("\t".join(row) for row in rows) + "\n"


def default_rtc_tsv_path() -> Path:
    OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
    return OUTPUT_DIR / "dashboard_rtc.tsv"


def export_rtc_tsv(projects: list[dict], output_path: Path | str | None = None) -> Path:
    """Genera el archivo TSV local que replica el formato publicado en Google Sheets."""
    path = Path(output_path) if output_path else default_rtc_tsv_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(projects_to_tsv(projects), encoding="utf-8")
    return path