bvandeusen
db092b113e
- rss_classifier: strip <think>...</think> blocks (qwen3 reasoning output) before JSON parse; use strict=False for control chars; bump timeout 30s→120s - briefing_pipeline: list_events returns dicts not Event objects — fix attribute access (.all_day/.start_dt/.title) to dict access - embeddings: guard upsert_note_embedding and semantic_search_notes against empty-string input to prevent Ollama embed 400 errors Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
477 lines
18 KiB
Python
477 lines
18 KiB
Python
"""
|
||
Briefing pipeline: parallel data gather + two-lane LLM synthesis.
|
||
|
||
Slot names: 'compilation' (4am), 'morning' (8am), 'midday' (12pm), 'afternoon' (4pm)
|
||
"""
|
||
|
||
import asyncio
|
||
import hashlib
|
||
import logging
|
||
from datetime import datetime, timezone
|
||
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
||
|
||
import httpx
|
||
|
||
from fabledassistant.models import async_session
|
||
|
||
from fabledassistant.config import Config
|
||
from fabledassistant.services.settings import get_setting
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
SLOT_NAMES = ("compilation", "morning", "midday", "afternoon")
|
||
|
||
|
||
|
||
def format_task(task: dict) -> str:
|
||
parts = [task.get("title", "Untitled")]
|
||
if task.get("status"):
|
||
parts.append(f"[{task['status'].replace('_', ' ').title()}]")
|
||
if task.get("due_date"):
|
||
parts.append(f"due {task['due_date']}")
|
||
if task.get("priority") and task["priority"] not in ("none", ""):
|
||
parts.append(f"priority: {task['priority']}")
|
||
return " — ".join(parts)
|
||
|
||
|
||
def compute_task_hash(task: dict) -> str:
|
||
"""Stable SHA-256 of the task's key change-detectable fields."""
|
||
key = "|".join([
|
||
str(task.get("status") or ""),
|
||
str(task.get("priority") or ""),
|
||
str(task.get("due_date") or ""),
|
||
str(task.get("title") or ""),
|
||
])
|
||
return hashlib.sha256(key.encode()).hexdigest()
|
||
|
||
|
||
async def split_changed_tasks(
|
||
user_id: int,
|
||
tasks: list[dict],
|
||
) -> tuple[list[dict], int]:
|
||
"""
|
||
Compare tasks against the briefing_task_snapshot table.
|
||
Returns (changed_tasks, unchanged_count).
|
||
changed_tasks includes new tasks (no snapshot row) and tasks whose hash differs.
|
||
"""
|
||
from sqlalchemy import text
|
||
|
||
if not tasks:
|
||
return [], 0
|
||
|
||
task_ids = [t["task_id"] for t in tasks if t.get("task_id")]
|
||
|
||
async with async_session() as session:
|
||
result = await session.execute(
|
||
text("""
|
||
SELECT task_id, snapshot_hash
|
||
FROM briefing_task_snapshot
|
||
WHERE user_id = :uid AND task_id = ANY(:ids)
|
||
""").bindparams(uid=user_id, ids=task_ids)
|
||
)
|
||
snapshots = {row.task_id: row.snapshot_hash for row in result}
|
||
|
||
changed = []
|
||
unchanged_count = 0
|
||
for task in tasks:
|
||
current_hash = compute_task_hash(task)
|
||
stored_hash = snapshots.get(task.get("task_id"))
|
||
if stored_hash is None or stored_hash != current_hash:
|
||
changed.append(task)
|
||
else:
|
||
unchanged_count += 1
|
||
|
||
return changed, unchanged_count
|
||
|
||
|
||
async def upsert_task_snapshots(user_id: int, tasks: list[dict]) -> None:
|
||
"""Upsert snapshot hashes for all tasks included in this briefing."""
|
||
from sqlalchemy import text
|
||
|
||
if not tasks:
|
||
return
|
||
|
||
now = datetime.now(timezone.utc)
|
||
async with async_session() as session:
|
||
for task in tasks:
|
||
task_id = task.get("task_id")
|
||
if not task_id:
|
||
continue
|
||
await session.execute(
|
||
text("""
|
||
INSERT INTO briefing_task_snapshot (user_id, task_id, snapshot_hash, last_briefed)
|
||
VALUES (:uid, :tid, :hash, :now)
|
||
ON CONFLICT (user_id, task_id)
|
||
DO UPDATE SET snapshot_hash = EXCLUDED.snapshot_hash,
|
||
last_briefed = EXCLUDED.last_briefed
|
||
""").bindparams(
|
||
uid=user_id,
|
||
tid=task_id,
|
||
hash=compute_task_hash(task),
|
||
now=now,
|
||
)
|
||
)
|
||
await session.commit()
|
||
|
||
|
||
# ── Internal data gather ──────────────────────────────────────────────────────
|
||
|
||
async def _gather_internal(user_id: int) -> dict:
|
||
"""Collect tasks, calendar events, and project data."""
|
||
from fabledassistant.services.notes import list_notes
|
||
from fabledassistant.services.projects import list_projects
|
||
from fabledassistant.services.caldav import is_caldav_configured, list_events
|
||
|
||
tz_name = await get_setting(user_id, "user_timezone") or "UTC"
|
||
try:
|
||
user_tz = ZoneInfo(tz_name)
|
||
except ZoneInfoNotFoundError:
|
||
user_tz = ZoneInfo("UTC")
|
||
|
||
today = datetime.now(user_tz).date().isoformat()
|
||
|
||
# Tasks: overdue, due today, high priority in-progress
|
||
all_tasks: list[dict] = []
|
||
try:
|
||
all_task_objs, _total = await list_notes(user_id, is_task=True, limit=100)
|
||
all_tasks = [
|
||
{
|
||
"task_id": t.id,
|
||
"title": t.title,
|
||
"status": t.status,
|
||
"due_date": t.due_date.isoformat() if t.due_date else None,
|
||
"priority": t.priority,
|
||
}
|
||
for t in all_task_objs
|
||
]
|
||
overdue = [
|
||
format_task(t) for t in all_tasks
|
||
if t.get("due_date") and t["due_date"] < today and t.get("status") != "done"
|
||
]
|
||
due_today = [
|
||
format_task(t) for t in all_tasks
|
||
if t.get("due_date") == today and t.get("status") != "done"
|
||
]
|
||
high_priority = [
|
||
format_task(t) for t in all_tasks
|
||
if t.get("priority") == "high" and t.get("status") not in ("done",) and
|
||
format_task(t) not in overdue and format_task(t) not in due_today
|
||
][:5]
|
||
except Exception:
|
||
logger.warning("Failed to gather tasks for briefing", exc_info=True)
|
||
overdue, due_today, high_priority = [], [], []
|
||
|
||
# Calendar events today — internal store
|
||
calendar_events: list[str] = []
|
||
try:
|
||
from fabledassistant.services.events import list_events as list_internal_events
|
||
today_date = datetime.now(user_tz).date()
|
||
day_start = datetime(today_date.year, today_date.month, today_date.day, 0, 0, 0, tzinfo=user_tz)
|
||
day_end = datetime(today_date.year, today_date.month, today_date.day, 23, 59, 59, tzinfo=user_tz)
|
||
internal_events = await list_internal_events(
|
||
user_id=user_id, date_from=day_start, date_to=day_end
|
||
)
|
||
for e in internal_events:
|
||
if e.get("all_day"):
|
||
time_str = "all day"
|
||
elif e.get("start_dt"):
|
||
from datetime import datetime as _dt
|
||
start = e["start_dt"]
|
||
if isinstance(start, str):
|
||
start = _dt.fromisoformat(start)
|
||
local_dt = start.astimezone(user_tz) if start.tzinfo else start.replace(tzinfo=timezone.utc).astimezone(user_tz)
|
||
time_str = local_dt.strftime("%-I:%M %p")
|
||
else:
|
||
time_str = "unknown time"
|
||
calendar_events.append(f"{e.get('title', 'Event')} at {time_str}")
|
||
except Exception:
|
||
logger.warning("Failed to gather internal calendar events for briefing", exc_info=True)
|
||
# Also pull CalDAV events (deduped)
|
||
try:
|
||
if await is_caldav_configured(user_id):
|
||
caldav_evs = await list_events(user_id, start=today, end=today)
|
||
for e in (caldav_evs or []):
|
||
summary = f"{e.get('summary', 'Event')} at {e.get('dtstart', 'unknown time')}"
|
||
if summary not in calendar_events:
|
||
calendar_events.append(summary)
|
||
except Exception:
|
||
logger.warning("Failed to gather CalDAV calendar events for briefing", exc_info=True)
|
||
|
||
# Projects: active projects
|
||
projects_summary = []
|
||
try:
|
||
projects = await list_projects(user_id)
|
||
for p in projects[:5]:
|
||
projects_summary.append(p.title or "Untitled project")
|
||
except Exception:
|
||
logger.warning("Failed to gather projects for briefing", exc_info=True)
|
||
|
||
return {
|
||
"date": today,
|
||
"overdue_tasks": overdue,
|
||
"due_today": due_today,
|
||
"high_priority": high_priority,
|
||
"calendar_events": calendar_events,
|
||
"active_projects": projects_summary,
|
||
"all_tasks_raw": all_tasks,
|
||
}
|
||
|
||
|
||
# ── External data gather ──────────────────────────────────────────────────────
|
||
|
||
async def _gather_external(user_id: int) -> dict:
|
||
"""Collect RSS items and weather."""
|
||
from fabledassistant.services.rss import get_recent_items
|
||
from fabledassistant.services.weather import get_cached_weather
|
||
|
||
rss_items, weather = await asyncio.gather(
|
||
get_recent_items(user_id, limit=20),
|
||
get_cached_weather(user_id),
|
||
return_exceptions=True,
|
||
)
|
||
return {
|
||
"rss_items": rss_items if not isinstance(rss_items, Exception) else [],
|
||
"weather": weather if not isinstance(weather, Exception) else [],
|
||
}
|
||
|
||
|
||
# ── LLM synthesis ─────────────────────────────────────────────────────────────
|
||
|
||
async def _llm_synthesise(system_prompt: str, user_prompt: str, model: str, num_ctx: int = 4096) -> str:
|
||
"""Single non-streaming LLM call. Returns the assistant's response text."""
|
||
payload = {
|
||
"model": model,
|
||
"messages": [
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": user_prompt},
|
||
],
|
||
"stream": False,
|
||
"options": {"num_ctx": num_ctx, "temperature": 0.4},
|
||
}
|
||
try:
|
||
async with httpx.AsyncClient(timeout=120.0) as client:
|
||
resp = await client.post(f"{Config.OLLAMA_URL}/api/chat", json=payload)
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
return data.get("message", {}).get("content", "").strip()
|
||
except Exception:
|
||
logger.warning("LLM synthesis failed", exc_info=True)
|
||
return ""
|
||
|
||
|
||
def _unified_system_prompt(profile_body: str) -> str:
|
||
return (
|
||
"You are a personal assistant delivering a daily briefing. "
|
||
"Speak naturally and conversationally — as if talking to the user, not writing a report. "
|
||
"Use no markdown: no headers, no bullet points, no bold, no lists. Write in flowing prose. "
|
||
"Weave together what matters today: mention the weather in a sentence, note any calendar "
|
||
"events or tasks due today, and briefly reference one or two noteworthy news stories. "
|
||
"Only mention projects if a task from one is specifically due today. "
|
||
"Be warm, concise, and human — aim for 3 to 5 sentences. "
|
||
"Future context like emails and messages will be added over time — keep the tone open and helpful.\n\n"
|
||
+ (f"User profile:\n{profile_body}\n" if profile_body else "")
|
||
)
|
||
|
||
|
||
def _unified_user_prompt(internal_data: dict, external_data: dict, slot: str, temp_unit: str = "C") -> str:
|
||
lines = [f"Date: {internal_data['date']}", f"Slot: {slot}", ""]
|
||
|
||
# Weather (brief — card handles detail)
|
||
weather = external_data.get("weather") or []
|
||
if weather:
|
||
loc = weather[0]
|
||
days = loc.get("days") or []
|
||
if days:
|
||
d = days[0]
|
||
t_min = _format_temp(d["temp_min"], temp_unit)
|
||
t_max = _format_temp(d["temp_max"], temp_unit)
|
||
unit_sym = f"°{temp_unit}"
|
||
lines.append(
|
||
f"WEATHER: {loc['location_label']} — {d['description']}, "
|
||
f"{t_min}–{t_max}{unit_sym}"
|
||
)
|
||
lines.append("")
|
||
|
||
# Today's calendar events
|
||
if internal_data.get("calendar_events"):
|
||
lines.append("TODAY'S EVENTS:")
|
||
lines.extend(f" - {e}" for e in internal_data["calendar_events"])
|
||
lines.append("")
|
||
|
||
# Tasks due today
|
||
if internal_data.get("due_today"):
|
||
lines.append("DUE TODAY:")
|
||
lines.extend(f" - {t}" for t in internal_data["due_today"])
|
||
lines.append("")
|
||
|
||
# Overdue tasks (brief mention only)
|
||
if internal_data.get("overdue_tasks"):
|
||
overdue = internal_data["overdue_tasks"]
|
||
lines.append(f"OVERDUE ({len(overdue)} task{'s' if len(overdue) != 1 else ''}):")
|
||
lines.extend(f" - {t}" for t in overdue[:3])
|
||
if len(overdue) > 3:
|
||
lines.append(f" (and {len(overdue) - 3} more)")
|
||
lines.append("")
|
||
|
||
# News highlights (top 3 with excerpts — right panel shows full list)
|
||
rss = external_data.get("rss_items") or []
|
||
if rss:
|
||
lines.append("NEWS HIGHLIGHTS (weave 1-2 into your briefing naturally; the full list is shown separately):")
|
||
for item in rss[:3]:
|
||
source = item.get("feed_title") or item.get("source") or "News"
|
||
title = item.get("title", "")
|
||
excerpt = (item.get("content") or item.get("snippet") or "")[:500].strip()
|
||
lines.append(f" [{source}] {title}")
|
||
if excerpt:
|
||
lines.append(f" {excerpt}")
|
||
lines.append("")
|
||
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _format_temp(value: float, unit: str) -> str:
|
||
"""Convert Celsius to the requested unit and format as an integer string."""
|
||
if unit == "F":
|
||
return f"{value * 9 / 5 + 32:.0f}"
|
||
return f"{value:.0f}"
|
||
|
||
|
||
|
||
|
||
# ── Main entry point ───────────────────────────────────────────────────────────
|
||
|
||
async def _get_temp_unit(user_id: int) -> str:
|
||
"""Read the user's preferred temperature unit from briefing_config ('C' or 'F')."""
|
||
import json
|
||
raw = await get_setting(user_id, "briefing_config", "{}")
|
||
try:
|
||
config = json.loads(raw) if isinstance(raw, str) else (raw or {})
|
||
unit = config.get("temp_unit", "C")
|
||
return unit if unit in ("C", "F") else "C"
|
||
except Exception:
|
||
return "C"
|
||
|
||
|
||
async def run_compilation(
|
||
user_id: int,
|
||
slot: str,
|
||
model: str | None = None,
|
||
) -> tuple[str, dict]:
|
||
"""
|
||
Run the full two-lane briefing pipeline for a user and slot.
|
||
Returns (briefing_text, metadata_dict) where metadata contains
|
||
weather card data and rss_item_ids for frontend rendering.
|
||
"""
|
||
if model is None:
|
||
model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL)
|
||
|
||
from fabledassistant.services.user_profile import build_profile_context
|
||
from fabledassistant.services.briefing_preferences import (
|
||
load_topic_preferences,
|
||
load_topic_reaction_scores,
|
||
score_and_filter_items,
|
||
)
|
||
from fabledassistant.services.weather import parse_weather_card_data, get_cached_weather_rows
|
||
|
||
profile_context, temp_unit = await asyncio.gather(
|
||
build_profile_context(user_id),
|
||
_get_temp_unit(user_id),
|
||
)
|
||
|
||
# ── Pre-processing ──────────────────────────────────────────────────────────
|
||
include_topics, exclude_topics = await load_topic_preferences(user_id)
|
||
topic_scores = await load_topic_reaction_scores(user_id)
|
||
|
||
# Parallel raw gather — weather rows fetched in same gather to avoid extra DB round-trip
|
||
internal_data, external_data, weather_rows = await asyncio.gather(
|
||
_gather_internal(user_id),
|
||
_gather_external(user_id),
|
||
get_cached_weather_rows(user_id),
|
||
)
|
||
|
||
# Task change detection
|
||
all_tasks = internal_data.get("all_tasks_raw", [])
|
||
changed_tasks, unchanged_count = await split_changed_tasks(user_id, all_tasks)
|
||
|
||
# RSS filtering
|
||
raw_rss = external_data.get("rss_items") or []
|
||
filtered_rss = score_and_filter_items(
|
||
raw_rss,
|
||
include_topics=include_topics,
|
||
exclude_topics=exclude_topics,
|
||
topic_scores=topic_scores,
|
||
max_items=10,
|
||
)
|
||
rss_item_ids = [item["id"] for item in filtered_rss if item.get("id")]
|
||
rss_items_meta = [
|
||
{
|
||
"id": item["id"],
|
||
"title": item.get("title", ""),
|
||
"url": item.get("url", ""),
|
||
"source": item.get("feed_title", ""),
|
||
"snippet": (item.get("content") or "")[:300],
|
||
"published_at": item.get("published_at"),
|
||
}
|
||
for item in filtered_rss
|
||
if item.get("id")
|
||
]
|
||
|
||
# Weather staleness gate — returns None if data is >24h old
|
||
weather_card = parse_weather_card_data(weather_rows[0], temp_unit) if weather_rows else None
|
||
|
||
# ── LLM Synthesis ──────────────────────────────────────────────────────────
|
||
# Build filtered internal data with only changed tasks
|
||
internal_data_filtered = dict(internal_data)
|
||
internal_data_filtered["unchanged_task_count"] = unchanged_count
|
||
internal_data_filtered["changed_tasks"] = [format_task(t) for t in changed_tasks]
|
||
|
||
# Build filtered external data (suppress weather prose — card handles it)
|
||
external_data_filtered = {
|
||
"rss_items": filtered_rss,
|
||
"weather": [],
|
||
}
|
||
|
||
briefing_text = await _llm_synthesise(
|
||
_unified_system_prompt(profile_context),
|
||
_unified_user_prompt(internal_data_filtered, external_data_filtered, slot, temp_unit),
|
||
model,
|
||
num_ctx=8192,
|
||
)
|
||
|
||
# ── Post-processing ─────────────────────────────────────────────────────────
|
||
await upsert_task_snapshots(user_id, all_tasks)
|
||
|
||
metadata: dict = {"rss_item_ids": rss_item_ids, "rss_items": rss_items_meta, "weather": weather_card}
|
||
|
||
if not briefing_text:
|
||
logger.warning("Briefing compilation produced no content for user %d slot %s", user_id, slot)
|
||
return "", metadata
|
||
|
||
return briefing_text, metadata
|
||
|
||
|
||
async def run_slot_injection(user_id: int, slot: str, model: str | None = None) -> str:
|
||
"""
|
||
Lighter update for 8am/12pm/4pm — gathers fresh data and produces a slot-specific
|
||
update prompt. Returns the text to inject as a new user→assistant exchange.
|
||
"""
|
||
if model is None:
|
||
model = await get_setting(user_id, "default_model", Config.OLLAMA_MODEL)
|
||
|
||
internal_data, external_data, temp_unit = await asyncio.gather(
|
||
_gather_internal(user_id),
|
||
_gather_external(user_id),
|
||
_get_temp_unit(user_id),
|
||
)
|
||
|
||
system = (
|
||
f"You are a personal assistant giving a brief {slot} check-in. "
|
||
"The user already had their morning briefing — focus only on what's changed or newly relevant. "
|
||
"Speak naturally in 2-3 sentences, no markdown formatting, no headers or bullet points."
|
||
)
|
||
return await _llm_synthesise(
|
||
system,
|
||
_unified_user_prompt(internal_data, external_data, slot, temp_unit),
|
||
model,
|
||
)
|