22.13M
Category: databasedatabase

pr_inj

1.

Проект по теме «Приложение для
мониторинга
экологической
обстановки»
Telegram-бот ‘Ecomonitor bot’
Крюков С.С ИВТ-262

2.

Отсканируйте QR чтобы протестировать бота:

3.

Актуальность и применение проекта:
В условиях роста урбанизации и загрязнения атмосферы контроль качества воздуха
становится важной задачей. Особенно это актуально для крупных городов, где
высокий уровень выбросов негативно влияет на здоровье населения.
Разрабатываемое приложение в виде Telegram-бота позволяет пользователям
быстро получать актуальную информацию о качестве воздуха в выбранном городе
без установки отдельного программного обеспечения. Благодаря использованию
популярного мессенджера Telegram доступ к данным возможен с любого устройства,
обеспечивая кроссплатформенность.
Назначение и функциональные возможности:
Основной задачей проекта является предоставление пользователю информации о качестве
атмосферного воздуха в выбранном городе.
Бот реализует удобный пошаговый интерфейс: сначала пользователь выбирает континент,
затем конкретный город. После этого он может получить текущие экологические показатели,
такие как индекс качества воздуха (AQI), температура, влажность, а также концентрация
мелкодисперсных частиц PM2.5 и PM10.
Данные могут быть обновлены по запросу пользователя, что позволяет отслеживать
изменения экологической обстановки в режиме реального времени.

4.

Используемые технологии
Проект реализован на языке программирования Python, который широко
используется для разработки сетевых и серверных приложений.
Для создания Telegram-бота применяется фреймворк aiogram, обеспечивающий
асинхронную обработку сообщений и высокую производительность.
Получение данных о качестве воздуха осуществляется через внешний API сервиса
World Air Quality Index (WAQI).
Для хранения информации о городах используется легковесная реляционная база
данных SQLite, что упрощает развертывание и не требует отдельного сервера баз
данных.
World's Air Pollution:
Real-time Air Quality Index

5.

Архитектура и структура проекта
import asyncio
import requests
from aiogram import Bot, Dispatcher
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton, CallbackQuery, Message
from aiogram.filters import CommandStart
from aiogram.client.default import DefaultBotProperties
from city_database import CityDatabase
class AirQualityService:
def __init__(self, api_key: str):
self.api_key = api_key
def fetch_data(self, city_api_name: str):
url = f"https://api.waqi.info/feed/{city_api_name}/?token={self.api_key}"
response = requests.get(url)
data = response.json()
if data.get("status") != "ok":
raise ValueError("Не удалось получить данные о воздухе.")
iaqi = data["data"].get("iaqi", {})
return {
"aqi": data["data"].get("aqi"),
"temp": iaqi.get("t", {}).get("v"),
"humidity": iaqi.get("h", {}).get("v"),
"pm25": iaqi.get("pm25", {}).get("v"),
"pm10": iaqi.get("pm10", {}).get("v")
}
def evaluate(self, aqi: int):
if aqi is None:
return "нет данных"
if aqi <= 50:
return "Хорошо ✅"
if aqi <= 100:
return "Средне ⚠ "
if aqi <= 150:
return "Вредно для чувствительных ❗"
if aqi <= 200:
return "Вредно ❗❗"
return "Опасно ❌"
Разделение логики на модули
Отдельный класс для работы с API
Отдельный класс для работы с БД
Повышение читаемости и
масштабируемости

6.

Код БД
import sqlite3
class CityDatabase:
_instance = None
def __new__(cls, db_path="cities.db"):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, db_path="cities.db"):
self.db_path = db_path
# Для in-memory используем одно соединение
self.conn = sqlite3.connect(db_path) if db_path == ":memory:" else None
self._create_table()
self._seed()
def _connect(self):
if self.conn:
return self.conn
return sqlite3.connect(self.db_path)
def _create_table(self):
conn = self._connect()
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS cities (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
waqi_name TEXT NOT NULL,
region TEXT NOT NULL,
continent TEXT NOT NULL
)
""")
conn.commit()
if self.conn is None:
conn.close()
def _seed(self):
conn = self._connect()
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM cities")
count = cursor.fetchone()[0]
if count == 0:
cities = [
("Москва", "Moscow", "Россия", "Европа"),
("Санкт-Петербург", "Saint Petersburg", "Россия", "Европа"),
("Берлин", "Berlin", "Германия", "Европа"),
("Париж", "Paris", "Франция", "Европа"),
("Рим", "Rome", "Италия", "Европа"),
("Мадрид", "Madrid", "Испания", "Европа"),
("Варшава", "Warsaw", "Польша", "Европа"),
("Пекин", "Beijing", "Китай", "Азия"),
("Шанхай", "Shanghai", "Китай", "Азия"),
("Токио", "Tokyo", "Япония", "Азия"),
("Сеул", "Seoul", "Южная Корея", "Азия"),
("Бангкок", "Bangkok", "Таиланд", "Азия"),
("Дели", "Delhi", "Индия", "Азия"),
("Нью-Йорк", "New York", "США", "Северная Америка"),
("Лос-Анджелес", "Los Angeles", "США", "Северная Америка"),
("Чикаго", "Chicago", "США", "Северная Америка"),
]
cursor.executemany(
"INSERT INTO cities (name, waqi_name, region, continent) VALUES (?, ?, ?, ?)",
cities
)
conn.commit()
if self.conn is None:
conn.close()
def get_continents(self):
conn = self._connect()
cursor = conn.cursor()
cursor.execute("SELECT DISTINCT continent FROM cities ORDER BY continent")
rows = cursor.fetchall()
if self.conn is None:
conn.close()
return [r[0] for r in rows]
def get_cities_by_continent(self, continent):
conn = self._connect()
cursor = conn.cursor()
cursor.execute("SELECT name FROM cities WHERE continent = ? ORDER BY name", (continent,))
rows = cursor.fetchall()
if self.conn is None:
conn.close()
return [r[0] for r in rows]
def get_waqi_name(self, name):
conn = self._connect()
cursor = conn.cursor()
cursor.execute("SELECT waqi_name FROM cities WHERE name = ?", (name,))
row = cursor.fetchone()
if self.conn is None:
conn.close()
return row[0] if row else None

7.

UNIT тестирование
Юнит-тесты проверяют ключевые
компоненты системы мониторинга
качества воздуха:
1.
Оценка AQI – преобразует числовой
индекс в текстовые категории (от
"Хорошо" до "Опасно")
2.
Получение данных – тестирует
успешные и ошибочные сценарии
работы с внешним API
3.
База данных – проверяет работу с
городами и континентами (inmemory SQLite)
4.
Telegram-бот – валидирует
генерацию интерфейса и кнопок
меню.

8.

Паттерны
Фабричный метод (factory) - применяется для
создания различных сервисов мониторинга
качества воздуха через единый интерфейс. В
проекте этот паттерн позволяет гибко
подключать дополнительные источники данных
— например, WAQI API, OpenWeatherMap или
тестовые данные — без изменения основной
логики приложения. Это обеспечивает легкое
расширение системы новыми провайдерами и
упрощает модульное тестирование.
Паттерн Singleton гарантирует существование только
одного экземпляра класса city_database
на всё время работы приложения. Это предотвращает
создание множественных подключений к базе данных
SQLite, экономит ресурсы и обеспечивает
согласованность операций с данными. Все
компоненты системы работают с единой, глобально
доступной точкой доступа к хранилищу городов и
настроек.

9.

Обзор приложения

10.

11.

Выводы:
В рамках данного проекта был разработан Telegram-бот
для мониторинга качества воздуха в различных городах
мира. Бот предоставляет пользователю удобный и
интуитивно понятный интерфейс на основе inlineклавиатуры, позволяя выбирать континент и город без
использования текстовых команд.
Функциональность приложения была подтверждена с
помощью модульного тестирования, что повысило
надёжность системы и снизило вероятность
возникновения ошибок при дальнейших изменениях
кода.
В результате проект представляет собой
масштабируемое, поддерживаемое и готовое к
расширению программное решение,
демонстрирующее практическое применение
объектно-ориентированного программирования и
паттернов проектирования в реальной задаче.

12.

Спасибо за внимание!
English     Русский Rules