Повна система сповіщень через Telegram з використанням FastAPI і Telethon

Наша мета – створити систему, яка:

1. Підключається до Telegram через Telethon для моніторингу чатів.
2. Дозволяє керувати сповіщеннями через API, використовуючи FastAPI.
3. Повідомляє про події (наприклад, нові повідомлення) через Telegram або інші канали (наприклад, email).

Цю систему можна використовувати для моніторингу чатів співробітників, отримання сповіщень про важливі події або навіть для аналітики активності користувачів.

---

Розробка системи сповіщень

Архітектура системи

Для початку розберемо загальний план:

1. Telethon: Підключається до Telegram і відстежує події, такі як нові повідомлення або згадки.
2. FastAPI: Відповідає за взаємодію з користувачем через HTTP API. Через API ми можемо:
   • Налаштовувати, які події відстежувати.
   • Підключати користувачів для отримання сповіщень.
3. SQLAlchemy (або будь‑яка інша база даних): Зберігання даних і налаштувань.
4. Асинхронні задачі: Для обробки подій в реальному часі.

Архітектурна схема:

+-------------------------------+
|   Користувач (HTTP-клієнт)   |
+-------------------------------+
                |
        (API-запити)
                |
                v
+-------------------------------+
|            FastAPI            |
+-------------------------------+
                |
                v
+-------------------------------+
|         Telethon Client       |
+-------------------------------+
                |
         (Telegram-події)
                |
                v
+-------------------------------+
|           База даних          |
+-------------------------------+
                |
      (збереження налаштувань)
                |
                v
+-------------------------------+
|         Сповіщення            |
+-------------------------------+

Налаштування проекту

Створимо структуру проекту для легкої організації коду:

project/
├── main.py         # Точка входу в додаток FastAPI
├── bot/
│   ├── client.py   # Конфігурація Telethon клієнта
│   ├── handlers.py # Обробники подій Telethon
│   ├── tasks.py    # Асинхронні задачі
├── api/
│   ├── routes.py   # Роути для API FastAPI
│   ├── models.py   # Pydantic-моделі для API
│   ├── db.py       # Підключення до бази даних
├── requirements.txt

Почнемо з встановлення необхідних бібліотек:

pip install fastapi uvicorn telethon sqlalchemy aiosqlite

Налаштування Telethon клієнта

Створимо файл bot/client.py, де налаштуємо Telethon-клієнта:

from telethon import TelegramClient

# Ваші дані для підключення до Telegram API
API_ID = "your_api_id"
API_HASH = "your_api_hash"
SESSION_NAME = "session_name"

# Створюємо Telethon-клієнта
client = TelegramClient(SESSION_NAME, API_ID, API_HASH)

# Запуск клієнта
async def start_client():
    await client.start()
    print("Telethon клієнт запущено!")

Код вище запускає Telegram-клієнта, використовуючи ваші API ID і Hash, які можна отримати тут.

Обробка подій Telegram

Тепер додамо обробники подій у bot/handlers.py. Наприклад, обробка нових повідомлень:

from telethon import events
from .client import client

@client.on(events.NewMessage)  # Відстежуємо нові повідомлення
async def new_message_handler(event):
    print(f"Нове повідомлення в чаті {event.chat_id}: {event.text}")
    # Тут можна додати сповіщення або інші дії

Налаштування FastAPI і маршрутів

Перейдемо до створення API у файлі api/routes.py:

from fastapi import APIRouter

router = APIRouter()

@router.get("/")
async def read_root():
    return {"message": "Система сповіщень працює!"}

@router.post("/add_chat/")
async def add_chat(chat_id: int):
    # Додаємо чат до списку моніторингу
    # Тут можна зберегти chat_id у базі даних
    return {"message": f"Чат {chat_id} додано до моніторингу."}

Підключимо ці маршрути в кореневому файлі main.py:

from fastapi import FastAPI
from bot.client import start_client
from api.routes import router

app = FastAPI()
app.include_router(router)

# Запуск Telethon клієнта при старті FastAPI
@app.on_event("startup")
async def startup_event():
    await start_client()

Запускаємо FastAPI:

uvicorn main:app --reload

Тепер ваш API доступний, і всі події Telegram обробляються.

Збереження даних у базі

Використовуємо SQLAlchemy для збереження даних (наприклад, сповіщення про нові повідомлення). У файлі api/db.py:

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker, declarative_base

DATABASE_URL = "sqlite+aiosqlite:///./test.db"

Base = declarative_base()
engine = create_async_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine, class_=AsyncSession)

async def init_db():
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

Створимо таблицю для збереження налаштувань чатів:

from sqlalchemy import Column, Integer, String
from .db import Base

class Chat(Base):
    __tablename__ = "chats"

    id = Column(Integer, primary_key=True, index=True)
    chat_id = Column(String, unique=True, index=True)
    notification_type = Column(String)  # Наприклад, "email" або "telegram"

Сповіщення

Тепер реалізуємо відправку сповіщень. Наприклад, сповіщення в Telegram:

from bot.client import client

async def send_notification(chat_id, message):
    await client.send_message(chat_id, message)

Логіка взаємодії

Об'єднуємо все: при отриманні нового повідомлення в Telethon викликаємо функцію запису в базу або відправки сповіщень. У bot/handlers.py:

from .notifications import send_notification

@client.on(events.NewMessage)
async def new_message_handler(event):
    chat_id = event.chat_id
    text = event.message.message
    print(f"Нове повідомлення в чаті {chat_id}: {text}")

    # Відправляємо сповіщення (наприклад, через Telegram)
    await send_notification(chat_id, f"Нове повідомлення: {text}")

Тестування системи

Тестування:

1. Запускаємо FastAPI (uvicorn main:app --reload).
2. Моніторимо повідомлення в чаті.
3. Перевіряємо відправку сповіщень.

Життя без багів — це утопія, але тепер ваші сповіщення завжди будуть працювати!