import json from typing import Any from app.db.models import PomodoroSession from app.pomodoro.cycle import PHASE_LONG_BREAK, PHASE_SHORT_BREAK, PHASE_WORK PHASE_LABELS = { PHASE_WORK: "Работа", PHASE_SHORT_BREAK: "Короткий перерыв", PHASE_LONG_BREAK: "Длинный перерыв", } def _format_time(seconds: int) -> str: minutes, secs = divmod(max(0, seconds), 60) return f"{minutes:02d}:{secs:02d}" def format_phase_completed_notice( session: PomodoroSession, next_phase: str | None, ) -> str: phase_label = PHASE_LABELS.get(session.phase, session.phase) task = session.task_note or "без описания" lines = [f"⏱ **{phase_label} завершена** · {session.duration_min} мин · _{task}_"] if next_phase == PHASE_SHORT_BREAK: lines.append("Дальше: короткий перерыв ☕") elif next_phase == PHASE_LONG_BREAK: lines.append("Дальше: длинный перерыв 🌴 · цикл почти завершён") elif next_phase == PHASE_WORK: lines.append("Дальше: снова работа 💪") else: lines.append("Цикл сброшен. Можно отдохнуть и начать заново.") return "\n".join(lines) POMODORO_TOOL_NAMES = frozenset({ "get_pomodoro_status", "start_pomodoro", "start_short_break", "start_long_break", "stop_pomodoro", "skip_pomodoro_phase", "reset_pomodoro_cycle", "get_pomodoro_history", }) MEMORY_TOOL_NAMES = frozenset({ "remember_fact", "recall_memories", "forget_memory", "update_profile", "update_session_summary", }) FITNESS_TOOL_NAMES = frozenset({ "get_fitness_summary", "get_fitness_history", "set_fitness_profile", "calc_fitness_targets", "calc_body_composition", "log_meal", "log_water", "log_weight", "log_workout", "lookup_food", "lookup_exercise", "set_fitness_reminder", }) # Не засорять чат служебными ответами REMINDER_TOOL_NAMES = frozenset({ "list_reminders", "create_reminder", "update_reminder", "delete_reminder", "complete_reminder", }) SHOPPING_TOOL_NAMES = frozenset({ "list_shopping_lists", "create_shopping_list", "add_shopping_items", "check_shopping_item", "remove_shopping_item", "delete_shopping_list", }) TOOLS_SKIP_CHAT_NOTICE = frozenset({ "get_pomodoro_status", "recall_memories", "get_fitness_summary", "get_fitness_history", "lookup_food", "lookup_exercise", "calc_fitness_targets", "calc_body_composition", "get_weather", "get_morning_briefing", "list_shopping_lists", "list_reminders", }) def _format_body_composition_notice(computed: dict[str, Any], *, headline: str) -> str: parts: list[str] = [] bf = computed.get("body_fat_pct") if bf is not None: method = computed.get("body_fat_method") if method == "navy": parts.append(f"жир ≈{bf}% (Navy)") elif method == "manual": parts.append(f"жир {bf}%") else: parts.append(f"жир ≈{bf}%") if computed.get("whr") is not None: parts.append(f"WHR {computed.get('whr')}") if computed.get("ffmi") is not None: parts.append(f"FFMI {computed.get('ffmi')}") if parts: return f"{headline} — {', '.join(parts)}" return headline def format_tool_notice(tool_name: str, raw_result: str) -> str | None: if tool_name in TOOLS_SKIP_CHAT_NOTICE: return None try: data = json.loads(raw_result) except json.JSONDecodeError: return None if isinstance(data, dict) and "error" in data: if tool_name in POMODORO_TOOL_NAMES: prefix = "⏱" elif tool_name in MEMORY_TOOL_NAMES: prefix = "🧠" elif tool_name in FITNESS_TOOL_NAMES: prefix = "💪" elif tool_name in SHOPPING_TOOL_NAMES: prefix = "🛒" elif tool_name in REMINDER_TOOL_NAMES: prefix = "📅" else: prefix = "📋" return f"{prefix} {data['error']}" if tool_name == "reset_pomodoro_cycle": cycle = data.get("cycle", data) return ( "⏱ **Цикл помидоро сброшен** · " f"прогресс: {cycle.get('completed_work_sessions', 0)}/" f"{cycle.get('sessions_until_long_break', 4)}" ) if tool_name in ( "get_pomodoro_status", "start_pomodoro", "start_work", "start_short_break", "start_long_break", "stop_pomodoro", "skip_pomodoro_phase", ): return _format_status_notice(data) if tool_name == "get_pomodoro_history": return _format_history_notice(data) if tool_name == "create_work_item": return _format_work_item_notice(data) if tool_name == "list_work_items": return _format_work_items_list_notice(data) if tool_name == "list_taiga_tasks": return _format_taiga_tasks_notice(data) if tool_name == "sync_taiga_projects": return f"📋 Синхронизировано проектов Taiga: **{len(data)}**" if tool_name == "list_taiga_projects": if not isinstance(data, list) or not data: return "📋 Проекты Taiga не найдены. Вызовите sync_taiga_projects." lines = ["📋 **Проекты:**"] for p in data: gitea = f"{p.get('gitea_owner')}/{p.get('gitea_repo')}" if p.get("gitea_configured") else "—" lines.append(f"- `{p.get('slug')}`: {p.get('name')} · Gitea: {gitea}") return "\n".join(lines) if tool_name == "remember_fact" and data.get("ok"): action = "обновлено" if data.get("action") == "updated" else "сохранено" return f"🧠 **Память {action}** · #{data.get('memory_id')}: {data.get('content')}" if tool_name == "forget_memory" and data.get("ok"): return f"🧠 **Забыто** · #{data.get('memory_id')}: {data.get('forgotten')}" if tool_name == "update_profile" and data.get("ok"): profile = data.get("profile") or {} parts = [f"{k}={v}" for k, v in profile.items() if v] return f"🧠 **Профиль обновлён** · {', '.join(parts) or 'пусто'}" if tool_name == "update_session_summary" and data.get("ok"): return "🧠 **Сводка чата сохранена**" if tool_name == "log_meal" and data.get("ok"): meal = data.get("meal", {}) est = "≈" if meal.get("estimated") else "" return ( f"💪 **Приём пищи** · {meal.get('description')} · " f"{est}{meal.get('calories', 0):.0f} ккал " f"(Б{meal.get('protein_g', 0):.0f}/Ж{meal.get('fat_g', 0):.0f}/У{meal.get('carbs_g', 0):.0f})" ) if tool_name == "log_water" and data.get("ok"): w = data.get("water", {}) return f"💪 **Вода** +{w.get('amount_ml')} мл" if tool_name == "log_weight" and data.get("ok"): m = data.get("metric", {}) computed = data.get("computed") or {} headline = f"💪 **Вес** {m.get('weight_kg')} кг" return _format_body_composition_notice(computed, headline=headline) if tool_name == "calc_body_composition" and isinstance(data, dict) and "error" not in data: w = data.get("weight_kg") headline = "💪 **Состав тела** (расчёт)" if w is not None: headline += f" · {w} кг" msg = _format_body_composition_notice(data, headline=headline) warnings = data.get("warnings") or [] if warnings: msg += f" · {'; '.join(warnings[:2])}" return msg if tool_name == "log_workout" and data.get("ok"): wo = data.get("workout", {}) return f"💪 **Тренировка** · {wo.get('title')}" if tool_name == "set_fitness_profile" and data.get("ok"): p = data.get("profile", {}) return ( f"💪 **Профиль** · {p.get('calorie_target')} ккал, " f"вода {p.get('water_l')} л" ) if tool_name == "set_fitness_reminder" and data.get("ok"): r = data.get("reminder", {}) state = "вкл" if r.get("enabled") else "выкл" return f"💪 **Напоминание {r.get('kind')}** · {state}" if tool_name == "generate_image" and data.get("ok"): url = data.get("url", "") return f"🎨 **Картинка готова**\n\n![image]({url})" if tool_name == "create_shopping_list" and data.get("ok"): lst = data.get("list") or {} action = "создан" if data.get("created") else "уже был" return f"🛒 **Список {action}** · «{lst.get('name')}» (#{lst.get('id')})" if tool_name == "add_shopping_items" and data.get("ok"): added = data.get("added") or [] names = ", ".join(i.get("text", "") for i in added[:5]) extra = f" +{len(added) - 5}" if len(added) > 5 else "" return f"🛒 **Добавлено в «{data.get('list_name')}»** · {names}{extra}" if tool_name == "check_shopping_item" and data.get("ok"): item = data.get("item") or {} state = "куплено" if item.get("checked") else "снята отметка" return f"🛒 **{state}** · #{item.get('id')} {item.get('text')}" if tool_name == "remove_shopping_item" and data.get("ok"): removed = data.get("removed") or {} return f"🛒 **Удалено** · {removed.get('text')}" if tool_name == "delete_shopping_list" and data.get("ok"): return f"🛒 **Список удалён** · «{data.get('name')}»" if tool_name == "create_reminder" and data.get("ok"): r = data.get("reminder") or {} rec = r.get("recurrence", "none") rec_label = f" · повтор {rec}" if rec and rec != "none" else "" return f"📅 **Напоминание создано** · {r.get('title')} · {r.get('due_at_local')}{rec_label}" if tool_name == "update_reminder" and data.get("ok"): r = data.get("reminder") or {} return f"📅 **Напоминание обновлено** · #{r.get('id')} {r.get('title')}" if tool_name == "delete_reminder" and data.get("ok"): return f"📅 **Напоминание удалено** · «{data.get('title')}»" if tool_name == "complete_reminder" and data.get("ok"): r = data.get("reminder") or {} return f"📅 **Готово** · {r.get('title')}" return None def _format_work_item_notice(data: dict[str, Any]) -> str | None: if data.get("error"): return f"📋 {data['error']}" if not data.get("ok"): return None taiga = data.get("taiga", {}) gitea = data.get("gitea", {}) lines = [ "📋 **Создано:**", f"- Taiga: #{taiga.get('ref')} — {taiga.get('subject')}", f"- URL: {taiga.get('url')}", ] if gitea.get("url"): lines.append(f"- Gitea: {gitea.get('url')}") if data.get("branch"): lines.append(f"- Ветка: `{data['branch']}`") subtasks = data.get("subtasks") or [] if subtasks: lines.append("**Подзадачи:**") for t in subtasks: lines.append(f"- #{t.get('ref')} {t.get('subject')}") return "\n".join(lines) def _format_work_items_list_notice(data: Any) -> str | None: if not isinstance(data, list) or not data: return "📋 Локальных work items (созданных ассистентом) нет." lines = ["📋 **Work items ассистента:**"] for item in data[:15]: lines.append( f"- [{item.get('status')}] #{item.get('taiga_ref')} {item.get('title')} " f"({item.get('taiga_slug')})" ) return "\n".join(lines) def _format_taiga_tasks_notice(data: Any) -> str | None: if not isinstance(data, dict): return None if data.get("error"): return f"📋 {data['error']}" blocks = data.get("projects") or [] total_stories = data.get("total_stories", 0) total_tasks = data.get("total_tasks", 0) if not blocks or (total_stories == 0 and total_tasks == 0): slug = blocks[0].get("slug") if len(blocks) == 1 else None if slug: return f"📋 В `{slug}` нет открытых user stories и tasks в Taiga." return "📋 Открытых задач в Taiga не найдено." lines = [f"📋 **Открытые задачи Taiga** (stories: {total_stories}, tasks: {total_tasks}):"] for block in blocks: stories = block.get("stories") or [] tasks = block.get("tasks") or [] if not stories and not tasks: continue lines.append(f"**{block.get('name')}** (`{block.get('slug')}`):") for s in stories: lines.append(f"- story #{s.get('ref')} {s.get('subject')}") for t in tasks: lines.append(f"- task #{t.get('ref')} {t.get('subject')}") return "\n".join(lines) def _format_status_notice(data: dict[str, Any]) -> str: status = data.get("status", "idle") phase = data.get("phase", PHASE_WORK) phase_label = PHASE_LABELS.get(phase, phase) task = data.get("task_note") or "без описания" remaining = data.get("remaining_seconds", 0) duration = data.get("duration_min", 25) cycle = data.get("cycle", {}) cycle_info = "" if cycle: cycle_info = ( f" · цикл {cycle.get('completed_work_sessions', 0)}/" f"{cycle.get('sessions_until_long_break', 4)}" ) if status == "idle": return f"⏱ **Помидоро:** таймер не запущен{cycle_info}." if status == "running": return ( f"⏱ **{phase_label}** · осталось **{_format_time(remaining)}** " f"из {duration} мин · _{task}_{cycle_info}" ) if status == "paused": elapsed = data.get("elapsed_seconds", 0) return ( f"⏱ **{phase_label} на паузе** · прошло {_format_time(elapsed)} " f"из {duration} мин · _{task}_{cycle_info}" ) if status == "completed": return f"⏱ **{phase_label} завершена** · {duration} мин · _{task}_" if status == "cancelled": return f"⏱ **{phase_label} отменена** · _{task}_" return f"⏱ Помидоро: {status}" def _format_history_notice(data: Any) -> str: if not isinstance(data, list) or not data: return "⏱ **История помидоро** пуста." lines = ["⏱ **История помидоро:**"] for item in data[:10]: task = item.get("task_note") or "без описания" phase = PHASE_LABELS.get(item.get("phase", ""), item.get("phase", "?")) duration = item.get("duration_min", "?") lines.append(f"- {phase}: {task} ({duration} мин)") return "\n".join(lines) def format_pomodoro_context(status: dict[str, Any]) -> str: notice = _format_status_notice(status) cycle = status.get("cycle", {}) extra = "" if cycle: extra = ( f"\nНастройки цикла: работа {cycle.get('work_duration_min')} мин, " f"перерыв {cycle.get('short_break_min')} мин, " f"длинный {cycle.get('long_break_min')} мин." ) return f"[Актуальный статус помидоро]\n{notice}{extra}"