62 lines
2.5 KiB
Python
62 lines
2.5 KiB
Python
import os, requests, datetime
|
|
from typing import Dict, Any, Tuple
|
|
|
|
INTAKE_API_URL = os.getenv("INTAKE_API_URL", "http://intake:7080")
|
|
DEFAULT_SESSION_ID = os.getenv("DEFAULT_SESSION_ID", "default")
|
|
LOCAL_TZ_LABEL = os.getenv("LOCAL_TZ_LABEL", "America/New_York")
|
|
|
|
def fetch_intake_context(session_id: str | None) -> Dict[str, Any]:
|
|
sid = session_id or DEFAULT_SESSION_ID
|
|
try:
|
|
r = requests.get(f"{INTAKE_API_URL}/summaries", params={"session_id": sid}, timeout=4)
|
|
r.raise_for_status()
|
|
data = r.json() or {}
|
|
except Exception:
|
|
data = {}
|
|
# Normalize expected fields
|
|
return {
|
|
"summary_text": data.get("summary_text", ""),
|
|
"last_message_ts": data.get("last_message_ts"), # ISO8601 or None
|
|
"session_id": sid,
|
|
"exchange_count": data.get("exchange_count", 0),
|
|
}
|
|
|
|
def build_temporal_snapshot(last_ts_iso: str | None) -> Dict[str, Any]:
|
|
now = datetime.datetime.now() # system local time
|
|
now_str = now.strftime("%A, %b %-d, %Y, %H:%M")
|
|
elapsed_str = "unknown"
|
|
if last_ts_iso:
|
|
try:
|
|
# parse ISO (with/without tz). If it has a timezone offset, fromisoformat handles it.
|
|
last = datetime.datetime.fromisoformat(last_ts_iso.replace("Z", "+00:00"))
|
|
delta = now - last.replace(tzinfo=None)
|
|
mins = int(delta.total_seconds() // 60)
|
|
if mins < 60:
|
|
elapsed_str = f"{mins} min"
|
|
else:
|
|
hrs = mins // 60
|
|
rem = mins % 60
|
|
elapsed_str = f"{hrs} hr {rem} min"
|
|
except Exception:
|
|
pass
|
|
return {
|
|
"local_time_label": LOCAL_TZ_LABEL,
|
|
"local_time_now": now_str,
|
|
"elapsed_since_last": elapsed_str,
|
|
}
|
|
|
|
def get_intake_block(session_id: str | None) -> Tuple[str, Dict[str, Any]]:
|
|
ctx = fetch_intake_context(session_id)
|
|
temporal = build_temporal_snapshot(ctx.get("last_message_ts"))
|
|
# A short, ready-to-inject block for prompts:
|
|
intake_block = (
|
|
f"[Intake]\n"
|
|
f"Session: {ctx['session_id']}\n"
|
|
f"Exchanges: {ctx['exchange_count']}\n"
|
|
f"Local time ({temporal['local_time_label']}): {temporal['local_time_now']}\n"
|
|
f"Elapsed since last: {temporal['elapsed_since_last']}\n"
|
|
f"Recent summary: {ctx['summary_text'] or '(none)'}\n"
|
|
)
|
|
# Also return raw dicts if you want to use fields programmatically
|
|
return intake_block, {"intake": ctx, "temporal": temporal}
|