import pandas as pd
import requests
from bs4 import BeautifulSoup
import time
from datetime import datetime
import logging

from src.s3_tools import (
    upload_df_parquet, 
    upload_df_csv, 
    read_csv_from_s3, 
)

HEADERS = {
    'accept': 'application/json, text/plain, */*',
    'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/145.0.0.0 Safari/537.36'
}

URLS = {
    'earthquake': 'https://www.ngdc.noaa.gov/hazel/hazard-service/api/v1/earthquakes',
    'tsunami': 'https://www.ngdc.noaa.gov/hazel/hazard-service/api/v1/tsunamis/events',
    'volcano': 'https://www.ngdc.noaa.gov/hazel/hazard-service/api/v1/volcanoes?order=year%3Aasc'
}

# --- EXTRACT ---
def extract_noaa_generic(endpoint, s3_file_name):
    base_url = URLS[endpoint.lower()]
    all_items = []
    page = 1
    
    logging.info(f"Начинаем выгрузку NOAA: {endpoint}...")
    while True:
        max_retries = 3
        retry_count = 0
        success = False
        response = None
        
        # --- БЛОК ПОВТОРНЫХ ПОПЫТОК (RETRIES) ---
        while retry_count < max_retries:
            try:
                response = requests.get(base_url, params={'page': str(page)}, headers=HEADERS, timeout=15)
                success = True
                break # Запрос прошел успешно, выходим из цикла попыток
            except requests.exceptions.RequestException as e:
                retry_count += 1
                logging.warning(f"⚠️ Ошибка сети на NOAA {endpoint}, стр {page} (попытка {retry_count}/{max_retries}): {e}")
                time.sleep(5) # Ждем 5 секунд перед повтором (может быть временный сбой DNS)
                
        if not success or response is None:
            logging.error(f"Не удалось выгрузить NOAA {endpoint} стр {page}. Прерываем.")
            break
            
        if response.status_code != 200:
            break
            
        data = response.json()
        items = data.get('items', [])
        if not items:
            break
            
        all_items.extend(items)
        page += 1
        
        # Увеличим паузу до 1 секунды, чтобы сервер NOAA не блокировал нас
        time.sleep(1)

    df = pd.DataFrame(all_items)
    upload_df_csv(df, f"raw/noaa/{s3_file_name}.csv")
    logging.info(f"Выгрузка NOAA {endpoint} завершена. Собрано {len(df)} записей.")
def extract_all_noaa(**kwargs):
    extract_noaa_generic('earthquake', 'earthquakes_raw')
    extract_noaa_generic('tsunami', 'tsunamis_raw') 
    extract_noaa_generic('volcano', 'volcanoes_raw')

def extract_aviation(start_year=1990, end_year=None):
    # Если конечный год не указан, берем текущий
    if end_year is None:
        end_year = datetime.now().year

    all_rows = []
    table_headers = None

    for year in range(start_year, end_year + 1):
        page = 1
        
        while True:
            url = f'https://aviation-safety.net/database/year/{year}/{page}'
            logging.info(f"Парсинг: {year} год, страница {page} ({url})")
            
            # --- ЗАЩИТА ОТ БЛОКИРОВОК (RATE LIMITING) ---
            time.sleep(2) # Базовая пауза в 2 секунды между страницами
            
            max_retries = 3
            retry_count = 0
            success = False
            response = None
            
            while retry_count < max_retries:
                try:
                    # Добавили timeout, чтобы скрипт не висел бесконечно
                    response = requests.get(url, headers=HEADERS, timeout=15)
                    success = True
                    break # Успех, выходим из цикла попыток
                except requests.exceptions.RequestException as e:
                    retry_count += 1
                    logging.warning(f"⚠️ Ошибка соединения на {year} году, стр {page} (попытка {retry_count}/{max_retries}): {e}")
                    time.sleep(10) # Длинная пауза перед повтором, чтобы сервер "остыл"
                    
            if not success or response is None:
                logging.error(f"Не удалось загрузить {url} после {max_retries} попыток. Прерываем сбор для этого года.")
                break # Выходим из цикла страниц, переходим к следующему году
            
            if response.status_code != 200:
                break # Страницы для этого года закончились (получили 404 или другой статус)
                
            # --- ПАРСИНГ HTML ---
            soup = BeautifulSoup(response.text, "html.parser")

            if table_headers is None:
                headers = soup.find_all("th")
                if headers:
                    table_headers = [th.get_text(strip=True) for th in headers]
                    table_headers = [h if h != "\xa0" and h else f"extra_{i}" for i, h in enumerate(table_headers)]

            list_rows = soup.find_all("tr", class_="list")
            
            if not list_rows:
                break
                
            for tr in list_rows:
                cells = tr.find_all("td")
                row = []
                for td in cells:
                    link = td.find("a")
                    if link:
                        row.append(link.get_text(strip=True))
                    else:
                        img = td.find("img")
                        row.append(img["alt"] if img and img.get("alt") else td.get_text(strip=True))
                
                # Дополняем строку пустыми значениями, если ячеек меньше, чем колонок
                if table_headers:
                    while len(row) < len(table_headers):
                        row.append("")
                        
                # Добавляем год
                row.append(year)
                all_rows.append(row)
            
            # Переходим к следующей странице внутри текущего года
            page += 1

    # Создаем DataFrame
    if not all_rows:
        logging.warning("Данные по авиации не найдены!")
        return pd.DataFrame()

    # Задаем имена колонок
    columns = table_headers + ["Year"] if table_headers else [f"col_{i}" for i in range(len(all_rows[0])-1)] + ["Year"]
    df = pd.DataFrame(all_rows, columns=columns)

    date_col = next((col for col in df.columns if 'date' in str(col).lower()), df.columns[0])

    # Конвертируем текстовую дату в datetime
    df['Date'] = pd.to_datetime(df[date_col], format='%d %b %Y', errors='coerce')

    upload_df_csv(df, f"raw/aircraft/aviation_safety_raw.csv")
    logging.info(f"Успешно выгружено {len(df)} записей по авиакатастрофам.")


# --- TRANSFORM ---
def transform_noaa(raw_df, event_type):
    df = raw_df.copy()
    cols_to_keep = ['year', 'month', 'day', 'country', 'deaths', 'damageMillionsDollars', 'housesDestroyed']
    df = df[[c for c in cols_to_keep if c in df.columns]]
    df['Date'] = pd.to_datetime(df[['year', 'month', 'day']].fillna(1), errors='coerce')

    return pd.DataFrame({
        'Date': df['Date'],
        'Year': pd.to_numeric(df.get('year'), errors='coerce'),
        'Country': df.get('country', 'Unknown').astype(str).str.title(),
        'Event_Type': event_type,
        'Fatalities': pd.to_numeric(df.get('deaths', 0), errors='coerce').fillna(0),
        'Damage (mill dollars)': pd.to_numeric(df.get('damageMillionsDollars', 0), errors='coerce').fillna(0),
        'Houses Destroyed': pd.to_numeric(df.get('housesDestroyed', 0), errors='coerce').fillna(0),
        'Source': 'NOAA'
    }).dropna(subset=['Year'])

def transform_emdat(raw_df):
    df = raw_df.copy()
    
    df.columns = df.columns.str.strip()
    
    year_col = 'Start Year' if 'Start Year' in df.columns else 'Year'
    damage_col = "Total Damage ('000 US$)" if "Total Damage ('000 US$)" in df.columns else 'Total Damage'
    
    year_s = df.get(year_col, pd.Series(1970, index=df.index))
    month_s = df.get('Start Month', pd.Series(1, index=df.index))
    day_s = df.get('Start Day', pd.Series(1, index=df.index))
    
    df['Date'] = pd.to_datetime(
        pd.DataFrame({'year': year_s, 'month': month_s, 'day': day_s}).fillna(1), 
        errors='coerce'
    )
    

    fatalities = pd.to_numeric(df.get('Total Deaths', pd.Series(0, index=df.index)), errors='coerce').fillna(0)
    damage = pd.to_numeric(df.get(damage_col, pd.Series(0, index=df.index)), errors='coerce').fillna(0) / 1000
    houses = pd.to_numeric(df.get('Total Destroyed', pd.Series(0, index=df.index)), errors='coerce').fillna(0)
    
    # 3. Формируем финальный датафрейм
    return pd.DataFrame({
        'Date': df['Date'],
        'Year': pd.to_numeric(df.get(year_col), errors='coerce'),
        'Country': df.get('Country', pd.Series('Unknown', index=df.index)).astype(str),
        'Event_Type': df.get('Disaster Type', pd.Series('Disaster', index=df.index)).astype(str),
        'Fatalities': fatalities,
        'Damage (mill dollars)': damage,
        'Houses Destroyed': houses,
        'Source': 'EM-DAT'
    }).dropna(subset=['Year'])

def transform_gtd(raw_df):
    logging.info("Трансформация данных GTD...")
    df = raw_df.copy()
    
    month_series = df.get('imonth', pd.Series(1, index=df.index))
    day_series = df.get('iday', pd.Series(1, index=df.index))
    prop_series = df.get('propvalue', pd.Series(0, index=df.index))
    
    df['month_clean'] = month_series.apply(lambda x: int(x) if pd.notnull(x) and x > 0 else 1)
    df['day_clean'] = day_series.apply(lambda x: int(x) if pd.notnull(x) and x > 0 else 1)
    
    year_str = pd.to_numeric(df.get('iyear', 1970), errors='coerce').fillna(1970).astype(int).astype(str)
    
    df['Date'] = pd.to_datetime(
        year_str + '-' + df['month_clean'].astype(str) + '-' + df['day_clean'].astype(str), 
        errors='coerce'
    )
    
    damage = pd.to_numeric(prop_series, errors='coerce').fillna(0)
    damage = damage.apply(lambda x: x / 1_000_000 if x > 0 else 0.0)
    
    return pd.DataFrame({
        'Date': df['Date'],
        'Year': pd.to_numeric(df.get('iyear'), errors='coerce'),
        'Country': df.get('country_txt', 'Unknown').astype(str),
        'Event_Type': df.get('attacktype1_txt', 'Terrorism').astype(str),
        'Fatalities': pd.to_numeric(df.get('nkill', 0), errors='coerce').fillna(0),
        'Damage (mill dollars)': damage,
        'Houses Destroyed': 0.0,
        'Source': 'GTD'
    }).dropna(subset=['Year'])


def transform_aviation(raw_df):
    logging.info("Трансформация данных Aviation...")
    df = raw_df.copy()
    
    df.columns = df.columns.str.strip().str.lower()
    
    date_series = df.get('date', pd.Series(pd.NaT, index=df.index))
    df['Date'] = pd.to_datetime(date_series, errors='coerce')
    
    fat_col = "fat."
    
    if fat_col in df.columns:
        fatalities = pd.to_numeric(df[fat_col].astype(str).str.extract(r'(\d+)')[0], errors='coerce').fillna(0)
    else:
        fatalities = pd.Series(0, index=df.index)
        
    return pd.DataFrame({
        'Date': df['Date'],
        'Year': df['Date'].dt.year,
        'Country': df.get('location', pd.Series('Unknown', index=df.index)).astype(str),
        'Event_Type': pd.Series('Aviation Accident', index=df.index).astype(str),
        'Fatalities': fatalities,
        'Damage (mill dollars)': 0.0,
        'Houses Destroyed': 0.0,
        'Source': 'Aviation Safety Network'
    }).dropna(subset=['Year'])


# --- LOAD ---
def transform_and_load_processed(**kwargs):
    logging.info("Чтение сырых данных...")
    
    try:
        logging.info("Поиск EM-DAT в S3...")
        raw_emdat = read_csv_from_s3("raw/emdat.csv", skiprows=6)
    except Exception as e:
        logging.warning(f"Файл EM-DAT в S3 не найден. Читаем локально: {e}")
        raw_emdat = pd.read_csv("/opt/airflow/data/emdat.csv", skiprows=6)
    print("ТАК ЗДЕСЬ ВРОДЕ ВСЕ НОРМ ЕСЛИ МЕНЯ НЕТУ ТЕБЕ ПИЗДЕС")
    try:
        logging.info("Поиск GTD в S3...")
        raw_gtd = read_csv_from_s3("raw/gtd.csv")
    except Exception as e:
        logging.warning(f"Файл GTD в S3 не найден. Читаем локально: {e}")
        raw_gtd = pd.read_csv("/opt/airflow/data/gtd.csv")

    raw_eq = read_csv_from_s3("raw/noaa/earthquakes_raw.csv")
    raw_ts = read_csv_from_s3("raw/noaa/tsunamis_raw.csv")
    raw_volc = read_csv_from_s3("raw/noaa/volcanoes_raw.csv")
    raw_avia = read_csv_from_s3("raw/aircraft/aviation_safety_raw.csv")

    # Трансформация
    clean_emdat = transform_emdat(raw_emdat)
    clean_gtd = transform_gtd(raw_gtd)
    clean_eq = transform_noaa(raw_eq, 'Earthquake')
    clean_ts = transform_noaa(raw_ts, 'Tsunami')
    clean_volc = transform_noaa(raw_volc, 'Volcano')
    clean_avia = transform_aviation(raw_avia)

    # Объединение
    final_df = pd.concat([clean_eq, clean_ts, clean_volc, clean_avia, clean_emdat, clean_gtd], ignore_index=True)
    
    final_df['Year'] = final_df['Year'].astype(int)
    final_df['Fatalities'] = final_df['Fatalities'].astype(int)
    final_df.drop_duplicates(inplace=True)
    
    # Загрузка финального файла в S3
    upload_df_parquet(final_df, "processed/global_risks_clean.parquet")
    logging.info(f"ETL завершен! Собрано записей: {len(final_df)}.")