121 lines
3.3 KiB
Python
121 lines
3.3 KiB
Python
# /// 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 <phone>
|
|
uv run utils/check_notifications_firestore.py <phone> --hours 24
|
|
"""
|
|
|
|
import sys
|
|
import time
|
|
from datetime import datetime
|
|
from typing import Any
|
|
|
|
import yaml
|
|
from google.cloud.firestore import Client
|
|
|
|
_SECONDS_PER_HOUR = 3600
|
|
_DEFAULT_WINDOW_HOURS = 48
|
|
|
|
|
|
def _extract_ts(n: dict[str, Any]) -> float:
|
|
"""Return the creation timestamp of a notification as epoch seconds."""
|
|
raw = n.get("timestamp_creacion", n.get("timestampCreacion", 0))
|
|
if isinstance(raw, (int, float)):
|
|
return float(raw)
|
|
if isinstance(raw, datetime):
|
|
return raw.timestamp()
|
|
if isinstance(raw, str):
|
|
try:
|
|
return float(raw)
|
|
except ValueError:
|
|
return 0.0
|
|
return 0.0
|
|
|
|
|
|
def main() -> None:
|
|
if len(sys.argv) < 2:
|
|
print(f"Usage: {sys.argv[0]} <phone> [--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)
|
|
|
|
recent = [n for n in all_notifications if _extract_ts(n) >= cutoff]
|
|
recent.sort(key=_extract_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 = _extract_ts(n)
|
|
ago = _format_time_ago(now, ts)
|
|
params = n.get("parameters", n.get("parametros", {}))
|
|
categoria = params.get("notification_po_Categoria", "")
|
|
texto = n.get("text", n.get("texto", ""))
|
|
print(f" [{i}] {ago}")
|
|
print(f" ID: {n.get('notificationId', 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()
|