File size: 4,345 Bytes
311871d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | """Background worker: sends periodic digest emails for digest-type subscriptions."""
from datetime import datetime, timezone
from config import SUPABASE_SERVICE_ROLE_KEY, SUPABASE_URL
from services.email_service import build_criteria_summary, send_alert_email
from services.observability import setup_logger
from services.supabase_service import execute_tool_rpc
logger = setup_logger(__name__)
def _get_service_client():
"""Lazy-import Supabase service client to avoid circular imports at module level."""
from supabase import create_client
return create_client(SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY)
async def _compile_market_digest(market: str) -> str:
"""Compile a market digest summary using existing RPCs."""
snapshot = await execute_tool_rpc("get_market_snapshot", {
"p_market": market,
"p_start_date": datetime.now(timezone.utc).strftime("%Y-%m-%d"),
"p_end_date": datetime.now(timezone.utc).strftime("%Y-%m-%d"),
})
averages = await execute_tool_rpc("get_market_averages", {"market_param": market})
spikes = await execute_tool_rpc("get_spike_alerts", {
"p_market": market,
"threshold_param": 25,
"days_param": 7,
"p_limit": 3,
})
body = f"<h3>📊 Market Digest — {market}</h3>"
body += "<h4>Snapshot</h4>"
snap_data = snapshot.get("data", snapshot)
if isinstance(snap_data, list) and snap_data:
snap_data = snap_data[0]
if isinstance(snap_data, dict):
body += f"<p>Active Properties: {snap_data.get('active_count', '?')}, "
body += f"Avg Rate: {snap_data.get('avg_rate', '?')}/night, "
body += f"Availability: {snap_data.get('availability_pct', '?')}%</p>"
avg_data = averages.get("data", averages)
if isinstance(avg_data, list) and avg_data:
body += "<h4>Market Averages</h4><ul>"
for item in avg_data[:5]:
body += f"<li>{item.get('property_name', '?')}: {item.get('avg_rate', '?')}/night</li>"
body += "</ul>"
spike_data = spikes.get("data", [])
if spike_data:
body += f"<h4>Notable Spikes (≥25%)</h4><ul>"
for s in spike_data[:3]:
body += f"<li>{s.get('property_name', '?')}: {s.get('pct_change', '?')}%</li>"
body += "</ul>"
else:
body += "<p>No significant spikes in the last 7 days.</p>"
return body
async def send_pending_digests() -> dict:
"""Fetch digest subscriptions due for send, compile market summary, send email."""
client = _get_service_client()
try:
result = (
client.table("alert_subscriptions")
.select("id, email, criteria, unsubscribe_token, last_fired_at")
.eq("status", "active")
.eq("confirmed", True)
.eq("criteria_type", "digest")
.execute()
)
except Exception as exc:
logger.error(f"[digest] failed to fetch digest subs: {exc}")
return {"status": "error", "message": str(exc)}
subs = result.data or []
sent_count = 0
for sub in subs:
criteria = sub["criteria"]
frequency = criteria.get("frequency", "weekly")
if not _is_due(sub.get("last_fired_at"), frequency):
continue
market = criteria.get("market", "All Markets")
digest_body = await _compile_market_digest(market)
summary = build_criteria_summary("digest", criteria)
subject = f"📊 {summary}"
await send_alert_email(sub["email"], subject, digest_body, sub["unsubscribe_token"])
client.table("alert_subscriptions").update({
"last_fired_at": datetime.now(timezone.utc).isoformat(),
"last_checked_at": datetime.now(timezone.utc).isoformat(),
}).eq("id", sub["id"]).execute()
sent_count += 1
logger.info(f"[digest] sent digest for sub {sub['id']}")
return {"status": "success", "sent": sent_count}
def _is_due(last_fired_at: str | None, frequency: str) -> bool:
"""Check if a digest subscription is due for sending."""
if not last_fired_at:
return True
fired_dt = datetime.fromisoformat(last_fired_at.replace("Z", "+00:00"))
elapsed = datetime.now(timezone.utc) - fired_dt
if frequency == "daily":
return elapsed.total_seconds() >= 23 * 3600
return elapsed.days >= 6
|