TG_autoposter/docs/API.md

API Документация - TG Autoposter

Обзор

Это документация по использованию репозиториев и основным компонентам бота для разработчиков.

Репозитории

GroupRepository

Работа с группами Telegram.

from app.database import AsyncSessionLocal
from app.database.repository import GroupRepository

async with AsyncSessionLocal() as session:
    repo = GroupRepository(session)
    
    # Добавить группу
    group = await repo.add_group(
        chat_id="-1001234567890",
        title="Название группы",
        slow_mode_delay=5  # в секундах
    )
    
    # Получить по chat_id
    group = await repo.get_group_by_chat_id("-1001234567890")
    
    # Получить все активные
    groups = await repo.get_all_active_groups()
    
    # Обновить slow mode
    await repo.update_group_slow_mode(group_id=1, delay=10)
    
    # Обновить время последнего сообщения
    await repo.update_last_message_time(group_id=1)
    
    # Деактивировать
    await repo.deactivate_group(group_id=1)
    
    # Активировать
    await repo.activate_group(group_id=1)

MessageRepository

Работа с сообщениями.

from app.database.repository import MessageRepository

async with AsyncSessionLocal() as session:
    repo = MessageRepository(session)
    
    # Создать сообщение
    msg = await repo.add_message(
        text="<b>Текст сообщения</b>",
        title="Название",
        parse_mode="HTML"  # HTML или Markdown
    )
    
    # Получить по ID
    msg = await repo.get_message(msg_id=1)
    
    # Получить все сообщения
    messages = await repo.get_all_messages(active_only=True)
    
    # Обновить
    await repo.update_message(
        message_id=1,
        text="Новый текст",
        title="Новое название"
    )
    
    # Деактивировать
    await repo.deactivate_message(message_id=1)
    
    # Удалить
    await repo.delete_message(message_id=1)

MessageGroupRepository

Связь между сообщениями и группами.

from app.database.repository import MessageGroupRepository

async with AsyncSessionLocal() as session:
    repo = MessageGroupRepository(session)
    
    # Добавить сообщение в группу
    link = await repo.add_message_to_group(
        message_id=1,
        group_id=1
    )
    
    # Получить неотправленные сообщения для отправки
    msg_groups = await repo.get_message_groups_to_send(message_id=1)
    
    # Получить неотправленные сообщения для группы
    msg_groups = await repo.get_unsent_messages_for_group(group_id=1)
    
    # Отметить как отправленное
    await repo.mark_as_sent(message_group_id=1)
    
    # Отметить как ошибка
    await repo.mark_as_sent(
        message_group_id=1,
        error="Бот не имеет прав в группе"
    )
    
    # Получить все сообщения для группы
    msg_groups = await repo.get_messages_for_group(group_id=1)
    
    # Удалить сообщение из группы
    await repo.remove_message_from_group(
        message_id=1,
        group_id=1
    )

Модели

Group

from app.models import Group

# Поля
group.id              # int (первичный ключ)
group.chat_id         # str (уникальный ID группы в Telegram)
group.title           # str (название группы)
group.slow_mode_delay # int (задержка между сообщениями в сек)
group.last_message_time # datetime (время последнего отправленного)
group.is_active       # bool (активна ли группа)
group.created_at      # datetime
group.updated_at      # datetime

# Связи
group.messages        # List[MessageGroup] (сообщения в этой группе)

Message

from app.models import Message

# Поля
msg.id           # int (первичный ключ)
msg.text         # str (текст сообщения)
msg.title        # str (название)
msg.is_active    # bool (активно ли)
msg.parse_mode   # str (HTML, Markdown, None)
msg.created_at   # datetime
msg.updated_at   # datetime

# Связи
msg.groups       # List[MessageGroup] (группы для отправки)

MessageGroup

from app.models import MessageGroup

# Поля
mg.id           # int (первичный ключ)
mg.message_id   # int (FK на Message)
mg.group_id     # int (FK на Group)
mg.is_sent      # bool (отправлено ли)
mg.sent_at      # datetime (когда отправлено)
mg.error        # str (описание ошибки если была)
mg.created_at   # datetime

# Связи
mg.message      # Message (само сообщение)
mg.group        # Group (сама группа)

Обработчики (Handlers)

Команды

• start - Главное меню
• help_command - Справка

Callback обработчики

Главное меню

• main_menu - Вернуться в главное меню

Управление сообщениями

• manage_messages - Меню управления сообщениями
• create_message - Начало создания сообщения
• list_messages - Список всех сообщений
• send_msg_<id> - Отправить сообщение в группы
• delete_msg_<id> - Удалить сообщение

Управление группами

• manage_groups - Меню управления группами
• list_groups - Список всех групп
• group_messages_<id> - Сообщения для группы
• delete_group_<id> - Удалить группу

Выбор групп при создании сообщения

• select_group_<id> - Выбрать/отменить выбор группы
• done_groups - Завершить выбор групп

Утилиты

Проверка slow mode

from app.utils import can_send_message

# Проверить можно ли отправлять
can_send, wait_time = await can_send_message(group)

if can_send:
    # Отправляем сейчас
    pass
else:
    # Ждем wait_time секунд
    pass

Клавиатуры

from app.utils.keyboards import (
    get_main_keyboard,
    get_messages_keyboard,
    get_groups_keyboard,
    get_back_keyboard,
    get_message_actions_keyboard,
    get_group_actions_keyboard,
    get_yes_no_keyboard,
)

# Главное меню
keyboard = get_main_keyboard()

# Меню сообщений
keyboard = get_messages_keyboard()

# Меню групп
keyboard = get_groups_keyboard()

# Кнопка назад
keyboard = get_back_keyboard()

# Действия с сообщением
keyboard = get_message_actions_keyboard(message_id=1)

# Действия с группой
keyboard = get_group_actions_keyboard(group_id=1)

# Подтверждение
keyboard = get_yes_no_keyboard(action="delete_message_1")

Примеры использования

Пример 1: Создание сообщения и отправка в группу

import asyncio
from app.database import AsyncSessionLocal, init_db
from app.database.repository import (
    GroupRepository, MessageRepository, MessageGroupRepository
)

async def main():
    await init_db()
    
    async with AsyncSessionLocal() as session:
        # Создаем сообщение
        msg_repo = MessageRepository(session)
        msg = await msg_repo.add_message(
            text="Привет, это тестовое сообщение!",
            title="Тест"
        )
        
        # Получаем группу
        group_repo = GroupRepository(session)
        groups = await group_repo.get_all_active_groups()
        
        if groups:
            # Добавляем сообщение в группу
            mg_repo = MessageGroupRepository(session)
            await mg_repo.add_message_to_group(msg.id, groups[0].id)
            
            print(f"✅ Сообщение готово к отправке в {groups[0].title}")

asyncio.run(main())

Пример 2: Отправка сообщения с учетом slow mode

from app.utils import can_send_message
from telegram import Bot

async def send_to_group(bot: Bot, message, group):
    # Проверяем slow mode
    can_send, wait_time = await can_send_message(group)
    
    if not can_send:
        print(f"⏳ Ожидаем {wait_time} секунд...")
        await asyncio.sleep(wait_time)
    
    # Отправляем
    await bot.send_message(
        chat_id=group.chat_id,
        text=message.text,
        parse_mode=message.parse_mode
    )
    
    print(f"✅ Отправлено в {group.title}")

Пример 3: Получение статистики

async def get_statistics():
    async with AsyncSessionLocal() as session:
        msg_repo = MessageRepository(session)
        group_repo = GroupRepository(session)
        mg_repo = MessageGroupRepository(session)
        
        messages = await msg_repo.get_all_messages()
        groups = await group_repo.get_all_active_groups()
        
        print(f"📊 Статистика:")
        print(f"   Сообщений: {len(messages)}")
        print(f"   Групп: {len(groups)}")
        
        # Сообщения по отправкам
        for msg in messages:
            msg_groups = await mg_repo.get_messages_for_group(msg.id)
            sent = sum(1 for mg in msg_groups if mg.is_sent)
            print(f"   {msg.title}: {sent}/{len(msg_groups)} групп")

Логирование

import logging

logger = logging.getLogger(__name__)

logger.debug("Отладочное сообщение")
logger.info("Информационное сообщение")
logger.warning("Предупреждение")
logger.error("Ошибка")
logger.critical("Критическая ошибка")

Логи сохраняются в папку logs/ с ротацией по дням.

Обработка ошибок

try:
    await bot.send_message(
        chat_id=group.chat_id,
        text=message.text,
        parse_mode=message.parse_mode
    )
except TelegramError as e:
    logger.error(f"Ошибка Telegram: {e}")
    # Сохраняем ошибку в БД
    await mg_repo.mark_as_sent(mg.id, error=str(e))
except Exception as e:
    logger.error(f"Неожиданная ошибка: {e}")

Асинхронность

Весь код использует async/await. При работе с БД и ботом всегда используйте:

async def my_function():
    async with AsyncSessionLocal() as session:
        # Работа с БД
        pass

Типизация

Проект использует type hints для улучшения качества кода:

from typing import List, Optional
from app.models import Group, Message

async def get_active_groups() -> List[Group]:
    """Получить все активные группы"""
    pass

async def find_group(chat_id: str) -> Optional[Group]:
    """Найти группу по chat_id"""
    pass