""" 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 date, datetime, timezone 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 slot_greeting(slot: str) -> str: return { "compilation": "Good morning", "morning": "Good morning — you're at the office", "midday": "Midday check-in", "afternoon": "End of day wrap-up", }.get(slot, "Update") 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 today = date.today().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 calendar_events = [] try: if await is_caldav_configured(user_id): events = await list_events(user_id, start=today, end=today) calendar_events = [ f"{e.get('summary', 'Event')} at {e.get('dtstart', 'unknown time')}" for e in (events or []) ] except Exception: logger.warning("Failed to gather 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) -> 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": 4096, "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 _internal_system_prompt(profile_body: str) -> str: return ( "You are a personal briefing assistant. Your job is to give the user a clear, " "concise summary of their internal workload: tasks, calendar, and projects. " "Be direct and prioritised — lead with what's urgent. Use plain text with light " "markdown. Do not include weather or news.\n\n" + (f"User profile:\n{profile_body}\n" if profile_body else "") ) def _external_system_prompt() -> str: return ( "You are a briefing assistant for external information. Your job is to present " "selected news items and summarise any remaining RSS content. " "IMPORTANT: Weather is handled separately — do NOT include any weather section.\n\n" "Format each news item EXACTLY as:\n" "**[Headline text](source_url)**\n" "*Outlet Name · Day Month*\n" "One or two sentence summary.\n\n" "Present news items in the EXACT ORDER they are provided. Do not reorder them. " "After the news cards, add a brief paragraph for any remaining context." ) def _internal_user_prompt(data: dict, slot: str) -> str: lines = [f"Briefing slot: {slot}", f"Date: {data['date']}", ""] if data.get("unchanged_task_count", 0) > 0: lines.append( f"({data['unchanged_task_count']} tasks are unchanged since the last briefing " "— acknowledge briefly, do not list them.)" ) lines.append("") changed = data.get("changed_tasks") or data.get("overdue_tasks", []) if changed: lines.append(f"CHANGED/NEW TASKS ({len(changed)}):") lines.extend(f" - {t}" for t in changed) lines.append("") if data.get("due_today"): lines.append(f"DUE TODAY ({len(data['due_today'])}):") lines.extend(f" - {t}" for t in data["due_today"]) lines.append("") if data.get("high_priority"): lines.append("HIGH PRIORITY (in progress):") lines.extend(f" - {t}" for t in data["high_priority"]) lines.append("") if data["calendar_events"]: lines.append("CALENDAR TODAY:") lines.extend(f" - {e}" for e in data["calendar_events"]) lines.append("") if data["active_projects"]: lines.append(f"ACTIVE PROJECTS: {', '.join(data['active_projects'])}") 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}" def _external_user_prompt(data: dict, slot: str, temp_unit: str = "C") -> str: unit_sym = f"°{temp_unit}" lines = [f"Briefing slot: {slot}", ""] if data["weather"]: lines.append("WEATHER:") for loc in data["weather"]: lines.append(f" {loc['location_label']}:") for day in loc["days"][:3]: t_min = _format_temp(day["temp_min"], temp_unit) t_max = _format_temp(day["temp_max"], temp_unit) lines.append( f" {day['date']}: {day['description']}, " f"{t_min}–{t_max}{unit_sym}, {day['precip_mm']}mm rain" ) if loc["changes_since_last_fetch"]: lines.append(" FORECAST CHANGES:") lines.extend(f" - {c}" for c in loc["changes_since_last_fetch"]) lines.append("") if data["rss_items"]: lines.append(f"RSS DIGEST ({len(data['rss_items'])} items):") for item in data["rss_items"][:15]: lines.append(f" [{item.get('feed_title', 'Feed')}] {item['title']}") if item.get("content"): lines.append(f" {item['content'][:200]}") return "\n".join(lines) # ── 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.briefing_profile import get_profile_body 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_body, temp_unit = await asyncio.gather( get_profile_body(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")] # 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 today = internal_data["date"] 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": [], } internal_text, external_text = await asyncio.gather( _llm_synthesise( _internal_system_prompt(profile_body), _internal_user_prompt(internal_data_filtered, slot), model, ), _llm_synthesise( _external_system_prompt(), _external_user_prompt(external_data_filtered, slot, temp_unit), model, ), ) # ── Post-processing ───────────────────────────────────────────────────────── await upsert_task_snapshots(user_id, all_tasks) metadata: dict = {"rss_item_ids": rss_item_ids, "weather": weather_card} if not internal_text and not external_text: logger.warning("Briefing compilation produced no content for user %d slot %s", user_id, slot) return "", metadata greeting = slot_greeting(slot) parts = [f"**{greeting} — {today}**", ""] if internal_text: parts += ["## Your Day", "", internal_text, ""] if external_text: parts += ["## The World", "", external_text] return "\n".join(parts).strip(), 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 briefing assistant providing a {slot} update. Be brief — " "the user has already seen the morning briefing. Focus on what's changed or new." ) user_prompt = ( f"Slot: {slot}\n\n" + _internal_user_prompt(internal_data, slot) + "\n\n" + _external_user_prompt(external_data, slot, temp_unit) ) return await _llm_synthesise(system, user_prompt, model)