Spaces:
Sleeping
Sleeping
File size: 2,074 Bytes
9af2b22 | 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 | """Derive schedule risk signals from entries + daily check-ins.
Shared by human plan routes and agent context — server-owned, not LLM.
"""
from __future__ import annotations
from datetime import date, datetime, timedelta, timezone
from typing import Any
RISK_TAGS = frozenset(
{
"urge",
"court",
"comparison",
"corn",
"daydream",
"bully",
"family",
"shame",
"rerun",
}
)
LAST_HOUR_HIGH_TAGS = frozenset({"urge", "court", "corn", "rerun"})
def risk_from_stores(
entry_store: Any,
daily_store: Any,
*,
now: datetime | None = None,
target_day: date | None = None,
) -> dict[str, Any]:
"""Return risk_1h, triggers_yesterday, and last_hour_high flags."""
now = now or datetime.now(timezone.utc)
target = target_day or now.date()
hour_ago = now - timedelta(hours=1)
yesterday = (target - timedelta(days=1)).isoformat()
tags_1h: list[str] = []
triggers_y: list[str] = []
for entry in entry_store._load():
ts = entry.ts
if ts.tzinfo is None:
ts = ts.replace(tzinfo=timezone.utc)
entry_tags = set(entry.tags) | set(entry.emotions)
hit = sorted(entry_tags & RISK_TAGS)
if ts >= hour_ago:
tags_1h.extend(hit)
if entry.ts.date().isoformat() == yesterday:
triggers_y.extend(hit)
y_daily = daily_store.get(date.fromisoformat(yesterday))
if y_daily:
if y_daily.court == "court":
triggers_y.append("court")
if y_daily.corn_sessions >= 2:
triggers_y.append("corn")
if y_daily.daydream == "fc":
triggers_y.append("daydream")
tags_1h = sorted(set(tags_1h))
triggers_y = sorted(set(triggers_y))
last_hour_high = bool(LAST_HOUR_HIGH_TAGS & set(tags_1h))
return {
"risk_1h_score": float(len(tags_1h)),
"risk_1h_tags": tags_1h,
"triggers_yesterday": triggers_y,
"yesterday_trigger": len(triggers_y) > 0,
"last_hour_high": last_hour_high,
}
|