Набор самодостаточных промтов, которые любой человек разворачивает у себя — в любой IDE (VS Code + Codex, Claude Code, Antigravity, Cursor) и под любой LLM. Никаких чужих аккаунтов, токенов и ключей внутри. Человек кладёт файл, отвечает на вопросы — и получает работающий проект.
Все три системы устроены одинаково: человек не пишет код руками — он отвечает на вопросы, а агент собирает проект по вшитому эталону.
Один промт-файл в пустую папку проекта. Для Codex — назвать AGENTS.md, агент подхватит сам.
Сначала интервью: ниша, аудитория, оффер, каналы, запреты. Тему за тебя не выдумывает.
Из ответов заполняет контент и доступы. Движок берёт из эталона как есть — логику не переписывает.
Тест-режим с ускоренными задержками: вся механика проверяется на себе за минуты, а не за дни.
Локально или на дешёвом VPS под супервизором, чтобы жил после перезагрузки. Секреты — на сервере.
Это общий костяк. Он одинаков для контент-пайплайна, команды публикации и бота воронки — меняется только начинка.
Агент задаёт вопросы, ждёт ответов и только потом создаёт файлы. Нишу и контент за человека не сочиняет — иначе получится чужой шаблон, а не его проект.
Логика (планировщик, БД, оркестровка) берётся из эталона неизменной. Под нишу меняется только контент и конфиг — posts.py, config.yaml, .env.
Внутри промта нет чужих токенов, доменов, юзернеймов и личных данных. Человек подставляет своё. Бренды и имена клиентов в контенте обезличиваются.
Токены и ключи живут в .env (в .gitignore) или на сервере с правами chmod 600. Никогда в коде, в git и — главное — не диктуются в чат ИИ-ассистенту.
Работаем в живой сессии под подпиской. В окружении нет ANTHROPIC_API_KEY / OPENAI_API_KEY, иначе агент улетит на pay-as-you-go. Публикация — детерминированный скрипт: ноль токенов.
Вся логика — в тексте промта, не в инструменте. Файл автозагрузки только переименовать: AGENTS.md для Codex, CLAUDE.md для Claude, GEMINI.md для Gemini.
Данные — в локальном SQLite или в файлах очереди. Без Google Sheets, CRM и внешних дашбордов. Меньше зависимостей — меньше того, что утечёт или отвалится.
При старте процесс переподнимает запланированные задачи из БД. Перезапуск не теряет расписание, не дублирует и не пропускает шаг. Проверяется в самотесте.
Тест-режим сжимает сутки в секунды — вся воронка или пайплайн прогоняется на себе за минуты. «Готово» — это проверяемое состояние, а не ощущение.
Помощнику — отдельный пользователь (не root) и его собственный SSH-ключ. Твой приватный ключ и личные файлы не передаются. На сервер уезжает только готовый обезличенный контент.
Тексты в человеческом голосе, без AI-штампов и канцелярита, тире — человеческой плотности. Никаких обещаний того, чего нет в оффере, и гарантий результата.
Каждая — отдельный самодостаточный промт. Отдаёшь человеку нужный (или все) — он разворачивает у себя по единому сценарию из пяти шагов.
Находит свежие новости по заданной нише, пишет из них посты и ставит в очередь на публикацию. Роли: researcher → writer → editor.
Берёт готовый пост из очереди, проверяет по запретам и публикует детерминированным скриптом. Отправка не тратит токены LLM.
Ведёт персональную серию сообщений с каждым, кто нажал /start: приветствие → анкета → тема в день → мягкие дожимы. Ловит горячих лидов.
AGENTS.md, говорит «разверни проект по этому файлу» — и отвечает на вопросы. Всё остальное агент создаёт сам по инструкции внутри промта. Токены и ключи владельца не передаются никогда.
Сами тексты — здесь, прямо в этом файле. Разверни нужный блок, нажми «Копировать промт» и вставь в своего ИИ-агента. Ничего скачивать отдельно не нужно.
Стартер: разворачивает проект, ищет новости по теме, пишет и публикует посты.
# Контент-пайплайн — самоустанавливающийся шаблон проекта
> **Что это.** Один файл, который разворачивает целую систему: находит свежие новости по твоей
> теме, пишет из них посты и публикует по расписанию. Работает в любой агентной среде (VS Code +
> Codex/Cline, Antigravity, Cursor, CLI) и под любой LLM (Claude, GPT/Codex, Gemini) — потому что
> вся логика лежит в тексте промптов, а не в конкретном инструменте.
>
> **Как это «подтягивается само».** Положи этот файл в корень пустой папки проекта. Агент в IDE
> читает файлы проекта → видит этот блюпринт и `config.yaml` → берёт оттуда твою тему → и дальше
> ищет новости и пишет посты **строго по ней**. Ты меняешь только `config.yaml` — система
> подстраивается под любую нишу.
---
## Запуск за 5 минут
1. Создай пустую папку — это твой проект. Положи в неё **этот файл**.
2. Открой папку в IDE с ИИ-агентом.
- **VS Code + Codex** → переименуй/скопируй этот файл в **`AGENTS.md`** в корне. Codex читает
`AGENTS.md` автоматически при каждом запросе — тема и правила подтянутся сами.
- **Claude Code** → имя **`CLAUDE.md`**. **Gemini CLI** → **`GEMINI.md`**. **Antigravity/Cursor** →
оставь как есть и один раз скажи агенту «прочитай этот файл».
3. Скажи агенту: **«Прочитай этот файл и разверни проект по нему»**. Он сначала задаст тебе вопросы
из раздела «Онбординг-интервью» — ответь на них обычными словами. Из ответов он соберёт
`config.yaml`, промпты ролей и всю структуру папок. (Не хочешь интервью — заполни блок **КОНФИГ**
руками сам.)
4. Дальше работаешь двумя фразами: **«Найди новости»** и **«Опубликуй новость»**.
---
## Онбординг-интервью (агент задаёт эти вопросы человеку)
**Инструкция агенту:** прежде чем что-то создавать — задай человеку вопросы ниже (можно по одному
или пачкой). Дождись ответов. Тему за него **не выдумывай**. Из ответов собери `config.yaml` и
покажи его человеку на подтверждение, потом разворачивай проект. Если на вопрос нет ответа — предложи
разумное значение по умолчанию и пометь, что это допущение.
**Про что проект**
1. О чём проект — какая тема или ниша? Сформулируй как можно уже. *(плохо: «маркетинг»; хорошо:
«AI-инструменты для стоматологических клиник»)* → `project_topic`
2. Для кого пишем — кто твой читатель? Кто он, чем занят, какой у него уровень? → `audience`
3. Какую его боль или потребность закрываем? Почему ему вообще это читать? → `value`
4. В каком тоне писать — строго/по-деловому, разговорно, экспертно, с юмором? → `tone`
5. На каком языке посты? → `language`
**Где брать новости**
6. Есть источники, которым ты доверяешь по этой теме — сайты, блоги, Telegram-каналы, RSS?
(Нет — агент подберёт сам веб-поиском.) → `sources`
7. По каким запросам искать свежее? Назови 3–5 фраз, как гуглил бы сам. → `search_queries`
**Куда и когда публиковать**
8. Куда выкладываем — Telegram-канал, сайт, оба? → `channels`
9. В каком часовом поясе живёшь и во сколько посты выходят? → `timezone`, `schedule`
10. Сколько постов в день/неделю комфортно? → влияет на `schedule`
**Границы и запреты**
11. Что писать категорически нельзя? Какие бренды/имена нельзя называть, каких обещаний избегать? →
`forbidden`
12. Есть слова-штампы или темы, которые тебя бесят и не должны попадать в текст? → `forbidden`
**Техника (чтобы правильно настроить среду)**
13. В какой программе работаешь — VS Code с Codex, Claude Code, Gemini, что-то ещё? *(от этого зависит
имя файла автозагрузки: `AGENTS.md` / `CLAUDE.md` / `GEMINI.md`)*
14. Нужен автопостинг 24/7, когда компьютер выключен (тогда разворачиваем VPS), или запускаешь
публикацию руками? → влияет на раздел про VPS
15. Кто-то ещё будет вести этот проект кроме тебя? *(если да — см. раздел про передачу без личных
данных)*
---
## КОНФИГ (агент заполняет из ответов интервью; можно и руками)
Агент при развёртывании превращает этот блок в файл `./config.yaml`. Все роли читают его в начале
работы — поэтому новости ищутся и посты пишутся **именно по твоей теме**.
```yaml
# --- О ЧЁМ ПРОЕКТ ---
project_topic: "___" # ниша строго: напр. "AI-инструменты для стоматологических клиник"
audience: "___" # для кого пишем: напр. "владельцы частных клиник, 1-3 кресла"
value: "___" # какую боль закрываем: напр. "кто не внедрит AI — потеряет пациентов"
tone: "разговорный, без канцелярита, без эмодзи, живой"
language: "ru"
# --- ГДЕ ИСКАТЬ НОВОСТИ ---
sources: # RSS/сайты по теме (агент дополнит веб-поиском по project_topic)
- "https://___"
- "https://___"
search_queries: # запросы для веб-поиска свежего
- "___ новости 2026"
- "___ кейсы"
# --- КУДА ПУБЛИКОВАТЬ ---
channels: ["telegram"] # telegram | site | оба
timezone: "Europe/Moscow"
schedule: ["09:00", "19:00"] # во сколько выходят посты
# --- ЗАПРЕТЫ (жёсткие, проверяются перед публикацией) ---
forbidden:
- "гарантии результата, охватов, продаж"
- "реальные имена/бренды клиентов — обезличивать"
- "AI-штампы: революционн, уникальн, экосистем, давайте погрузимся"
- "личные данные: телефоны, адреса, счета"
```
---
## Что агент создаёт в папке (инструкция по развёртыванию)
Когда человек говорит «разверни проект», агент **сначала проводит онбординг-интервью** (раздел
выше), собирает `config.yaml`, показывает его на подтверждение — и только потом создаёт остальное.
Итоговая структура в корне:
```
./config.yaml ← из блока КОНФИГ выше
./roles/ ← системные промпты ролей (из раздела «Промпты ролей»)
researcher.md
writer.md
editor.md
./state/ ← рабочие данные пайплайна
news-<дата>.json ← собранные новости
post-<время>/ ← готовый пост: text.md, meta.json
queue/ ← очередь на публикацию: <id>.json
./posts/ ← тексты постов, на которые ссылается очередь
./published/ ← опубликованное уезжает сюда
./log.md ← журнал: что и когда вышло
./publish.py ← публикатор (детерминированный, без LLM) — по желанию
```
Правило для агента: **результат каждой роли — файлы на диске по этим путям**, не текст в чат.
Следующая роль читает файлы предыдущей. Это контракт, который держит систему вместе независимо от
LLM и IDE.
---
## Как это работает (конвейер)
```
ты: «Найди новости»
└─ роль researcher → читает config.yaml → ищет по теме → ./state/news-<дата>.json
└─ роль writer → берёт новости → пишет пост → ./state/post-<время>/text.md + meta.json
└─ роль editor → проверяет по forbidden → если ок → кладёт в очередь ./state/queue/<id>.json
ты: «Опубликуй новость»
└─ публикатор → берёт из очереди то, чему пора → постит в канал → двигает в ./published/ → пишет в log.md
```
Генерация (researcher + writer + editor) и публикация — **разнесены**. Генерацию гоняешь, когда
удобно (хоть на неделю вперёд), публикация идёт по расписанию из очереди.
---
## Роли — кто за что отвечает (универсальный контракт)
| Роль | Читает | Делает | Пишет | Нужные возможности |
|---|---|---|---|---|
| **researcher** | `config.yaml`, интернет | 5–10 свежих новостей строго по теме за 24 ч | `state/news-<дата>.json` | веб-поиск, веб-фетч, запись файлов |
| **writer** | новости + `config.yaml` | 1 пост в нужном тоне (+ версия для сайта) | `state/post-<время>/text.md`, `meta.json` | чтение/запись файлов |
| **editor** | папку поста + `forbidden` | проверка запретов, ставит в очередь или на правку | `state/queue/<id>.json` или `revise.md` | чтение/запись файлов |
| **publisher** *(скрипт)* | очередь | публикует по расписанию, детерминированно | `published/`, `log.md` | HTTP к API канала (LLM не нужна) |
---
## Промпты ролей (агент кладёт их в `./roles/`)
Промпты **не привязаны к теме** — они каждый раз читают `config.yaml`. Один шаблон работает для
любой ниши. Вставляй как есть.
### roles/researcher.md
```
Ты — сборщик новостей контент-пайплайна.
1. Прочитай ./config.yaml — возьми project_topic, audience, value, sources, search_queries.
2. Найди 5–10 свежих новостей за последние 24 часа СТРОГО по project_topic, полезных для audience.
Источники: sources из конфига + веб-поиск по search_queries.
3. Жёсткий фильтр: бери только то, что реально касается project_topic и даёт audience пользу или
тревогу по value. Офтоп, реклама, вода — выбрасывай. Лучше 5 точных, чем 10 с мусором.
4. Для каждой новости: заголовок, 2–3 предложения сути, ссылка-источник, «почему это важно audience».
5. РЕЗУЛЬТАТ ЗАПИШИ В ФАЙЛ ./state/news-<сегодня>.json (JSON, массив items). В чат — одна строка
статуса. Содержимое новостей в чат не выводи.
```
### roles/writer.md
```
Ты — редактор-копирайтер.
1. Прочитай ./config.yaml (tone, audience, language, channels) и свежий ./state/news-<дата>.json.
2. Выбери ОДНУ новость (самую сильную по пользе/боли для audience) и слепи из неё пост.
3. Создай ./state/post-<время>/text.md:
- для telegram: 300–500 слов, живой тон из config, жирный заголовок-хук (через боль/потребность
audience, не «Обзор» и не «Топ-5»), 1–2 важные мысли выделены, перечисления цитатой, в конце
призыв к действию;
- если channels включает site — второй блок 600–900 слов с подзаголовками.
4. Создай ./state/post-<время>/meta.json: title (≤70 симв), description (≤160), slug, дата.
5. Стиль строго по tone из config: без AI-штампов, без канцелярита, тире — человеческой плотности.
6. Пиши СРАЗУ В ФАЙЛЫ, не в чат. После записи проверь, что файлы созданы и не пустые.
```
### roles/editor.md
```
Ты — финальный редактор и «юрист» пайплайна. Ничего не выходит без твоего одобрения.
1. Прочитай ./config.yaml (forbidden) и папку последнего поста ./state/post-<время>/.
2. Проверь по пунктам:
- нет ничего из forbidden (гарантии, реальные бренды/имена, личные данные, AI-штампы);
- заголовок цепляет боль/потребность audience (не нейтральный);
- структура и длина в норме, есть призыв к действию;
- тон живой, соответствует config.
3. Если ОК → поставь в очередь: создай ./state/queue/<id>.json с полями:
{publish_at, channel, text_file (путь к text.md), parse_mode:"HTML", log_slug} и допиши строку
в ./log.md.
4. Если НЕ ОК → создай ./state/post-<время>/revise.md с конкретными правками. В чат — вердикт одной
строкой.
```
### Оркестровка (одной фразой агенту)
```
Выполни по очереди роли из ./roles/: researcher → writer → editor. После каждой проверь, что её
выходные файлы созданы и не пустые; если файла нет — повтори роль (до 3 раз), потом остановись и
скажи, где сломалось. Сам контент не сочиняй — только запускай роли и проверяй файлы.
```
---
## Публикация: по подписке, без платного API (так дешевле всего)
Три факта, которые делают публикацию почти бесплатной:
1. **Отправка поста не тратит LLM.** Публикатор — обычный скрипт: взял готовый текст → отправил в
API канала. Никакая модель не вызывается. Рутинная выкладка по расписанию стоит **ноль** токенов.
2. **Генерацию гоняй в живой сессии под подпиской.** researcher/writer/editor жгут токены, но их
запускаешь вручную в сессии, где выполнен вход по подписке (не по API-ключу). Пока в окружении
**нет** переменной API-ключа провайдера — всё идёт на подписку.
3. **Не автозапускай headless-скрипт, который сам дёргает LLM** — так легко случайно улететь на
платный API. Вместо этого человек сам говорит «Найди новости» / «Опубликуй новость» в живой сессии.
> Проверь перед работой, что API-ключ провайдера **не выставлен** в окружении (для Claude это
> `ANTHROPIC_API_KEY`, для OpenAI/Codex — `OPENAI_API_KEY`). Иначе агент уйдёт на платный API.
### Команда «Опубликуй новость» (сохранённый промпт)
В Claude Code это скилл (`.claude/skills/publish-news/`), в Codex — блок в `AGENTS.md`, в других —
просто сохранённая фраза. Суть одна:
```
Прочитай ./state/queue/*.json. Для каждого, чей publish_at <= сейчас: прочитай text_file, проверь
что там нет ничего из config.forbidden, опубликуй в channel, перенеси файл в ./published/ и допиши
строку в ./log.md. В конце покажи: что опубликовано, ссылку, что осталось в очереди. Токены не трать
на сочинение — публикуешь готовое.
```
---
## Публикация с VPS без выдачи личных данных (по желанию, для автопостинга 24/7)
Если хочешь, чтобы посты выходили сами, даже когда компьютер выключен, — очередь и публикатор
переносятся на дешёвый VPS, а маленький бот там постит по расписанию (systemd-таймер каждые 5 минут).
Это же — способ **отдать проект помощнику, не показывая ему ничего личного**.
**Модель доступа — что даёшь / что не даёшь:**
| Даёшь помощнику | Не даёшь никогда |
|---|---|
| IP сервера | свой приватный SSH-ключ |
| отдельного пользователя (не root) | root-пароль |
| право развернуть/перезапустить его бота | доступ к твоим личным файлам |
| готовую обезличенную очередь постов | реальные токены/пароли в переписке с ИИ |
**Три правила, закрывающие «личные данные»:**
1. **Свой ключ.** Помощник генерирует свою пару SSH-ключей, присылает только **публичную** часть.
Твой приватный ключ не передаётся, вход по паролю выключается.
2. **Секреты — на сервере, не в чате.** Токен бота вписывается руками на сервере в файл `.env`
(`chmod 600`), а не диктуется ИИ-ассистенту (всё в чат ассистента уходит вендору LLM). Ассистент
создаёт `.env` **пустым**, значения человек вписывает сам.
3. **На VPS уезжает только готовый обезличенный контент** — очередь и публикатор. Черновики,
источники, любые личные материалы остаются локально.
**Развёртывание одной задачей ИИ-ассистенту** (после того как вход по ключу настроен):
```
На сервере <пользователь>@<IP> разверни бота-публикатора из текущей папки: поставь python3/venv/git,
залей код и папку queue в /opt/publisher (без .git, .venv, .env, секретов), сделай venv и поставь
зависимости, создай ПУСТОЙ .env (chmod 600), настрой systemd-таймер запускать публикатор каждые 5
минут, включи и запусти. Показывай каждую команду перед выполнением. Боевой .env не трогай. В конце —
статус таймера и последние 30 строк логов.
```
Обновление плана потом = «залей новую очередь на сервер», бот подхватит сам.
---
## Перенос на конкретный стек
Что переносится **без изменений**: промпты ролей, структура папок, формат очереди, config.yaml.
Что зависит от среды — как запускать роли и как хранить команду «Опубликуй новость»:
| Среда | Файл автозагрузки | Команда публикации |
|---|---|---|
| VS Code + **Codex** | `AGENTS.md` в корне (читается сам) | блок-инструкция в `AGENTS.md` |
| **Claude Code** | `CLAUDE.md` | скилл `.claude/skills/publish-news/` |
| **Gemini CLI** | `GEMINI.md` | сохранённый промпт |
| **Antigravity / Cursor** | сказать «прочитай этот файл» | сохранённая фраза/воркфлоу |
Проверка «роль отработала» везде одинаковая: **на диске появились нужные файлы и они не пустые** —
нет файлов, повтори роль.
---
## Чеклист первого запуска
- [ ] Файл лежит в корне пустой папки проекта (для Codex — переименован в `AGENTS.md`).
- [ ] Агент задал вопросы из онбординг-интервью, ты ответил, `config.yaml` собран из ответов.
- [ ] `config.yaml` заполнен: тема, аудитория, тон, источники, каналы, часовой пояс, запреты.
- [ ] Агент развернул структуру: `./roles/`, `./state/queue/`, `./posts/`, `./published/`, `./log.md`.
- [ ] API-ключ провайдера **не выставлен** в окружении — работаем по подписке.
- [ ] «Найди новости» → в `./state/news-<дата>.json` появились релевантные теме новости.
- [ ] «Опубликуй новость» → пост ушёл в канал, файл уехал в `./published/`, есть строка в `log.md`.
- [ ] (Если нужен автопостинг) очередь и бот развёрнуты на VPS по правилам выше.
---
## Частые грабли
- **Новости не по теме** → researcher не прочитал `config.yaml` или тема размыта. Уточни
`project_topic` до конкретной ниши, повтори.
- **Роль вернула текст в чат вместо файла** → повтори с явным «пиши в файл по пути X, не в чат».
- **`409 Conflict` при постинге** → публикатор запущен в двух местах на один токен канала. Оставь один.
- **Счёт за платный API** → в окружении торчит API-ключ провайдера. Убери, работай по подписке.
- **`.env` не подхватился на сервере** → не в той папке или нет прав (`chmod 600` и правильный путь).
```
Скилл публикации из очереди по подписке. Отправка не тратит токены LLM.
---
name: publish-news
description: Опубликовать готовый пост из очереди контент-пайплайна в подключённый канал (Telegram и/или сайт) — по подписке, без платного API. Человек говорит «Опубликуй новость» / «опубликуй сегодняшний пост» / «выложи что в очереди» / «сделай публикацию дня» — скилл берёт готовый контент из очереди, проверяет по запретам, публикует детерминированным скриптом (0 токенов LLM), отмечает в журнале, даёт отчёт. Работает из корня проекта, развёрнутого по AGENTS.md.
---
# Публикация новости из очереди
Скилл делает ровно одно: **берёт уже готовый пост из очереди и публикует его**. Контент готовится
ЗАРАНЕЕ (ролями researcher → writer → editor или руками) и лежит проверенным. Здесь — только
публикация, отметка, отчёт. Работать из корня проекта (там, где лежит `config.yaml`).
## ⛔ ГЛАВНОЕ ПРАВИЛО — ПОДПИСКА, А НЕ ПЛАТНЫЙ API
Смысл скилла — чтобы всё шло **по подписке LLM и не жгло деньги на pay-as-you-go API**.
1. **Запускать в живой интерактивной сессии**, где выполнен вход по подписке (не по API-ключу).
2. **В окружении НЕ должно быть API-ключа провайдера.** Для Claude это `ANTHROPIC_API_KEY`, для
OpenAI/Codex — `OPENAI_API_KEY`. Если ключ выставлен — агент уйдёт на платный API. Проверить и
убрать перед работой:
```bash
echo "$ANTHROPIC_API_KEY $OPENAI_API_KEY" # должно быть пусто
unset ANTHROPIC_API_KEY OPENAI_API_KEY
```
3. **Сама отправка поста НЕ тратит токены LLM.** Публикатор (`./publish.py`) — обычный скрипт: берёт
готовый текст, конвертит разметку, шлёт в API канала. Модель не вызывается. Поэтому «нажал
Опубликуй новость» стоит почти ноль — токены жжёт только генерация, а она сделана заранее.
**Вывод:** тяжёлую генерацию — заранее и на подписке; публикацию — детерминированным скриптом;
платный API — нигде.
## 0. Перед стартом — прочитать контракт
1. `./config.yaml` — берёт `forbidden` (запреты), `channels`, `timezone`.
2. `./log.md` — что уже опубликовано (не дублировать слот).
3. Секреты — только в `./.env` в корне (например `BOT_TOKEN`, `OWNER_CHAT`), права `chmod 600`.
Токен **никогда** не вписывать в чат агенту и не коммитить в git (добавь `.env` в `.gitignore`).
## 1. Шаг 1 — что в очереди на публикацию
Очередь — папка `./state/queue/*.json`. Один файл = один пост. Формат:
```json
{
"publish_at": "2026-07-10T19:00:00", // когда публиковать (местное время, ISO)
"channel": "telegram", // telegram | site | оба (из config.channels)
"text_file": "state/post-.../text.md", // путь к готовому тексту поста
"parse_mode": "HTML", // HTML (рекомендуется) | MarkdownV2
"log_slug": "kratkiy-slug", // идентификатор для журнала
"site_url": null // null или "/news/.../"
}
```
Показать, что готово к публикации СЕЙЧАС (`publish_at <= now`) и что запланировано позже:
```bash
python3 - <<'PY'
import json, glob
from datetime import datetime
now = datetime.now()
for f in sorted(glob.glob("state/queue/*.json")):
d = json.load(open(f, encoding="utf-8"))
due = datetime.fromisoformat(d["publish_at"])
print(("● ПОРА " if due <= now else "○ позже"), d["publish_at"], d["log_slug"], "→", d["channel"])
PY
```
**Если очередь пуста** — постить нечего. Сообщить человеку «очередь пуста, нужен свежий контент» и
не выдумывать пост на ходу (генерация — отдельная задача: «Найди новости»).
## 2. Шаг 2 — проверка перед публикацией (быстрый гейт)
Перед отправкой прочитать текст поста (`text_file`) и сверить с `config.forbidden`:
- нет ничего из списка запретов (гарантии результата, реальные бренды/имена, личные данные, штампы);
- есть заголовок-хук и призыв к действию;
- тон соответствует `config.tone`, тире — человеческой плотности.
Нашёл нарушение — не публиковать, вернуть на правку, сообщить человеку что именно не так.
## 3. Шаг 3 — публикация
**Вариант A — публикуем локально (просто):** запустить публикатор, он возьмёт всё, чему
`publish_at <= now`, опубликует, перенесёт в `./published/`, допишет `./log.md`:
```bash
python3 publish.py
```
**Вариант B — публикует бот на VPS (для автопостинга 24/7):** очередь и публикатор живут на VPS,
systemd-таймер дёргает публикатор каждые 5 минут — тогда постить руками не нужно, скилл только
**заливает свежую очередь на сервер**:
```bash
scp state/queue/*.json {{VPS_USER}}@{{VPS_IP}}:/opt/publisher/queue/
scp -r state/post-* {{VPS_USER}}@{{VPS_IP}}:/opt/publisher/state/
```
Бот подхватит на следующем тике. Токены LLM здесь = 0 (публикует скрипт на сервере).
## 4. Шаг 4 — подтвердить и отчитаться
- Убедиться, что пост реально появился в канале (открыть его, не верить «ok=True» вслепую).
- Проверить, что файл уехал из `state/queue/` в `published/` и в `log.md` есть строка.
- Отчёт человеку: что опубликовано, куда, ссылка, что осталось в очереди.
## Как это выглядит в других средах (скилл ≠ только Claude Code)
Здесь это оформлено как скилл Claude Code (`.claude/skills/publish-news/`). Механика та же везде:
- **VS Code + Codex** — вставь текст этого файла блоком «Команда: Опубликовать новость» в `AGENTS.md`;
Codex выполнит те же шаги по фразе-триггеру.
- **Gemini / Cursor / прочее** — сохрани как готовую фразу-команду. Логика (очередь → гейт →
публикация → отметка) не меняется.
## Грабли
- **`409 Conflict`** — публикатор запущен в двух местах на один токен канала. Оставить один инстанс.
- **Счёт за платный API** — в окружении торчит API-ключ провайдера. Убрать (см. главное правило).
- **Пост ушёл черновиком/битой разметкой** — проверить `parse_mode` (HTML надёжнее) и что разметка
корректно сконвертилась.
- **`.env` не найден** — токен ищется в `./.env` в корне; проверить путь и права `chmod 600`.
## Журнал прогонов
<!-- дописывать сюда после каждой боевой публикации: дата, что опубликовано, грабли, фиксы -->
Личная серия прогрева лидов в Telegram: приветствие, анкета, тема в день, дожимы.
# Промт: собери и запусти Telegram-бот воронки прогрева
> **Как пользоваться.** Скопируй это сообщение целиком и вставь в своего ИИ-агента (Codex в VS Code, Cline, Cursor, Claude Code — любой). Агент сначала задаст тебе вопросы, ты ответишь, и он соберёт рабочий проект. Ничего из чужих аккаунтов, токенов и ключей внутри нет — всё своё.
---
## Роль
Ты — инженер, который разворачивает Telegram-бот **воронки прогрева лидов**. Это standalone-приложение на Python (`python-telegram-bot` + JobQueue, long-polling). Бот не постит в канал — он ведёт **личную серию сообщений** с каждым, кто нажал `/start`: приветствие → короткая анкета → цепочка сообщений «по одной теме в день» → мягкие дожимы. Цель серии — довести тёплого лида до целевого действия (написать / купить / оставить заявку).
**Жёсткое правило: сначала интервью, потом код.** Сперва задай все вопросы блока «Интервью», дождись ответов, только потом создавай файлы. Движок (`config.py`, `db.py`, `bot.py`) бери из раздела «Эталонный код» как есть — логику не переписывай. Меняются только `.env` и `posts.py` (контент из ответов).
---
## Интервью (задай это первым, дождись ответов)
**A. Доступы бота**
1. Токен основного бота от [@BotFather](https://t.me/BotFather) (создай бота командой `/newbot`, если ещё нет).
2. Username бота (без `@`).
3. Твой личный `chat_id` для уведомлений о горячих лидах (узнать: напиши [@userinfobot](https://t.me/userinfobot) → он пришлёт число).
4. *(Опц.)* Токен **второго** бота — служебного, чтобы уведомления о лидах не мешались с клиентскими чатами. Нет — уведомления пойдут основным ботом.
5. *(Опц.)* Нужен ли прокси для Telegram API (только если сервер в регионе с блокировкой; обычно — нет).
**B. Суть воронки**
6. Ниша и что продаёшь — одной фразой.
7. Это прогрев к продаже — или другой тип (онбординг, напоминания, обучающая серия)? Если другой — опиши.
8. Целевое действие лида (написать / позвонить / купить / забронировать / заявка).
9. Оффер и цены (если показываешь цифры — какие и в каких сообщениях).
10. Trust-маркеры: сертификаты, годы опыта, число клиентов, кейсы — что вплетать.
11. Стиль и язык: на «ты» или «вы», разговорный или деловой, язык контента.
**C. Структура серии**
12. Сколько сообщений в основной серии (например 14).
13. Интервал между ними (по умолчанию 24 часа).
14. Нужна ли догоняющая weekly-серия для тех, кто не купил (сколько сообщений, интервал, например 4 сообщения через 4 дня). Не нужна — оставим пустой.
15. Темы по дням — по одной боли/теме на день (список).
16. Лид-магнит: есть ли ссылка (гид / PDF / страница)? Дай URL или скажи «нет».
**D. Анкета на входе (день 0, сегментация)**
17. Нужна ли анкета из 1–3 вопросов? Если да — по каждому вопросу: текст + варианты ответов. Каждый вариант станет тегом сегментации (по нему потом фильтруешь и делаешь адресные рассылки). Нет — пропустим, серия начнётся сразу.
**E. Ссылки и каналы (для кнопок и подписей)**
18. Контакт для связи: ссылка `t.me/username` и/или телефон.
19. Ссылка на сайт / страницу с ценами / оплату (если есть).
20. Соцсети для кнопок «подписаться» (IG / TG-канал / VK / прочее) — если есть.
21. Ссылка на отзывы (если есть).
**F. Расписание**
22. Тихие часы (по умолчанию не пишем с 22:00 до 09:00) и таймзона (например `Europe/Moscow`).
После ответов создай проект по разделам ниже, заполнив `posts.py` и `.env` из ответов.
---
## Принципы, которые нельзя терять (проверь по чек-листу в конце)
- **SQLite — источник истины.** Подписчики, теги, события в `data/funnel.sqlite`. Google Sheets/CRM не нужны.
- **Рестарт-устойчивость.** При старте бот переподнимает запланированные отправки из БД — перезапуск процесса не теряет и не сдвигает расписание, не дублирует и не пропускает день.
- **Тихие часы** со сдвигом на утро и рандомом 0–30 мин (чтобы не было всплеска у всех в 09:00:00).
- **Кнопка «Следующий день сейчас»** — нетерпеливый пропускает таймер.
- **Лимит подписи Telegram 1024** — длинный текст уходит отдельным сообщением после медиа.
- **Горячий лид** (кнопки «написать»/«цены») → тег `lead_hot` + статус + уведомление владельцу по-русски, без техно-кодов.
- **Авто-стоп** при `Forbidden`/«chat not found»/удалён/кикнут — чтобы не долбить API годами.
- **Секреты только в `.env`** (в `.gitignore`), в коде — никогда. `.env` грузится из папки самого бота.
- **Тест-режим** `FUNNEL_TEST_MODE=1`: 24ч→60с, weekly→2 мин — прогнать всю воронку на себе за минуты.
- **Контент:** одна тема на сообщение; твой живой голос; без обещаний того, чего нет в оффере; цены там, где о деньгах; в каждом дне уместная CTA-кнопка.
---
## Эталонный код (создать эти файлы; движок не переписывать)
### `config.py`
```python
"""Конфигурация бота воронки. Все секреты и ссылки — из .env в этой же папке."""
from __future__ import annotations
import os
from pathlib import Path
from dotenv import load_dotenv
ROOT = Path(__file__).resolve().parent
load_dotenv(ROOT / ".env")
# --- Telegram ---
BOT_TOKEN = os.getenv("BOT_TOKEN")
BOT_USERNAME = os.getenv("BOT_USERNAME", "my_funnel_bot")
OWNER_CHAT_ID = int(os.getenv("OWNER_CHAT_ID") or 0)
# Прокси (http://user:pass@host:port или socks5://…). Пусто — без прокси.
TELEGRAM_PROXY = os.getenv("TELEGRAM_PROXY") or None
# Токен отдельного служебного бота для уведомлений владельцу. Пусто — шлём основным.
NOTIFY_BOT_TOKEN = os.getenv("NOTIFY_BOT_TOKEN") or None
# --- Бренд / контакты / ссылки (подставляются в тексты как {ПЛЕЙСХОЛДЕР}) ---
OWNER_NAME = os.getenv("OWNER_NAME", "нам")
CONTACT_LINK = os.getenv("CONTACT_LINK", "") # https://t.me/username
CONTACT_PHONE = os.getenv("CONTACT_PHONE", "")
SITE_MAIN = os.getenv("SITE_MAIN", "")
SITE_PAY = os.getenv("SITE_PAY", "")
# --- Лид-магнит с UTM ---
GUIDE_URL_BASE = os.getenv("GUIDE_URL", "")
UTM_SOURCE = "telegram"
UTM_MEDIUM = "bot"
UTM_CAMPAIGN = os.getenv("UTM_CAMPAIGN", "funnel")
def guide_url(tag: str = "hello_day0") -> str:
"""Ссылка на лид-магнит с UTM-метками (для аналитики переходов из воронки)."""
if not GUIDE_URL_BASE:
return ""
sep = "&" if "?" in GUIDE_URL_BASE else "?"
return (f"{GUIDE_URL_BASE}{sep}utm_source={UTM_SOURCE}&utm_medium={UTM_MEDIUM}"
f"&utm_campaign={UTM_CAMPAIGN}&utm_content={tag}")
GUIDE_URL = guide_url("hello_day0")
# --- Медиа (Telegram file_id) ---
# Любая env-переменная вида MEDIA_<KEY>_FILE_ID автоматически попадает сюда.
# В posts.py у поста media={"kind": "...", "video_key": "<KEY>"}.
# Нет ключа — пост уходит просто текстом (медиа добавляются без правок кода).
VIDEO_FILE_IDS: dict[str, str] = {
k[len("MEDIA_"):-len("_FILE_ID")]: v
for k, v in os.environ.items()
if k.startswith("MEDIA_") and k.endswith("_FILE_ID") and v
}
def update_file_id(key: str, file_id: str) -> None:
"""Сохраняет file_id медиа в память и в .env (переживает рестарт).
Дёргается командой /savemedia."""
VIDEO_FILE_IDS[key] = file_id
var = f"MEDIA_{key}_FILE_ID"
env_path = ROOT / ".env"
lines = env_path.read_text(encoding="utf-8").splitlines() if env_path.exists() else []
lines = [ln for ln in lines if not ln.startswith(f"{var}=")]
lines.append(f"{var}={file_id}")
env_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
# --- Расписание ---
TEST_MODE = os.getenv("FUNNEL_TEST_MODE", "0") == "1"
DAY_DELAY_SECONDS = 60 if TEST_MODE else 24 * 60 * 60
WEEKLY_DELAY_SECONDS = 120 if TEST_MODE else 4 * 24 * 60 * 60
START_QUESTIONS_DELAY_SECONDS = int(
os.getenv("START_QUESTIONS_DELAY_SECONDS") or (60 if TEST_MODE else 5 * 60)
)
# --- Тихие часы ---
QUIET_HOURS_ENABLED = os.getenv("QUIET_HOURS_ENABLED", "1") == "1"
QUIET_START_HOUR = int(os.getenv("QUIET_START_HOUR") or 22) # включительно
QUIET_END_HOUR = int(os.getenv("QUIET_END_HOUR") or 9) # не включительно
QUIET_TZ = os.getenv("QUIET_TZ", "Europe/Moscow")
# --- Хранилище ---
DATA_DIR = ROOT / "data"
DATA_DIR.mkdir(exist_ok=True)
DB_PATH = DATA_DIR / "funnel.sqlite"
def assert_ready() -> None:
if not BOT_TOKEN:
raise SystemExit("BOT_TOKEN не задан в .env")
if not OWNER_CHAT_ID:
raise SystemExit("OWNER_CHAT_ID не задан в .env")
```
### `db.py`
```python
"""SQLite: subscribers (подписчик + его день и статус), tags (сегментация),
events (лог действий). Источник истины."""
from __future__ import annotations
import json
import sqlite3
from contextlib import contextmanager
from datetime import datetime
from pathlib import Path
from config import DB_PATH
SCHEMA = """
CREATE TABLE IF NOT EXISTS subscribers (
chat_id INTEGER PRIMARY KEY,
first_name TEXT,
last_name TEXT,
username TEXT,
joined_at DATETIME,
last_interaction DATETIME,
current_day INTEGER DEFAULT 0,
next_send_at DATETIME,
status TEXT DEFAULT 'new', -- new|active|hot|converted|dead
stopped INTEGER DEFAULT 0
);
CREATE TABLE IF NOT EXISTS tags (
chat_id INTEGER,
tag TEXT,
added_at DATETIME,
PRIMARY KEY (chat_id, tag)
);
CREATE INDEX IF NOT EXISTS idx_tags_chat ON tags(chat_id);
CREATE TABLE IF NOT EXISTS events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
chat_id INTEGER,
event TEXT,
meta TEXT,
ts DATETIME
);
CREATE INDEX IF NOT EXISTS idx_events_chat ON events(chat_id, ts);
"""
def init_db(path: Path = DB_PATH) -> None:
conn = sqlite3.connect(path)
try:
conn.executescript(SCHEMA)
conn.commit()
finally:
conn.close()
@contextmanager
def _conn():
c = sqlite3.connect(DB_PATH)
c.row_factory = sqlite3.Row
try:
yield c
c.commit()
finally:
c.close()
def upsert_subscriber(chat_id, first_name, last_name, username) -> dict:
now = datetime.now().isoformat(timespec="seconds")
with _conn() as c:
existing = c.execute("SELECT * FROM subscribers WHERE chat_id=?", (chat_id,)).fetchone()
if existing:
c.execute("""UPDATE subscribers SET first_name=?, last_name=?, username=?,
last_interaction=? WHERE chat_id=?""",
(first_name, last_name, username, now, chat_id))
return dict(existing)
c.execute("""INSERT INTO subscribers(chat_id, first_name, last_name, username,
joined_at, last_interaction, current_day, status)
VALUES (?,?,?,?,?,?,0,'new')""",
(chat_id, first_name, last_name, username, now, now))
return {"chat_id": chat_id, "is_new": True}
def set_next_send(chat_id, day, when) -> None:
with _conn() as c:
c.execute("""UPDATE subscribers SET current_day=?, next_send_at=?, last_interaction=?
WHERE chat_id=?""",
(day, when.isoformat(timespec="seconds") if when else None,
datetime.now().isoformat(timespec="seconds"), chat_id))
def set_status(chat_id, status) -> None:
with _conn() as c:
c.execute("UPDATE subscribers SET status=?, last_interaction=? WHERE chat_id=?",
(status, datetime.now().isoformat(timespec="seconds"), chat_id))
def stop(chat_id) -> None:
with _conn() as c:
c.execute("UPDATE subscribers SET stopped=1, next_send_at=NULL WHERE chat_id=?", (chat_id,))
def add_tag(chat_id, tag) -> None:
with _conn() as c:
c.execute("INSERT OR REPLACE INTO tags(chat_id, tag, added_at) VALUES (?,?,?)",
(chat_id, tag, datetime.now().isoformat(timespec="seconds")))
def get_tags(chat_id) -> list[str]:
with _conn() as c:
rows = c.execute("SELECT tag FROM tags WHERE chat_id=?", (chat_id,)).fetchall()
return [r["tag"] for r in rows]
def log_event(chat_id, event, meta=None) -> None:
with _conn() as c:
c.execute("INSERT INTO events(chat_id, event, meta, ts) VALUES (?,?,?,?)",
(chat_id, event, json.dumps(meta or {}, ensure_ascii=False),
datetime.now().isoformat(timespec="seconds")))
def active_subscribers_with_pending() -> list[dict]:
with _conn() as c:
rows = c.execute("""SELECT * FROM subscribers
WHERE stopped=0 AND next_send_at IS NOT NULL
AND status NOT IN ('converted')""").fetchall()
return [dict(r) for r in rows]
def get_subscriber(chat_id) -> dict | None:
with _conn() as c:
r = c.execute("SELECT * FROM subscribers WHERE chat_id=?", (chat_id,)).fetchone()
return dict(r) if r else None
def stats() -> dict:
with _conn() as c:
total = c.execute("SELECT COUNT(*) FROM subscribers").fetchone()[0]
by_status = {r["status"]: r["n"] for r in c.execute(
"SELECT status, COUNT(*) AS n FROM subscribers GROUP BY status")}
active = c.execute(
"SELECT COUNT(*) FROM subscribers WHERE stopped=0 AND next_send_at IS NOT NULL").fetchone()[0]
hot = c.execute("SELECT COUNT(*) FROM subscribers WHERE status='hot'").fetchone()[0]
return {"total": total, "by_status": by_status, "active_jobs": active, "hot": hot}
def list_by_filter(tag=None, status=None, limit=50) -> list[dict]:
q = "SELECT s.* FROM subscribers s"
params: list = []
if tag:
q += " WHERE chat_id IN (SELECT chat_id FROM tags WHERE tag=?)"
params.append(tag)
if status:
q += " AND status=?"; params.append(status)
elif status:
q += " WHERE status=?"; params.append(status)
q += " ORDER BY last_interaction DESC LIMIT ?"
params.append(limit)
with _conn() as c:
rows = [dict(r) for r in c.execute(q, params).fetchall()]
for row in rows:
row["tags"] = [r["tag"] for r in c.execute(
"SELECT tag FROM tags WHERE chat_id=?", (row["chat_id"],))]
return rows
def all_active_chat_ids(tag=None) -> list[int]:
with _conn() as c:
if tag:
rows = c.execute("""SELECT DISTINCT s.chat_id FROM subscribers s
JOIN tags t ON t.chat_id=s.chat_id
WHERE s.stopped=0 AND t.tag=?""", (tag,)).fetchall()
else:
rows = c.execute("SELECT chat_id FROM subscribers WHERE stopped=0").fetchall()
return [r[0] for r in rows]
def export_csv_rows() -> list[dict]:
with _conn() as c:
rows = [dict(r) for r in c.execute(
"SELECT * FROM subscribers ORDER BY joined_at").fetchall()]
for row in rows:
row["tags"] = ",".join(r["tag"] for r in c.execute(
"SELECT tag FROM tags WHERE chat_id=? ORDER BY tag", (row["chat_id"],)))
return rows
if __name__ == "__main__":
init_db()
print(f"[OK] DB initialized: {DB_PATH}")
```
### `bot.py`
```python
"""Движок воронки прогрева. python-telegram-bot (JobQueue, long-polling).
Запуск:
python bot.py # боевой
FUNNEL_TEST_MODE=1 python bot.py # ускоренные задержки для теста на себе
Движок не трогаем — контент только в posts.py, доступы в .env.
"""
from __future__ import annotations
import html
import logging
import re
import sys
from datetime import datetime, timedelta
from pathlib import Path
from telegram import Bot, InlineKeyboardButton, InlineKeyboardMarkup, Update
from telegram.constants import UpdateType
from telegram.error import BadRequest, Forbidden
from telegram.ext import Application, CommandHandler, CallbackQueryHandler, ContextTypes
from telegram.request import HTTPXRequest
sys.path.insert(0, str(Path(__file__).resolve().parent))
import config as cfg
import db
import posts
logging.basicConfig(format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", level=logging.INFO)
log = logging.getLogger("funnel_bot")
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)
# Границы серии из posts.py: дни 1..MAIN_LAST — основные, далее — weekly.
MAIN_LAST = max(posts.DAYS) if posts.DAYS else 0
WEEKLY_LAST = max(posts.WEEKLY) if posts.WEEKLY else MAIN_LAST
# ---------- Уведомления владельцу ----------
_notify_bot: Bot | None = None
def _get_notify_bot() -> Bot | None:
global _notify_bot
if _notify_bot is None and cfg.NOTIFY_BOT_TOKEN:
req = HTTPXRequest(proxy=cfg.TELEGRAM_PROXY) if cfg.TELEGRAM_PROXY else None
_notify_bot = Bot(token=cfg.NOTIFY_BOT_TOKEN, request=req)
return _notify_bot
async def _notify_owner(text: str, main_bot: Bot | None = None, **kw) -> None:
bot = _get_notify_bot() or main_bot
if bot is None:
log.warning("Некуда слать уведомление владельцу"); return
try:
await bot.send_message(cfg.OWNER_CHAT_ID, text, **kw)
except Exception as e: # noqa: BLE001
log.warning("notify_owner failed: %s", e)
def _fmt_user_header(sub: dict) -> str:
name = f"{sub.get('first_name') or ''} {sub.get('last_name') or ''}".strip() or "—"
uname = f"@{sub['username']}" if sub.get("username") else "—"
return f"Имя: {name}\nUsername: {uname}\nchat_id: {sub['chat_id']}"
def _human_tags(tags: list[str]) -> str:
labeled = [posts.TAG_LABELS.get(t, t) for t in tags
if not t.startswith("utm:") and not t.startswith("lead_")]
return "\n".join(labeled) if labeled else "—"
# ---------- Рендер плейсхолдеров ----------
_GUIDE_PATTERN = re.compile(r"\{GUIDE_URL(?::([a-z0-9_]+))?\}", re.IGNORECASE)
def _render(text: str, sub: dict) -> str:
text = _GUIDE_PATTERN.sub(lambda m: cfg.guide_url(m.group(1) or "hello_day0"), text)
safe = {
"first_name": html.escape(sub.get("first_name") or "друг"),
"OWNER_NAME": cfg.OWNER_NAME,
"CONTACT_LINK": cfg.CONTACT_LINK,
"CONTACT_PHONE": cfg.CONTACT_PHONE,
"SITE_MAIN": cfg.SITE_MAIN,
"SITE_PAY": cfg.SITE_PAY,
}
for k, v in safe.items():
text = text.replace("{" + k + "}", str(v))
return text
def _build_kb(buttons: list, sub: dict) -> InlineKeyboardMarkup | None:
if not buttons:
return None
rows = []
for label, data in buttons:
if data.startswith("url:"):
url = _render(data[4:], sub)
if not url:
continue # пустая ссылка — кнопку пропускаем
rows.append([InlineKeyboardButton(label, url=url)])
else:
rows.append([InlineKeyboardButton(label, callback_data=data)])
return InlineKeyboardMarkup(rows) if rows else None
async def _send_post(ctx, chat_id, post, sub, with_cta=False, current_day=None):
text = _render(post["text"], sub)
buttons = list(post.get("buttons") or [])
if with_cta:
if current_day is not None and current_day < WEEKLY_LAST:
buttons.insert(0, ("⏭ Следующий день сейчас", f"next:{current_day}"))
buttons.append((f"💬 Написать {cfg.OWNER_NAME}", "hot:daily_cta"))
buttons.append(("🚪 Отписаться", "unsub:now"))
kb = _build_kb(buttons, sub)
media = post.get("media")
CAPTION_LIMIT = 1024
if media and (file_id := cfg.VIDEO_FILE_IDS.get(media["video_key"])):
kind = media["kind"]
long_text = len(text) > CAPTION_LIMIT
if kind == "video_note":
await ctx.bot.send_video_note(chat_id, video_note=file_id)
await ctx.bot.send_message(chat_id, text, reply_markup=kb,
parse_mode="HTML", disable_web_page_preview=True)
return
if kind in ("video", "photo"):
send = ctx.bot.send_video if kind == "video" else ctx.bot.send_photo
arg = "video" if kind == "video" else "photo"
if long_text:
await send(chat_id, **{arg: file_id})
await ctx.bot.send_message(chat_id, text, reply_markup=kb,
parse_mode="HTML", disable_web_page_preview=True)
else:
await send(chat_id, **{arg: file_id}, caption=text,
parse_mode="HTML", reply_markup=kb)
return
await ctx.bot.send_message(chat_id, text, reply_markup=kb,
parse_mode="HTML", disable_web_page_preview=True)
def _subscriber_or_new(update: Update) -> dict:
u = update.effective_user
chat_id = update.effective_chat.id
db.upsert_subscriber(chat_id, u.first_name or "", u.last_name, u.username)
return db.get_subscriber(chat_id) or {"chat_id": chat_id}
# ---------- Планировщик ----------
def _cancel_start_questions(ctx, chat_id) -> None:
for j in ctx.job_queue.get_jobs_by_name(f"d0_questions_{chat_id}"):
j.schedule_removal()
def _next_delay(day: int) -> int:
if day < MAIN_LAST:
return cfg.DAY_DELAY_SECONDS
if MAIN_LAST <= day < WEEKLY_LAST:
return cfg.WEEKLY_DELAY_SECONDS
return 0
def _adjust_for_quiet_hours(delay_seconds: int) -> int:
"""Если плановое время попадает в тихие часы — сдвигаем на утро + рандом 0–30 мин."""
if not cfg.QUIET_HOURS_ENABLED or delay_seconds <= 0:
return delay_seconds
import random as _r
try:
from zoneinfo import ZoneInfo
now = datetime.now(ZoneInfo(cfg.QUIET_TZ))
except Exception:
now = datetime.now()
target = now + timedelta(seconds=delay_seconds)
h = target.hour
in_quiet = (h >= cfg.QUIET_START_HOUR) or (h < cfg.QUIET_END_HOUR)
if not in_quiet:
return delay_seconds
if h < cfg.QUIET_END_HOUR:
wakeup = target.replace(hour=cfg.QUIET_END_HOUR, minute=0, second=0, microsecond=0)
else:
wakeup = (target + timedelta(days=1)).replace(
hour=cfg.QUIET_END_HOUR, minute=0, second=0, microsecond=0)
wakeup += timedelta(minutes=_r.randint(0, 30))
return max(int((wakeup - now).total_seconds()), delay_seconds)
def _series_already_started(sub: dict) -> bool:
current_day = int(sub.get("current_day") or 0)
return (not sub.get("stopped")
and (current_day >= 1 or bool(sub.get("next_send_at"))
or sub.get("status") in {"active", "converted"}))
def _plan_day_one(ctx, chat_id) -> None:
delay = _adjust_for_quiet_hours(cfg.DAY_DELAY_SECONDS)
when = datetime.now() + timedelta(seconds=delay)
db.set_status(chat_id, "active")
db.set_next_send(chat_id, 1, when)
for j in ctx.job_queue.get_jobs_by_name(f"day_{chat_id}_1"):
j.schedule_removal()
ctx.job_queue.run_once(_job_send_day, when=delay, chat_id=chat_id,
name=f"day_{chat_id}_1", data={"day": 1})
db.log_event(chat_id, "series_started", {"day": 1})
async def _notify_new_lead(ctx, sub, suffix="") -> None:
tags = db.get_tags(sub["chat_id"])
await _notify_owner(
f"✅ Новый лид в воронке{suffix}\n\n{_fmt_user_header(sub)}\n\n{_human_tags(tags)}",
main_bot=ctx.bot)
async def _job_send_start_questions(ctx) -> None:
chat_id = ctx.job.chat_id
force = bool((ctx.job.data or {}).get("force"))
sub = db.get_subscriber(chat_id)
if not sub or sub.get("stopped"):
return
if not force and int(sub.get("current_day") or 0) != 0:
return
if posts.SURVEY:
await _send_post(ctx, chat_id, posts.SURVEY_INTRO, sub)
await _send_post(ctx, chat_id, posts.SURVEY[0]["post"], sub)
if int(sub.get("current_day") or 0) == 0:
db.set_next_send(chat_id, 0, None)
db.log_event(chat_id, "survey_started")
else:
await _send_post(ctx, chat_id, posts.SURVEY_DONE, sub)
fresh = db.get_subscriber(chat_id) or sub
if not _series_already_started(fresh):
_plan_day_one(ctx, chat_id)
await _notify_new_lead(ctx, fresh)
def _plan_start_questions(ctx, chat_id, force=False) -> None:
delay = _adjust_for_quiet_hours(cfg.START_QUESTIONS_DELAY_SECONDS)
when = datetime.now() + timedelta(seconds=delay)
if not force:
db.set_next_send(chat_id, 0, when)
_cancel_start_questions(ctx, chat_id)
ctx.job_queue.run_once(_job_send_start_questions, when=delay, chat_id=chat_id,
name=f"d0_questions_{chat_id}", data={"day": 0, "force": force})
db.log_event(chat_id, "start_questions_planned", {"delay_seconds": delay, "force": force})
async def _job_send_day(ctx) -> None:
chat_id = ctx.job.chat_id
day = ctx.job.data["day"]
sub = db.get_subscriber(chat_id)
if not sub or sub.get("stopped") or sub.get("status") == "converted":
return
post = posts.get_post(day)
if not post:
log.warning("no post for day %s", day); return
try:
await _send_post(ctx, chat_id, post, sub, with_cta=True, current_day=day)
db.log_event(chat_id, "day_sent", {"day": day})
except Forbidden as e:
db.stop(chat_id)
db.log_event(chat_id, "unsub_forbidden", {"day": day, "reason": str(e)[:200]})
return
except BadRequest as e:
msg = str(e).lower()
if any(x in msg for x in ("chat not found", "user is deleted", "bot was kicked")):
db.stop(chat_id)
db.log_event(chat_id, "unsub_bad_request", {"day": day, "reason": str(e)[:200]})
return
log.exception("send day %s to %s: %s", day, chat_id, e)
return
except Exception as e:
log.exception("send day %s to %s: %s", day, chat_id, e)
return
if day < WEEKLY_LAST:
next_day = day + 1
delay = _adjust_for_quiet_hours(_next_delay(day))
when = datetime.now() + timedelta(seconds=delay)
db.set_next_send(chat_id, next_day, when)
ctx.job_queue.run_once(_job_send_day, when=delay, chat_id=chat_id,
name=f"day_{chat_id}_{next_day}", data={"day": next_day})
else:
db.set_next_send(chat_id, day, None)
db.log_event(chat_id, "funnel_complete")
# ---------- Deeplink /start payload → UTM ----------
def _parse_start_payload(payload: str) -> dict | None:
import base64, json as _json
SOURCE_PREFIX = {
"vk": ("vk_ads", "cpc"), "vkads": ("vk_ads", "cpc"),
"yd": ("yandex_direct", "cpc"), "yandex": ("yandex_direct", "cpc"),
"tg": ("telegram_ads", "cpc"), "telegram": ("telegram_ads", "cpc"),
"ig": ("instagram", "social"), "instagram": ("instagram", "social"),
"fb": ("facebook", "social"), "facebook": ("facebook", "social"),
"site": ("site", "organic"), "popup": ("site", "popup"),
"hero": ("site", "hero_cta"), "page": ("site", "page_cta"),
}
p = payload.strip()
if p.startswith("c_"):
try:
raw = base64.urlsafe_b64decode(p[2:] + "=" * (-len(p[2:]) % 4)).decode("utf-8")
data = _json.loads(raw)
return data if isinstance(data, dict) else None
except Exception:
return None
parts = p.split("_", 1)
if len(parts) != 2:
return None
prefix, camp = parts[0].lower(), parts[1]
pair = SOURCE_PREFIX.get(prefix)
if not pair or not camp:
return None
src, medium = pair
return {"utm_source": src, "utm_medium": medium, "utm_campaign": camp}
# ---------- Команды ----------
async def start_cmd(update: Update, ctx) -> None:
sub = _subscriber_or_new(update)
db.log_event(sub["chat_id"], "start")
payload = (ctx.args or [""])[0] if ctx.args else ""
if payload:
utm = _parse_start_payload(payload)
if utm:
for k, v in utm.items():
if v:
db.add_tag(sub["chat_id"], f"utm:{k}:{v}")
db.log_event(sub["chat_id"], f"utm:{payload[:32]}")
else:
db.log_event(sub["chat_id"], f"utm:unknown:{payload[:32]}")
await _send_post(ctx, sub["chat_id"], posts.HELLO, sub)
is_fresh = int(sub.get("current_day") or 0) == 0 and sub.get("status") == "new"
_plan_start_questions(ctx, sub["chat_id"], force=not is_fresh)
async def guide_cmd(update: Update, ctx) -> None:
sub = _subscriber_or_new(update)
await _send_post(ctx, sub["chat_id"], posts.HELLO, sub)
async def contact_cmd(update: Update, ctx) -> None:
sub = _subscriber_or_new(update)
txt = _render("Пишите напрямую: {CONTACT_LINK}", sub)
if cfg.CONTACT_PHONE:
txt += f"\nТелефон: {cfg.CONTACT_PHONE}"
await ctx.bot.send_message(sub["chat_id"], txt)
async def menu_cmd(update: Update, ctx) -> None:
sub = _subscriber_or_new(update)
rows = []
if cfg.GUIDE_URL:
rows.append([InlineKeyboardButton("📖 Открыть гид", url=cfg.GUIDE_URL)])
rows.append([InlineKeyboardButton(f"💬 Написать {cfg.OWNER_NAME}", callback_data="hot:menu_msg")])
if cfg.SITE_MAIN:
rows.append([InlineKeyboardButton("Подробнее", url=cfg.SITE_MAIN)])
await ctx.bot.send_message(sub["chat_id"], "Что интересует?",
reply_markup=InlineKeyboardMarkup(rows))
def _owner_only(handler):
async def wrapped(update: Update, ctx):
if update.effective_chat.id != cfg.OWNER_CHAT_ID:
return
await handler(update, ctx)
return wrapped
def _format_sub_row(r: dict) -> str:
name = f"{r.get('first_name') or ''} {r.get('last_name') or ''}".strip() or "—"
uname = f"@{r['username']}" if r.get("username") else "—"
tags = ", ".join(r.get("tags") or []) or "—"
return (f"• [{r['chat_id']}] {name} {uname}\n"
f" день {r.get('current_day', 0)} · {r.get('status', 'new')} · {tags}")
@_owner_only
async def stats_cmd(update: Update, ctx) -> None:
s = db.stats()
lines = ["📊 Статистика воронки", f"Всего: {s['total']}",
f"Активные джобы: {s['active_jobs']}", f"Горячих: {s['hot']}", "", "По статусам:"]
for st, n in sorted(s["by_status"].items(), key=lambda x: -x[1]):
lines.append(f" {st:10} {n}")
await ctx.bot.send_message(update.effective_chat.id, "\n".join(lines))
@_owner_only
async def hot_cmd(update: Update, ctx) -> None:
rows = db.list_by_filter(status="hot", limit=50)
if not rows:
await ctx.bot.send_message(update.effective_chat.id, "Горячих лидов пока нет."); return
body = "\n\n".join(_format_sub_row(r) for r in rows)
await ctx.bot.send_message(update.effective_chat.id, f"🔥 Горячие ({len(rows)}):\n\n{body}")
@_owner_only
async def filter_cmd(update: Update, ctx) -> None:
args = ctx.args or []
if not args:
await ctx.bot.send_message(update.effective_chat.id,
"Usage: /filter <тег>\nТеги смотри в posts.py (TAG_LABELS) + lead_hot, lead_converted.")
return
tag = args[0].strip().lower()
rows = db.list_by_filter(tag=tag, limit=50)
if not rows:
await ctx.bot.send_message(update.effective_chat.id, f"По тегу «{tag}» — 0."); return
body = "\n\n".join(_format_sub_row(r) for r in rows)
await ctx.bot.send_message(update.effective_chat.id, f"Тег «{tag}» ({len(rows)}):\n\n{body}")
@_owner_only
async def broadcast_cmd(update: Update, ctx) -> None:
args = ctx.args or []
if not args:
await ctx.bot.send_message(update.effective_chat.id,
"Usage:\n /broadcast текст\n /broadcast tag:<тег> текст")
return
tag = None
if args[0].startswith("tag:"):
tag = args[0][4:].strip().lower() or None
args = args[1:]
text = " ".join(args).strip()
if not text:
await ctx.bot.send_message(update.effective_chat.id, "Пустой текст."); return
chat_ids = db.all_active_chat_ids(tag=tag)
await ctx.bot.send_message(update.effective_chat.id,
f"Рассылка на {len(chat_ids)} чатов" + (f" (тег: {tag})" if tag else "") + "…")
sent = failed = 0
for cid in chat_ids:
try:
await ctx.bot.send_message(cid, text, disable_web_page_preview=True)
db.log_event(cid, "broadcast", {"tag": tag})
sent += 1
except Forbidden:
db.stop(cid); failed += 1
except Exception as e: # noqa: BLE001
log.warning("broadcast %s: %s", cid, e); failed += 1
await ctx.bot.send_message(update.effective_chat.id,
f"Готово. Отправлено: {sent}, ошибок: {failed}.")
@_owner_only
async def export_cmd(update: Update, ctx) -> None:
import csv, io
rows = db.export_csv_rows()
if not rows:
await ctx.bot.send_message(update.effective_chat.id, "Подписчиков пока нет."); return
buf = io.StringIO()
w = csv.DictWriter(buf, fieldnames=list(rows[0].keys()))
w.writeheader(); w.writerows(rows)
await ctx.bot.send_document(update.effective_chat.id,
document=io.BytesIO(buf.getvalue().encode("utf-8-sig")),
filename="subscribers.csv", caption=f"Всего: {len(rows)}")
@_owner_only
async def savemedia_cmd(update: Update, ctx) -> None:
"""Ответь на фото/видео/кружок командой /savemedia КЛЮЧ — сохранит file_id
в .env под MEDIA_КЛЮЧ_FILE_ID (переживает рестарт)."""
chat_id = update.effective_chat.id
args = ctx.args or []
reply = update.message.reply_to_message if update.message else None
if not args or not reply:
await ctx.bot.send_message(chat_id,
"Пришли мне фото/видео, затем ответь на него: /savemedia КЛЮЧ (напр. /savemedia INTRO).")
return
key = args[0].strip().upper()
file_id = None
if reply.photo:
file_id = reply.photo[-1].file_id
elif reply.video:
file_id = reply.video.file_id
elif reply.video_note:
file_id = reply.video_note.file_id
if not file_id:
await ctx.bot.send_message(chat_id, "В сообщении нет фото/видео."); return
cfg.update_file_id(key, file_id)
await ctx.bot.send_message(chat_id,
f"✅ Сохранено под ключом {key}. В posts.py: media={{'kind': 'video', 'video_key': '{key}'}}.")
# ---------- Callback-кнопки ----------
async def cb_handler(update: Update, ctx) -> None:
q = update.callback_query
await q.answer()
sub = _subscriber_or_new(update)
data = q.data or ""
chat_id = sub["chat_id"]
# Анкета дня 0: d0:<field>:<value>
if data.startswith("d0:") and data != "d0:start_chain":
_cancel_start_questions(ctx, chat_id)
if int(sub.get("current_day") or 0) == 0:
db.set_next_send(chat_id, 0, None)
_, field, value = data.split(":", 2)
db.add_tag(chat_id, f"{field}_{value}")
db.log_event(chat_id, "d0_tag", {"tag": f"{field}_{value}"})
idx = next((i for i, item in enumerate(posts.SURVEY) if item["field"] == field), None)
nxt = posts.SURVEY[idx + 1] if idx is not None and idx + 1 < len(posts.SURVEY) else None
if nxt:
await _send_post(ctx, chat_id, nxt["post"], sub)
else:
await _send_post(ctx, chat_id, posts.SURVEY_DONE, sub)
fresh = db.get_subscriber(chat_id) or sub
if not _series_already_started(fresh):
_plan_day_one(ctx, chat_id)
await _notify_new_lead(ctx, fresh)
return
if data == "d0:start_chain":
fresh = db.get_subscriber(chat_id) or sub
if not _series_already_started(fresh):
_plan_day_one(ctx, chat_id)
await _notify_new_lead(ctx, fresh)
await ctx.bot.send_message(chat_id,
"Хорошо. Первое сообщение пришлю в ближайшее время." if not cfg.TEST_MODE
else "🧪 Тест-режим: следующее сообщение через минуту.")
return
if data.startswith("info:"):
await ctx.bot.send_message(chat_id, "Эта кнопка пока в работе."); return
# next:N — «не хочу ждать», шлём день N+1 сразу
if data.startswith("next:"):
try:
current_day = int(data.split(":", 1)[1])
except ValueError:
return
next_day = current_day + 1
if next_day > WEEKLY_LAST:
await ctx.bot.send_message(chat_id, "Это было последнее сообщение серии."); return
if current_day == 0 and not _series_already_started(sub):
db.set_status(chat_id, "active")
await _notify_new_lead(ctx, sub, " (смотрит без ожидания)")
for j in ctx.job_queue.get_jobs_by_name(f"day_{chat_id}_{next_day}"):
j.schedule_removal()
post = posts.get_post(next_day)
if not post:
return
await _send_post(ctx, chat_id, post, sub, with_cta=True, current_day=next_day)
db.log_event(chat_id, "day_sent", {"day": next_day, "source": "next_btn"})
if next_day < WEEKLY_LAST:
delay = _adjust_for_quiet_hours(_next_delay(next_day))
when = datetime.now() + timedelta(seconds=delay)
db.set_next_send(chat_id, next_day + 1, when)
ctx.job_queue.run_once(_job_send_day, when=delay, chat_id=chat_id,
name=f"day_{chat_id}_{next_day + 1}", data={"day": next_day + 1})
else:
db.set_next_send(chat_id, next_day, None)
db.log_event(chat_id, "funnel_complete")
return
if data == "unsub:now":
db.stop(chat_id)
db.log_event(chat_id, "unsub", {"day": sub.get("current_day")})
await ctx.bot.send_message(chat_id,
"Отписал. Передумаешь — напиши /start, продолжим с того же места.")
await _notify_owner(
f"🚪 Отписка\n\n{_fmt_user_header(sub)}\nДень: {sub.get('current_day', 0)}",
main_bot=ctx.bot)
return
# hot:* — горячий лид из любой CTA-кнопки
if data.startswith("hot:"):
db.add_tag(chat_id, "lead_hot")
db.set_status(chat_id, "hot")
db.log_event(chat_id, "lead_hot", {"trigger": data})
await ctx.bot.send_message(chat_id,
_render("Отлично. Напишу тебе в ближайшее время.\n\n"
"Если удобнее — напиши сам: {CONTACT_LINK}", sub))
tags = db.get_tags(chat_id)
trigger_ru = posts.HOT_LABELS.get(data, data)
await _notify_owner(
f"🔥 Горячий лид\n\n{_fmt_user_header(sub)}\n\n"
f"Откуда: {trigger_ru}\n\n{_human_tags(tags)}",
main_bot=ctx.bot)
return
# ---------- Восстановление расписания при старте ----------
async def _restore_jobs(app: Application) -> None:
now = datetime.now()
for sub in db.active_subscribers_with_pending():
chat_id = sub["chat_id"]
next_day = sub["current_day"] or 0
try:
nxt = datetime.fromisoformat(sub["next_send_at"])
except (TypeError, ValueError):
continue
delay = max(1, int((nxt - now).total_seconds()))
if next_day == 0:
app.job_queue.run_once(_job_send_start_questions, when=delay, chat_id=chat_id,
name=f"d0_questions_{chat_id}", data={"day": 0})
continue
if next_day < 1 or next_day > WEEKLY_LAST:
continue
app.job_queue.run_once(_job_send_day, when=delay, chat_id=chat_id,
name=f"day_{chat_id}_{next_day}", data={"day": next_day})
log.info("restored: chat=%s day=%s in %ss", chat_id, next_day, delay)
# ---------- Main ----------
def main() -> None:
cfg.assert_ready()
db.init_db()
builder = (Application.builder().token(cfg.BOT_TOKEN)
.connection_pool_size(16).pool_timeout(15.0)
.read_timeout(30.0).connect_timeout(15.0)
.get_updates_connection_pool_size(2).get_updates_pool_timeout(15.0))
if cfg.TELEGRAM_PROXY:
builder = builder.proxy(cfg.TELEGRAM_PROXY).get_updates_proxy(cfg.TELEGRAM_PROXY)
app = builder.build()
app.add_handler(CommandHandler("start", start_cmd))
app.add_handler(CommandHandler("guide", guide_cmd))
app.add_handler(CommandHandler("contact", contact_cmd))
app.add_handler(CommandHandler("menu", menu_cmd))
app.add_handler(CommandHandler("stats", stats_cmd))
app.add_handler(CommandHandler("hot", hot_cmd))
app.add_handler(CommandHandler("filter", filter_cmd))
app.add_handler(CommandHandler("broadcast", broadcast_cmd))
app.add_handler(CommandHandler("export", export_cmd))
app.add_handler(CommandHandler("savemedia", savemedia_cmd))
app.add_handler(CallbackQueryHandler(cb_handler))
async def _on_error(_u, context):
err = context.error
if isinstance(err, (BadRequest, __import__("telegram").error.TimedOut)):
log.warning("handler: %s", err); return
log.exception("handler error", exc_info=err)
app.add_error_handler(_on_error)
async def _post_init(a: Application):
await _restore_jobs(a)
log.info("Бот @%s запущен. TEST_MODE=%s", cfg.BOT_USERNAME, cfg.TEST_MODE)
app.post_init = _post_init
app.run_polling(allowed_updates=[UpdateType.MESSAGE, UpdateType.CALLBACK_QUERY])
if __name__ == "__main__":
main()
```
### `posts.py` — ШАБЛОН (заполни из ответов интервью)
Формат текста — Telegram HTML: `<b>жирный</b>`, `<i>курсив</i>`, `<blockquote>цитата</blockquote>`, `<tg-spoiler>скрытое</tg-spoiler>`, `<a href="...">ссылка</a>`.
Плейсхолдеры в тексте: `{first_name}`, `{GUIDE_URL}` (или `{GUIDE_URL:метка}` для UTM), `{OWNER_NAME}`, `{CONTACT_LINK}`, `{CONTACT_PHONE}`, `{SITE_MAIN}`, `{SITE_PAY}`.
Кнопки: `(подпись, "d0:поле:значение")` — анкета; `(подпись, "hot:...")` — горячий лид; `(подпись, "url:https://...")` — ссылка.
```python
"""Контент воронки. Здесь — ВСЁ, что меняется под нишу. Движок не трогаем.
Структура: HELLO → SURVEY_INTRO → SURVEY[...] → SURVEY_DONE → DAYS[1..N] → WEEKLY[...].
Дни в DAYS должны быть подряд: 1,2,3,…N. WEEKLY (если есть) — подряд с N+1.
"""
from __future__ import annotations
# --- День 0: приветствие (уходит сразу на /start) ---
HELLO = {
"text": (
"Привет, <b>{first_name}</b>!\n\n"
"<i>[2–4 строки: узнай боль/сомнение лида, пообещай честный разбор без давления.]</i>"
),
# Кнопка на лид-магнит. Нет лид-магнита — оставь "buttons": [].
"buttons": [("📖 Открыть гид", "url:{GUIDE_URL}")],
}
# --- Подводка к анкете (уходит через паузу после приветствия) ---
SURVEY_INTRO = {
"text": "Позволь пару вопросов, чтобы писать по делу, а не спамить.",
"buttons": [],
}
# --- Анкета. Каждый ответ → тег «поле_значение». Не нужна анкета — SURVEY = [] ---
SURVEY = [
{
"field": "goal",
"post": {
"text": "🎯 <b>Что тебе сейчас важнее?</b>",
"buttons": [
("Вариант A", "d0:goal:a"),
("Вариант B", "d0:goal:b"),
("Вариант C", "d0:goal:c"),
],
},
},
{
"field": "timing",
"post": {
"text": "📅 <b>Когда планируешь?</b>",
"buttons": [
("Скоро", "d0:timing:soon"),
("Через пару месяцев", "d0:timing:later"),
("Пока присматриваюсь", "d0:timing:undecided"),
],
},
},
]
# --- Финал анкеты (после последнего вопроса) ---
SURVEY_DONE = {
"text": (
"🙏 <b>Спасибо!</b>\n\n"
"Завтра пришлю первое сообщение. А не хочешь ждать — начнём сейчас 👇"
),
"buttons": [
("⏭ Начать сейчас, без ожидания", "next:0"),
("💬 Написать {OWNER_NAME}", "hot:d0_ready"),
("✅ Хорошо, жду", "d0:start_chain"),
],
}
# --- Человекочитаемые подписи тегов (для уведомлений владельцу) ---
TAG_LABELS = {
"goal_a": "Цель: вариант A",
"goal_b": "Цель: вариант B",
"goal_c": "Цель: вариант C",
"timing_soon": "Когда: скоро",
"timing_later": "Когда: через пару месяцев",
"timing_undecided": "Когда: присматривается",
}
# --- Подписи для «откуда пришёл горячий лид» ---
HOT_LABELS = {
"hot:d0_ready": "после анкеты — захотел поговорить",
"hot:daily_cta": "из сообщения серии",
"hot:menu_msg": "из меню /menu",
"hot:d2_price": "День 2 — узнать цены и условия",
"hot:w1_msg": "Weekly — написать",
# добавь свои hot:* из кнопок в DAYS/WEEKLY
}
# --- Основная серия: по одному сообщению = одна тема ---
DAYS: dict[int, dict] = {
1: {
# media опционально: {"kind": "video"|"photo"|"video_note", "video_key": "INTRO"}
# video_key должен совпасть с MEDIA_<KEY>_FILE_ID в .env (грузится через /savemedia)
"text": (
"🎯 <b>ДЕНЬ 1. [ЗАГОЛОВОК]</b>\n\n"
"<i>[Тело: одна тема/боль. Твой живой голос. Без обещаний того, чего нет в оффере.]</i>\n\n"
"Завтра — <b>[тизер следующей темы]</b>."
),
"buttons": [],
},
2: {
"text": (
"💰 <b>ДЕНЬ 2. [ТЕМА ПРО ЦЕНУ/ОФФЕР]</b>\n\n"
"<i>[Цифры и оффер — здесь, где о деньгах.]</i>"
),
# «денежный» день — уместна CTA к целевому действию (создаст горячий лид)
"buttons": [("💬 Узнать цены и условия", "hot:d2_price")],
},
# … догенерируй дни 3..N из тем интервью (по одной теме на день).
}
# --- Weekly-дожимы (опц.). Не нужны — WEEKLY = {}. Ключи — подряд с N+1. ---
WEEKLY: dict[int, dict] = {
3: { # пример: если основная серия = 2 дня, weekly начинается с 3
"text": (
"<b>{first_name}</b>, короткое напоминание.\n\n"
"<i>[Мягкий дожим тех, кто ещё думает. Без давления.]</i>"
),
"buttons": [("💬 Написать {OWNER_NAME}", "hot:w1_msg")],
},
}
def get_post(day: int):
if day == 0:
return HELLO
if day in DAYS:
return DAYS[day]
if day in WEEKLY:
return WEEKLY[day]
return None
```
> **Важно про нумерацию.** `DAYS` — подряд `1..N`. `WEEKLY` (если есть) продолжает нумерацию с `N+1` подряд. Не должно быть дыр — иначе серия оборвётся на пропуске. Каждый `hot:<код>` из кнопок добавь в `HOT_LABELS`, каждый тег анкеты — в `TAG_LABELS`.
### `.env.example`
```dotenv
# --- Обязательное ---
BOT_TOKEN=123456:PASTE_TOKEN_FROM_BOTFATHER
BOT_USERNAME=my_funnel_bot
OWNER_CHAT_ID=0 # твой chat_id из @userinfobot
# --- Уведомления владельцу (опц.) ---
NOTIFY_BOT_TOKEN= # токен второго бота; пусто = слать основным
# --- Бренд / контакты / ссылки (подставляются в тексты) ---
OWNER_NAME=нам
CONTACT_LINK=https://t.me/your_username
CONTACT_PHONE=
SITE_MAIN=
SITE_PAY=
GUIDE_URL= # ссылка на лид-магнит; пусто = кнопка «гид» скрыта
UTM_CAMPAIGN=funnel
# --- Расписание / тихие часы ---
FUNNEL_TEST_MODE=0 # 1 = ускоренные задержки для теста на себе
QUIET_HOURS_ENABLED=1
QUIET_START_HOUR=22
QUIET_END_HOUR=9
QUIET_TZ=Europe/Moscow
# --- Прокси для Telegram API (опц., только если регион блокирует) ---
TELEGRAM_PROXY=
# --- Медиа (заполняются автоматически командой /savemedia) ---
# MEDIA_INTRO_FILE_ID=...
```
### `requirements.txt`
```text
python-telegram-bot[job-queue]>=21.0
python-dotenv>=1.0.0
```
### `README.md` (коротко)
```markdown
# Бот воронки прогрева
Standalone Python + python-telegram-bot (JobQueue). Контент — в `posts.py`, доступы — в `.env`.
## Запуск
1. `python -m venv .venv && . .venv/bin/activate` (Windows: `.venv\Scripts\activate`)
2. `pip install -r requirements.txt`
3. `cp .env.example .env` и заполнить `BOT_TOKEN`, `OWNER_CHAT_ID`.
4. Тест на себе: `FUNNEL_TEST_MODE=1 python bot.py` → в Telegram `/start`.
5. Боевой: `python bot.py` (лучше под systemd/pm2/nssm, чтобы жил после перезагрузки).
`.env` и `data/` — в `.gitignore`. БД `data/funnel.sqlite` — источник истины, её бэкапь.
```
Добавь `.gitignore` со строками: `.env`, `data/`, `.venv/`, `__pycache__/`.
---
## Настройка бота в @BotFather (сделать один раз)
1. `/newbot` → имя и username → получишь **токен** (в `BOT_TOKEN`).
2. `/setcommands` для бота, вставить:
```
start - Начать
guide - Получить гид ещё раз
contact - Связаться
menu - Меню
```
3. `/setdescription` и `/setabouttext` — короткий текст про бота.
4. `/setprivacy` → `Enable` (бот видит только адресованные сообщения). `/setjoingroups` → `Disable`.
---
## Запуск и самотест (обязательно перед боевым)
1. Заполни `.env` (`BOT_TOKEN`, `OWNER_CHAT_ID`) и `posts.py` из ответов интервью.
2. `FUNNEL_TEST_MODE=1 python bot.py`.
3. В Telegram открой бота → `/start`:
- пришло приветствие (+ кнопка гида, если задан);
- через ~минуту — подводка и первый вопрос анкеты;
- ответь на анкету → пришёл финал, тебе (владельцу) упало «✅ Новый лид»;
- дни идут ускоренно (60с); нажми «⏭ Следующий день сейчас» — приходит следующий;
- нажми «💬 Написать …» — тебе упало «🔥 Горячий лид»;
- `/stats`, `/hot`, `/export` — работают только у владельца.
4. Проверь рестарт-устойчивость: во время серии `Ctrl+C`, снова `python bot.py` — расписание восстановилось (в логах `restored: …`), день не задвоился и не пропал.
5. Медиа: пришли боту фото/видео → ответь на него `/savemedia INTRO` → пропиши `media` в нужном дне `posts.py`.
6. Всё ок → `FUNNEL_TEST_MODE=0`, боевой запуск под супервизором.
---
## Чек-лист развязки (ничего чужого)
- [ ] `.env` заполнен своими `BOT_TOKEN` и `OWNER_CHAT_ID`.
- [ ] В `posts.py` — свой контент, свои ссылки/цены/маркеры; чужих доменов и юзернеймов нет.
- [ ] `CONTACT_LINK`, `SITE_*`, `GUIDE_URL` — свои (или пустые).
- [ ] Нет внешних сервисов (Google Sheets, CRM, S3, чужие дашборды) — только Telegram + локальный SQLite.
- [ ] `.env` и `data/` в `.gitignore`; токен нигде в коде.
- [ ] Каждый `hot:<код>` есть в `HOT_LABELS`, каждый тег анкеты — в `TAG_LABELS`.
- [ ] `DAYS` пронумерованы подряд `1..N`, `WEEKLY` (если есть) — подряд с `N+1`.
```
Пройти по пунктам — чтобы у человека завёлся его собственный проект, а не копия твоего с торчащими личными данными.
.env, он в .gitignore; в коде токенов нет