"""Evening daily recap: LLM text + contextual character image posted to chat.""" from __future__ import annotations import json import logging from datetime import date, datetime, time, timezone from typing import Any from zoneinfo import ZoneInfo from sqlalchemy import select from sqlalchemy.orm import Session from app.character.service import CharacterService from app.db.models import ChatSession, PomodoroSession, Reminder, SessionSummary from app.fitness.service import FitnessService from app.homelab.comfyui import random_rofl_pose from app.homelab.image_gen import generate_image from app.llm.client import LLMClient from app.pomodoro.cycle import PHASE_WORK from app.projects.structuring import strip_markdown_json logger = logging.getLogger(__name__) DEFAULT_SCENE = "full_body, standing, smile, looking_at_viewer" LLM_TIMEOUT_SEC = 90 def _day_bounds(day: date, tz: ZoneInfo) -> tuple[datetime, datetime]: start = datetime.combine(day, time.min, tzinfo=tz) end = datetime.combine(day, time.max, tzinfo=tz) return start, end def collect_day_context(db: Session, user_id: int, day: date, tz: ZoneInfo) -> dict[str, Any]: fitness = FitnessService(db, user_id) summary = fitness.get_daily_summary(day) totals = summary.get("totals") or {} targets = summary.get("targets") or {} meals = summary.get("meals") or [] workouts = summary.get("workouts") or [] steps_total = summary.get("steps_total") or 0 start_dt, end_dt = _day_bounds(day, tz) pomodoro_rows = db.scalars( select(PomodoroSession).where( PomodoroSession.user_id == user_id, PomodoroSession.phase == PHASE_WORK, PomodoroSession.completed.is_(True), PomodoroSession.finished_at >= start_dt, PomodoroSession.finished_at <= end_dt, ) ).all() pomodoro_tasks = [r.task_note.strip() for r in pomodoro_rows if (r.task_note or "").strip()] reminder_rows = db.scalars( select(Reminder).where( Reminder.user_id == user_id, Reminder.completed_at.is_not(None), Reminder.completed_at >= start_dt, Reminder.completed_at <= end_dt, ) ).all() completed_reminders = [r.title for r in reminder_rows] session_summary_text = "" latest_session = db.scalar( select(ChatSession) .where(ChatSession.user_id == user_id) .order_by(ChatSession.updated_at.desc()) .limit(1) ) if latest_session: row = db.scalar( select(SessionSummary).where(SessionSummary.session_id == latest_session.id) ) if row and row.summary.strip() and row.updated_at: updated = row.updated_at if updated.tzinfo is None: updated = updated.replace(tzinfo=timezone.utc) if updated.astimezone(tz).date() == day: session_summary_text = row.summary.strip()[:500] calorie_delta = float(totals.get("calories", 0)) - float(targets.get("calories", 0)) return { "date": day.isoformat(), "meals_count": len(meals), "workouts_count": len(workouts), "steps_total": steps_total, "calories": round(float(totals.get("calories", 0))), "calorie_target": round(float(targets.get("calories", 0))), "calorie_delta": round(calorie_delta), "protein_g": round(float(totals.get("protein_g", 0))), "protein_target": round(float(targets.get("protein_g", 0))), "water_l": round(float(totals.get("water_ml", 0)) / 1000, 1), "pomodoro_sessions": len(pomodoro_rows), "pomodoro_tasks": pomodoro_tasks[:8], "completed_reminders": completed_reminders[:8], "session_summary": session_summary_text, "has_fitness_data": bool(meals or workouts or steps_total), } def _format_context_block(ctx: dict[str, Any]) -> str: lines = [f"Дата: {ctx['date']}"] if ctx["has_fitness_data"]: lines.append( f"Питание: {ctx['meals_count']} приёмов, {ctx['calories']}/{ctx['calorie_target']} ккал " f"({'+' if ctx['calorie_delta'] >= 0 else ''}{ctx['calorie_delta']} к цели), " f"белок {ctx['protein_g']}/{ctx['protein_target']} г, вода {ctx['water_l']} л" ) lines.append(f"Шаги: {ctx['steps_total']}, тренировок: {ctx['workouts_count']}") else: lines.append("Фитнес-данных за день нет.") if ctx["pomodoro_sessions"]: tasks = "; ".join(ctx["pomodoro_tasks"]) or "без описания задач" lines.append(f"Помидоро: {ctx['pomodoro_sessions']} сессий — {tasks}") if ctx["completed_reminders"]: lines.append(f"Выполненные напоминания: {', '.join(ctx['completed_reminders'])}") if ctx["session_summary"]: lines.append(f"Тема чата сегодня: {ctx['session_summary']}") return "\n".join(lines) def _fallback_summary(ctx: dict[str, Any]) -> tuple[str, str]: parts: list[str] = [] if ctx["pomodoro_sessions"]: parts.append(f"Сегодня {ctx['pomodoro_sessions']} помидоро-сессий.") if ctx["has_fitness_data"]: parts.append( f"По питанию {ctx['calories']} ккал из {ctx['calorie_target']}." ) if ctx["completed_reminders"]: parts.append(f"Закрыто напоминаний: {len(ctx['completed_reminders'])}.") if not parts: parts.append("День прошёл спокойно — мало записей, но это тоже результат.") text = " ".join(parts) scene = ( "sportswear, stretching, energetic" if ctx["workouts_count"] else "cozy, holding_cup, relaxed" if not ctx["has_fitness_data"] else "thumbs_up, smile" ) return text, scene def _parse_llm_payload(raw: str) -> tuple[str, str] | None: try: data = json.loads(strip_markdown_json(raw)) except (json.JSONDecodeError, TypeError): return None if not isinstance(data, dict): return None text = str(data.get("text") or "").strip() scene = str(data.get("scene_tags") or "").strip() if not text: return None return text, scene or DEFAULT_SCENE async def generate_evening_summary_content( db: Session, user_id: int, day: date, tz: ZoneInfo, ) -> dict[str, Any]: ctx = collect_day_context(db, user_id, day, tz) character = CharacterService(db, user_id) system = character.get_system_prompt() facts = _format_context_block(ctx) user_prompt = f"""Сводка дня пользователя: {facts} Напиши короткую вечернюю сводку (3–5 предложений) на русском от лица персонажа: - что было сделано и изучено (помидоро, напоминания, чат); - как прошёл день по фитнесу (если есть данные); - мягкая поддержка или лёгкий юмор, без морализаторства. Ответь ТОЛЬКО JSON без markdown: {{"text": "...", "scene_tags": "full_body, pose, mood, ..."}} scene_tags — 5–12 booru-тегов для иллюстрации персонажа по настроению дня (тренировка → sportswear, stretching; продуктивность → victory_pose; спокойный день → cozy, holding_cup).""" llm = LLMClient() try: result = await llm.complete( [ {"role": "system", "content": system}, {"role": "user", "content": user_prompt}, ], temperature=0.85, visible_reply=True, ) parsed = _parse_llm_payload(result.get("content") or "") except Exception as exc: logger.warning("Evening summary LLM failed: %s", exc) parsed = None if parsed: text, scene_tags = parsed else: text, scene_tags = _fallback_summary(ctx) return {"text": text, "scene_tags": scene_tags, "context": ctx} def format_evening_notice(text: str, image_url: str | None = None) -> str: body = f"💪 **Сводка дня**\n\n{text}" if image_url: body += f"\n\n![day]({image_url})" return body async def build_evening_notice( db: Session, user_id: int, day: date, tz: ZoneInfo, *, image_enabled: bool = True, ) -> str | None: card = CharacterService(db, user_id).get_card().get("data", {}) if not (card.get("appearance_tags") or "").strip(): logger.info("Evening summary skipped: no character appearance for user %s", user_id) return None content = await generate_evening_summary_content(db, user_id, day, tz) text = content["text"] scene_tags = content["scene_tags"] image_url: str | None = None if image_enabled: try: pose = scene_tags or random_rofl_pose() result = await generate_image( db, user_id=user_id, draw_self=True, scene_description=pose, ) if result.get("ok"): image_url = result.get("url") else: logger.warning("Evening summary image failed: %s", result.get("error")) except Exception as exc: logger.warning("Evening summary image skipped: %s", exc) return format_evening_notice(text, image_url)