# /// script # requires-python = ">=3.12" # dependencies = ["google-cloud-firestore>=2.0", "pyyaml>=6.0"] # /// """Check recent notifications in Firestore for a phone number. Usage: uv run utils/check_notifications_firestore.py uv run utils/check_notifications_firestore.py --hours 24 """ import sys import time import yaml from google.cloud.firestore import Client _SECONDS_PER_HOUR = 3600 _DEFAULT_WINDOW_HOURS = 48 def main() -> None: if len(sys.argv) < 2: print(f"Usage: {sys.argv[0]} [--hours N]") sys.exit(1) phone = sys.argv[1] window_hours = _DEFAULT_WINDOW_HOURS if "--hours" in sys.argv: idx = sys.argv.index("--hours") window_hours = float(sys.argv[idx + 1]) with open("config.yaml") as f: cfg = yaml.safe_load(f) db = Client( project=cfg["google_cloud_project"], database=cfg["firestore_db"], ) collection_path = cfg["notifications_collection_path"] doc_ref = db.collection(collection_path).document(phone) doc = doc_ref.get() if not doc.exists: print(f"📭 No notifications found for {phone}") sys.exit(0) data = doc.to_dict() or {} all_notifications = data.get("notificaciones", []) if not all_notifications: print(f"📭 No notifications found for {phone}") sys.exit(0) cutoff = time.time() - (window_hours * _SECONDS_PER_HOUR) def _ts(n: dict) -> float: return n.get("timestamp_creacion", n.get("timestampCreacion", 0)) recent = [n for n in all_notifications if _ts(n) >= cutoff] recent.sort(key=_ts, reverse=True) if not recent: print( f"📭 No notifications within the last" f" {window_hours:.0f}h for {phone}" ) sys.exit(0) print( f"🔔 {len(recent)} notification(s) for {phone}" f" (last {window_hours:.0f}h)\n" ) now = time.time() for i, n in enumerate(recent, 1): ts = _ts(n) ago = _format_time_ago(now, ts) categoria = n.get("parametros", {}).get( "notification_po_Categoria", "" ) texto = n.get("texto", "") print(f" [{i}] {ago}") print(f" ID: {n.get('id_notificacion', '?')}") if categoria: print(f" Category: {categoria}") print(f" {texto[:120]}{'…' if len(texto) > 120 else ''}") print() def _format_time_ago(now: float, ts: float) -> str: diff = max(now - ts, 0) minutes = int(diff // 60) hours = int(diff // _SECONDS_PER_HOUR) if minutes < 1: return "justo ahora" if minutes < 60: return f"hace {minutes} min" if hours < 24: return f"hace {hours}h" days = hours // 24 return f"hace {days}d" if __name__ == "__main__": main()