Telegram-ассистент для закупки рекламы: пошаговая сборка с кодом и промтами

</> Версия для ИИ

Пошаговое руководство: как собрать бота, который принимает голосовые сообщения, превращает их в задачи, ведёт таблицу закупок, напоминает о замерах, анализирует каналы перед покупкой и пишет черновики рекламных постов.

Инструкция рассчитана на человека, который раньше не поднимал ботов. Все команды копируются целиком. Если шаг не получился — в конце каждого раздела есть проверка и разбор типичных ошибок.

Что нужно иметь до начала:

аккаунт Telegram
любой VPS на Ubuntu 22.04 или новее (хватит самого дешёвого тарифа, 1 ядро и 1 ГБ памяти)
аккаунт Google (для таблицы)
около 40 минут времени

Во что обойдётся в месяц: сервер — от 4 долларов; Telegram — бесплатно; Google Sheets — бесплатно; распознавание голоса и разбор задач — меньше доллара при обычной нагрузке.

Важно про выбор сервера. Часть хостингов (особенно шаред-хостинги для сайтов) блокируют исходящие соединения к api.telegram.org. На таком сервере бот не заработает, и понять это сложно — он просто молчит. Проверка одной командой есть в шаге 3.

Путь голосового сообщения: от диктовки до строки в таблице

Часть 1. Процесс, который мы автоматизируем

Бот бесполезен, если непонятно, какую работу он поддерживает. Ниже — короткая версия процесса закупки. Полная версия со всеми шаблонами сообщений, формулами и правилами безопасности лежит в отдельном документе «Регламент: закупка рекламы в Telegram-каналах», его нужно прочитать до начала работы.

Закупка состоит из семи этапов:

Отбор канала. Смотрим не количество подписчиков, а средний охват поста и вовлечённость. Отсеиваем накрутку.
Переговоры. Пишем админу: продукт, формат, просьба назвать цену, свободные даты и кошелёк.
Оплата. Платим, записываем хеш транзакции, отправляем хеш админу в Telegram. Только новые хеши, повторно — никогда.
Креатив и ссылка. Создаём уникальную пригласительную ссылку до оплаты, готовим пост, отправляем админу.
Контроль выхода. Проверяем в оговорённое время, забираем ссылку на пост, следим, чтобы ссылку в тексте не вырезали.
Замеры. Через 24 часа — охват. Через 7 дней — сколько людей вступило по уникальной ссылке.
Отчёт. Считаем цену подписчика, выносим вердикт: повторять, торговаться или в чёрный список.

Три места здесь ломаются чаще всего, и именно их закрывает бот:

забыли создать уникальную ссылку до размещения — приход не с чем связать;
не сняли замер вовремя — цифры смешались с другими источниками и восстановить их нельзя;
потеряли, какие хеши уже отправлены админам.

Бот берёт на себя память и счёт. Решения и переписку с людьми оставляем человеку.

Часть 2. Подготовка

Шаг 1. Создаём бота

Открываем Telegram, в поиске находим @BotFather — официальный бот Telegram для создания ботов. У настоящего BotFather есть галочка верификации, у подделок её нет.

Нажимаем «Запустить».
Отправляем команду /newbot.
BotFather спросит имя бота — это то, что видно в заголовке чата. Например: Ассистент закупки.
Затем спросит username — он должен заканчиваться на bot и быть свободным. Например: zakupka_assist_bot.
В ответ приходит токен вида 8123456789:AAF... — длинная строка.

Токен — это пароль от бота. Кто им владеет, тот управляет ботом. Не отправляем его в общие чаты, не коммитим в репозиторий. Если случайно засветили — в BotFather команда /revoke выпускает новый.

Сразу настраиваем приватность, чтобы бот не собирал лишнее:

/setprivacy → выбрать бота → Enable
/setdescription → короткое описание
/setcommands → отправить блоком:

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 (нужен всегда):

Открываем console.mistral.ai, регистрируемся.
Раздел API Keys → Create new key.
Копируем ключ. Показывается он один раз.

Ключ OpenAI (рекомендуется, для распознавания):

Открываем platform.openai.com, раздел API keys → Create new secret key.
Копируем ключ, пополняем баланс на минимальную сумму.

Распознавание стоит около половины цента за минуту речи. При двадцати голосовых в день это меньше доллара в месяц. Почему рекомендуется именно он — в шаге 8, там замеры на одном и том же файле.

Если платить за второй сервис не хочется — работает и на одном ключе Mistral, и есть полностью бесплатный вариант без внешних сервисов вообще. Оба описаны там же.

Пять листов таблицы — вся память ассистента

Шаг 5. Готовим таблицу

Таблица — это база данных ассистента. Структура повторяет регламент.

Создаём Google-таблицу с названием Закупка Telegram и пятью листами. Названия листов должны совпадать буква в букву:

Лист Задачи — первая строка:

Дата создания | Задача | Тип | Канал | Срок | Приоритет | Статус | Заметка

Лист Каналы:

Название | Ссылка | Тематика | Подписчиков | Средний охват | ER % | Админ | Кошелёк | Прайс | Статус | Последний контакт | Заметки

Лист Размещения:

ID | Канал | Дата брони | Дата выхода | Формат | Цена | Валюта | Кошелёк | Хеш | Хеш отправлен | Креатив | Инвайт-ссылка | Ссылка на пост | Статус | Охват 24ч | Подписчиков 7д | CPM | CPS | Вердикт

Лист Платежи:

Дата | Сумма | Валюта | Кошелёк | Хеш | Канал | Хеш отправлен

Лист Креативы:

ID | Текст | Медиа | Где использовался | Лучший CPS | Заметки

Теперь выдаём боту доступ к этой таблице.

Открываем console.cloud.google.com, создаём проект (название любое).
В поиске сверху вводим Google Sheets API → Enable. То же самое для Google Drive API.
Слева IAM и администрирование → Сервисные аккаунты → Создать сервисный аккаунт. Имя любое, роли не нужны, жмём «Готово».
Открываем созданный аккаунт → вкладка Ключи → Добавить ключ → Создать ключ → тип JSON. Файл скачается сам.
Открываем скачанный файл, находим строку "client_email" — там адрес вида имя@проект.iam.gserviceaccount.com.
Возвращаемся в таблицу, жмём «Настройки доступа» и добавляем этот адрес как Редактора.

Шаг, который забывают все. Если не дать сервисному аккаунту доступ к таблице (пункт 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:

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:

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.

Если ответа нет:

в консоли Unauthorized — неверный токен, перепроверяем .env;
бот молчит и в логах пусто — сервер не видит Telegram, возвращаемся к проверке из шага 3;
бот отвечает «Нет доступа» — свой ID не попал в ALLOWED_IDS.

Замер шести вариантов распознавания на одном голосовом

Шаг 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:

openai — рекомендуемый. Лучшее качество на терминах, около половины цента за минуту речи.
mistral — один ключ на весь проект, дешевле. Термины чинятся позже, правилом 6 в промте разбора, так что вариант полностью рабочий.
local — бесплатно, без внешних сервисов, речь не уходит с вашего сервера. Берите его, если в голосовых звучат суммы, кошельки и договорённости, которые нельзя отдавать наружу.

Честно про бесплатный вариант. В замерах выше он прогнан на модели 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:

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:

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:

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."""

Разбор каждого правила — зачем оно нужно:

Правило 4 и календарь. Языковые модели плохо считают даты. В проверке без готового календаря «в пятницу» превратилось в 25 июля вместо 24-го — ошибка на сутки, из-за которой замер был бы сорван. С подставленным календарём дата верная. Никогда не позволяйте модели считать даты самой.
Правило 3 с примером. Без примера название «Нейросети и деньги» обрезалось до «Нейросети», и задача уходила не на тот канал. Пример в промте важнее формулировки правила.
Правило 5 с последним предложением. Без явного запрета оплата иногда получала третий приоритет и тонула в списке.
Правило 6. Лечит то самое «1.24» из распознавания.
Правило 8. Без него модель на любую мутную фразу придумывает задачу. Проверено: с этим правилом на фразу «там по этому каналу надо бы что-то решить» бот не создаёт ничего, а переспрашивает.

Файл brain.py:

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": ["Не удалось разобрать сообщение."]}

Что видит менеджер после того, как надиктовал поручения

Шаг 10. Запись задач в таблицу

Файл sheets.py:

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()

Подключаем к боту. Заменяем обработчик голосовых на полный:

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:

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:

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 не даёт ботам доступ к каналам, где они не админ. Нужен доступ от имени пользователя.

Получаем ключи:

Открываем my.telegram.org, входим по номеру телефона.
API development tools → заполняем название приложения → Create application.
Копируем api_id и api_hash в .env.

Предупреждение, которое нельзя пропускать. Такой доступ работает от вашего личного аккаунта. Читать публичные каналы и считать статистику — безопасно. Рассылать сообщения незнакомым людям с него нельзя. Массовая отправка незнакомым — прямой триггер антиспама: сначала запрет писать тем, кого нет в контактах, затем блокировка аккаунта вместе со всей перепиской по закупкам. Сообщения админам отправляет человек руками. Всегда.

Файл channels.py:

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:

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:

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 — функция для примеров постов:

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:

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)

Диспетчер с состояниями требует хранилища. Меняем создание диспетчера в начале файла:

from aiogram.fsm.storage.memory import MemoryStorage
dp = Dispatcher(storage=MemoryStorage())

Черновик — это черновик. Модель может написать красиво и при этом переврать факт. Перед отправкой админу текст читает человек. Это не перестраховка: цифра, которой нет в брифе, в рекламе означает претензию к вам, а не к боту.

Боту достаточно одного права — публиковать

Шаг 14. Публикация в свой канал

Здесь всё безопасно: бот публикует в канал, где он администратор. Личный аккаунт не задействован, риска блокировки нет.

Открываем свой канал → Управление → Администраторы → Добавить администратора.
Находим бота по username, добавляем.
Оставляем только право Публикация сообщений, остальное выключаем.
В .env в CHANNEL_ID пишем @имя_канала (для публичного) или числовой ID вида -100... (для приватного).

@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 означает, что после сбоя или перезагрузки сервера бот поднимется сам.

Семь этапов закупки. Розовым — то, что теряется без бота

Часть 4. Финальная проверка

Проходим по списку. Каждый пункт должен отработать.

/start — приходит приветствие.
Голосовое из трёх поручений — приходит расшифровка, затем три задачи, в таблице три новые строки.
Текстом «оплатить размещение в канале Х, 2000» — задача с типом «оплата» и первым приоритетом.
Мутная фраза без конкретики — бот переспрашивает, задач не создаёт.
/today — показывает задачи на сегодня.
/channel @любой_публичный_канал — приходит статистика и красные флаги.
/draft @канал | бриф | ссылка — приходит черновик в манере канала.
/post проверка — сообщение появляется в вашем канале.
systemctl restart tg-assistant — бот отвечает и после перезапуска.
Заполняем в листе «Размещения» строку с датой выхода вчерашним днём, вызываем сводку — приходит напоминание снять охват.

Если все десять прошли — ассистент собран.

Часть 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. Кейс: похожая система в работе

Всё, что описано выше, — не теория. У нас работает система той же архитектуры, только вокруг контента, а не закупки. Полезна она здесь тем, что показывает, что ломается на дистанции.

Из чего состоит:

Читалка каналов. Тот же Telethon из шага 12. Забирает посты нужных публичных каналов, по ним видно, какие темы сейчас обсуждают и что уже разобрано конкурентами.
Очередь тем. Хранится в базе, а не в голове. Каждая тема имеет статус: ждёт, в работе, опубликована.
Генератор текстов. Тот же принцип, что в шаге 13: промт с правилами тона и запретами, на выходе черновик.
Проверка на дубли перед генерацией. Каждый готовый материал переводится в числовой отпечаток и складывается в отдельную базу. Прежде чем писать новый текст, система сверяет тему с уже написанными. Похожая есть — тема не берётся в работу.
Публикация ботом по расписанию — ровно как в шаге 14.

Три вещи, которые пришлось чинить уже в работе. Заложите их сразу.

Ограничение по времени публикации оказалось важнее качества текстов. Сначала бот публиковал, когда материал был готов, — то есть иногда ночью и по два поста подряд. Со стороны это выглядит как спам-лента, независимо от того, что написано. Сейчас жёсткое окно: не раньше девяти утра, не позже десяти вечера, минимум четыре с половиной часа между постами. Если слот не подходит — публикация переносится, а не отменяется.

Без проверки на дубли генератор ходит по кругу. Через несколько десятков материалов он начинает писать про одно и то же разными словами. Обнаруживается это поздно, а разгребается долго. Проверка на схожесть должна появиться раньше, чем накопится объём, — потом её внедрять больнее.

Человек на выходе обязателен. Всё, что уходит наружу от нашего имени, читает человек. Отдельно это касается всего, что нельзя проверить глазами: например, озвучку роликов человек прослушивает до публикации, потому что синтез речи регулярно ломает произношение терминов и цифр — ровно та же болезнь, что 1.24 в шаге 8, только в обратную сторону.

Чего система принципиально не делает: не пишет незнакомым людям от личного аккаунта. Это ограничение не техническое, а осознанное, по причинам из шага 12.

Часть 7. Что дальше

Ассистент собран, но он пока принимает задачи и считает. Дальше его наращивают под процесс:

Автозаполнение охватов. Бот сам читает вышедший пост по ссылке из таблицы и проставляет просмотры через сутки — без ручного ввода.
Расчёт CPM и CPS. Формулы в таблице, бот присылает недельную сводку с вердиктами по каждому каналу.
Черновики сообщений админам. Бот подставляет название канала, цену и нужный шаблон из регламента, кладёт готовый текст. Отправляет человек.
Контроль хешей. Отдельная команда показывает список неотправленных хешей одним сообщением.

Порядок наращивания определяется тем, что чаще всего теряется в работе. Начинать стоит с автозаполнения охватов — это самая рутинная и самая забываемая операция.

Приложение. Все промты одним списком

Промты вынесены в prompts.py, чтобы их можно было править без правки кода. Менять их — нормально, но после каждой правки прогоняйте проверку из части 4, пункты 2-4: промт легко испортить одной формулировкой.

TASKS_SYSTEM — разбор надиктованного в задачи. Шаг 9.
date_table — не промт, а подставляемый календарь. Без него даты будут неверными.
DRAFT_SYSTEM и DRAFT_USER — черновик рекламного поста. Шаг 13.

Andrey Krasovskiy

Андрей Красовский — программист и дата-сайентист с опытом создания сложных автоматизированных систем на Python, Google Colab и n8n. Его экспертиза охватывает построение SEO-экосистем, интеграцию API (Ahrefs, Google Ads, Search Console) и построение контент-пайплайнов. Андрей сочетает техническую точность с предпринимательским видением, создавая решения, работающие на результат.