Парсинг погоды на Python: от Visual Crossing API до PostgreSQL

CREATE TABLE IF NOT EXISTS weather (
    id SERIAL PRIMARY KEY,
    city VARCHAR(255),
    date DATE NOT NULL,
    tempmax NUMERIC,
    temp NUMERIC,
    tempmin NUMERIC,
    precip NUMERIC,
    humidity NUMERIC,
    windspeed NUMERIC,
    cloudcover NUMERIC,
    solarenergy NUMERIC,
    uvindex NUMERIC,
    preciptype VARCHAR(255),
    latitude DOUBLE PRECISION NOT NULL,
    longitude DOUBLE PRECISION NOT NULL,
    UNIQUE (date, latitude, longitude)
);

import time
import logging
import psycopg2
import pandas as pd
import urllib.parse
import urllib.request
from datetime import datetime
from geopy.geocoders import Nominatim
from typing import Optional, Dict, Any

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def filter_weather(df_weather: pd.DataFrame) -> pd.DataFrame:
    """
    Фильтрует нужные колонки, сортирует данные по дате и переводит типы осадков на русский язык.

    :param df_weather: DataFrame с данными о погоде.
    :return: Обработанный DataFrame.
    """
    columns = [
        'datetime', 'tempmax', 'temp', 'tempmin', 'precip', 'humidity',
        'windspeed', 'cloudcover', 'solarenergy', 'uvindex', 'preciptype'
    ]
    df_weather = df_weather[columns]
    translations = {
        'rain': 'дождь',
        'rain,snow': 'дождь, снег',
        'snow': 'снег',
        'rain,freezingrain,snow': 'дождь, заморозки, снег',
        'freezingrain': 'заморозки',
        'rain,snow,ice': 'дождь, снег, лед',
        'freezingrain,snow': 'замерзающий дождь, снег',
        'snow,ice': 'снег, лед',
        'freezingrain,snow,ice': 'замерзающий дождь, снег, лед',
        'rain,freezingrain,snow,ice': 'дождь, замерзающий дождь, снег, лед',
        'rain,freezingrain': 'дождь, град',
        'ice': 'лед',
        'rain,ice': 'дождь, лед'
    }
    df_weather.loc[:, 'preciptype'] = df_weather['preciptype'].replace(translations)
    return df_weather.sort_values(by='datetime')


def historical_weather(latitude: float, longitude: float, start_date: str,
                           end_date: str, api_key: str, pause: int = 2) -> Optional[pd.DataFrame]:
    """
    Получает исторические данные о погоде для заданных координат и диапазона дат.

    :param latitude: Широта.
    :param longitude: Долгота.
    :param start_date: Начальная дата (YYYY-MM-DD).
    :param end_date: Конечная дата (YYYY-MM-DD).
    :param api_key: API-ключ Visual Crossing.
    :param pause: Пауза между запросами в секундах.
    :return: DataFrame с данными или None при ошибке.
    """
    coords = f"{latitude},{longitude}"
    url = (
        f"https://weather.visualcrossing.com/VisualCrossingWebServices/rest/services/timeline/"
        f"{urllib.parse.quote(coords)}/{start_date}/{end_date}"
        f"?unitGroup=metric&include=days&key={api_key}&contentType=csv"
    )
    logging.info(f"Запрос: {url}")
    try:
        with urllib.request.urlopen(url) as response:
            df_weather = pd.read_csv(response)
        df_weather = filter_weather(df_weather)
        df_weather['latitude'] = latitude
        df_weather['longitude'] = longitude
        time.sleep(pause)
        return df_weather
    except urllib.error.HTTPError as e:
        logging.error(f"HTTPError (код {e.code}) для координат {latitude}, {longitude}")
    except urllib.error.URLError as e:
        logging.error(f"URLError ({e.reason}) для координат {latitude}, {longitude}")
    return None


def get_city_by_coordinates(latitude: float, longitude: float) -> str:
    """
    Определяет название города по координатам с использованием Nominatim.

    :param latitude: Широта.
    :param longitude: Долгота.
    :return: Название города или 'Не определен'.
    """
    try:
        geolocator = Nominatim(user_agent="weather_parser")
        location = geolocator.reverse((latitude, longitude), language='ru')
        if location:
            address = location.raw.get('address', {})
            return address.get('city') or address.get('town') or address.get('village') or "Не определен"
    except Exception as e:
        logging.error(f"Ошибка геокодирования ({latitude}, {longitude}): {e}")
    return "Не определен"


def process_data(df_input: pd.DataFrame, api_key: str,
                                   db_config: Dict[str, Any]) -> None:
    """
    Для каждой записи входного DataFrame получает данные о погоде, добавляет название города,
    и записывает данные в PostgreSQL.

    :param df_input: DataFrame с колонками 'latitude', 'longitude', 'start_date', 'end_date'.
    :param api_key: API-ключ Visual Crossing.
    :param db_config: Параметры подключения к БД.
    """
    all_data = pd.DataFrame()

    for idx, row in df_input.iterrows():
        lat, lon = row['latitude'], row['longitude']
        start, end = row['start_date'], row['end_date']
        logging.info(f"Обработка {idx+1}/{len(df_input)}: ({lat}, {lon}) с {start} по {end}")
        df_weather = historical_weather(lat, lon, start, end, api_key)
        if df_weather is not None:
            df_weather['city'] = get_city_by_coordinates(lat, lon)
            all_data = pd.concat([all_data, df_weather], ignore_index=True)
        else:
            logging.error(f"Нет данных для координат ({lat}, {lon})")

    if all_data.empty:
        logging.error("Нет данных для записи в БД.")
        return

    all_data.rename(columns={'datetime': 'date'}, inplace=True)
    cols = ['city', 'date', 'tempmax', 'temp', 'tempmin', 'precip', 'humidity',
            'windspeed', 'cloudcover', 'solarenergy', 'uvindex', 'preciptype',
            'latitude', 'longitude']
    data_to_insert = all_data[cols]

    insert_query = """
        INSERT INTO weather (
            city, date, tempmax, temp, tempmin, precip, humidity, windspeed,
            cloudcover, solarenergy, uvindex, preciptype, latitude, longitude
        )
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
        ON CONFLICT (date, latitude, longitude) DO UPDATE SET
            city = EXCLUDED.city,
            tempmax = EXCLUDED.tempmax,
            temp = EXCLUDED.temp,
            tempmin = EXCLUDED.tempmin,
            precip = EXCLUDED.precip,
            humidity = EXCLUDED.humidity,
            windspeed = EXCLUDED.windspeed,
            cloudcover = EXCLUDED.cloudcover,
            solarenergy = EXCLUDED.solarenergy,
            uvindex = EXCLUDED.uvindex,
            preciptype = EXCLUDED.preciptype
    """

    try:
        conn = psycopg2.connect(
            host=db_config['host'],
            port=db_config.get('port', 5432),
            dbname=db_config['dbname'],
            user=db_config['user'],
            password=db_config['password']
        )
        conn.autocommit = True
        cursor = conn.cursor()
        logging.info("Соединение с БД установлено.")
    except Exception as e:
        logging.error(f"Ошибка подключения к БД: {e}")
        return

    for _, row in data_to_insert.iterrows():
        params = (
            row['city'], row['date'], row['tempmax'], row['temp'],
            row['tempmin'], row['precip'], row['humidity'], row['windspeed'],
            row['cloudcover'], row['solarenergy'], row['uvindex'], row['preciptype'],
            row['latitude'], row['longitude']
        )
        try:
            cursor.execute(insert_query, params)
            logging.info(f"Данные за {row['date']} для ({row['latitude']}, {row['longitude']}), город: {row['city']} записаны")
        except Exception as e:
            logging.error(f"Ошибка вставки данных: {e}")

    cursor.close()
    conn.close()
    logging.info("Соединение с БД закрыто.")


def main() -> None:
    """
    Основная функция: получает данные о погоде по заданным координатам и датам, 
    и сохраняет их в базу данных PostgreSQL.
    """
    api_key = "YOUR_API_KEY_HERE"

    db_config = {
        'host': "your_host",
        'port': 5432,
        'dbname': "your_dbname",
        'user': "your_user",
        'password': "your_password"
    }
    process_data(df_input, api_key, db_config)
    logging.info("Сбор данных завершен.")


if __name__ == "__main__":
    main()

api_key = "YOUR_API_KEY_HERE"

db_config = {
    'host': "your_host",
    'port': 5432,
    'dbname': "your_dbname",
    'user': "your_user",
    'password': "your_password"
}