Add support for prev notification collection structure
All checks were successful
CI / ci (pull_request) Successful in 19s
All checks were successful
CI / ci (pull_request) Successful in 19s
This commit is contained in:
@@ -11,6 +11,8 @@ Usage:
|
||||
|
||||
import sys
|
||||
import time
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
import yaml
|
||||
from google.cloud.firestore import Client
|
||||
@@ -19,6 +21,21 @@ _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]")
|
||||
@@ -55,11 +72,8 @@ def main() -> None:
|
||||
|
||||
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)
|
||||
recent = [n for n in all_notifications if _extract_ts(n) >= cutoff]
|
||||
recent.sort(key=_extract_ts, reverse=True)
|
||||
|
||||
if not recent:
|
||||
print(
|
||||
@@ -74,14 +88,13 @@ def main() -> None:
|
||||
)
|
||||
now = time.time()
|
||||
for i, n in enumerate(recent, 1):
|
||||
ts = _ts(n)
|
||||
ts = _extract_ts(n)
|
||||
ago = _format_time_ago(now, ts)
|
||||
categoria = n.get("parametros", {}).get(
|
||||
"notification_po_Categoria", ""
|
||||
)
|
||||
texto = n.get("texto", "")
|
||||
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('id_notificacion', '?')}")
|
||||
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 ''}")
|
||||
|
||||
Reference in New Issue
Block a user