--- project: vibe-coding type: plan date: 2026-05-31 status: ready-to-execute related_spec: "{vibe-coding} {plan} tasks-bot 2.0 ticktick – 2026-05-30.md" --- # tasks-bot 2.0 — Core (Capture + Commands) Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** реализовать ядро бота v2.0 — TickTick-клиент, capture-handler (LLM → routing → write), event log и 9 команд из спека. После выполнения бот умеет принимать текст из Telegram, классифицировать через LLM, писать в правильный TickTick-проект, и реагировать на команды /today /inbox /backlog /done /defer /text /proj /date /del. **Архитектура:** живёт в v1-репо рядом со старым кодом. Новый код — в пакетах `src/tasks_bot/ticktick/` (HTTPx-клиент + mapping) и `src/tasks_bot/v2/` (routing, capture, commands, event log, кэши). Включается через флаг `V2_CAPTURE_ENABLED=true`. Когда флаг включён — v2-хендлеры регистрируются вместо v1, conversational mode отключается, v1-код остаётся в репо нетронутым (для миграции в Плане 4). Когда флаг выключен — бот работает как сейчас. **Tech Stack:** Python 3.11, python-telegram-bot 21+ (async), httpx (async), pydantic + pydantic-settings, aiosqlite, pytest + pytest-asyncio. **Не входит в скоуп этого плана:** - Scheduler (утренние/вечерние/воскресные сводки) — План 3 - Migration v1 → TickTick — План 4 - Monthly archive — План 5 - Evals — План 6 - Удаление v1-кода — План 7 - `/defer all` — кнопка появится в Плане 3 (вечерняя сводка) --- ## File Structure | File | Responsibility | |------|----------------| | `src/tasks_bot/ticktick/__init__.py` | Package marker | | `src/tasks_bot/ticktick/client.py` | Async httpx-клиент TickTick API: bearer auth, retries, structured exceptions, async create/get/update/complete/delete | | `src/tasks_bot/ticktick/mapping.py` | Загружает `projects_mapping.json`. Lookup `(brand, list_name) → project_id` и обратно | | `src/tasks_bot/ticktick/models.py` | Pydantic-модели: `TickTickTask`, `TickTickProject`, `TickTickGroup` | | `src/tasks_bot/v2/__init__.py` | Package marker | | `src/tasks_bot/v2/routing.py` | Чистые routing-правила: `(brand, due, platform_hint, strategic_flag) → list_name` | | `src/tasks_bot/v2/extraction.py` | LLM-извлечение через OpenRouter: prompt, schema, `extract_fields(text)` | | `src/tasks_bot/v2/pending.py` | Pending capture cache (in-memory, TTL=300s) для inline-кнопок | | `src/tasks_bot/v2/capture.py` | Capture handler: extract → route → write OR pending+buttons | | `src/tasks_bot/v2/event_log.py` | Append-only `events.jsonl` writer | | `src/tasks_bot/v2/summary_cache.py` | N→task_id маппинг (aiosqlite, для рестарта) | | `src/tasks_bot/v2/summary_format.py` | Форматирование списка задач для `/today` и (в Плане 3) scheduled-сводок | | `src/tasks_bot/v2/commands.py` | 9 commands: `/today /inbox /backlog /done /defer /text /proj /date /del` | | `src/tasks_bot/v2/app.py` | `register_v2_handlers(app, ...)` — регистрирует все handlers условно | | `src/tasks_bot/config.py` (modify) | Добавить `v2_capture_enabled`, `ticktick_access_token`, `ticktick_mapping_path`, `event_log_path` | | `src/tasks_bot/bot.py` (modify) | В `build_application` ветвление: если v2 включён — `register_v2_handlers` вместо v1-хендлеров | | `tests/v2/test_routing.py` | Unit-тесты для routing rules (10 кейсов) | | `tests/v2/test_pending.py` | Pending cache TTL behavior | | `tests/v2/test_event_log.py` | Event log writer (idempotent append, JSON shape) | | `tests/v2/test_summary_cache.py` | N→task_id round-trip | | `tests/v2/test_summary_format.py` | Форматтер: пустой, один проект, несколько, со статусами | | `tests/v2/test_extraction.py` | Mock-тест extraction (не live) | | `tests/v2/test_capture.py` | Capture handler с mock TickTick client + mock LLM | | `tests/v2/test_commands.py` | Commands против mock client | | `tests/ticktick/test_mapping.py` | Mapping loader | | `.env.example` (modify) | Новые переменные | | `README.md` (modify) | Заметка о v2-режиме | --- ## Manual prerequisites Уже выполнено (2026-05-31): - ✅ OAuth-app зарегистрирован, токены в `.env.preflight` - ✅ 15 проектов и 5 групп созданы в TickTick - ✅ `scripts/preflight/projects_mapping.json` сгенерирован Перед стартом — перенести значения из `.env.preflight` в основной `.env`: - `TICKTICK_ACCESS_TOKEN` - (опционально) `TICKTICK_CLIENT_ID`, `TICKTICK_CLIENT_SECRET` — пригодятся при перевыдаче токена --- ## Task 1: TickTick async client **Files:** - Create: `src/tasks_bot/ticktick/__init__.py` - Create: `src/tasks_bot/ticktick/client.py` - Create: `tests/ticktick/__init__.py` - Create: `tests/ticktick/test_client_unit.py` Production-grade async client. Не путать с `scripts/preflight/http_client.py` — тот был sync probe для разовых проверок. Этот будет работать в боте 24/7 в asyncio-loop, ему нужны: httpx, retry на сетевых ошибках, ясные exceptions. - [ ] **Step 1.1: Создать `src/tasks_bot/ticktick/__init__.py`** ```python """TickTick API client and mapping.""" ``` - [ ] **Step 1.2: Написать падающий тест на client** `tests/ticktick/test_client_unit.py`: ```python import httpx import pytest from tasks_bot.ticktick.client import TickTickClient, TickTickHTTPError @pytest.mark.asyncio async def test_get_returns_parsed_json(): transport = httpx.MockTransport(lambda req: httpx.Response(200, json=[{"id": "a"}])) client = TickTickClient( access_token="t", base_url="https://api.example.com", transport=transport, ) async with client: result = await client.get("project") assert result == [{"id": "a"}] @pytest.mark.asyncio async def test_non_2xx_raises(): transport = httpx.MockTransport(lambda req: httpx.Response(401, text='{"err":"unauth"}')) client = TickTickClient(access_token="t", base_url="https://x.com", transport=transport) async with client: with pytest.raises(TickTickHTTPError) as exc: await client.get("project") assert exc.value.status == 401 @pytest.mark.asyncio async def test_post_sends_json_body_and_auth(): captured = {} def handler(req: httpx.Request) -> httpx.Response: captured["headers"] = dict(req.headers) captured["content"] = req.content.decode() return httpx.Response(200, json={"id": "new"}) transport = httpx.MockTransport(handler) client = TickTickClient(access_token="abc", base_url="https://x.com", transport=transport) async with client: result = await client.post("task", {"title": "hi"}) assert result == {"id": "new"} assert captured["headers"]["authorization"] == "Bearer abc" assert '"title": "hi"' in captured["content"] or '"title":"hi"' in captured["content"] ``` - [ ] **Step 1.3: Прогнать — должны упасть** Run: `.venv/bin/pytest tests/ticktick/test_client_unit.py -v` Expected: FAIL (ModuleNotFoundError) - [ ] **Step 1.4: Написать `client.py`** ```python """Async TickTick API client. Уровень: production. Используется внутри бота в asyncio-loop. Не путать с scripts/preflight/http_client.py — тот был sync probe. Не делаем здесь auto-refresh: TickTick на текущем OAuth не возвращает refresh_token, access живёт ~180 дней. При 401 пишем ERROR в лог и бросаем TickTickHTTPError — пользователю надо перезапустить oauth_setup и обновить .env. """ from typing import Any import logging import httpx log = logging.getLogger(__name__) class TickTickHTTPError(Exception): def __init__(self, method: str, path: str, status: int, body: str): super().__init__(f"{method} {path} -> {status}: {body[:300]}") self.status = status self.body = body class TickTickClient: def __init__( self, access_token: str, base_url: str = "https://api.ticktick.com/open/v1", timeout_seconds: float = 20.0, transport: httpx.AsyncBaseTransport | None = None, ): self._token = access_token self._base = base_url.rstrip("/") self._client = httpx.AsyncClient( timeout=timeout_seconds, transport=transport, headers={ "Authorization": f"Bearer {access_token}", "Content-Type": "application/json", }, ) async def __aenter__(self) -> "TickTickClient": return self async def __aexit__(self, *args) -> None: await self._client.aclose() def _url(self, path: str) -> str: return self._base + "/" + path.lstrip("/") async def _request( self, method: str, path: str, json_body: Any = None, ) -> Any: try: resp = await self._client.request(method, self._url(path), json=json_body) except httpx.TimeoutException as e: log.warning("ticktick %s %s timeout: %s", method, path, e) raise TickTickHTTPError(method, path, -1, "timeout") from e if resp.is_success: if resp.status_code == 204 or not resp.text: return None return resp.json() if resp.status_code == 401: log.error("ticktick auth failed (401), token may have expired") raise TickTickHTTPError(method, path, resp.status_code, resp.text) async def get(self, path: str) -> Any: return await self._request("GET", path) async def post(self, path: str, body: Any = None) -> Any: return await self._request("POST", path, body) async def delete(self, path: str) -> Any: return await self._request("DELETE", path) ``` - [ ] **Step 1.5: Прогнать тесты — должны пройти** Run: `.venv/bin/pytest tests/ticktick/test_client_unit.py -v` Expected: 3 passed. - [ ] **Step 1.6: Commit** ```bash git add src/tasks_bot/ticktick/__init__.py src/tasks_bot/ticktick/client.py \ tests/ticktick/__init__.py tests/ticktick/test_client_unit.py git commit -m "v2: async TickTick HTTP client (httpx-based)" ``` --- ## Task 2: Mapping loader **Files:** - Create: `src/tasks_bot/ticktick/mapping.py` - Create: `tests/ticktick/test_mapping.py` - Create: `tests/ticktick/fixtures/projects_mapping.json` (test fixture) Загружает `projects_mapping.json`. Структура файла после Task setup_ticktick_groups: `{"Mars - To Do": {"id": "...", "groupId": "..."}, ...}`. Lookup API: - `mapping.id_for(brand="mars", list_name="To Do") → "tk_id"` - `mapping.reverse(project_id) → ("Mars", "To Do")` (если найден) - [ ] **Step 2.1: Создать fixture** `tests/ticktick/fixtures/projects_mapping.json`: ```json { "Mars - To Do": {"id": "p_mars_todo", "groupId": "g_mars"}, "Mars - Бэклог": {"id": "p_mars_back", "groupId": "g_mars"}, "Mars - Стратегия": {"id": "p_mars_strat", "groupId": "g_mars"}, "Channel - To Do": {"id": "p_ch_todo", "groupId": "g_ch"}, "Channel - Threads": {"id": "p_ch_threads", "groupId": "g_ch"}, "Channel - Instagram": {"id": "p_ch_ig", "groupId": "g_ch"}, "Channel - Канал": {"id": "p_ch_tg", "groupId": "g_ch"}, "Channel - Бэклог": {"id": "p_ch_back", "groupId": "g_ch"}, "Consulting - To Do": {"id": "p_co_todo", "groupId": "g_co"}, "Consulting - Бэклог": {"id": "p_co_back", "groupId": "g_co"}, "Life - To Do": {"id": "p_li_todo", "groupId": "g_li"}, "Life - Бэклог": {"id": "p_li_back", "groupId": "g_li"}, "Vibe Coding - To Do": {"id": "p_vc_todo", "groupId": "g_vc"}, "Vibe Coding - Бэклог": {"id": "p_vc_back", "groupId": "g_vc"}, "Vibe Coding - Стратегия": {"id": "p_vc_strat", "groupId": "g_vc"} } ``` - [ ] **Step 2.2: Написать тесты** `tests/ticktick/test_mapping.py`: ```python from pathlib import Path import pytest from tasks_bot.ticktick.mapping import ProjectMapping, UnknownProjectError FIXTURE = Path(__file__).parent / "fixtures" / "projects_mapping.json" def test_id_for_known_brand_and_list(): m = ProjectMapping.load(FIXTURE) assert m.id_for("mars", "To Do") == "p_mars_todo" assert m.id_for("channel", "Threads") == "p_ch_threads" assert m.id_for("vibe-coding", "Стратегия") == "p_vc_strat" def test_id_for_unknown_combination_raises(): m = ProjectMapping.load(FIXTURE) with pytest.raises(UnknownProjectError): m.id_for("mars", "Threads") def test_reverse_lookup(): m = ProjectMapping.load(FIXTURE) assert m.reverse("p_ch_threads") == ("channel", "Threads") assert m.reverse("p_mars_strat") == ("mars", "Стратегия") assert m.reverse("unknown") is None def test_all_brands_listed(): m = ProjectMapping.load(FIXTURE) assert set(m.brands()) == {"mars", "channel", "consulting", "life", "vibe-coding"} ``` - [ ] **Step 2.3: Запустить — упадут** Run: `.venv/bin/pytest tests/ticktick/test_mapping.py -v` Expected: FAIL. - [ ] **Step 2.4: Написать `mapping.py`** ```python """Маппинг `projects_mapping.json` → TickTick project_id. `brand` (например `mars`) — внутреннее имя в боте (lowercase, со слэш-нижним `vibe-coding`). В JSON-ключах оно с большой буквы и пробелом (`Vibe Coding - To Do`). Эта обёртка нормализует. """ import json from pathlib import Path # Внутренний brand → имя группы в TickTick (как в JSON-ключах) BRAND_TO_GROUP_PREFIX: dict[str, str] = { "mars": "Mars", "channel": "Channel", "consulting": "Consulting", "life": "Life", "vibe-coding": "Vibe Coding", } GROUP_PREFIX_TO_BRAND: dict[str, str] = {v: k for k, v in BRAND_TO_GROUP_PREFIX.items()} class UnknownProjectError(LookupError): pass class ProjectMapping: def __init__(self, raw: dict[str, dict]): # raw: { "Mars - To Do": {"id": "...", "groupId": "..."}, ... } self._by_name = raw self._by_id: dict[str, tuple[str, str]] = {} for full_name, meta in raw.items(): # split " - " (space-dash-space) into prefix + list if " - " not in full_name: continue prefix, list_name = full_name.split(" - ", 1) brand = GROUP_PREFIX_TO_BRAND.get(prefix) if brand: self._by_id[meta["id"]] = (brand, list_name) @classmethod def load(cls, path: Path) -> "ProjectMapping": with open(path, encoding="utf-8") as f: return cls(json.load(f)) def id_for(self, brand: str, list_name: str) -> str: prefix = BRAND_TO_GROUP_PREFIX.get(brand) if not prefix: raise UnknownProjectError(f"unknown brand: {brand}") key = f"{prefix} - {list_name}" if key not in self._by_name: raise UnknownProjectError(f"no mapping for {key}") return self._by_name[key]["id"] def reverse(self, project_id: str) -> tuple[str, str] | None: return self._by_id.get(project_id) def brands(self) -> list[str]: return list(BRAND_TO_GROUP_PREFIX.keys()) ``` - [ ] **Step 2.5: Прогнать — должны пройти** Run: `.venv/bin/pytest tests/ticktick/test_mapping.py -v` Expected: 4 passed. - [ ] **Step 2.6: Commit** ```bash git add src/tasks_bot/ticktick/mapping.py tests/ticktick/test_mapping.py \ tests/ticktick/fixtures/projects_mapping.json git commit -m "v2: ProjectMapping loader for projects_mapping.json" ``` --- ## Task 3: Routing rules **Files:** - Create: `src/tasks_bot/v2/__init__.py` - Create: `src/tasks_bot/v2/routing.py` - Create: `tests/v2/__init__.py` - Create: `tests/v2/test_routing.py` Чистая функция без зависимостей: `(brand, due, platform_hint, strategic_flag) → list_name`. Реализует правила из спека (см. раздел «Routing — детерминированный»). - [ ] **Step 3.1: Создать `src/tasks_bot/v2/__init__.py`** ```python """tasks-bot v2 — TickTick-based core.""" ``` - [ ] **Step 3.2: Написать тесты** `tests/v2/__init__.py`: ```python ``` `tests/v2/test_routing.py`: ```python from datetime import date from tasks_bot.v2.routing import resolve_list def test_channel_with_platform_threads_goes_to_threads(): assert resolve_list("channel", due=None, platform_hint="threads", strategic_flag=False) == "Threads" def test_channel_with_platform_telegram_goes_to_kanal(): assert resolve_list("channel", due=date(2026, 6, 1), platform_hint="telegram", strategic_flag=False) == "Канал" def test_channel_without_platform_but_with_due_goes_to_to_do(): assert resolve_list("channel", due=date(2026, 6, 1), platform_hint=None, strategic_flag=False) == "To Do" def test_channel_without_platform_no_due_goes_to_backlog(): assert resolve_list("channel", due=None, platform_hint=None, strategic_flag=False) == "Бэклог" def test_mars_strategic_goes_to_strategy_regardless_of_due(): assert resolve_list("mars", due=None, platform_hint=None, strategic_flag=True) == "Стратегия" assert resolve_list("mars", due=date(2026, 6, 1), platform_hint=None, strategic_flag=True) == "Стратегия" def test_mars_with_due_goes_to_to_do(): assert resolve_list("mars", due=date(2026, 6, 1), platform_hint=None, strategic_flag=False) == "To Do" def test_mars_without_due_goes_to_backlog(): assert resolve_list("mars", due=None, platform_hint=None, strategic_flag=False) == "Бэклог" def test_vibe_coding_strategic(): assert resolve_list("vibe-coding", due=None, platform_hint=None, strategic_flag=True) == "Стратегия" def test_life_with_due_to_do(): assert resolve_list("life", due=date(2026, 6, 1), platform_hint=None, strategic_flag=False) == "To Do" def test_life_without_due_backlog(): assert resolve_list("life", due=None, platform_hint=None, strategic_flag=False) == "Бэклог" def test_consulting_routing(): assert resolve_list("consulting", due=date(2026, 6, 1), platform_hint=None, strategic_flag=False) == "To Do" assert resolve_list("consulting", due=None, platform_hint=None, strategic_flag=False) == "Бэклог" ``` - [ ] **Step 3.3: Запустить — упадут** Run: `.venv/bin/pytest tests/v2/test_routing.py -v` Expected: FAIL. - [ ] **Step 3.4: Написать `routing.py`** ```python """Детерминированные routing-правила для capture handler. LLM извлекает (brand, due, platform_hint, strategic_flag, content), а в какой именно список TickTick положить — решает эта функция через простые if-ы. Реализует правила из спека «Routing — детерминированный». """ from datetime import date def resolve_list( brand: str, *, due: date | None, platform_hint: str | None, strategic_flag: bool, ) -> str: """Вернуть имя списка внутри TickTick-проекта. Возвращаемое значение — `To Do` / `Бэклог` / `Стратегия` / `Threads` / `Instagram` / `Канал`. Используется потом mapping.id_for для получения project_id. """ if brand == "channel": if platform_hint in ("threads", "instagram", "telegram"): return { "threads": "Threads", "instagram": "Instagram", "telegram": "Канал", }[platform_hint] return "To Do" if due else "Бэклог" if brand in ("mars", "vibe-coding"): if strategic_flag: return "Стратегия" return "To Do" if due else "Бэклог" if brand in ("consulting", "life"): return "To Do" if due else "Бэклог" raise ValueError(f"unknown brand: {brand}") ``` - [ ] **Step 3.5: Прогнать — должны пройти** Run: `.venv/bin/pytest tests/v2/test_routing.py -v` Expected: 11 passed. - [ ] **Step 3.6: Commit** ```bash git add src/tasks_bot/v2/__init__.py src/tasks_bot/v2/routing.py \ tests/v2/__init__.py tests/v2/test_routing.py git commit -m "v2: deterministic routing rules" ``` --- ## Task 4: Config additions **Files:** - Modify: `src/tasks_bot/config.py` - Modify: `.env.example` Добавить env vars для v2. Не ломать v1 — все новые поля имеют дефолты. - [ ] **Step 4.1: Дописать поля в `Settings`** В `src/tasks_bot/config.py` в класс `Settings` добавить: ```python # === v2 settings === v2_capture_enabled: bool = False ticktick_access_token: str | None = None ticktick_api_base: str = "https://api.ticktick.com/open/v1" ticktick_mapping_path: str = "scripts/preflight/projects_mapping.json" event_log_path: str = "data/events.jsonl" summary_cache_db_path: str = "data/summary_cache.sqlite" pending_capture_ttl_seconds: int = 300 project_confidence_threshold: float = 0.7 openrouter_v2_model: str = "anthropic/claude-haiku-4.5" ``` - [ ] **Step 4.2: Обновить `.env.example`** Дописать в конец `.env.example`: ``` # === v2 === V2_CAPTURE_ENABLED=false TICKTICK_ACCESS_TOKEN= TICKTICK_API_BASE=https://api.ticktick.com/open/v1 TICKTICK_MAPPING_PATH=scripts/preflight/projects_mapping.json EVENT_LOG_PATH=data/events.jsonl SUMMARY_CACHE_DB_PATH=data/summary_cache.sqlite PENDING_CAPTURE_TTL_SECONDS=300 PROJECT_CONFIDENCE_THRESHOLD=0.7 OPENROUTER_V2_MODEL=anthropic/claude-haiku-4.5 ``` - [ ] **Step 4.3: Проверить, что v1-тесты по-прежнему зелёные** Run: `.venv/bin/pytest 2>&1 | tail -3` Expected: 274+ passed (текущий уровень + v2 тесты добавились). - [ ] **Step 4.4: Commit** ```bash git add src/tasks_bot/config.py .env.example git commit -m "v2: add settings (feature flag + ticktick + paths)" ``` --- ## Task 5: LLM extraction **Files:** - Create: `src/tasks_bot/v2/extraction.py` - Create: `tests/v2/test_extraction.py` Вызывает OpenRouter с tool/JSON-mode, извлекает `Extracted(brand, content, due, platform_hint, strategic_flag, project_confidence)`. Системный промпт описывает структуру TickTick (5 brands) и просит вернуть JSON. - [ ] **Step 5.1: Написать тест** `tests/v2/test_extraction.py`: ```python from datetime import date from unittest.mock import AsyncMock import pytest from tasks_bot.v2.extraction import Extracted, Extractor @pytest.mark.asyncio async def test_extractor_parses_llm_json_output(): fake_response = { "brand": "channel", "content": "пост про утренний ритуал", "due": None, "platform_hint": "threads", "strategic_flag": False, "project_confidence": 0.9, } llm = AsyncMock(return_value=fake_response) extractor = Extractor(llm_call=llm, today=date(2026, 6, 1)) result = await extractor.extract("идея для треда про утренний ритуал") assert isinstance(result, Extracted) assert result.brand == "channel" assert result.platform_hint == "threads" assert result.due is None assert result.project_confidence == 0.9 @pytest.mark.asyncio async def test_extractor_parses_iso_date(): fake_response = { "brand": "life", "content": "налоги", "due": "2026-06-05", "platform_hint": None, "strategic_flag": False, "project_confidence": 0.95, } llm = AsyncMock(return_value=fake_response) extractor = Extractor(llm_call=llm, today=date(2026, 6, 1)) result = await extractor.extract("налоги до пятницы") assert result.due == date(2026, 6, 5) @pytest.mark.asyncio async def test_extractor_low_confidence_passes_through(): fake_response = { "brand": "mars", "content": "что-то непонятное", "due": None, "platform_hint": None, "strategic_flag": False, "project_confidence": 0.4, } llm = AsyncMock(return_value=fake_response) extractor = Extractor(llm_call=llm, today=date(2026, 6, 1)) result = await extractor.extract("foo bar") assert result.project_confidence == 0.4 ``` - [ ] **Step 5.2: Прогнать — упадут** Run: `.venv/bin/pytest tests/v2/test_extraction.py -v` Expected: FAIL. - [ ] **Step 5.3: Написать `extraction.py`** ```python """LLM-извлечение полей из сообщения пользователя. Вход: текст. Выход: Extracted с brand, content, due, platform_hint, strategic_flag, project_confidence. LLM-промпт построен на JSON-mode (modern OpenAI-compatible API). """ from dataclasses import dataclass from datetime import date from typing import Any, Awaitable, Callable LLMCall = Callable[[str, dict], Awaitable[dict]] SYSTEM_PROMPT = """Ты помощник по классификации задач. Получаешь короткое сообщение от пользователя и извлекаешь из него поля для записи в TickTick. Бренды (выбери ОДИН): - "mars" — школа музыки «Марс» (учёба, продукт, педагоги, ученики) - "channel" — личный Telegram-канал и контент (посты, идеи) - "consulting" — консалтинг 1-1 (клиенты, лиды, гайды) - "life" — личное (быт, документы, здоровье) - "vibe-coding" — скрипты и автоматизация (боты, AI, разработка) - "inbox" — если непонятно `platform_hint` (только для brand=channel): - "threads" / "instagram" / "telegram" — если в тексте явно указана платформа - null — иначе `strategic_flag` (только для mars/vibe-coding): - true — если речь о стратегии, направлении, долгосрочной идее без конкретного срока - false — иначе `due`: - ISO дата (YYYY-MM-DD) — если в тексте есть срок («завтра», «пятница», «до 5 июня») - null — иначе - Интерпретируй даты относительно: {today} (часовой пояс Asia/Yekaterinburg) `content`: - Чистый текст задачи (без меток типа «идея для», «срочно», «в марс») `project_confidence`: - 0..1, твоя уверенность что brand определён правильно Верни ТОЛЬКО JSON, никакого текста.""" @dataclass class Extracted: brand: str content: str due: date | None platform_hint: str | None strategic_flag: bool project_confidence: float class Extractor: def __init__(self, llm_call: LLMCall, today: date): self._llm = llm_call self._today = today def _system_prompt(self) -> str: return SYSTEM_PROMPT.format(today=self._today.isoformat()) async def extract(self, user_text: str) -> Extracted: json_schema = { "type": "object", "properties": { "brand": { "type": "string", "enum": ["mars", "channel", "consulting", "life", "vibe-coding", "inbox"], }, "content": {"type": "string"}, "due": {"type": ["string", "null"], "description": "ISO YYYY-MM-DD"}, "platform_hint": { "type": ["string", "null"], "enum": ["threads", "instagram", "telegram", None], }, "strategic_flag": {"type": "boolean"}, "project_confidence": {"type": "number", "minimum": 0, "maximum": 1}, }, "required": [ "brand", "content", "due", "platform_hint", "strategic_flag", "project_confidence", ], } response = await self._llm( self._system_prompt(), {"user_message": user_text, "schema": json_schema}, ) return Extracted( brand=response["brand"], content=response["content"], due=date.fromisoformat(response["due"]) if response["due"] else None, platform_hint=response.get("platform_hint"), strategic_flag=bool(response.get("strategic_flag", False)), project_confidence=float(response["project_confidence"]), ) ``` - [ ] **Step 5.4: Прогнать — должны пройти** Run: `.venv/bin/pytest tests/v2/test_extraction.py -v` Expected: 3 passed. - [ ] **Step 5.5: Commit** ```bash git add src/tasks_bot/v2/extraction.py tests/v2/test_extraction.py git commit -m "v2: LLM extractor with JSON-schema output" ``` --- ## Task 6: OpenRouter LLM adapter **Files:** - Create: `src/tasks_bot/v2/openrouter.py` Тонкая обёртка над OpenRouter chat-completions API, возвращающая dict из JSON-ответа. Реальная реализация Callable[[str, dict], Awaitable[dict]] которую Extractor получает. В v1 уже есть `src/tasks_bot/conversational/openrouter.py` — посмотри для паттерна, но не импортируй (он tool-calling, наш — JSON-mode). - [ ] **Step 6.1: Написать** ```python """OpenRouter chat-completions с JSON-mode. Используется Extractor'ом v2. Отличается от conversational/openrouter.py тем, что не делает tool-calling, а просит structured JSON-output. """ import json import logging import httpx log = logging.getLogger(__name__) OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions" class OpenRouterError(Exception): pass def make_extractor_llm( *, api_key: str, model: str, http_client: httpx.AsyncClient | None = None, ): """Вернуть async-callable, совместимый с Extractor. Сигнатура: async def(system_prompt, ctx) -> dict где ctx = {"user_message": str, "schema": dict}. """ own_client = http_client is None client = http_client or httpx.AsyncClient(timeout=30.0) async def call(system_prompt: str, ctx: dict) -> dict: try: resp = await client.post( OPENROUTER_URL, headers={"Authorization": f"Bearer {api_key}"}, json={ "model": model, "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": ctx["user_message"]}, ], "response_format": {"type": "json_object"}, }, ) if not resp.is_success: raise OpenRouterError(f"{resp.status_code}: {resp.text[:300]}") body = resp.json() raw = body["choices"][0]["message"]["content"] try: return json.loads(raw) except json.JSONDecodeError as e: raise OpenRouterError(f"non-JSON response: {raw[:300]}") from e finally: if own_client: pass # клиент управляется снаружи, не закрываем тут return call ``` - [ ] **Step 6.2: Commit (без тестов — это адаптер, тест появится в live-eval в Плане 6)** ```bash git add src/tasks_bot/v2/openrouter.py git commit -m "v2: OpenRouter JSON-mode adapter" ``` --- ## Task 7: Pending capture cache **Files:** - Create: `src/tasks_bot/v2/pending.py` - Create: `tests/v2/test_pending.py` Кэш для inline-кнопок: когда LLM не уверен в проекте, мы запоминаем извлечённые поля + текст пользователя, и ждём 5 минут когда он нажмёт на кнопку. После таймаута — задача всё равно создаётся в Inbox. - [ ] **Step 7.1: Тесты** ```python import asyncio from datetime import datetime, timedelta, timezone import pytest from tasks_bot.v2.pending import PendingCaptureCache, PendingEntry @pytest.mark.asyncio async def test_put_and_pop(): cache = PendingCaptureCache(ttl_seconds=60) entry = PendingEntry(user_id=1, payload={"x": 1}) key = cache.put(entry) assert key popped = cache.pop(key) assert popped is entry assert cache.pop(key) is None # second pop returns None @pytest.mark.asyncio async def test_ttl_expiry(monkeypatch): now = datetime(2026, 6, 1, 12, 0, 0, tzinfo=timezone.utc) monkeypatch.setattr( "tasks_bot.v2.pending._now", lambda: now, ) cache = PendingCaptureCache(ttl_seconds=10) key = cache.put(PendingEntry(user_id=1, payload={"x": 1})) later = now + timedelta(seconds=15) monkeypatch.setattr("tasks_bot.v2.pending._now", lambda: later) assert cache.pop(key) is None # expired @pytest.mark.asyncio async def test_pop_expired_keys_during_iteration(monkeypatch): now = datetime(2026, 6, 1, 12, 0, 0, tzinfo=timezone.utc) monkeypatch.setattr("tasks_bot.v2.pending._now", lambda: now) cache = PendingCaptureCache(ttl_seconds=10) k1 = cache.put(PendingEntry(user_id=1, payload={"a": 1})) later = now + timedelta(seconds=20) monkeypatch.setattr("tasks_bot.v2.pending._now", lambda: later) k2 = cache.put(PendingEntry(user_id=2, payload={"b": 2})) # k1 expired, k2 alive assert cache.pop(k1) is None assert cache.pop(k2).payload == {"b": 2} ``` - [ ] **Step 7.2: Прогнать — упадут** Run: `.venv/bin/pytest tests/v2/test_pending.py -v` Expected: FAIL. - [ ] **Step 7.3: Написать** ```python """In-memory кэш pending-capture'ов с TTL. Бот при низком project_confidence сохраняет {user_id, payload}, выдаёт ключ, шлёт inline-кнопки с этим ключом. По callback — pop по ключу. По таймауту — задача создаётся в системный Inbox (это делает капчер, не кэш). """ import secrets from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any def _now() -> datetime: return datetime.now(timezone.utc) @dataclass class PendingEntry: user_id: int payload: dict[str, Any] created_at: datetime = field(default_factory=_now) class PendingCaptureCache: def __init__(self, ttl_seconds: int = 300): self._ttl = ttl_seconds self._store: dict[str, PendingEntry] = {} def put(self, entry: PendingEntry) -> str: key = secrets.token_urlsafe(10) self._store[key] = entry return key def pop(self, key: str) -> PendingEntry | None: entry = self._store.pop(key, None) if entry is None: return None age = (_now() - entry.created_at).total_seconds() if age > self._ttl: return None return entry ``` - [ ] **Step 7.4: Прогнать — должны пройти** Run: `.venv/bin/pytest tests/v2/test_pending.py -v` Expected: 3 passed. - [ ] **Step 7.5: Commit** ```bash git add src/tasks_bot/v2/pending.py tests/v2/test_pending.py git commit -m "v2: pending capture cache with TTL" ``` --- ## Task 8: Event log writer **Files:** - Create: `src/tasks_bot/v2/event_log.py` - Create: `tests/v2/test_event_log.py` Append-only `events.jsonl`. Каждая мутация (create/complete/delete/update) пишет одну строку. Используется Плане 5 для monthly archive (поскольку TickTick API не отдаёт closed-таски). - [ ] **Step 8.1: Тесты** `tests/v2/test_event_log.py`: ```python import json from datetime import datetime, timezone from pathlib import Path import pytest from tasks_bot.v2.event_log import EventLog def test_writes_jsonl_line(tmp_path: Path): log_path = tmp_path / "events.jsonl" log = EventLog(log_path) ts = datetime(2026, 6, 1, 9, 0, 0, tzinfo=timezone.utc) log.write( op="complete", task_id="tk_123", ts=ts, title="налоги", brand="life", list_name="To Do", ) content = log_path.read_text(encoding="utf-8") line = json.loads(content.strip()) assert line["op"] == "complete" assert line["task_id"] == "tk_123" assert line["ts"] == ts.isoformat() assert line["title"] == "налоги" assert line["brand"] == "life" assert line["list"] == "To Do" def test_multiple_writes_append(tmp_path: Path): log_path = tmp_path / "events.jsonl" log = EventLog(log_path) log.write(op="create", task_id="a", ts=datetime.now(timezone.utc), title="x", brand="life", list_name="To Do") log.write(op="complete", task_id="a", ts=datetime.now(timezone.utc), title="x", brand="life", list_name="To Do") lines = log_path.read_text(encoding="utf-8").strip().splitlines() assert len(lines) == 2 assert json.loads(lines[0])["op"] == "create" assert json.loads(lines[1])["op"] == "complete" def test_creates_parent_dir(tmp_path: Path): log_path = tmp_path / "nested" / "dir" / "events.jsonl" log = EventLog(log_path) log.write(op="create", task_id="a", ts=datetime.now(timezone.utc), title="x", brand="life", list_name="To Do") assert log_path.exists() ``` - [ ] **Step 8.2: Прогнать — упадут** Run: `.venv/bin/pytest tests/v2/test_event_log.py -v` Expected: FAIL. - [ ] **Step 8.3: Написать** ```python """Append-only event log для monthly archive (план B). Каждая мутация бота над TickTick-задачей пишет одну строку. Скрипт архива (План 5) читает .jsonl за прошлый месяц, фильтрует op=complete, группирует и пишет markdown. """ import json from datetime import datetime from pathlib import Path from typing import Any class EventLog: def __init__(self, path: Path | str): self._path = Path(path) def write( self, *, op: str, task_id: str, ts: datetime, title: str, brand: str, list_name: str, extra: dict[str, Any] | None = None, ) -> None: self._path.parent.mkdir(parents=True, exist_ok=True) record = { "ts": ts.isoformat(), "op": op, "task_id": task_id, "title": title, "brand": brand, "list": list_name, } if extra: record.update(extra) with self._path.open("a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False)) f.write("\n") ``` - [ ] **Step 8.4: Прогнать** Run: `.venv/bin/pytest tests/v2/test_event_log.py -v` Expected: 3 passed. - [ ] **Step 8.5: Commit** ```bash git add src/tasks_bot/v2/event_log.py tests/v2/test_event_log.py git commit -m "v2: append-only event log (jsonl)" ``` --- ## Task 9: Capture handler (high confidence path) **Files:** - Create: `src/tasks_bot/v2/capture.py` - Create: `tests/v2/test_capture.py` Главный класс `CaptureHandler` со state'ом: `extractor`, `mapping`, `ticktick`, `event_log`, `pending`, `confidence_threshold`. Метод `handle_text(user_id, text) → CaptureResult` где `CaptureResult` это либо `Created(task_id, brand, list_name)` либо `NeedsBrand(pending_key, content_preview)`. Inline-кнопки и фактический Telegram-API живут в следующей задаче (Task 11 — wiring) — здесь только бизнес-логика. - [ ] **Step 9.1: Тесты (с моками)** `tests/v2/test_capture.py`: ```python from datetime import date from pathlib import Path from unittest.mock import AsyncMock, MagicMock import pytest from tasks_bot.ticktick.mapping import ProjectMapping from tasks_bot.v2.capture import CaptureHandler, Created, NeedsBrand from tasks_bot.v2.extraction import Extracted FIXTURE = Path(__file__).parent.parent / "ticktick" / "fixtures" / "projects_mapping.json" def _mapping() -> ProjectMapping: return ProjectMapping.load(FIXTURE) def _mock_ticktick(created_id: str = "tk_new") -> MagicMock: client = MagicMock() client.post = AsyncMock(return_value={"id": created_id}) return client def _mock_event_log() -> MagicMock: log = MagicMock() log.write = MagicMock() return log def _mock_extractor(extracted: Extracted) -> MagicMock: ex = MagicMock() ex.extract = AsyncMock(return_value=extracted) return ex @pytest.mark.asyncio async def test_high_confidence_writes_to_ticktick(): extracted = Extracted( brand="channel", content="пост про утренний ритуал", due=None, platform_hint="threads", strategic_flag=False, project_confidence=0.92, ) client = _mock_ticktick("tk_post") log = _mock_event_log() handler = CaptureHandler( extractor=_mock_extractor(extracted), mapping=_mapping(), ticktick=client, event_log=log, pending_cache=None, # not used in this path confidence_threshold=0.7, ) result = await handler.handle_text(user_id=1, text="идея для треда про утренний ритуал") assert isinstance(result, Created) assert result.task_id == "tk_post" assert result.brand == "channel" assert result.list_name == "Threads" # POST /task был вызван с правильным projectId args, kwargs = client.post.call_args assert args[0] == "task" body = args[1] assert body["projectId"] == "p_ch_threads" assert body["title"] == "пост про утренний ритуал" # event log записал create log.write.assert_called_once() @pytest.mark.asyncio async def test_low_confidence_returns_needs_brand(): extracted = Extracted( brand="mars", content="посмотреть отчёт", due=None, platform_hint=None, strategic_flag=False, project_confidence=0.4, ) from tasks_bot.v2.pending import PendingCaptureCache cache = PendingCaptureCache(ttl_seconds=300) handler = CaptureHandler( extractor=_mock_extractor(extracted), mapping=_mapping(), ticktick=_mock_ticktick(), event_log=_mock_event_log(), pending_cache=cache, confidence_threshold=0.7, ) result = await handler.handle_text(user_id=1, text="посмотреть отчёт") assert isinstance(result, NeedsBrand) assert result.pending_key # не пустой assert "посмотреть отчёт" in result.content_preview @pytest.mark.asyncio async def test_due_set_picks_to_do_list(): extracted = Extracted( brand="life", content="налоги", due=date(2026, 6, 5), platform_hint=None, strategic_flag=False, project_confidence=0.95, ) client = _mock_ticktick("tk_taxes") handler = CaptureHandler( extractor=_mock_extractor(extracted), mapping=_mapping(), ticktick=client, event_log=_mock_event_log(), pending_cache=None, confidence_threshold=0.7, ) result = await handler.handle_text(user_id=1, text="налоги до пятницы") assert isinstance(result, Created) assert result.list_name == "To Do" body = client.post.call_args[0][1] assert body["projectId"] == "p_li_todo" assert body["dueDate"] == "2026-06-05" ``` - [ ] **Step 9.2: Прогнать — упадут** Run: `.venv/bin/pytest tests/v2/test_capture.py -v` Expected: FAIL. - [ ] **Step 9.3: Написать `capture.py`** ```python """Capture handler — основная бизнес-логика приёма текста. Не знает про Telegram. Принимает (user_id, text), отдаёт Created или NeedsBrand. Telegram-wiring (inline buttons) — в v2/app.py. """ import logging from dataclasses import dataclass from datetime import datetime, timezone from tasks_bot.ticktick.mapping import ProjectMapping from tasks_bot.v2.event_log import EventLog from tasks_bot.v2.extraction import Extractor from tasks_bot.v2.pending import PendingCaptureCache, PendingEntry from tasks_bot.v2.routing import resolve_list log = logging.getLogger(__name__) @dataclass class Created: task_id: str brand: str list_name: str @dataclass class NeedsBrand: pending_key: str content_preview: str CaptureResult = Created | NeedsBrand class CaptureHandler: def __init__( self, *, extractor: Extractor, mapping: ProjectMapping, ticktick, # TickTickClient, duck-typed для тестов с моком event_log: EventLog, pending_cache: PendingCaptureCache | None, confidence_threshold: float = 0.7, ): self._extractor = extractor self._mapping = mapping self._ticktick = ticktick self._event_log = event_log self._pending = pending_cache self._threshold = confidence_threshold async def handle_text(self, user_id: int, text: str) -> CaptureResult: extracted = await self._extractor.extract(text) log.info( "extracted brand=%s conf=%.2f due=%s platform=%s strat=%s", extracted.brand, extracted.project_confidence, extracted.due, extracted.platform_hint, extracted.strategic_flag, ) # Low confidence → pending + buttons (заносится в кэш, реальные кнопки # шлёт wiring-слой) if extracted.project_confidence < self._threshold: assert self._pending is not None key = self._pending.put(PendingEntry( user_id=user_id, payload={ "content": extracted.content, "due": extracted.due.isoformat() if extracted.due else None, "platform_hint": extracted.platform_hint, "strategic_flag": extracted.strategic_flag, }, )) return NeedsBrand(pending_key=key, content_preview=extracted.content) return await self._create( brand=extracted.brand, content=extracted.content, due=extracted.due, platform_hint=extracted.platform_hint, strategic_flag=extracted.strategic_flag, ) async def create_from_pending( self, *, brand: str, pending: PendingEntry, ) -> Created: """Вызывается wiring-слоем после нажатия кнопки выбора brand.""" from datetime import date as _date due_iso = pending.payload.get("due") return await self._create( brand=brand, content=pending.payload["content"], due=_date.fromisoformat(due_iso) if due_iso else None, platform_hint=pending.payload.get("platform_hint"), strategic_flag=bool(pending.payload.get("strategic_flag", False)), ) async def _create( self, *, brand: str, content: str, due, platform_hint, strategic_flag: bool, ) -> Created: if brand == "inbox": # Fall through to system Inbox: пустой projectId → TickTick кладёт в inbox body = {"title": content} if due: body["dueDate"] = due.isoformat() result = await self._ticktick.post("task", body) self._log_create(result["id"], content, "inbox", "Inbox") return Created(task_id=result["id"], brand="inbox", list_name="Inbox") list_name = resolve_list( brand, due=due, platform_hint=platform_hint, strategic_flag=strategic_flag, ) project_id = self._mapping.id_for(brand, list_name) body = {"title": content, "projectId": project_id} if due: body["dueDate"] = due.isoformat() result = await self._ticktick.post("task", body) self._log_create(result["id"], content, brand, list_name) return Created(task_id=result["id"], brand=brand, list_name=list_name) def _log_create(self, task_id: str, title: str, brand: str, list_name: str) -> None: self._event_log.write( op="create", task_id=task_id, ts=datetime.now(timezone.utc), title=title, brand=brand, list_name=list_name, ) ``` - [ ] **Step 9.4: Прогнать** Run: `.venv/bin/pytest tests/v2/test_capture.py -v` Expected: 3 passed. - [ ] **Step 9.5: Commit** ```bash git add src/tasks_bot/v2/capture.py tests/v2/test_capture.py git commit -m "v2: CaptureHandler — extract → route → write" ``` --- ## Task 10: Summary cache (N → task_id) **Files:** - Create: `src/tasks_bot/v2/summary_cache.py` - Create: `tests/v2/test_summary_cache.py` aiosqlite-таблица: `(user_id, n, task_id, summary_kind, created_at)`. Когда бот шлёт сводку — каждый номер `N` мапится на TickTick `task_id`, сохраняется. Когда пользователь шлёт `/done 3` — читаем последнюю запись с n=3 и берём task_id. Свежесть: только записи, созданные сегодня (по TZ Yekaterinburg) считаются актуальными. Иначе «Номера устарели». - [ ] **Step 10.1: Тесты** ```python from datetime import datetime, timedelta, timezone from pathlib import Path import pytest from tasks_bot.v2.summary_cache import SummaryCache, StaleNumberError YEKB = timezone(timedelta(hours=5)) @pytest.mark.asyncio async def test_store_and_resolve_roundtrip(tmp_path: Path): cache = SummaryCache(tmp_path / "cache.sqlite", tz=YEKB) await cache.connect() try: now = datetime.now(YEKB) await cache.replace_summary( user_id=1, kind="today", entries=[(1, "tk_a"), (2, "tk_b")], at=now, ) assert await cache.resolve(user_id=1, n=2, now=now) == "tk_b" finally: await cache.close() @pytest.mark.asyncio async def test_resolve_stale_raises(tmp_path: Path): cache = SummaryCache(tmp_path / "cache.sqlite", tz=YEKB) await cache.connect() try: yesterday = datetime.now(YEKB) - timedelta(days=2) await cache.replace_summary( user_id=1, kind="today", entries=[(1, "tk_old")], at=yesterday, ) with pytest.raises(StaleNumberError): await cache.resolve(user_id=1, n=1, now=datetime.now(YEKB)) finally: await cache.close() @pytest.mark.asyncio async def test_replace_summary_overwrites_prev(tmp_path: Path): cache = SummaryCache(tmp_path / "cache.sqlite", tz=YEKB) await cache.connect() try: now = datetime.now(YEKB) await cache.replace_summary(user_id=1, kind="today", entries=[(1, "tk_a"), (2, "tk_b")], at=now) # Заменяем — теперь только одна запись await cache.replace_summary(user_id=1, kind="today", entries=[(1, "tk_c")], at=now) assert await cache.resolve(user_id=1, n=1, now=now) == "tk_c" with pytest.raises(KeyError): await cache.resolve(user_id=1, n=2, now=now) finally: await cache.close() ``` - [ ] **Step 10.2: Прогнать — упадут** Run: `.venv/bin/pytest tests/v2/test_summary_cache.py -v` Expected: FAIL. - [ ] **Step 10.3: Написать** ```python """Кэш `N → task_id` для нумерованных команд (/done 3 и др.). Persistent (aiosqlite) чтобы переживать рестарт Pi. Запись считается устаревшей, если она из другого календарного дня (TZ Екб). """ from datetime import datetime, timezone from pathlib import Path import aiosqlite class StaleNumberError(LookupError): pass SCHEMA = """ CREATE TABLE IF NOT EXISTS summary_entries ( user_id INTEGER NOT NULL, n INTEGER NOT NULL, task_id TEXT NOT NULL, kind TEXT NOT NULL, created_at TEXT NOT NULL, PRIMARY KEY (user_id, n) ); """ class SummaryCache: def __init__(self, db_path: Path | str, tz: timezone): self._path = str(db_path) self._tz = tz self._db: aiosqlite.Connection | None = None async def connect(self) -> None: Path(self._path).parent.mkdir(parents=True, exist_ok=True) self._db = await aiosqlite.connect(self._path) await self._db.executescript(SCHEMA) await self._db.commit() async def close(self) -> None: if self._db: await self._db.close() self._db = None async def replace_summary( self, *, user_id: int, kind: str, entries: list[tuple[int, str]], # [(n, task_id), ...] at: datetime, ) -> None: assert self._db await self._db.execute( "DELETE FROM summary_entries WHERE user_id = ?", (user_id,) ) for n, task_id in entries: await self._db.execute( "INSERT INTO summary_entries (user_id, n, task_id, kind, created_at) " "VALUES (?, ?, ?, ?, ?)", (user_id, n, task_id, kind, at.isoformat()), ) await self._db.commit() async def resolve(self, *, user_id: int, n: int, now: datetime) -> str: assert self._db async with self._db.execute( "SELECT task_id, created_at FROM summary_entries " "WHERE user_id = ? AND n = ?", (user_id, n), ) as cur: row = await cur.fetchone() if not row: raise KeyError(f"no entry for n={n}") task_id, created_str = row created_at = datetime.fromisoformat(created_str) if created_at.astimezone(self._tz).date() != now.astimezone(self._tz).date(): raise StaleNumberError(f"summary is stale (was {created_at})") return task_id ``` - [ ] **Step 10.4: Прогнать** Run: `.venv/bin/pytest tests/v2/test_summary_cache.py -v` Expected: 3 passed. - [ ] **Step 10.5: Commit** ```bash git add src/tasks_bot/v2/summary_cache.py tests/v2/test_summary_cache.py git commit -m "v2: persistent N→task_id summary cache" ``` --- ## Task 11: Summary formatter **Files:** - Create: `src/tasks_bot/v2/summary_format.py` - Create: `tests/v2/test_summary_format.py` Pure-function форматтер: на вход `list[TickTickTask]` (просто dict-ы) + сегодняшняя дата (TZ Екб), на выходе `(text, [task_ids])` — текст для Telegram и список task_id-ов в том же порядке, чтобы кэшить N→id. Группирует по brand в фиксированном порядке: mars → channel → consulting → life → vibe-coding. Внутри — по статусу: overdue → today → ahead. Эмодзи 🔴/🔥/📅. Если задача не в дефолтном списке (`To Do`), показываем `(Threads)` / `(Бэклог)` и т.п. - [ ] **Step 11.1: Тесты** ```python from datetime import date from tasks_bot.v2.summary_format import format_summary, TaskRow def _row(**kwargs) -> TaskRow: defaults = dict( task_id="t", title="x", brand="mars", list_name="To Do", due=None, ) defaults.update(kwargs) return TaskRow(**defaults) def test_empty_summary(): text, ids = format_summary([], today=date(2026, 6, 1)) assert "Сегодня свободно" in text or "Нет задач" in text assert ids == [] def test_grouped_by_brand_and_status(): rows = [ _row(task_id="m1", title="позвонить педагогу", brand="mars", list_name="To Do", due=date(2026, 6, 1)), _row(task_id="m2", title="сдать отчёт", brand="mars", list_name="To Do", due=date(2026, 5, 28)), _row(task_id="c1", title="пост про хор", brand="channel", list_name="Threads", due=date(2026, 6, 1)), ] text, ids = format_summary(rows, today=date(2026, 6, 1)) # Все три номера в выходе assert "1." in text and "2." in text and "3." in text # Mars впереди channel (порядок брендов) assert text.index("mars") < text.index("channel") # Overdue перед today assert text.index("🔴") < text.index("🔥") # Не дефолтный list — в скобках assert "(Threads)" in text # ids в порядке вывода assert ids == ["m2", "m1", "c1"] def test_status_emoji_for_ahead(): rows = [_row(task_id="a", title="завтра", brand="life", list_name="To Do", due=date(2026, 6, 3))] text, ids = format_summary(rows, today=date(2026, 6, 1)) assert "📅" in text assert ids == ["a"] ``` - [ ] **Step 11.2: Прогнать — упадут** Run: `.venv/bin/pytest tests/v2/test_summary_format.py -v` Expected: FAIL. - [ ] **Step 11.3: Написать** ```python """Форматтер списка задач для сводок и /today. Группирует по brand в фиксированном порядке (mars/channel/consulting/ life/vibe-coding), внутри — по статусу (overdue → today → ahead). """ from dataclasses import dataclass from datetime import date BRAND_ORDER = ["mars", "channel", "consulting", "life", "vibe-coding"] DEFAULT_LIST_PER_BRAND = { "mars": "To Do", "channel": "To Do", "consulting": "To Do", "life": "To Do", "vibe-coding": "To Do", } @dataclass class TaskRow: task_id: str title: str brand: str list_name: str due: date | None def _status_emoji(due: date | None, today: date) -> str: if due is None: return "📅" # на самом деле не должно показываться (без срока — не лезет в утро) if due < today: return "🔴" if due == today: return "🔥" return "📅" def _status_order(due: date | None, today: date) -> int: if due is None: return 3 if due < today: return 0 if due == today: return 1 return 2 def format_summary(rows: list[TaskRow], today: date) -> tuple[str, list[str]]: if not rows: return "Сегодня свободно 🎉", [] # Группировка by_brand: dict[str, list[TaskRow]] = {b: [] for b in BRAND_ORDER} for r in rows: if r.brand not in by_brand: continue # игнорируем неизвестные бренды (например inbox в сводке не показываем) by_brand[r.brand].append(r) # Внутри группы — сортируем по статусу for brand_rows in by_brand.values(): brand_rows.sort(key=lambda r: _status_order(r.due, today)) lines: list[str] = [] ids: list[str] = [] counter = 0 for brand in BRAND_ORDER: brand_rows = by_brand[brand] if not brand_rows: continue lines.append(brand) for r in brand_rows: counter += 1 emoji = _status_emoji(r.due, today) list_suffix = ( f" ({r.list_name})" if r.list_name != DEFAULT_LIST_PER_BRAND.get(brand) else "" ) lines.append(f" {emoji} {counter}. {r.title}{list_suffix}") ids.append(r.task_id) lines.append("") return "\n".join(lines).rstrip(), ids ``` - [ ] **Step 11.4: Прогнать** Run: `.venv/bin/pytest tests/v2/test_summary_format.py -v` Expected: 3 passed. - [ ] **Step 11.5: Commit** ```bash git add src/tasks_bot/v2/summary_format.py tests/v2/test_summary_format.py git commit -m "v2: summary formatter (groups by brand, sorts by status)" ``` --- ## Task 12: Commands — /today **Files:** - Create: `src/tasks_bot/v2/commands.py` - Create: `tests/v2/test_commands.py` В `commands.py` будут все 9 команд. Начнём с `/today` — он наиболее завязан на formatter и mapping. Другие команды добавляются последовательно в Tasks 13-19. `/today` собирает все открытые задачи в `To Do` всех проектов + платформенные Channel с `due ≤ today`, форматирует через `summary_format.format_summary`, сохраняет `N→task_id` в `summary_cache.replace_summary`. - [ ] **Step 12.1: Тест** ```python from datetime import date, datetime, timezone, timedelta from pathlib import Path from unittest.mock import AsyncMock, MagicMock import pytest from tasks_bot.ticktick.mapping import ProjectMapping from tasks_bot.v2.commands import Commands YEKB = timezone(timedelta(hours=5)) FIXTURE = Path(__file__).parent.parent / "ticktick" / "fixtures" / "projects_mapping.json" @pytest.mark.asyncio async def test_today_returns_grouped_summary(tmp_path): # TickTick возвращает 4 задачи в разных проектах mapping = ProjectMapping.load(FIXTURE) def projects_data(project_id): # Возвращаем разные dict-ы в зависимости от project_id bag = { "p_mars_todo": {"tasks": [ {"id": "m1", "title": "позвонить педагогу", "dueDate": "2026-06-01T00:00:00+0500", "status": 0}, ]}, "p_ch_threads": {"tasks": [ {"id": "c1", "title": "пост про хор", "dueDate": "2026-06-01T00:00:00+0500", "status": 0}, ]}, "p_li_todo": {"tasks": [ {"id": "l1", "title": "налоги", "dueDate": "2026-06-05T00:00:00+0500", "status": 0}, ]}, } return bag.get(project_id, {"tasks": []}) ticktick = MagicMock() async def get_side_effect(path): if path.startswith("project/") and path.endswith("/data"): pid = path.split("/")[1] return projects_data(pid) return None ticktick.get = AsyncMock(side_effect=get_side_effect) from tasks_bot.v2.summary_cache import SummaryCache cache = SummaryCache(tmp_path / "c.sqlite", tz=YEKB) await cache.connect() try: cmds = Commands( ticktick=ticktick, mapping=mapping, summary_cache=cache, tz=YEKB, ) # Зафиксируем «сегодня» = 2026-06-01 text, ids = await cmds.today(user_id=1, today=date(2026, 6, 1)) assert "1." in text and "позвонить педагогу" in text assert "пост про хор" in text # Channel/Threads тоже показан # Налоги в life с due=2026-06-05 — попадает (≤ end_of_week) assert "налоги" in text # Кэш заполнен assert await cache.resolve(user_id=1, n=1, now=datetime.now(YEKB)) finally: await cache.close() ``` - [ ] **Step 12.2: Прогнать — упадут** Run: `.venv/bin/pytest tests/v2/test_commands.py -v` Expected: FAIL. - [ ] **Step 12.3: Написать `commands.py` (только `/today` и инициализация)** ```python """V2 commands: /today /inbox /backlog /done /defer /text /proj /date /del. Все методы возвращают tuple[str, list[str]] или str. Telegram-wiring (превращение в reply) делается в v2/app.py. """ import logging from dataclasses import dataclass from datetime import date, datetime, timedelta, timezone from tasks_bot.ticktick.mapping import ProjectMapping from tasks_bot.v2.summary_cache import SummaryCache from tasks_bot.v2.summary_format import TaskRow, format_summary log = logging.getLogger(__name__) # Список с дефолтными списками каждого бренда для /today TO_DO_LISTS = ["To Do"] PLATFORM_LISTS_CHANNEL = ["Threads", "Instagram", "Канал"] def _end_of_week(today: date) -> date: # Неделя начинается с понедельника; конец — воскресенье days_to_sunday = 6 - today.weekday() return today + timedelta(days=days_to_sunday) def _parse_due(raw: str | None) -> date | None: if not raw: return None # TickTick возвращает ISO с offset — берём только дату return date.fromisoformat(raw[:10]) class Commands: def __init__( self, *, ticktick, mapping: ProjectMapping, summary_cache: SummaryCache, tz: timezone, ): self._tk = ticktick self._mapping = mapping self._cache = summary_cache self._tz = tz async def today(self, *, user_id: int, today: date) -> tuple[str, list[str]]: eow = _end_of_week(today) rows: list[TaskRow] = [] # 1. To Do для каждого brand for brand in self._mapping.brands(): try: pid = self._mapping.id_for(brand, "To Do") except LookupError: continue data = await self._tk.get(f"project/{pid}/data") for t in (data or {}).get("tasks", []): if t.get("status", 0) != 0: continue due = _parse_due(t.get("dueDate")) if due is None or due > eow: continue rows.append(TaskRow( task_id=t["id"], title=t["title"], brand=brand, list_name="To Do", due=due, )) # 2. Платформенные списки Channel с due ≤ today for lst in PLATFORM_LISTS_CHANNEL: try: pid = self._mapping.id_for("channel", lst) except LookupError: continue data = await self._tk.get(f"project/{pid}/data") for t in (data or {}).get("tasks", []): if t.get("status", 0) != 0: continue due = _parse_due(t.get("dueDate")) if due is None or due > today: continue rows.append(TaskRow( task_id=t["id"], title=t["title"], brand="channel", list_name=lst, due=due, )) text, ids = format_summary(rows, today=today) # Сохраняем кэш N→task_id entries = [(i + 1, tid) for i, tid in enumerate(ids)] await self._cache.replace_summary( user_id=user_id, kind="today", entries=entries, at=datetime.now(self._tz), ) return text, ids ``` - [ ] **Step 12.4: Прогнать** Run: `.venv/bin/pytest tests/v2/test_commands.py -v` Expected: 1 passed. - [ ] **Step 12.5: Commit** ```bash git add src/tasks_bot/v2/commands.py tests/v2/test_commands.py git commit -m "v2: /today command with grouping and cache" ``` --- ## Task 13: Commands — /inbox, /backlog (read-only) **Files:** - Modify: `src/tasks_bot/v2/commands.py` - Modify: `tests/v2/test_commands.py` `/inbox` показывает системный TickTick Inbox (project_id отсутствует у задач TickTick Inbox). По API: `GET /project` обычно НЕ возвращает системный Inbox (это особый). Тогда `GET /project/{inbox_id}/data` либо специальный endpoint. В preflight мы это не проверяли — нужно понять опытным путём. Пока: в `commands.py` добавим метод `inbox()` который читает специальный `inbox` (TickTick задачи без projectId часто адресуются через `inbox_id` — Наташа может узнать его из своего mapping или из `GET /project`). Если не найдётся — fallback на «inbox пока не поддержан, открой TickTick». `/backlog [project]` — показывает `Бэклог` и `Стратегия` (если есть в проекте). Параметр опциональный: без него — по всем проектам. - [ ] **Step 13.1: Тесты для /backlog** В `tests/v2/test_commands.py` добавить: ```python @pytest.mark.asyncio async def test_backlog_all_brands(tmp_path): mapping = ProjectMapping.load(FIXTURE) bag = { "p_mars_back": {"tasks": [ {"id": "mb1", "title": "пересмотреть конкурс", "status": 0, "dueDate": None}, ]}, "p_mars_strat": {"tasks": [ {"id": "ms1", "title": "развитие выпускников", "status": 0, "dueDate": None}, ]}, } ticktick = MagicMock() async def get(path): if path.startswith("project/") and path.endswith("/data"): pid = path.split("/")[1] return bag.get(pid, {"tasks": []}) return None ticktick.get = AsyncMock(side_effect=get) cache = SummaryCache(tmp_path / "c.sqlite", tz=YEKB) await cache.connect() try: cmds = Commands(ticktick=ticktick, mapping=mapping, summary_cache=cache, tz=YEKB) text, ids = await cmds.backlog(user_id=1, brand=None, today=date(2026, 6, 1)) assert "пересмотреть конкурс" in text assert "развитие выпускников" in text assert len(ids) == 2 finally: await cache.close() ``` - [ ] **Step 13.2: Добавить методы в `commands.py`** После `today` добавить: ```python async def backlog( self, *, user_id: int, brand: str | None, today: date, ) -> tuple[str, list[str]]: """Показать `Бэклог` и `Стратегия`. Если brand=None — по всем.""" rows: list[TaskRow] = [] brands_to_check = [brand] if brand else self._mapping.brands() for b in brands_to_check: for lst in ("Бэклог", "Стратегия"): try: pid = self._mapping.id_for(b, lst) except LookupError: continue data = await self._tk.get(f"project/{pid}/data") for t in (data or {}).get("tasks", []): if t.get("status", 0) != 0: continue rows.append(TaskRow( task_id=t["id"], title=t["title"], brand=b, list_name=lst, due=_parse_due(t.get("dueDate")), )) text, ids = format_summary(rows, today=today) entries = [(i + 1, tid) for i, tid in enumerate(ids)] await self._cache.replace_summary( user_id=user_id, kind="backlog", entries=entries, at=datetime.now(self._tz), ) return text, ids async def inbox(self, *, user_id: int, today: date) -> tuple[str, list[str]]: """Read-only TickTick Inbox. TickTick Inbox = задачи без projectId или с projectId = "inbox" (зависит от API). Сначала пробуем GET /project и ищем системный Inbox (обычно name == "Inbox" и/или kind == "INBOX"). Если такой найден — читаем его /data. Иначе сообщаем что не нашли. """ all_projects = await self._tk.get("project") or [] inbox_id = None for p in all_projects: if p.get("name", "").lower() == "inbox" or p.get("kind") == "INBOX": inbox_id = p["id"] break if not inbox_id: return "Системный Inbox не найден в API.", [] data = await self._tk.get(f"project/{inbox_id}/data") rows: list[TaskRow] = [] for t in (data or {}).get("tasks", []): if t.get("status", 0) != 0: continue rows.append(TaskRow( task_id=t["id"], title=t["title"], brand="inbox", list_name="Inbox", due=_parse_due(t.get("dueDate")), )) if not rows: return "Inbox пуст 📥", [] # Не используем общий format_summary (он группирует по brand); # для inbox — простой нумерованный список. lines = ["📥 inbox"] ids: list[str] = [] for i, r in enumerate(rows, 1): lines.append(f" {i}. {r.title}") ids.append(r.task_id) entries = [(i + 1, tid) for i, tid in enumerate(ids)] await self._cache.replace_summary( user_id=user_id, kind="inbox", entries=entries, at=datetime.now(self._tz), ) return "\n".join(lines), ids ``` - [ ] **Step 13.3: Прогнать** Run: `.venv/bin/pytest tests/v2/test_commands.py -v` Expected: 2 passed. - [ ] **Step 13.4: Commit** ```bash git add src/tasks_bot/v2/commands.py tests/v2/test_commands.py git commit -m "v2: /inbox and /backlog commands" ``` --- ## Task 14: Commands — /done (with re-check) **Files:** - Modify: `src/tasks_bot/v2/commands.py` - Modify: `tests/v2/test_commands.py` `/done N` — закрывает задачу N. Перед мутацией читает её из TickTick: если не существует или уже closed — отвечает понятно. Если ок — `POST /project/{pid}/task/{tid}/complete` + пишет в event log `op=complete`. - [ ] **Step 14.1: Тест** В `test_commands.py` добавить: ```python @pytest.mark.asyncio async def test_done_completes_task(tmp_path): mapping = ProjectMapping.load(FIXTURE) ticktick = MagicMock() # Сначала /today заполнит кэш async def get(path): if path == "project/p_li_todo/data": return {"tasks": [ {"id": "tk_taxes", "title": "налоги", "status": 0, "dueDate": "2026-06-01T00:00:00+0500", "projectId": "p_li_todo"}, ]} if path == "task/tk_taxes": return {"id": "tk_taxes", "projectId": "p_li_todo", "status": 0, "title": "налоги"} if path.endswith("/data"): return {"tasks": []} return None ticktick.get = AsyncMock(side_effect=get) ticktick.post = AsyncMock(return_value=None) from tasks_bot.v2.event_log import EventLog event_log = EventLog(tmp_path / "events.jsonl") cache = SummaryCache(tmp_path / "c.sqlite", tz=YEKB) await cache.connect() try: cmds = Commands(ticktick=ticktick, mapping=mapping, summary_cache=cache, tz=YEKB, event_log=event_log) await cmds.today(user_id=1, today=date(2026, 6, 1)) # /done 1 result = await cmds.done(user_id=1, n=1, now=datetime.now(YEKB)) assert "✓" in result or "закрыта" in result.lower() # POST /project/p_li_todo/task/tk_taxes/complete был вызван ticktick.post.assert_called_with( "project/p_li_todo/task/tk_taxes/complete", None, ) finally: await cache.close() ``` - [ ] **Step 14.2: Добавить `done` в `Commands` (+ принять `event_log` в конструкторе)** Заменить конструктор `Commands.__init__` на: ```python def __init__( self, *, ticktick, mapping: ProjectMapping, summary_cache: SummaryCache, tz: timezone, event_log=None, # Optional; required for mutations ): self._tk = ticktick self._mapping = mapping self._cache = summary_cache self._tz = tz self._event_log = event_log ``` И добавить метод: ```python async def done(self, *, user_id: int, n: int, now: datetime) -> str: try: task_id = await self._cache.resolve(user_id=user_id, n=n, now=now) except StaleNumberError: return "Номера устарели, пришли /today" except KeyError: return f"Не нашла задачу #{n}" # Re-check task = await self._tk.get(f"task/{task_id}") if not task: return f"Задача #{n} уже удалена в TickTick" if task.get("status", 0) != 0: return f"Задача #{n} уже закрыта" project_id = task["projectId"] await self._tk.post(f"project/{project_id}/task/{task_id}/complete", None) if self._event_log: brand_list = self._mapping.reverse(project_id) brand = brand_list[0] if brand_list else "?" list_name = brand_list[1] if brand_list else "?" self._event_log.write( op="complete", task_id=task_id, ts=now, title=task.get("title", ""), brand=brand, list_name=list_name, ) return f"✓ #{n} закрыта" ``` Также добавь импорт `StaleNumberError` в начале `commands.py`: ```python from tasks_bot.v2.summary_cache import StaleNumberError, SummaryCache ``` - [ ] **Step 14.3: Прогнать** Run: `.venv/bin/pytest tests/v2/test_commands.py -v` Expected: 3 passed. - [ ] **Step 14.4: Commit** ```bash git add src/tasks_bot/v2/commands.py tests/v2/test_commands.py git commit -m "v2: /done command with re-check before mutation" ``` --- ## Task 15: Commands — /defer, /date (alias) **Files:** - Modify: `src/tasks_bot/v2/commands.py` - Modify: `tests/v2/test_commands.py` `/defer N ` — изменить срок. `when`: `today`, `week`, `none`, `YYYY-MM-DD`. - `today` → due = сегодня (TZ Екб) - `week` → due = воскресенье текущей недели - `none` → due = null (убрать срок). Это означает что задача переедет визуально из `To Do` в `Бэклог`, если applies. **Бот её НЕ переносит**, только сбрасывает `dueDate` — это уже UI-задача пользователя. Если бренд = `mars/vibe-coding/consulting/life` и `due → none`, корректнее было бы переместить в `Бэклог` — реализуем именно так. - `YYYY-MM-DD` → due = эта дата `/date` — алиас. - [ ] **Step 15.1: Тест** ```python @pytest.mark.asyncio async def test_defer_to_specific_date(tmp_path): mapping = ProjectMapping.load(FIXTURE) ticktick = MagicMock() async def get(path): if path == "project/p_li_todo/data": return {"tasks": [{"id": "t1", "title": "x", "status": 0, "dueDate": "2026-06-01T00:00:00+0500", "projectId": "p_li_todo"}]} if path == "task/t1": return {"id": "t1", "projectId": "p_li_todo", "status": 0, "title": "x", "dueDate": "2026-06-01T00:00:00+0500"} if path.endswith("/data"): return {"tasks": []} return None ticktick.get = AsyncMock(side_effect=get) ticktick.post = AsyncMock(return_value={"id": "t1"}) from tasks_bot.v2.event_log import EventLog cache = SummaryCache(tmp_path / "c.sqlite", tz=YEKB) await cache.connect() try: cmds = Commands(ticktick=ticktick, mapping=mapping, summary_cache=cache, tz=YEKB, event_log=EventLog(tmp_path / "e.jsonl")) await cmds.today(user_id=1, today=date(2026, 6, 1)) result = await cmds.defer(user_id=1, n=1, when="2026-06-10", now=datetime(2026, 6, 1, 9, 0, tzinfo=YEKB)) assert "10" in result body = ticktick.post.call_args[0][1] assert body["dueDate"].startswith("2026-06-10") finally: await cache.close() ``` - [ ] **Step 15.2: Добавить `defer` + alias `date`** ```python async def defer( self, *, user_id: int, n: int, when: str, now: datetime, ) -> str: try: task_id = await self._cache.resolve(user_id=user_id, n=n, now=now) except StaleNumberError: return "Номера устарели, пришли /today" except KeyError: return f"Не нашла задачу #{n}" task = await self._tk.get(f"task/{task_id}") if not task: return f"Задача #{n} уже удалена" if task.get("status", 0) != 0: return f"Задача #{n} уже закрыта" # Парсим when today_local = now.astimezone(self._tz).date() new_due: date | None if when == "today": new_due = today_local elif when == "week": new_due = _end_of_week(today_local) elif when == "none": new_due = None else: try: new_due = date.fromisoformat(when) except ValueError: return ( f"Не поняла «{when}». Используй: today | week | none | " f"YYYY-MM-DD" ) body: dict = {"id": task_id, "projectId": task["projectId"]} if new_due: body["dueDate"] = f"{new_due.isoformat()}T00:00:00+0500" else: body["dueDate"] = None await self._tk.post(f"task/{task_id}", body) if self._event_log: brand_list = self._mapping.reverse(task["projectId"]) self._event_log.write( op="update", task_id=task_id, ts=now, title=task.get("title", ""), brand=brand_list[0] if brand_list else "?", list_name=brand_list[1] if brand_list else "?", extra={"due": new_due.isoformat() if new_due else None}, ) if new_due: return f"📅 #{n} → {new_due.isoformat()}" return f"📅 #{n} → без срока" # alias async def date( self, *, user_id: int, n: int, when: str, now: datetime, ) -> str: return await self.defer(user_id=user_id, n=n, when=when, now=now) ``` - [ ] **Step 15.3: Прогнать** Run: `.venv/bin/pytest tests/v2/test_commands.py -v` Expected: 4 passed. - [ ] **Step 15.4: Commit** ```bash git add src/tasks_bot/v2/commands.py tests/v2/test_commands.py git commit -m "v2: /defer and /date (alias) commands" ``` --- ## Task 16: Commands — /text **Files:** - Modify: `src/tasks_bot/v2/commands.py` - Modify: `tests/v2/test_commands.py` `/text N ` — изменить заголовок задачи. - [ ] **Step 16.1: Тест** ```python @pytest.mark.asyncio async def test_text_updates_title(tmp_path): mapping = ProjectMapping.load(FIXTURE) ticktick = MagicMock() async def get(path): if path == "project/p_li_todo/data": return {"tasks": [{"id": "t1", "title": "old", "status": 0, "dueDate": "2026-06-01T00:00:00+0500"}]} if path == "task/t1": return {"id": "t1", "projectId": "p_li_todo", "title": "old", "status": 0} if path.endswith("/data"): return {"tasks": []} return None ticktick.get = AsyncMock(side_effect=get) ticktick.post = AsyncMock(return_value={"id": "t1"}) from tasks_bot.v2.event_log import EventLog cache = SummaryCache(tmp_path / "c.sqlite", tz=YEKB) await cache.connect() try: cmds = Commands(ticktick=ticktick, mapping=mapping, summary_cache=cache, tz=YEKB, event_log=EventLog(tmp_path / "e.jsonl")) await cmds.today(user_id=1, today=date(2026, 6, 1)) result = await cmds.text(user_id=1, n=1, new_text="new title", now=datetime.now(YEKB)) assert "new title" in result body = ticktick.post.call_args[0][1] assert body["title"] == "new title" finally: await cache.close() ``` - [ ] **Step 16.2: Метод** ```python async def text( self, *, user_id: int, n: int, new_text: str, now: datetime, ) -> str: try: task_id = await self._cache.resolve(user_id=user_id, n=n, now=now) except StaleNumberError: return "Номера устарели, пришли /today" except KeyError: return f"Не нашла задачу #{n}" task = await self._tk.get(f"task/{task_id}") if not task: return f"Задача #{n} уже удалена" if task.get("status", 0) != 0: return f"Задача #{n} уже закрыта" await self._tk.post( f"task/{task_id}", {"id": task_id, "projectId": task["projectId"], "title": new_text}, ) if self._event_log: brand_list = self._mapping.reverse(task["projectId"]) self._event_log.write( op="update", task_id=task_id, ts=now, title=new_text, brand=brand_list[0] if brand_list else "?", list_name=brand_list[1] if brand_list else "?", ) return f"✏ #{n} → «{new_text}»" ``` - [ ] **Step 16.3: Прогнать + commit** ```bash .venv/bin/pytest tests/v2/test_commands.py -v git add src/tasks_bot/v2/commands.py tests/v2/test_commands.py git commit -m "v2: /text command" ``` --- ## Task 17: Commands — /proj (with routing) **Files:** - Modify: `src/tasks_bot/v2/commands.py` - Modify: `tests/v2/test_commands.py` `/proj N ` — перенести в другой бренд. По спеку: routing-правила пересчитываются (`due есть → To Do`, нет → `Бэклог`); platform_hint и strategic_flag сбрасываются. - [ ] **Step 17.1: Тест** ```python @pytest.mark.asyncio async def test_proj_moves_task(tmp_path): mapping = ProjectMapping.load(FIXTURE) ticktick = MagicMock() async def get(path): if path == "project/p_li_todo/data": return {"tasks": [{"id": "t1", "title": "x", "status": 0, "dueDate": "2026-06-01T00:00:00+0500"}]} if path == "task/t1": return {"id": "t1", "projectId": "p_li_todo", "title": "x", "status": 0, "dueDate": "2026-06-01T00:00:00+0500"} if path.endswith("/data"): return {"tasks": []} return None ticktick.get = AsyncMock(side_effect=get) ticktick.post = AsyncMock(return_value={"id": "t1"}) from tasks_bot.v2.event_log import EventLog cache = SummaryCache(tmp_path / "c.sqlite", tz=YEKB) await cache.connect() try: cmds = Commands(ticktick=ticktick, mapping=mapping, summary_cache=cache, tz=YEKB, event_log=EventLog(tmp_path / "e.jsonl")) await cmds.today(user_id=1, today=date(2026, 6, 1)) # Move life → mars; due есть, значит → Mars To Do result = await cmds.proj(user_id=1, n=1, target_brand="mars", now=datetime.now(YEKB)) assert "mars" in result.lower() body = ticktick.post.call_args[0][1] assert body["projectId"] == "p_mars_todo" finally: await cache.close() ``` - [ ] **Step 17.2: Метод** ```python async def proj( self, *, user_id: int, n: int, target_brand: str, now: datetime, ) -> str: from tasks_bot.v2.routing import resolve_list try: task_id = await self._cache.resolve(user_id=user_id, n=n, now=now) except StaleNumberError: return "Номера устарели, пришли /today" except KeyError: return f"Не нашла задачу #{n}" task = await self._tk.get(f"task/{task_id}") if not task: return f"Задача #{n} уже удалена" if task.get("status", 0) != 0: return f"Задача #{n} уже закрыта" # Routing без platform_hint/strategic_flag due_raw = task.get("dueDate") due = _parse_due(due_raw) list_name = resolve_list( target_brand, due=due, platform_hint=None, strategic_flag=False, ) try: new_pid = self._mapping.id_for(target_brand, list_name) except LookupError: return f"Не знаю проект {target_brand} / {list_name}" await self._tk.post( f"task/{task_id}", {"id": task_id, "projectId": new_pid}, ) if self._event_log: self._event_log.write( op="update", task_id=task_id, ts=now, title=task.get("title", ""), brand=target_brand, list_name=list_name, ) return f"📂 #{n} → {target_brand} / {list_name}" ``` - [ ] **Step 17.3: Прогнать + commit** ```bash .venv/bin/pytest tests/v2/test_commands.py -v git add src/tasks_bot/v2/commands.py tests/v2/test_commands.py git commit -m "v2: /proj command (with routing recompute)" ``` --- ## Task 18: Commands — /del **Files:** - Modify: `src/tasks_bot/v2/commands.py` - Modify: `tests/v2/test_commands.py` `/del N` — удалить задачу. DELETE `/project/{pid}/task/{tid}`. - [ ] **Step 18.1: Тест** ```python @pytest.mark.asyncio async def test_del_removes_task(tmp_path): mapping = ProjectMapping.load(FIXTURE) ticktick = MagicMock() async def get(path): if path == "project/p_li_todo/data": return {"tasks": [{"id": "t1", "title": "x", "status": 0, "dueDate": "2026-06-01T00:00:00+0500"}]} if path == "task/t1": return {"id": "t1", "projectId": "p_li_todo", "title": "x", "status": 0} if path.endswith("/data"): return {"tasks": []} return None ticktick.get = AsyncMock(side_effect=get) ticktick.delete = AsyncMock(return_value=None) from tasks_bot.v2.event_log import EventLog cache = SummaryCache(tmp_path / "c.sqlite", tz=YEKB) await cache.connect() try: cmds = Commands(ticktick=ticktick, mapping=mapping, summary_cache=cache, tz=YEKB, event_log=EventLog(tmp_path / "e.jsonl")) await cmds.today(user_id=1, today=date(2026, 6, 1)) result = await cmds.delete(user_id=1, n=1, now=datetime.now(YEKB)) assert "удалена" in result.lower() ticktick.delete.assert_called_with("project/p_li_todo/task/t1") finally: await cache.close() ``` - [ ] **Step 18.2: Метод** ```python async def delete( self, *, user_id: int, n: int, now: datetime, ) -> str: try: task_id = await self._cache.resolve(user_id=user_id, n=n, now=now) except StaleNumberError: return "Номера устарели, пришли /today" except KeyError: return f"Не нашла задачу #{n}" task = await self._tk.get(f"task/{task_id}") if not task: return f"Задача #{n} уже удалена" await self._tk.delete(f"project/{task['projectId']}/task/{task_id}") if self._event_log: brand_list = self._mapping.reverse(task["projectId"]) self._event_log.write( op="delete", task_id=task_id, ts=now, title=task.get("title", ""), brand=brand_list[0] if brand_list else "?", list_name=brand_list[1] if brand_list else "?", ) return f"🗑 #{n} удалена" ``` - [ ] **Step 18.3: Прогнать + commit** ```bash .venv/bin/pytest tests/v2/test_commands.py -v git add src/tasks_bot/v2/commands.py tests/v2/test_commands.py git commit -m "v2: /del command" ``` --- ## Task 19: Telegram wiring — v2 app module **Files:** - Create: `src/tasks_bot/v2/app.py` - Modify: `src/tasks_bot/bot.py` - Create: `tests/v2/test_app_smoke.py` Регистрирует MessageHandler (для capture) и 9 CommandHandlers против `Commands`. Также CallbackQueryHandler для inline-кнопок выбора brand. В `bot.py` `build_application` — ветвление: если `settings.v2_capture_enabled` — вызывает `register_v2_handlers(app, ...)` и **пропускает** v1-хендлеры. - [ ] **Step 19.1: Написать `v2/app.py`** ```python """Регистрация v2-хендлеров в python-telegram-bot Application. Включается в bot.py при v2_capture_enabled=True. Регистрирует: - MessageHandler(filters.TEXT & ~filters.COMMAND) → CaptureHandler - CallbackQueryHandler(pattern=brand_pick) → выбор brand из inline-кнопок - CommandHandler для каждой из 9 команд """ import logging from datetime import datetime from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update from telegram.ext import ( Application, CallbackQueryHandler, CommandHandler, ContextTypes, MessageHandler, filters, ) from tasks_bot.v2.capture import CaptureHandler, Created, NeedsBrand from tasks_bot.v2.commands import Commands log = logging.getLogger(__name__) CALLBACK_PREFIX = "v2pick:" def _build_brand_keyboard(pending_key: str) -> InlineKeyboardMarkup: rows = [ [ InlineKeyboardButton("mars", callback_data=f"{CALLBACK_PREFIX}mars:{pending_key}"), InlineKeyboardButton("channel", callback_data=f"{CALLBACK_PREFIX}channel:{pending_key}"), ], [ InlineKeyboardButton("consulting", callback_data=f"{CALLBACK_PREFIX}consulting:{pending_key}"), InlineKeyboardButton("life", callback_data=f"{CALLBACK_PREFIX}life:{pending_key}"), ], [ InlineKeyboardButton("vibe-coding", callback_data=f"{CALLBACK_PREFIX}vibe-coding:{pending_key}"), InlineKeyboardButton("📥 inbox", callback_data=f"{CALLBACK_PREFIX}inbox:{pending_key}"), ], ] return InlineKeyboardMarkup(rows) def register_v2_handlers( app: Application, *, capture: CaptureHandler, commands: Commands, owner_id: int, tz, ) -> None: def _owner_only(update: Update) -> bool: return update.effective_user and update.effective_user.id == owner_id async def on_text(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: if not _owner_only(update) or not update.message: return result = await capture.handle_text( user_id=update.effective_user.id, text=update.message.text, ) if isinstance(result, Created): list_suffix = f" / {result.list_name}" if result.list_name != "To Do" else "" await update.message.reply_text( f"✓ создано в {result.brand}{list_suffix}", ) elif isinstance(result, NeedsBrand): await update.message.reply_text( f"Куда положить «{result.content_preview}»?", reply_markup=_build_brand_keyboard(result.pending_key), ) async def on_brand_pick(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: if not _owner_only(update): return q = update.callback_query await q.answer() # callback_data = "v2pick::" rest = q.data[len(CALLBACK_PREFIX):] brand, pending_key = rest.split(":", 1) pending = capture._pending.pop(pending_key) if capture._pending else None if not pending: await q.edit_message_text("⌛ слишком долго думали, повтори сообщение") return created = await capture.create_from_pending(brand=brand, pending=pending) list_suffix = f" / {created.list_name}" if created.list_name != "To Do" else "" await q.edit_message_text(f"✓ создано в {created.brand}{list_suffix}") # === Commands === async def cmd_today(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: if not _owner_only(update): return text, _ = await commands.today( user_id=update.effective_user.id, today=datetime.now(tz).date(), ) await update.message.reply_text(text) async def cmd_inbox(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: if not _owner_only(update): return text, _ = await commands.inbox( user_id=update.effective_user.id, today=datetime.now(tz).date(), ) await update.message.reply_text(text) async def cmd_backlog(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: if not _owner_only(update): return arg = (ctx.args or [None])[0] text, _ = await commands.backlog( user_id=update.effective_user.id, brand=arg, today=datetime.now(tz).date(), ) await update.message.reply_text(text) def _parse_n(args: list[str]) -> int | None: if not args: return None try: return int(args[0]) except ValueError: return None async def cmd_done(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: if not _owner_only(update): return n = _parse_n(ctx.args) if n is None: await update.message.reply_text("Использование: /done N") return result = await commands.done( user_id=update.effective_user.id, n=n, now=datetime.now(tz), ) await update.message.reply_text(result) async def cmd_defer(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: if not _owner_only(update): return if not ctx.args or len(ctx.args) < 2: await update.message.reply_text( "Использование: /defer N today|week|none|YYYY-MM-DD", ) return try: n = int(ctx.args[0]) except ValueError: await update.message.reply_text("N должно быть числом") return when = ctx.args[1] result = await commands.defer( user_id=update.effective_user.id, n=n, when=when, now=datetime.now(tz), ) await update.message.reply_text(result) async def cmd_text(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: if not _owner_only(update): return if not ctx.args or len(ctx.args) < 2: await update.message.reply_text("Использование: /text N новый текст") return try: n = int(ctx.args[0]) except ValueError: await update.message.reply_text("N должно быть числом") return new_text = " ".join(ctx.args[1:]) result = await commands.text( user_id=update.effective_user.id, n=n, new_text=new_text, now=datetime.now(tz), ) await update.message.reply_text(result) async def cmd_proj(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: if not _owner_only(update): return if not ctx.args or len(ctx.args) < 2: await update.message.reply_text( "Использование: /proj N mars|channel|consulting|life|vibe-coding", ) return try: n = int(ctx.args[0]) except ValueError: await update.message.reply_text("N должно быть числом") return brand = ctx.args[1] result = await commands.proj( user_id=update.effective_user.id, n=n, target_brand=brand, now=datetime.now(tz), ) await update.message.reply_text(result) async def cmd_date(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: return await cmd_defer(update, ctx) # alias async def cmd_del(update: Update, ctx: ContextTypes.DEFAULT_TYPE) -> None: if not _owner_only(update): return n = _parse_n(ctx.args) if n is None: await update.message.reply_text("Использование: /del N") return result = await commands.delete( user_id=update.effective_user.id, n=n, now=datetime.now(tz), ) await update.message.reply_text(result) # === Register === app.add_handler(CommandHandler("today", cmd_today)) app.add_handler(CommandHandler("inbox", cmd_inbox)) app.add_handler(CommandHandler("backlog", cmd_backlog)) app.add_handler(CommandHandler("done", cmd_done)) app.add_handler(CommandHandler("defer", cmd_defer)) app.add_handler(CommandHandler("date", cmd_date)) app.add_handler(CommandHandler("text", cmd_text)) app.add_handler(CommandHandler("proj", cmd_proj)) app.add_handler(CommandHandler("del", cmd_del)) app.add_handler( CallbackQueryHandler(on_brand_pick, pattern=f"^{CALLBACK_PREFIX}"), ) app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, on_text)) log.info("v2 handlers registered") ``` - [ ] **Step 19.2: Изменить `bot.py` — ветвление по флагу** В `src/tasks_bot/bot.py`, в начале `build_application` (после прочтения settings), добавить ветвление перед регистрацией v1-хендлеров. Найди функцию `build_application` и оберни регистрацию старых хендлеров в условие: ```python def build_application(token: str, owner_id: int, db: Database) -> Application: settings = get_settings() app = Application.builder().token(token).build() if settings.v2_capture_enabled: # v2 mode — пропускаем v1-хендлеры from tasks_bot.v2.app import register_v2_handlers from tasks_bot.v2 import ( commands as cmd_mod, capture as cap_mod, event_log as elog_mod, extraction as ext_mod, openrouter as or_mod, pending as pending_mod, summary_cache as cache_mod, ) from tasks_bot.ticktick import client as tk_client_mod from tasks_bot.ticktick import mapping as tk_map_mod import zoneinfo from datetime import date as _date_cls tz = zoneinfo.ZoneInfo(settings.timezone) ticktick = tk_client_mod.TickTickClient( access_token=settings.ticktick_access_token, base_url=settings.ticktick_api_base, ) mapping = tk_map_mod.ProjectMapping.load( __import__("pathlib").Path(settings.ticktick_mapping_path), ) event_log = elog_mod.EventLog(settings.event_log_path) pending = pending_mod.PendingCaptureCache( ttl_seconds=settings.pending_capture_ttl_seconds, ) llm = or_mod.make_extractor_llm( api_key=settings.openrouter_api_key, model=settings.openrouter_v2_model, ) # Extractor today обновляется на каждый вызов через partial-обёртку class _DynamicExtractor: async def extract(self, text): from datetime import datetime as _dt ex = ext_mod.Extractor(llm_call=llm, today=_dt.now(tz).date()) return await ex.extract(text) capture = cap_mod.CaptureHandler( extractor=_DynamicExtractor(), mapping=mapping, ticktick=ticktick, event_log=event_log, pending_cache=pending, confidence_threshold=settings.project_confidence_threshold, ) summary_cache = cache_mod.SummaryCache( settings.summary_cache_db_path, tz=tz, ) async def _post_init(application: Application) -> None: await summary_cache.connect() application.bot_data["v2_summary_cache"] = summary_cache application.bot_data["v2_ticktick"] = ticktick async def _post_shutdown(application: Application) -> None: await summary_cache.close() await ticktick.__aexit__(None, None, None) app.post_init = _post_init app.post_shutdown = _post_shutdown commands = cmd_mod.Commands( ticktick=ticktick, mapping=mapping, summary_cache=summary_cache, tz=tz, event_log=event_log, ) register_v2_handlers(app, capture=capture, commands=commands, owner_id=owner_id, tz=tz) return app # ... (старый v1-код регистрации хендлеров — оставить как было) ``` (Точное расположение блока — перед существующим `app.add_handler(...)` вызовами; v1-блок не трогать.) - [ ] **Step 19.3: Дымовой smoke-тест** `tests/v2/test_app_smoke.py`: ```python """Smoke: register_v2_handlers вызывается без ошибок.""" from unittest.mock import MagicMock from telegram.ext import Application from tasks_bot.v2.app import register_v2_handlers def test_register_v2_handlers_does_not_raise(): app = Application.builder().token("123:abc").build() capture = MagicMock() capture._pending = MagicMock() commands = MagicMock() register_v2_handlers( app, capture=capture, commands=commands, owner_id=1, tz=None, ) # Проверяем, что несколько хендлеров зарегистрировались total = sum(len(h_list) for h_list in app.handlers.values()) assert total >= 11 # 9 commands + callback + message ``` - [ ] **Step 19.4: Прогнать smoke + полный тест-сьют** Run: `.venv/bin/pytest -v 2>&1 | tail -15` Expected: все v1 + v2 тесты зелёные. - [ ] **Step 19.5: Commit** ```bash git add src/tasks_bot/v2/app.py src/tasks_bot/bot.py tests/v2/test_app_smoke.py git commit -m "v2: register handlers in bot.py under V2_CAPTURE_ENABLED flag" ``` --- ## Task 20: README + AGENTS.md note **Files:** - Modify: `README.md` - Modify: `AGENTS.md` - [ ] **Step 20.1: README** — добавить раздел про v2: ```markdown ## v2 (TickTick) При установленной `V2_CAPTURE_ENABLED=true` бот работает в режиме v2: captures пишутся в TickTick через Open API, команды управляют задачами там же. v1-код остаётся в репо для миграции и не задействуется. Конфигурация — переменные `TICKTICK_*` в `.env`. Маппинг проектов: `scripts/preflight/projects_mapping.json`. Event log: `data/events.jsonl` (используется для monthly archive, План 5). ``` - [ ] **Step 20.2: AGENTS.md** — добавить: ```markdown ## v2 architecture (active) - `src/tasks_bot/ticktick/` — async TickTick client + mapping - `src/tasks_bot/v2/` — capture, routing, commands, event log - `src/tasks_bot/v2/app.py` — wiring handlers - Активируется флагом `V2_CAPTURE_ENABLED` - v1-код (`conversational/`, `mac/`, `git_sync`, etc.) — в репо, не запущен - Полный спек: `docs/{vibe-coding} {plan} tasks-bot 2.0 ticktick – 2026-05-30.md` ``` - [ ] **Step 20.3: Commit** ```bash git add README.md AGENTS.md git commit -m "v2: README and AGENTS notes about v2 mode" ``` --- ## Task 21: Deploy to Pi (manual) Без правок кода. После всех тестов локально: - [ ] **Step 21.1: Перенести токен и mapping** Скопировать на Pi: ```bash scp scripts/preflight/projects_mapping.json pi:~/tasks_bot/scripts/preflight/ ``` - [ ] **Step 21.2: Обновить `.env` на Pi** В `~/tasks_bot/.env` на Pi дописать новые поля (см. `.env.example`). НЕ выставлять `V2_CAPTURE_ENABLED=true` пока не убедились что v2 работает в боте. - [ ] **Step 21.3: Деплой кода** ```bash git push origin main ssh pi "cd ~/tasks_bot && git pull && sudo systemctl restart tasks-bot" ``` Бот должен запуститься в v1-режиме (флаг false). Проверить `sudo systemctl status tasks-bot`. - [ ] **Step 21.4: Включить v2 и проверить** ```bash ssh pi "sed -i 's/V2_CAPTURE_ENABLED=false/V2_CAPTURE_ENABLED=true/' ~/tasks_bot/.env && sudo systemctl restart tasks-bot" ssh pi "sudo journalctl -u tasks-bot -f --since '1 min ago'" ``` В Telegram отправить тестовое сообщение типа «идея для треда про утренний ритуал». Ожидание: ответ `✓ создано в channel / Threads`. В TickTick — новая задача в Channel / Threads. Если что-то идёт не так — `V2_CAPTURE_ENABLED=false` + restart, копаем логи. --- ## Self-Review **1. Spec coverage:** - Capture handler: Tasks 5–9 ✓ - 9 команд: Tasks 12–18 (`/today /inbox /backlog /done /defer /text /proj /date /del`) ✓ - `/defer all` — НЕ в этом плане (требует scheduled-сводки, см. План 3) - Routing rules: Task 3 ✓ - TickTick client + mapping: Tasks 1-2 ✓ - Event log: Task 8 ✓ - Inline-кнопки fallback при низком confidence: Tasks 7+9+19 ✓ - Re-check перед мутацией: реализован в Tasks 14-18 в каждой команде ✓ - TZ единый: используется `Asia/Yekaterinburg` через `settings.timezone` ✓ - Confidence threshold через env: ✓ - Feature flag для безопасной выкатки: ✓ **Gap'ы (осознанные, в Планах 3-7):** - Scheduled сводки (утро/вечер/воскресная celebration) — План 3 - Refresh-token flow — пока access на 180 дней - Migration v1 → TickTick — План 4 - Monthly archive script — План 5 - Evals — План 6 - v1 cleanup — План 7 **2. Placeholder scan:** В плане нет «TODO/TBD/fill in details». В Step 19.2 указано «(точное расположение блока — перед существующим `app.add_handler(...)` вызовами)» — это уточнение, не плейсхолдер, разрешается читать v1-код для встраивания. **3. Type consistency:** - `Extracted`, `Created`, `NeedsBrand`, `CaptureResult`, `TaskRow`, `PendingEntry`, `CheckResult` — определены один раз, переиспользуются. - `Commands.__init__` принимает `event_log=None` (Task 12, расширено в Task 14) — изменение совместимое. - `_parse_due`, `_end_of_week` — helpers в `commands.py`, не дублируются. - `resolve_list` — единственная routing-функция. - `TickTickClient.get/post/delete` — стабильный интерфейс через все команды. План готов к исполнению.