tg-bot/bot-tg.py

193 lines
7.5 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import logging
from aiogram import Bot, Dispatcher, types, F
from aiogram.filters import Command
from aiogram.utils.keyboard import ReplyKeyboardBuilder
import requests
import asyncio
from config import BOT_TOKEN, WEATHER_API_KEY
# настройка логирования
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# пасхалочка
CUSTOM_CITY_PHOTOS = {
"киев": {
"photo_url": "https://imgur.com/a/jzVHNoF",
"author": "Wikipedia"
},
"kyiv": {
"photo_url": "https://imgur.com/a/jzVHNoF",
"author": "Wikipedia"
}
}
# инициализация бота
bot = Bot(token=BOT_TOKEN)
dp = Dispatcher()
# клавиатура
def get_keyboard():
builder = ReplyKeyboardBuilder()
builder.button(text="🌤️ Узнать погоду")
builder.button(text="📍 Мое местоположение", request_location=True)
return builder.as_markup(resize_keyboard=True)
async def get_coordinates(city: str):
"""Получение координат города"""
url = f"https://nominatim.openstreetmap.org/search?city={city}&format=json"
headers = {"User-Agent": "CityPhotoBot/1.0"}
try:
response = requests.get(url, headers=headers, timeout=10)
if response.status_code == 200:
data = response.json()
if data:
for item in data:
if item.get('type') in ['city', 'town']:
return float(item['lon']), float(item['lat'])
return float(data[0]['lon']), float(data[0]['lat'])
except Exception as e:
logger.error(f"Geocoding error: {e}")
return None
async def get_weather(lat: float, lon: float, city_name: str) -> str:
"""Получение данных о погоде"""
url = f"https://api.openweathermap.org/data/2.5/weather?lat={lat}&lon={lon}&appid={WEATHER_API_KEY}&units=metric&lang=ru"
try:
response = requests.get(url, timeout=10)
if response.status_code == 200:
data = response.json()
weather_info = (
f"🌆 Город: {city_name}\n"
f"🌡 Температура: {data['main']['temp']:.1f}°C\n"
f"🧭 Ощущается как: {data['main']['feels_like']:.1f}°C\n"
f"💨 Ветер: {data['wind']['speed']} м/с\n"
f"☁️ Погода: {data['weather'][0]['description'].capitalize()}\n"
f"💧 Влажность: {data['main']['humidity']}%"
)
return weather_info
except Exception as e:
logger.error(f"Weather API error: {e}")
return "Не удалось получить данные о погоде"
async def get_city_photo(city: str):
"""Получение фото города с приоритетом ваших собственных фото"""
city_lower = city.lower()
# проверочка посхалки
if city_lower in CUSTOM_CITY_PHOTOS:
custom = CUSTOM_CITY_PHOTOS[city_lower]
return custom["photo_url"], custom["author"]
# генерация фото по городам
wikipedia_url = f"https://ru.wikipedia.org/w/api.php?action=query&prop=pageimages&titles={city}&pithumbsize=800&format=json"
try:
response = requests.get(wikipedia_url, timeout=10)
if response.status_code == 200:
data = response.json()
pages = data.get('query', {}).get('pages', {})
for page in pages.values():
if 'thumbnail' in page:
return page['thumbnail']['source'], "Wikipedia"
except Exception as e:
logger.error(f"Wikipedia API error: {e}")
return None, None
# ответ на /start
@dp.message(Command("start", "help"))
async def cmd_start(message: types.Message):
await message.answer(
"🏙️ Бот Погоды с фотографиями городов\n\n"
"Отправьте мне название города или поделитесь геопозицией:",
reply_markup=get_keyboard()
)
@dp.message(F.text == "🌤️ Узнать погоду")
async def weather_button(message: types.Message):
await message.answer("Введите название города:")
@dp.message(F.location)
async def handle_location(message: types.Message):
"""Обработка геолокации"""
try:
lat = message.location.latitude
lon = message.location.longitude
# принятие названия города
url = f"https://nominatim.openstreetmap.org/reverse?lat={lat}&lon={lon}&format=json"
headers = {"User-Agent": "CityPhotoBot/1.0"}
response = requests.get(url, headers=headers, timeout=10)
if response.status_code == 200:
data = response.json()
city_name = data.get('address', {}).get('city',
data.get('address', {}).get('town',
data.get('address', {}).get('village', "это место")))
weather_data = await get_weather(lat, lon, city_name)
photo_url, author = await get_city_photo(city_name)
if photo_url:
caption = f"{weather_data}\n\n📷 Фото: {author}" if author else weather_data
await message.answer_photo(photo_url, caption=caption)
else:
await message.answer(weather_data)
except Exception as e:
logger.error(f"Location error: {e}")
await message.answer("Ошибка обработки местоположения")
@dp.message(F.text)
async def city_handler(message: types.Message):
"""Обработка текстовых запросов"""
city = message.text.strip()
if city.startswith('/'):
return
logger.info(f"Processing city: {city}")
try:
# проверка не киев ли это
city_lower = city.lower()
if city_lower in CUSTOM_CITY_PHOTOS:
custom = CUSTOM_CITY_PHOTOS[city_lower]
weather_data = await get_weather(50.45, 30.52, "Киев") # корды киева
await message.answer_photo(
custom["photo_url"],
caption=f"{weather_data}\n\n📷 Фото: {custom['author']}"
)
return
# обработка других городов
coords = await get_coordinates(city)
if not coords:
await message.answer("Город не найден. Попробуйте уточнить название")
return
lon, lat = coords
weather_data = await get_weather(lat, lon, city)
photo_url, author = await get_city_photo(city)
if photo_url:
caption = f"{weather_data}\n\n📷 Фото: {author}" if author else weather_data
await message.answer_photo(photo_url, caption=caption)
else:
await message.answer(weather_data)
except Exception as e:
logger.error(f"City handler error: {e}")
await message.answer("Произошла ошибка при обработке запроса")
async def main():
await dp.start_polling(bot)
if __name__ == "__main__":
asyncio.run(main())