Spaces:
Sleeping
Sleeping
| from firebase_admin import credentials, firestore | |
| import firebase_admin | |
| from collections import Counter | |
| from datetime import datetime | |
| # # --- Inicialización de Firebase (asegúrate de que esto se ejecute una sola vez) --- | |
| if not firebase_admin._apps: | |
| try: | |
| firebase_cred = credentials.Certificate('archivos/config.json') | |
| firebase_admin.initialize_app(firebase_cred) | |
| print("Firebase Admin SDK inicializada exitosamente.") | |
| except Exception as e: | |
| print(f"Error al inicializar Firebase Admin SDK: {e}") | |
| exit() | |
| db = firestore.client(database_id='nowme') | |
| def sacar_estadistica_horas_filtrada(coleccion_nombre): | |
| """ | |
| Analiza la hora de creación de los documentos en una colección | |
| y devuelve un recuento de cuántos documentos se crearon en cada hora del día, | |
| excluyendo aquellos cuyo campo 'servidor' sea 'buzzword' y tengan un timestamp. | |
| Args: | |
| coleccion_nombre (str): El nombre de la colección a analizar. | |
| Returns: | |
| dict: Un diccionario donde las claves son las horas (0-23) y los | |
| valores son el número de documentos creados en esa hora. | |
| """ | |
| print(f"\n--- Sacando estadísticas de creación por hora en la colección '{coleccion_nombre}' ---") | |
| try: | |
| # 1. Obtener los documentos que cumplen el filtro de exclusión principal | |
| docs_filtrados = db.collection(coleccion_nombre).where( | |
| filter=firestore.FieldFilter('servidor', '!=', 'buzzword') | |
| ).stream() | |
| # 2. Extraer la hora de cada timestamp, filtrando en Python | |
| horas_creacion = [] | |
| for doc in docs_filtrados: | |
| doc_data = doc.to_dict() | |
| # Filtramos en el código de Python | |
| if 'timestamp_creacion' in doc_data and doc_data['timestamp_creacion'] is not None: | |
| timestamp = doc_data['timestamp_creacion'] | |
| hora = timestamp.hour | |
| horas_creacion.append(hora) | |
| if not horas_creacion: | |
| print("❌ No se encontraron timestamps de creación para analizar después de aplicar el filtro.") | |
| return {} | |
| # 3. Contar la frecuencia de cada hora | |
| conteo_horas = Counter(horas_creacion) | |
| resultados_ordenados = dict(sorted(conteo_horas.items())) | |
| print("✔️ Estadísticas de creación por hora calculadas.") | |
| return resultados_ordenados | |
| except Exception as e: | |
| print(f"❌ Error al sacar las estadísticas: {e}") | |
| return {} | |
| # --- Ejemplo de uso --- | |
| if __name__ == "__main__": | |
| # Asegúrate de que tu colección 'logs' exista y tenga documentos con los campos necesarios | |
| estadisticas_filtradas = sacar_estadistica_horas_filtrada('logs') | |
| if estadisticas_filtradas: | |
| print("\n--- Conteo de documentos creados por hora (sin 'buzzword') ---") | |
| for hora, cantidad in estadisticas_filtradas.items(): | |
| print(f" Hora {hora:02d}:00 - {hora:02d}:59 -> {cantidad} documentos") | |
| else: | |
| print("\nNo se pudieron generar las estadísticas.") |