pm-dashboard/airtable_client.py

from pyairtable import Api
import config


def _collaborator(value):
    if isinstance(value, dict):
        return {
            "id": value.get("id", ""),
            "name": value.get("name") or value.get("email") or "",
            "email": value.get("email", ""),
        }
    return None


def _collaborators(values):
    if not values:
        return []
    return [c for c in (_collaborator(v) for v in values) if c]


class AirtableClient:
    def __init__(self):
        self.api = Api(config.AIRTABLE_TOKEN)
        self.projects = self.api.table(config.AIRTABLE_BASE_ID, config.PROJECTS_TABLE)
        self.tasks = self.api.table(config.AIRTABLE_BASE_ID, config.TASKS_TABLE)
        self.clients = self.api.table(config.AIRTABLE_BASE_ID, config.CLIENTS_TABLE)
        self.areas = self.api.table(config.AIRTABLE_BASE_ID, config.AREAS_TABLE)
        self.products = self.api.table(config.AIRTABLE_BASE_ID, config.PRODUCTS_TABLE)

    def get_projects(self) -> list[dict]:
        records = self.projects.all(fields=[
            "Project", "Tags", "Status", "Client", "Project lead",
            "Working team", "Kickoff date", "Due date", "Budget",
            "Internal", "Code", "📝 Tasks",
        ])
        result = []
        for r in records:
            f = r["fields"]
            project_lead = _collaborator(f.get("Project lead"))
            working_team = _collaborators(f.get("Working team"))
            result.append({
                "id": r["id"],
                "project": f.get("Project", ""),
                "tags": f.get("Tags", []),
                "status": f.get("Status", ""),
                "client_ids": f.get("Client", []),
                "project_lead": project_lead,
                "project_lead_name": project_lead["name"] if project_lead else "",
                "working_team": working_team,
                "working_team_names": [c["name"] for c in working_team],
                "kickoff_date": f.get("Kickoff date"),
                "due_date": f.get("Due date"),
                "budget": f.get("Budget", 0),
                "internal": f.get("Internal", False),
                "code": f.get("Code", ""),
                "task_ids": f.get("📝 Tasks", []),
            })
        return result

    def get_tasks(self) -> list[dict]:
        records = self.tasks.all(fields=[
            "Task", "Project", "Area", "Status Task", "Assigned to",
            "Kick off", "Due date", "Days to complete", "Horas estimadas",
            "Prioridad", "INTERNO/EXTERNO", "Recurrencia",
            "Client (from Project)",
        ])
        result = []
        for r in records:
            f = r["fields"]
            assigned_to = _collaborators(f.get("Assigned to"))
            result.append({
                "id": r["id"],
                "task": f.get("Task", ""),
                "project_ids": f.get("Project", []),
                "area_ids": f.get("Area", []),
                "status": f.get("Status Task", ""),
                "assigned_to": assigned_to,
                "assigned_to_names": [c["name"] for c in assigned_to],
                "kick_off": f.get("Kick off"),
                "due_date": f.get("Due date"),
                "days_to_complete": f.get("Days to complete"),
                "horas_estimadas": f.get("Horas estimadas", 0),
                "prioridad": f.get("Prioridad", ""),
                "interno_externo": f.get("INTERNO/EXTERNO", ""),
                "recurrencia": f.get("Recurrencia", ""),
                "client_names": f.get("Client (from Project)", []),
            })
        return result

    def get_clients(self) -> list[dict]:
        records = self.clients.all(fields=["Display Name", "Company Name", "Projects", "Tasks"])
        result = []
        for r in records:
            f = r["fields"]
            result.append({
                "id": r["id"],
                "display_name": f.get("Display Name", ""),
                "company_name": f.get("Company Name", ""),
                "project_ids": f.get("Projects", []),
                "task_ids": f.get("Tasks", []),
            })
        return result

    def get_areas(self) -> list[dict]:
        records = self.areas.all(fields=["Área", "Descripción", "Tasks"])
        result = []
        for r in records:
            f = r["fields"]
            result.append({
                "id": r["id"],
                "area": f.get("Área", ""),
                "descripcion": f.get("Descripción", ""),
                "task_ids": f.get("Tasks", []),
            })
        return result

    @staticmethod
    def get_known_collaborators(projects: list[dict], tasks: list[dict]) -> dict:
        """Colaboradores conocidos por haber aparecido en algún proyecto o tarea, indexados por email."""
        known = {}
        for p in projects:
            candidates = list(p["working_team"])
            if p["project_lead"]:
                candidates.append(p["project_lead"])
            for c in candidates:
                if c["email"]:
                    known[c["email"]] = c
        for t in tasks:
            for c in t["assigned_to"]:
                if c["email"]:
                    known[c["email"]] = c
        return known

    # ── Escritura ─────────────────────────────────────────────────────────── #

    def create_project(self, fields: dict) -> dict:
        return self.projects.create(fields, typecast=True)

    def update_project(self, record_id: str, fields: dict) -> dict:
        return self.projects.update(record_id, fields, typecast=True)

    def delete_project(self, record_id: str) -> None:
        self.projects.delete(record_id)

    def create_task(self, fields: dict) -> dict:
        return self.tasks.create(fields, typecast=True)

    def update_task(self, record_id: str, fields: dict) -> dict:
        return self.tasks.update(record_id, fields, typecast=True)

    def delete_task(self, record_id: str) -> None:
        self.tasks.delete(record_id)