--- title: Telegram-ассистент для закупки рекламы: пошаговая сборка с кодом и промтами url: https://blog.krasovskiy.team/ru/telegram-assistent-zakupka-reklamy-instrukciya/ date: 2026-07-22 lang: ru source: blog.krasovskiy.team --- # Telegram-ассистент для закупки рекламы: пошаговая сборка с кодом и промтами Пошаговое руководство: как собрать бота, который принимает голосовые сообщения, превращает их в задачи, ведёт таблицу закупок, напоминает о замерах, анализирует каналы перед покупкой и пишет черновики рекламных постов. Инструкция рассчитана на человека, который раньше не поднимал ботов. Все команды копируются целиком. Если шаг не получился — в конце каждого раздела есть проверка и разбор типичных ошибок. **Что нужно иметь до начала:** **Во что обойдётся в месяц:** сервер — от 4 долларов; Telegram — бесплатно; Google Sheets — бесплатно; распознавание голоса и разбор задач — меньше доллара при обычной нагрузке. **Важно про выбор сервера.** Часть хостингов (особенно шаред-хостинги для сайтов) блокируют исходящие соединения к api.telegram.org. На таком сервере бот не заработает, и понять это сложно — он просто молчит. Проверка одной командой есть в шаге 3. ![Путь голосового сообщения: от диктовки до строки в таблице](https://blog.krasovskiy.team/wp-content/uploads/2026/07/ru-02-arch.png) ## Часть 1. Процесс, который мы автоматизируем Бот бесполезен, если непонятно, какую работу он поддерживает. Ниже — короткая версия процесса закупки. Полная версия со всеми шаблонами сообщений, формулами и правилами безопасности лежит в отдельном документе «Регламент: закупка рекламы в Telegram-каналах», его нужно прочитать до начала работы. Закупка состоит из семи этапов: Три места здесь ломаются чаще всего, и именно их закрывает бот: Бот берёт на себя память и счёт. Решения и переписку с людьми оставляем человеку. ## Часть 2. Подготовка ## Шаг 1. Создаём бота Открываем Telegram, в поиске находим **@BotFather** — официальный бот Telegram для создания ботов. У настоящего BotFather есть галочка верификации, у подделок её нет. **Токен — это пароль от бота.** Кто им владеет, тот управляет ботом. Не отправляем его в общие чаты, не коммитим в репозиторий. Если случайно засветили — в BotFather команда `/revoke` выпускает новый. Сразу настраиваем приватность, чтобы бот не собирал лишнее: ``` start - запустить help - что умею task - добавить задачу текстом today - задачи на сегодня channel - анализ канала перед закупкой draft - черновик рекламного поста ``` ## Шаг 2. Узнаём свой Telegram ID Бот должен отвечать только своим. Для этого нужен числовой ID каждого сотрудника. Находим в поиске бота **@userinfobot**, запускаем, он присылает ID — число вида `123456789`. Собираем ID всех, кто будет пользоваться ассистентом. Остальным бот будет отвечать отказом. ## Шаг 3. Готовим сервер Подключаемся к серверу по SSH: ``` ssh root@АДРЕС_СЕРВЕРА ``` **Первым делом проверяем, что сервер вообще может достучаться до Telegram:** ``` curl -s -o /dev/null -w "%{http_code}n" https://api.telegram.org ``` Должно вывести `200`. Если команда висит или возвращает `000` — этот сервер не подходит, Telegram с него заблокирован. Меняем хостинг, дальше идти бессмысленно. Создаём отдельного пользователя — работать ботом от root неправильно: ``` adduser --disabled-password --gecos "" assistant apt update && apt install -y python3-venv python3-pip ffmpeg su - assistant ``` Готовим папку проекта и виртуальное окружение: ``` mkdir -p ~/tg-assistant && cd ~/tg-assistant python3 -m venv venv source venv/bin/activate pip install --upgrade pip pip install aiogram==3.15.0 requests gspread google-auth apscheduler telethon python-dotenv ``` **Проверка шага:** команда `python -c "import aiogram, gspread, telethon; print('ok')"` должна вывести `ok`. ## Шаг 4. Получаем ключи Ассистенту нужны две способности: **распознать голос** и **разобрать смысл**. Разбор смысла делает Mistral, распознавание — на выбор, подробное сравнение с замерами в шаге 8. **Ключ Mistral (нужен всегда):** **Ключ OpenAI (рекомендуется, для распознавания):** Распознавание стоит около половины цента за минуту речи. При двадцати голосовых в день это меньше доллара в месяц. Почему рекомендуется именно он — в шаге 8, там замеры на одном и том же файле. Если платить за второй сервис не хочется — работает и на одном ключе Mistral, и есть полностью бесплатный вариант без внешних сервисов вообще. Оба описаны там же. ![Пять листов таблицы — вся память ассистента](https://blog.krasovskiy.team/wp-content/uploads/2026/07/ru-05-sheets.png) ## Шаг 5. Готовим таблицу Таблица — это база данных ассистента. Структура повторяет регламент. Создаём Google-таблицу с названием `Закупка Telegram` и пятью листами. Названия листов должны совпадать буква в букву: **Лист `Задачи`** — первая строка: ``` Дата создания | Задача | Тип | Канал | Срок | Приоритет | Статус | Заметка ``` **Лист `Каналы`**: ``` Название | Ссылка | Тематика | Подписчиков | Средний охват | ER % | Админ | Кошелёк | Прайс | Статус | Последний контакт | Заметки ``` **Лист `Размещения`**: ``` ID | Канал | Дата брони | Дата выхода | Формат | Цена | Валюта | Кошелёк | Хеш | Хеш отправлен | Креатив | Инвайт-ссылка | Ссылка на пост | Статус | Охват 24ч | Подписчиков 7д | CPM | CPS | Вердикт ``` **Лист `Платежи`**: ``` Дата | Сумма | Валюта | Кошелёк | Хеш | Канал | Хеш отправлен ``` **Лист `Креативы`**: ``` ID | Текст | Медиа | Где использовался | Лучший CPS | Заметки ``` Теперь выдаём боту доступ к этой таблице. **Шаг, который забывают все.** Если не дать сервисному аккаунту доступ к таблице (пункт 6), бот будет падать с ошибкой `SpreadsheetNotFound`, хотя таблица существует. Сервисный аккаунт — это отдельный «пользователь», он не видит ваши файлы автоматически. Загружаем файл ключа на сервер. Команда выполняется **на своём компьютере**, не на сервере: ``` scp ~/Downloads/имя-файла-ключа.json assistant@АДРЕС_СЕРВЕРА:~/tg-assistant/google.json ``` ## Часть 3. Сборка Все файлы создаются в папке `~/tg-assistant`. ## Шаг 6. Настройки Создаём файл `.env`: ``` nano ~/tg-assistant/.env ``` Содержимое (подставляем свои значения): ``` BOT_TOKEN=8123456789:AAF... MISTRAL_KEY=... ALLOWED_IDS=123456789,987654321 SHEET_NAME=Закупка Telegram GOOGLE_CREDS=/home/assistant/tg-assistant/google.json CHANNEL_ID=@moy_kanal TG_API_ID= TG_API_HASH= ``` Сохранение в nano: `Ctrl+O`, `Enter`, `Ctrl+X`. Закрываем файл от посторонних: ``` chmod 600 ~/tg-assistant/.env ~/tg-assistant/google.json ``` Файл `config.py`: ```python import os from dotenv import load_dotenv load_dotenv() BOT_TOKEN = os.environ["BOT_TOKEN"] MISTRAL_KEY = os.environ["MISTRAL_KEY"] ALLOWED_IDS = {int(x) for x in os.environ["ALLOWED_IDS"].split(",") if x.strip()} SHEET_NAME = os.environ["SHEET_NAME"] GOOGLE_CREDS = os.environ["GOOGLE_CREDS"] CHANNEL_ID = os.getenv("CHANNEL_ID", "") TG_API_ID = os.getenv("TG_API_ID", "") TG_API_HASH = os.getenv("TG_API_HASH", "") MISTRAL_URL = "https://api.mistral.ai/v1" MODEL_TEXT = "mistral-large-latest" MODEL_STT = "voxtral-mini-latest" ``` ## Шаг 7. Скелет бота Файл `bot.py`: ```python import asyncio import logging from aiogram import Bot, Dispatcher, F from aiogram.filters import Command from aiogram.types import Message import config logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") bot = Bot(token=config.BOT_TOKEN) dp = Dispatcher() def allowed(message: Message) -> bool: return message.from_user.id in config.ALLOWED_IDS @dp.message(Command("start")) async def cmd_start(message: Message): if not allowed(message): await message.answer("Нет доступа. Обратитесь к администратору.") return await message.answer( "Ассистент закупки на связи.nn" "Надиктуйте голосовое — разберу на задачи.n" "/today — задачи на сегодняn" "/channel ссылка — анализ каналаn" "/draft — черновик рекламного поста" ) async def main(): await dp.start_polling(bot) if __name__ == "__main__": asyncio.run(main()) ``` Запускаем: ``` cd ~/tg-assistant && source venv/bin/activate && python bot.py ``` **Проверка шага:** находим своего бота в Telegram по username, жмём «Запустить». Должно прийти приветствие. В консоли сервера появятся строки лога. Остановить бота — `Ctrl+C`. **Если ответа нет:** ![Замер шести вариантов распознавания на одном голосовом](https://blog.krasovskiy.team/wp-content/uploads/2026/07/ru-03-stt.png) ## Шаг 8. Голосовое сообщение в текст Здесь легко ошибиться с выбором сервиса, поэтому ниже замеры, а не мнения. Одно и то же голосовое — три поручения, два названия каналов латиницей и произнесённый вслух формат «один двадцать четыре» — прогнано через четыре варианта. ВариантЧто получилось на спорных местахMistral Voxtral`1.24`, `DigitalBourse`, `Telegram Pro IT`OpenAI whisper-1`1.24`, `DigitalBors`, `Telegram про айти`OpenAI gpt-4o-transcribe`1.24`, потерян союз: «нейросети деньги»**OpenAI gpt-4o-transcribe со словарём****`1/24`**, `Digital Bors`, `Telegram Pro IT`, «две тысячи рублей»faster-whisper `small` локально`1.24`, «напомним мне» вместо «напомни мне», «канал о нейросети», `Digital Bores`faster-whisper `small` со словарём`1.24`, «Ниросетий деньги» — словарь сделал хуже Вывод не в том, что одна модель умнее другой — по разборчивости речи они близки. Разница в одном: **OpenAI позволяет передать словарь терминов перед распознаванием, и это единственное, что исправило `1/24`.** Проверено отдельно: Mistral такую подсказку принимает молча и игнорирует — три прогона, с подсказкой и без, дали посимвольно одинаковый текст. Почему это важно именно в закупке: `1.24` вместо `1/24` — это формат размещения, то есть то, за что платят деньги. Ошибка тянется дальше в задачу, в переписку с админом и в отчёт. Поддерживаем три варианта, переключаются одной строкой в `.env`: **Честно про бесплатный вариант.** В замерах выше он прогнан на модели `small` — той, что работает на дешёвом сервере. Видно, что он путает окончания и глаголы: «напомним мне» вместо «напомни мне». Для задач, где важен смысл, а не буква, этого хватает — разбор в шаге 9 всё равно вытащит нужное. Но словарь терминов ему не помогает, а иногда вредит: с подсказкой название канала стало «Ниросетий деньги» вместо «нейросети и деньги». Модель `large-v3` заметно точнее, но требует около 3 ГБ на диске и сервера от 4 ГБ памяти — на ней замеров здесь нет, проверяйте на своих голосовых, прежде чем ставить в работу. Дописываем в `.env`: ``` STT_PROVIDER=openai OPENAI_KEY=sk-... ``` И в `config.py`: ``` STT_PROVIDER = os.getenv("STT_PROVIDER", "mistral") OPENAI_KEY = os.getenv("OPENAI_KEY", "") MODEL_STT_OPENAI = "gpt-4o-transcribe" MODEL_STT_LOCAL = "large-v3" ``` Файл `stt.py`: ```python import tempfile import requests import config # Словарь терминов. Распознавание опирается на него в спорных местах. # Дополняйте своими: названия ваших каналов, продуктов, имена админов. HINT = ( "Термины закупки рекламы в Telegram: формат 1/24, формат 2/48, посев, " "охват, инвайт-ссылка, CPM, CPS, хеш транзакции, слот, закреп, " "оплата в USDT, чёрный список." ) _local_model = None def _openai(audio: bytes, filename: str) -> str: response = requests.post( "https://api.openai.com/v1/audio/transcriptions", headers={"Authorization": f"Bearer {config.OPENAI_KEY}"}, files={"file": (filename, audio, "audio/ogg")}, data={"model": config.MODEL_STT_OPENAI, "prompt": HINT}, timeout=180, ) response.raise_for_status() return response.json().get("text", "").strip() def _mistral(audio: bytes, filename: str) -> str: response = requests.post( f"{config.MISTRAL_URL}/audio/transcriptions", headers={"Authorization": f"Bearer {config.MISTRAL_KEY}"}, files={"file": (filename, audio, "audio/ogg")}, data={"model": config.MODEL_STT}, timeout=180, ) response.raise_for_status() return response.json().get("text", "").strip() def _local(audio: bytes, filename: str) -> str: global _local_model from faster_whisper import WhisperModel if _local_model is None: _local_model = WhisperModel(config.MODEL_STT_LOCAL, device="cpu", compute_type="int8") with tempfile.NamedTemporaryFile(suffix=".ogg") as handle: handle.write(audio) handle.flush() segments, _ = _local_model.transcribe(handle.name, language="ru", initial_prompt=HINT) return " ".join(segment.text.strip() for segment in segments).strip() def transcribe(audio: bytes, filename: str = "voice.ogg") -> str: """Голосовое Telegram (ogg/opus) в текст. Конвертация не нужна.""" if config.STT_PROVIDER == "openai": return _openai(audio, filename) if config.STT_PROVIDER == "local": return _local(audio, filename) return _mistral(audio, filename) ``` Для варианта `local` доставляем библиотеку — это открытый проект, модель скачивается сама при первом запуске: ``` pip install faster-whisper ``` Первый запуск будет долгим: модель `large-v3` весит около трёх гигабайт и качается один раз. Если сервер слабый, поменяйте `MODEL_STT_LOCAL` на `medium` или `small` — качество ниже, зато работает на гигабайте памяти. Отдельно конвертировать голосовые не нужно ни в одном из вариантов — формат Telegram принимается как есть. Добавляем в `bot.py` обработчик. Импорты — вверх файла, обработчик — перед функцией `main`: ```python import stt @dp.message(F.voice) async def on_voice(message: Message): if not allowed(message): return note = await message.answer("Слушаю...") file = await bot.get_file(message.voice.file_id) buffer = await bot.download_file(file.file_path) text = stt.transcribe(buffer.read()) await note.edit_text(f"Расшифровка:nn{text}") ``` **Проверка шага:** перезапускаем бота, отправляем голосовое. Через несколько секунд приходит текст. **Особенность, о которой надо знать.** Названия каналов латиницей любое распознавание пишет по-своему: `DigitalBourse`, `DigitalBors`, `Digital Bors` — это три варианта одного и того же канала из замеров выше. Исправлять это в промте бессмысленно, модель не знает, как правильно. Поэтому в правилах разбора стоит прямой запрет «переноси как есть, не исправляй»: пусть в задаче будет узнаваемое искажение, чем уверенно выдуманное чужое название. Со словарём (вариант `openai`) термины закупки распознаются верно. На вариантах `mistral` и `local` формат вида `1.24` подчищается уже на следующем шаге — правилом 6 в промте разбора. Поэтому шаг 9 нужен в любом случае. ## Шаг 9. Разбор текста в задачи Это главный узел ассистента. Промт ниже — рабочий, он прошёл проверку на живых голосовых, включая описанные искажения. Файл `prompts.py`: ```python import datetime DAYS = ["понедельник", "вторник", "среда", "четверг", "пятница", "суббота", "воскресенье"] def date_table(today: datetime.date) -> str: """Готовый календарь для модели. Считать даты сама она не должна.""" lines = [ f"сегодня = {today.isoformat()} ({DAYS[today.weekday()]})", f"завтра = {(today + datetime.timedelta(1)).isoformat()}", f"послезавтра = {(today + datetime.timedelta(2)).isoformat()}", ] for index, name in enumerate(DAYS): shift = ((index - today.weekday()) % 7) or 7 lines.append(f"ближайший {name} = " f"{(today + datetime.timedelta(shift)).isoformat()}") monday = today + datetime.timedelta(days=7 - today.weekday()) lines.append(f"следующая неделя = с {monday.isoformat()} " f"по {(monday + datetime.timedelta(6)).isoformat()}") lines.append(f"через неделю = {(today + datetime.timedelta(7)).isoformat()}") return "n".join(lines) TASKS_SYSTEM = """Ты разбираешь надиктованные сообщения менеджера по закупке рекламы в Telegram-каналах и превращаешь их в задачи. Отвечаешь только валидным JSON, без markdown и пояснений. Календарь (бери даты только отсюда, сам не вычисляй): {calendar} Формат ответа: {{"tasks": [{{"title": "...", "type": "...", "channel": "...", "due": "YYYY-MM-DD", "priority": 1, "note": "..."}}], "questions": ["..."]}} Поле type — строго одно из: переговоры, оплата, креатив, контроль_выхода, замер_24ч, замер_7д, анализ_канала, отчёт, прочее. Правила: 1. Одна мысль — одна задача. Три поручения в сообщении — три объекта в tasks. 2. title — короткая формулировка в повелительном наклонении, без слов «надо», «нужно». 3. channel — ПОЛНОЕ название канала, все слова подряд, пока не начнётся следующее действие. «админу канала Нейросети и деньги, спросить цену» — название «Нейросети и деньги», а не «Нейросети». Слово «канал» отбрасывай. Если канал не назван — null. 4. due — только дата из календаря выше. Своей арифметики с датами не делай. Срок не назван — null. 5. priority: 1 — всё, что связано с деньгами (type = оплата), а также любой срок сегодня или завтра. 2 — задача со сроком дальше завтра. 3 — задача без срока. Оплата не может иметь приоритет ниже 1. 6. Нормализация форматов: «1.24», «1 24», «один двадцать четыре», «1/24» — записывай как 1/24. Аналогично 2/48. Формат указывай в note. 7. Распознавание речи искажает названия каналов и пишет их латиницей. Переноси как есть, не переводи и не исправляй. 8. Непонятно, что требуется — не выдумывай задачу, задай вопрос в questions.""" ``` Разбор каждого правила — зачем оно нужно: Файл `brain.py`: ```python import datetime import json import requests import config import prompts def ask(system: str, user: str, json_mode: bool = True) -> str: payload = { "model": config.MODEL_TEXT, "temperature": 0.2, "messages": [ {"role": "system", "content": system}, {"role": "user", "content": user}, ], } if json_mode: payload["response_format"] = {"type": "json_object"} response = requests.post( f"{config.MISTRAL_URL}/chat/completions", headers={"Authorization": f"Bearer {config.MISTRAL_KEY}", "Content-Type": "application/json"}, json=payload, timeout=120, ) response.raise_for_status() return response.json()["choices"][0]["message"]["content"] def parse_tasks(text: str) -> dict: system = prompts.TASKS_SYSTEM.format( calendar=prompts.date_table(datetime.date.today()) ) try: return json.loads(ask(system, text)) except json.JSONDecodeError: return {"tasks": [], "questions": ["Не удалось разобрать сообщение."]} ``` ![Что видит менеджер после того, как надиктовал поручения](https://blog.krasovskiy.team/wp-content/uploads/2026/07/ru-04-dialog.png) ## Шаг 10. Запись задач в таблицу Файл `sheets.py`: ```python import datetime import gspread from google.oauth2.service_account import Credentials import config SCOPES = ["https://www.googleapis.com/auth/spreadsheets", "https://www.googleapis.com/auth/drive"] _book = None def book(): global _book if _book is None: creds = Credentials.from_service_account_file( config.GOOGLE_CREDS, scopes=SCOPES) _book = gspread.authorize(creds).open(config.SHEET_NAME) return _book def add_tasks(tasks: list) -> int: sheet = book().worksheet("Задачи") today = datetime.date.today().isoformat() rows = [[today, task.get("title", ""), task.get("type", "прочее"), task.get("channel") or "", task.get("due") or "", task.get("priority", 3), "новая", task.get("note", "")] for task in tasks] if rows: sheet.append_rows(rows, value_input_option="USER_ENTERED") return len(rows) def tasks_for(date_iso: str) -> list: rows = book().worksheet("Задачи").get_all_records() return [row for row in rows if str(row.get("Срок", "")).startswith(date_iso) and row.get("Статус") != "готово"] def placements() -> list: return book().worksheet("Размещения").get_all_records() ``` Подключаем к боту. Заменяем обработчик голосовых на полный: ```python import brain import sheets PRIORITY_MARK = {1: "СРОЧНО", 2: "обычный", 3: "фон"} async def handle_text(message: Message, text: str): result = brain.parse_tasks(text) tasks = result.get("tasks", []) questions = result.get("questions", []) if tasks: sheets.add_tasks(tasks) lines = ["Записал в таблицу:", ""] for task in tasks: mark = PRIORITY_MARK.get(task.get("priority", 3), "фон") due = f", срок {task['due']}" if task.get("due") else "" channel = f" [{task['channel']}]" if task.get("channel") else "" lines.append(f"{mark}: {task['title']}{channel}{due}") await message.answer("n".join(lines)) if questions: await message.answer("Уточните:n" + "n".join(f"- {q}" for q in questions)) if not tasks and not questions: await message.answer("Задач не нашёл.") @dp.message(F.voice) async def on_voice(message: Message): if not allowed(message): return note = await message.answer("Слушаю...") file = await bot.get_file(message.voice.file_id) buffer = await bot.download_file(file.file_path) text = stt.transcribe(buffer.read()) await note.edit_text(f"Расшифровка:nn{text}") await handle_text(message, text) @dp.message(F.text & ~F.text.startswith("/")) async def on_text(message: Message): if not allowed(message): return await handle_text(message, message.text) @dp.message(Command("today")) async def cmd_today(message: Message): if not allowed(message): return today = datetime.date.today().isoformat() rows = sheets.tasks_for(today) if not rows: await message.answer("На сегодня задач нет.") return lines = ["Задачи на сегодня:", ""] for row in sorted(rows, key=lambda r: r.get("Приоритет", 3)): lines.append(f"- {row['Задача']} [{row.get('Канал', '')}]") await message.answer("n".join(lines)) ``` В начало `bot.py` добавляем `import datetime`. **Проверка шага:** отправляем голосовое с тремя поручениями сразу. Бот присылает расшифровку, затем список задач, а в таблице на листе «Задачи» появляются три новые строки. Так выглядит реальный ответ модели на голосовое из трёх поручений — это результат живой проверки, не пример из головы: ``` { "tasks": [ {"title": "Спросить цену на 1/24 и свободные слоты на следующую неделю", "type": "переговоры", "channel": "Нейросети и деньги", "due": null, "priority": 3, "note": "Формат: 1/24"}, {"title": "Снять охват по вчерашнему посеву", "type": "замер_24ч", "channel": "DigitalBourse", "due": "2026-07-24", "priority": 1, "note": ""}, {"title": "Оплатить размещение", "type": "оплата", "channel": "Telegram Pro IT", "due": null, "priority": 1, "note": "2000 рублей, кошелёк в закрепе"} ], "questions": [] } ``` ## Шаг 11. Напоминания Самая ценная функция. Замер, не снятый вовремя, восстановить нельзя. Файл `reminders.py`: ```python import datetime import sheets MANAGER_CHAT = None # подставляется при запуске def _date(value: str): try: return datetime.date.fromisoformat(str(value)[:10]) except ValueError: return None def build_digest() -> str: today = datetime.date.today() out_today, need_24h, need_7d = [], [], [] for row in sheets.placements(): published = _date(row.get("Дата выхода")) if not published: continue channel = row.get("Канал", "без названия") if published == today and row.get("Статус") != "вышло": out_today.append(f"- {channel}: сегодня выход, проверить") if published + datetime.timedelta(1) == today and not row.get("Охват 24ч"): need_24h.append(f"- {channel}: снять охват за сутки") if published + datetime.timedelta(7) == today and not row.get("Подписчиков 7д"): need_7d.append(f"- {channel}: снять подписчиков по инвайт-ссылке") unpaid = [f"- {row.get('Канал')}: хеш {row.get('Хеш')}" for row in sheets.book().worksheet("Платежи").get_all_records() if str(row.get("Хеш отправлен", "")).strip().lower() in ("", "нет")] blocks = [] if out_today: blocks.append("ВЫХОД СЕГОДНЯn" + "n".join(out_today)) if need_24h: blocks.append("ЗАМЕР ЗА СУТКИn" + "n".join(need_24h)) if need_7d: blocks.append("ЗАМЕР ЗА НЕДЕЛЮn" + "n".join(need_7d)) if unpaid: blocks.append("ХЕШИ НЕ ОТПРАВЛЕНЫ АДМИНАМn" + "n".join(unpaid)) return "nn".join(blocks) ``` Подключаем расписание в `bot.py`: ```python from apscheduler.schedulers.asyncio import AsyncIOScheduler import reminders async def send_digest(): text = reminders.build_digest() if not text: return for user_id in config.ALLOWED_IDS: try: await bot.send_message(user_id, "Сводка на сегодняnn" + text) except Exception as error: logging.warning("Не отправилось %s: %s", user_id, error) async def main(): scheduler = AsyncIOScheduler(timezone="Europe/Kyiv") scheduler.add_job(send_digest, "cron", hour=10, minute=0) scheduler.add_job(send_digest, "cron", hour=18, minute=0) scheduler.start() await dp.start_polling(bot) ``` Часовой пояс ставим свой. Сводка приходит дважды в день: утром — план, вечером — что не закрыто. **Проверка шага:** временно меняем время задачи на ближайшую минуту, перезапускаем бота, ждём. Сводка должна прийти. Затем возвращаем нормальное время. ## Шаг 12. Анализ канала перед закупкой Читать чужие каналы обычный бот не может — Telegram не даёт ботам доступ к каналам, где они не админ. Нужен доступ от имени пользователя. Получаем ключи: **Предупреждение, которое нельзя пропускать.** Такой доступ работает от вашего личного аккаунта. Читать публичные каналы и считать статистику — безопасно. **Рассылать сообщения незнакомым людям с него нельзя.** Массовая отправка незнакомым — прямой триггер антиспама: сначала запрет писать тем, кого нет в контактах, затем блокировка аккаунта вместе со всей перепиской по закупкам. Сообщения админам отправляет человек руками. Всегда. Файл `channels.py`: ```python import statistics from telethon.sync import TelegramClient import config def analyze(username: str, limit: int = 50) -> dict: """Считает статистику публичного канала по последним постам.""" with TelegramClient("assistant_session", int(config.TG_API_ID), config.TG_API_HASH) as client: entity = client.get_entity(username) views, reactions, ad_posts = [], [], 0 for post in client.iter_messages(entity, limit=limit): if post.views: views.append(post.views) count = 0 if getattr(post, "reactions", None): count = sum(r.count for r in post.reactions.results) reactions.append(count) text = (post.text or "").lower() if any(word in text for word in ("реклама", "партнёр", "erid", "промокод")): ad_posts += 1 subscribers = getattr(entity, "participants_count", 0) or 0 if not views: return {"error": "Нет постов с просмотрами. Канал закрыт или пуст."} average = statistics.mean(views) spread = (statistics.pstdev(views) / average * 100) if average else 0 engagement = (statistics.mean(reactions) / average * 100) if average else 0 return { "название": getattr(entity, "title", username), "подписчиков": subscribers, "средний охват": round(average), "охват к подписчикам, %": round(average / subscribers * 100, 1) if subscribers else 0, "ER, %": round(engagement, 2), "разброс просмотров, %": round(spread, 1), "рекламных постов из выборки": ad_posts, } def verdict(stats: dict) -> list: """Красные флаги. Считаются по правилам регламента.""" flags = [] if stats.get("ER, %", 0) < 1: flags.append("ER ниже 1 процента — аудитория мёртвая или накрученная") if stats.get("охват к подписчикам, %", 0) < 10: flags.append("Охват меньше 10 процентов от подписчиков — канал выгорел") if stats.get("разброс просмотров, %", 100) < 8: flags.append("Просмотры у всех постов почти одинаковые — признак накрутки") if stats.get("рекламных постов из выборки", 0) > 15: flags.append("Слишком много рекламы — наш пост утонет") return flags ``` Про разброс просмотров: у живого канала одни посты заходят, другие нет, и разброс большой. Если у всех постов просмотры почти совпадают — их накручивают ботами до одинакового значения. Это самый надёжный признак, его сложнее подделать, чем сам охват. Команда в `bot.py`: ```python import channels @dp.message(Command("channel")) async def cmd_channel(message: Message): if not allowed(message): return parts = message.text.split(maxsplit=1) if len(parts) < 2: await message.answer("Формат: /channel @username") return await message.answer("Считаю...") stats = channels.analyze(parts[1].strip()) if "error" in stats: await message.answer(stats["error"]) return lines = [f"{key}: {value}" for key, value in stats.items()] flags = channels.verdict(stats) if flags: lines += ["", "КРАСНЫЕ ФЛАГИ:"] + [f"- {flag}" for flag in flags] else: lines += ["", "Явных признаков накрутки нет."] await message.answer("n".join(lines)) ``` **Первый запуск потребует входа.** Telethon попросит номер телефона и код из Telegram — один раз, прямо в консоли сервера. После этого рядом появится файл `assistant_session.session`, и вход больше не потребуется. Этот файл равносилен доступу к аккаунту, его нельзя копировать и выкладывать. ## Шаг 13. Черновик рекламного поста Добавляем в `prompts.py`: ``` DRAFT_SYSTEM = """Ты пишешь рекламный пост для размещения в чужом Telegram-канале. Отвечаешь только текстом поста, без пояснений и без markdown. Требования: 1. Первая строка — крючок под аудиторию именно этого канала, а не общий. 2. Дальше два-четыре предложения по сути: что человек получит. 3. Одно конкретное доказательство: цифра, срок или результат. Не выдумывай цифры, бери только из брифа. Если цифр в брифе нет — обойдись без них. 4. Призыв к действию и ссылка. Ссылка ровно одна, вставь её как есть. 5. Объём — до 700 знаков. Пост должен читаться с телефона без раскрытия. 6. Подстройся под манеру канала из примеров ниже: длина фраз, обращение на ты или на вы, наличие заголовков. Не копируй их темы, копируй интонацию. 7. Без эмодзи. Без канцелярита. Без обещаний, которых нет в брифе.""" DRAFT_USER = """Бриф продукта: {brief} Ссылка: {link} Примеры постов канала, под который пишем: {samples}""" ``` В `brain.py`: ```python def draft_ad(brief: str, link: str, samples: list) -> str: joined = "nn---nn".join(samples[:5]) user = prompts.DRAFT_USER.format(brief=brief, link=link, samples=joined) return ask(prompts.DRAFT_SYSTEM, user, json_mode=False) ``` В `channels.py` — функция для примеров постов: ```python def samples(username: str, limit: int = 10) -> list: with TelegramClient("assistant_session", int(config.TG_API_ID), config.TG_API_HASH) as client: return [post.text for post in client.iter_messages(username, limit=limit) if post.text and len(post.text) > 120] ``` Команда работает в два шага. В `bot.py`: ```python from aiogram.fsm.context import FSMContext from aiogram.fsm.state import State, StatesGroup class Draft(StatesGroup): waiting = State() @dp.message(Command("draft")) async def cmd_draft(message: Message, state: FSMContext): if not allowed(message): return await state.set_state(Draft.waiting) await message.answer( "Пришлите одной строкой: @канал | бриф продукта | ссылкаnn" "Пример:n@ai_channel | сервис делает 3D-модели из фото | https://example.com" ) @dp.message(Draft.waiting) async def do_draft(message: Message, state: FSMContext): await state.clear() parts = [p.strip() for p in message.text.split("|")] if len(parts) < 3: await message.answer("Нужно три части через | ") return channel, brief, link = parts[0], parts[1], parts[2] await message.answer("Читаю канал и пишу черновик...") samples = channels.samples(channel) text = brain.draft_ad(brief, link, samples) await message.answer("Черновик. Проверьте факты перед отправкой админу:nn" + text) ``` Диспетчер с состояниями требует хранилища. Меняем создание диспетчера в начале файла: ```python from aiogram.fsm.storage.memory import MemoryStorage dp = Dispatcher(storage=MemoryStorage()) ``` **Черновик — это черновик.** Модель может написать красиво и при этом переврать факт. Перед отправкой админу текст читает человек. Это не перестраховка: цифра, которой нет в брифе, в рекламе означает претензию к вам, а не к боту. ![Боту достаточно одного права — публиковать](https://blog.krasovskiy.team/wp-content/uploads/2026/07/ru-06-rights.png) ## Шаг 14. Публикация в свой канал Здесь всё безопасно: бот публикует в канал, где он администратор. Личный аккаунт не задействован, риска блокировки нет. ``` @dp.message(Command("post")) async def cmd_post(message: Message): if not allowed(message): return text = message.text.split(maxsplit=1) if len(text) < 2: await message.answer("Формат: /post текст поста") return await bot.send_message(config.CHANNEL_ID, text[1]) await message.answer("Опубликовано.") ``` **Если приходит ошибка `chat not found`** — бот не добавлен в канал или username указан с ошибкой. **`not enough rights`** — не выдано право публикации. ## Шаг 15. Автозапуск и логи Пока бот запущен вручную, он умрёт вместе с закрытием SSH. Ставим его в автозапуск. Выходим в root (`exit` из-под пользователя `assistant`) и создаём файл службы: ``` nano /etc/systemd/system/tg-assistant.service ``` ``` [Unit] Description=Telegram assistant After=network-online.target [Service] Type=simple User=assistant WorkingDirectory=/home/assistant/tg-assistant ExecStart=/home/assistant/tg-assistant/venv/bin/python bot.py Restart=always RestartSec=10 [Install] WantedBy=multi-user.target ``` Включаем: ``` systemctl daemon-reload systemctl enable --now tg-assistant systemctl status tg-assistant ``` В статусе должно быть `active (running)`. Полезные команды на каждый день: ``` journalctl -u tg-assistant -f # смотреть логи вживую systemctl restart tg-assistant # перезапустить после правок journalctl -u tg-assistant --since today | grep -i error # ошибки за сегодня ``` `Restart=always` означает, что после сбоя или перезагрузки сервера бот поднимется сам. ![Семь этапов закупки. Розовым — то, что теряется без бота](https://blog.krasovskiy.team/wp-content/uploads/2026/07/ru-07-stages.png) ## Часть 4. Финальная проверка Проходим по списку. Каждый пункт должен отработать. Если все десять прошли — ассистент собран. ## Часть 5. Типичные ошибки СимптомПричинаЧто делатьБот молчит, в логах пустоХостинг блокирует api.telegram.orgПроверка из шага 3, менять сервер`Unauthorized` при стартеНеверный или отозванный токенВзять новый в BotFather через `/revoke``SpreadsheetNotFound`Сервисному аккаунту не выдан доступ к таблицеШаг 5, пункт 6`WorksheetNotFound`Название листа не совпадаетСверить буква в букву, включая регистрЗадачи с неверными датамиМодель считает даты самаПроверить, что календарь подставляется в промтНазвание канала обрезаноИз промта убран пример к правилу 3Вернуть примерБот придумывает задачиУбрано правило 8Вернуть правило`FloodWait` от TelethonСлишком частый анализ каналовЖдать указанное время, ставить паузы между запросамиБот отвечает «Нет доступа»ID не в `ALLOWED_IDS`Добавить ID и перезапустить службуБот пропал после перезагрузкиНе включён автозапуск`systemctl enable --now tg-assistant` ## Часть 6. Кейс: похожая система в работе Всё, что описано выше, — не теория. У нас работает система той же архитектуры, только вокруг контента, а не закупки. Полезна она здесь тем, что показывает, что ломается на дистанции. **Из чего состоит:** **Три вещи, которые пришлось чинить уже в работе. Заложите их сразу.** **Чего система принципиально не делает:** не пишет незнакомым людям от личного аккаунта. Это ограничение не техническое, а осознанное, по причинам из шага 12. ## Часть 7. Что дальше Ассистент собран, но он пока принимает задачи и считает. Дальше его наращивают под процесс: Порядок наращивания определяется тем, что чаще всего теряется в работе. Начинать стоит с автозаполнения охватов — это самая рутинная и самая забываемая операция. ## Приложение. Все промты одним списком Промты вынесены в `prompts.py`, чтобы их можно было править без правки кода. Менять их — нормально, но после каждой правки прогоняйте проверку из части 4, пункты 2-4: промт легко испортить одной формулировкой.