José Manuel Gómez
663f058c00
Streamlit dashboard con pestañas Proyectos/Tareas/Clientes/Áreas leyendo directamente de la base de Airtable de gestión de proyectos. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
106 lines
4.0 KiB
Python
106 lines
4.0 KiB
Python
from pyairtable import Api
|
|
import config
|
|
|
|
|
|
def _collaborator_name(value):
|
|
if isinstance(value, dict):
|
|
return value.get("name") or value.get("email") or ""
|
|
return ""
|
|
|
|
|
|
def _collaborator_names(values):
|
|
if not values:
|
|
return []
|
|
return [_collaborator_name(v) for v in values]
|
|
|
|
|
|
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"]
|
|
result.append({
|
|
"id": r["id"],
|
|
"project": f.get("Project", ""),
|
|
"tags": f.get("Tags", []),
|
|
"status": f.get("Status", ""),
|
|
"client_ids": f.get("Client", []),
|
|
"project_lead": _collaborator_name(f.get("Project lead")),
|
|
"working_team": _collaborator_names(f.get("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"]
|
|
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": _collaborator_names(f.get("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
|