""" Monitor de Microsoft Planner. Fuentes de datos (configurar en .env): DATA_SOURCE=powerautomate -> Puente HTTP de Power Automate (sin Azure) DATA_SOURCE=graph -> Microsoft Graph API directa Uso: python script.py # Extrae datos y genera Excel python run_tablero.py # Servidor local del tablero HTML v7 python sync_dashboard.py # Alternativa: Google Sheet del tablero HTML v7 python script.py --test # Prueba la conexion python script.py --probe # Solo modo graph: prueba endpoints HTTP """ import argparse import sys from config import DATA_SOURCE, AUTH_MODE, GRAPH_API_VERSION, data_source_label, validate_config from export.excel_exporter import default_output_path, export_to_excel from services.data_factory import create_data_service def test_connection() -> bool: missing = validate_config() if missing: print("[ERROR] Faltan variables en .env:") for var in missing: print(f" - {var}") if DATA_SOURCE == "powerautomate": print("\nConfigura POWER_AUTOMATE_URL con la URL del trigger HTTP.") print("Guia: docs/POWER_AUTOMATE_FLUJO.md") print("Expresiones: docs/POWER_AUTOMATE_EXPRESIONES.md") print("Diagnostico: python scripts/diagnostico_pa.py") else: print("\nCopia .env.example a .env y completa los valores de Azure.") return False print(f"[OK] Configuracion valida.") print(f" Fuente de datos: {data_source_label()}") print("[...] Conectando...") try: service = create_data_service() if DATA_SOURCE == "powerautomate": result = service.test_connection() print(f"[OK] Puente Power Automate conectado.") print(f" {result['mensaje']}") print(f" -> {result['proyectos']} proyecto(s), {result['tareas']} tarea(s)") for name in result["muestra"]: print(f" - {name}") if result["proyectos"] > 5: print(f" ... y {result['proyectos'] - 5} mas") else: from services.planner_service import PlannerService assert isinstance(service, PlannerService) if AUTH_MODE == "delegated": plans = service.get_my_plans() teams = service.get_my_teams() print("[OK] Conexion exitosa con Microsoft Graph.") print(f" GET https://graph.microsoft.com/{GRAPH_API_VERSION}/me/planner/plans") print(f" -> {len(plans)} plan(es)") print(f" GET https://graph.microsoft.com/{GRAPH_API_VERSION}/me/joinedTeams") print(f" -> {len(teams)} equipo(s)") for plan in plans[:5]: print(f" - {plan.get('title', plan['id'])}") else: groups = service.get_groups_with_planner() print(f"[OK] Conexion exitosa. {len(groups)} grupos con Teams.") for group in groups[:5]: print(f" - {group.get('displayName', group['id'])}") return True except Exception as e: print(f"[ERROR] Error de conexion: {e}") if DATA_SOURCE == "powerautomate": print("\nAyuda:") print(" python scripts/diagnostico_pa.py") print(" docs/POWER_AUTOMATE_EXPRESIONES.md") return False def probe_graph_endpoints() -> bool: """Solo disponible en modo graph.""" if DATA_SOURCE != "graph": print("[AVISO] --probe solo aplica cuando DATA_SOURCE=graph") return False from auth.graph_client import create_graph_client from auth.graph_endpoints import ( APPLICATION_PROBE_SEQUENCE, DELEGATED_PROBE_SEQUENCE, plan_buckets, plan_tasks, ) missing = validate_config() if missing: print("[ERROR] Faltan variables en .env:") for var in missing: print(f" - {var}") return False print("\nPatron HTTP de Microsoft Graph:") print(" {HTTP method} https://graph.microsoft.com/{version}/{resource}?{query-parameters}") print(f" Version: {GRAPH_API_VERSION} | Modo: {AUTH_MODE}\n") client = create_graph_client() sequence = ( DELEGATED_PROBE_SEQUENCE if AUTH_MODE == "delegated" else APPLICATION_PROBE_SEQUENCE ) all_ok = True first_plan_id = None for endpoint in sequence: result = client.probe(endpoint) status = "OK" if result["ok"] else "FALLO" print(f"[{status}] {result['method']} {result['url']}") print(f" {result['description']}") if result["ok"]: print(f" Respuesta: {result['items']} elemento(s)") if endpoint.resource == "me/planner/plans" and result.get("sample"): first_plan_id = result["sample"][0].get("id") else: all_ok = False print(f" Error {result.get('status')}: {result.get('error')}") print() if first_plan_id: for extra in [plan_tasks(first_plan_id), plan_buckets(first_plan_id)]: result = client.probe(extra) status = "OK" if result["ok"] else "FALLO" print(f"[{status}] {result['method']} {result['url']}") if not result["ok"]: all_ok = False print(f" Error: {result.get('error')}") print() return all_ok def extract_and_export() -> None: print("\n[...] Obteniendo proyectos y tareas...") service = create_data_service() projects = service.get_all_projects() if not projects: print("[AVISO] No se encontraron proyectos.") if DATA_SOURCE == "powerautomate": print(" Revisa que el flujo de Power Automate este activo y devuelva JSON.") return total_tasks = sum(p["total_tareas"] for p in projects) print(f"[OK] {len(projects)} proyectos, {total_tasks} tareas obtenidas.") output_path = default_output_path() export_to_excel(projects, output_path) print(f"[OK] Excel generado: {output_path}") def main() -> None: parser = argparse.ArgumentParser(description="Monitor de Microsoft Planner") parser.add_argument("--test", action="store_true", help="Solo probar conexion") parser.add_argument("--probe", action="store_true", help="Probar endpoints Graph (solo modo graph)") args = parser.parse_args() print("=" * 50) print(" Monitor Microsoft Planner") print(f" Fuente: {data_source_label()}") print("=" * 50) if args.probe: ok = probe_graph_endpoints() sys.exit(0 if ok else 1) if not test_connection(): sys.exit(1) if args.test: print("\n[OK] Prueba de conexion completada.") return extract_and_export() print("\n[OK] Proceso completado.") print(" Tablero HTML v7 (red): python run_tablero.py") print(" Dashboard Streamlit: python -m streamlit run dashboard/app.py") if __name__ == "__main__": main()