Spaces:
Running
Running
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.") |