use actix_web::{web, HttpResponse};
use serde::Deserialize;
use serde_json::{json, Value};
use std::collections::HashMap;
use std::time::Instant;

use crate::audit::HttpAuditMeta;
use crate::data::http_profiles;
use crate::request::client::build_client_merged;
use crate::request::headers::build_headers;
use crate::scrape::template;
use crate::AppState;

#[derive(Deserialize)]
pub struct ScrapeRequest {
    pub config_name: String,
    /// When set, expanded with `context` and used as the request URL. When omitted, `base_url` from config is used.
    #[serde(default)]
    pub url: Option<String>,
    /// Template placeholder values (v1: object with string or JSON-scalar values).
    #[serde(default)]
    pub context: Option<Value>,
}

#[derive(Deserialize)]
pub struct ScrapeQuery {
    pub url: Option<String>,
}

pub async fn scrape_post_handler(state: web::Data<AppState>, body: web::Json<ScrapeRequest>) -> HttpResponse {
    let b = body.into_inner();
    let url = b.url.as_deref().filter(|s| !s.trim().is_empty());
    run_scrape(
        state.get_ref(),
        &b.config_name,
        url,
        b.context.as_ref(),
    )
    .await
}

pub async fn scrape_get_handler(
    state: web::Data<AppState>,
    path: web::Path<String>,
    query: web::Query<ScrapeQuery>,
) -> HttpResponse {
    let config_name = path.into_inner();
    let url = query.url.as_deref().filter(|s| !s.trim().is_empty());
    run_scrape(state.get_ref(), &config_name, url, None).await
}

async fn run_scrape(
    state: &AppState,
    config_name: &str,
    url_override: Option<&str>,
    context: Option<&Value>,
) -> HttpResponse {
    let start = Instant::now();

    let Some(config) = state.resolve_scraper_config(config_name) else {
        state
            .metrics
            .scrape_requests_total
            .with_label_values(&[config_name, "not_found"])
            .inc();
        return HttpResponse::NotFound()
            .json(json!({"error": format!("Config '{}' not found", config_name)}));
    };

    let ctx_src = context.cloned();
    let ctx = match template::context_from_optional(&ctx_src) {
        Ok(c) => c,
        Err(e) => {
            return HttpResponse::BadRequest().json(json!({"error": e.to_string()}));
        }
    };

    let url_source = url_override
        .map(str::trim)
        .filter(|s| !s.is_empty())
        .map(String::from)
        .unwrap_or_else(|| config.base_url.clone());

    let final_url = match template::expand_template(&url_source, &ctx) {
        Ok(u) => u,
        Err(e) => {
            return HttpResponse::BadRequest().json(json!({"error": e.to_string()}));
        }
    };

    if url::Url::parse(&final_url).is_err() {
        return HttpResponse::BadRequest().json(json!({"error": "invalid final URL"}));
    }

    let mut config = config.clone();
    if let Some(ref mut headers) = config.request.headers {
        let mut next = HashMap::new();
        for (k, v) in headers.iter() {
            let nv = if v.contains("{{") {
                match template::expand_template(v, &ctx) {
                    Ok(x) => x,
                    Err(e) => return HttpResponse::BadRequest().json(json!({"error": e.to_string()})),
                }
            } else {
                v.clone()
            };
            next.insert(k.clone(), nv);
        }
        *headers = next;
    }

    let cache_key: String = format!("{}:{}", config_name, final_url);
    if let Some(cached) = state.moka.get(&cache_key).await {
        state.metrics.cache_hits_total.inc();
        state
            .metrics
            .scrape_requests_total
            .with_label_values(&[config_name, "cache_hit"])
            .inc();
        if let Ok(val) = serde_json::from_str::<Value>(&cached) {
            state
                .metrics
                .scrape_api_response_json_bytes
                .with_label_values(&[config_name, "cache_hit"])
                .observe(cached.len() as f64);
            return HttpResponse::Ok().json(val);
        }
    }

    let db_profile = if let Some(ref pname) = config.request_profile {
        if state.athena.is_none() {
            tracing::warn!(
                "request_profile '{}' ignored for config '{}': Athena not configured",
                pname,
                config_name
            );
            None
        } else {
            let p = http_profiles::resolve_profile_cached(
                state.athena.as_ref(),
                &state.http_profile_cache,
                pname,
            )
            .await;
            if p.is_none() {
                tracing::warn!(
                    "request_profile '{}' not found or disabled for config '{}'",
                    pname,
                    config_name
                );
            }
            p
        }
    } else {
        None
    };

    let profile_name_for_log = db_profile
        .as_ref()
        .map(|p| p.name.clone())
        .or_else(|| config.request_profile.clone());

    let client = match build_client_merged(
        db_profile.as_ref(),
        config.proxy.as_ref(),
        &config.request,
    ) {
        Ok(c) => c,
        Err(e) => {
            tracing::error!("Failed to build HTTP client: {}", e);
            return HttpResponse::InternalServerError()
                .json(json!({"error": "Failed to build client"}));
        }
    };

    let audit = state.audit.as_ref();

    let is_post = config.request.method.eq_ignore_ascii_case("POST");
    let method_str: &'static str = if is_post { "POST" } else { "GET" };

    let meta = HttpAuditMeta {
        source: "api_scrape",
        profile_name: profile_name_for_log,
        config_name: Some(config_name.to_string()),
        extra: None,
    };

    let mut req_builder = if is_post {
        client.post(final_url.as_str())
    } else {
        client.get(final_url.as_str())
    };

    if let Some(ref headers) = config.request.headers {
        let header_map = build_headers(headers);
        req_builder = req_builder.headers(header_map);
    }

    let response = match crate::audit::logged_send(
        audit,
        req_builder.send(),
        method_str,
        final_url.as_str(),
        meta,
    )
    .await
    {
        Ok(r) => r,
        Err(e) => {
            tracing::error!("HTTP request failed: {}", e);
            state
                .metrics
                .scrape_requests_total
                .with_label_values(&[config_name, "error"])
                .inc();
            return HttpResponse::BadGateway()
                .json(json!({"error": format!("Request failed: {}", e)}));
        }
    };

    let html = match response.text().await {
        Ok(t) => t,
        Err(e) => {
            tracing::error!("Failed to read response: {}", e);
            return HttpResponse::InternalServerError()
                .json(json!({"error": "Failed to read response"}));
        }
    };

    state
        .metrics
        .scrape_upstream_html_bytes
        .with_label_values(&[config_name])
        .observe(html.len() as f64);

    let document = scraper::Html::parse_document(&html);
    let mut result: HashMap<String, String> = HashMap::new();

    for selector_config in &config.selectors {
        let selector = match scraper::Selector::parse(&selector_config.css) {
            Ok(s) => s,
            Err(e) => {
                tracing::warn!("Invalid CSS selector '{}': {:?}", selector_config.css, e);
                continue;
            }
        };

        let element = document.select(&selector).next();
        let value = match element {
            Some(el) => {
                if selector_config.attribute == "text" {
                    el.text().collect::<Vec<_>>().join(" ")
                } else {
                    el.value()
                        .attr(&selector_config.attribute)
                        .unwrap_or("")
                        .to_string()
                }
            }
            None => {
                if selector_config.required.unwrap_or(false) {
                    tracing::warn!(
                        "Required selector '{}' not found on {}",
                        selector_config.name,
                        final_url
                    );
                }
                String::new()
            }
        };

        let value = match selector_config.transform.as_deref() {
            Some("trim") => value.trim().to_string(),
            Some("uppercase") => value.to_uppercase(),
            Some("lowercase") => value.to_lowercase(),
            _ => value,
        };

        result.insert(selector_config.name.clone(), value);
    }

    result.insert("url".to_string(), final_url.clone());
    result.insert("config_name".to_string(), config_name.to_string());

    let duration = start.elapsed().as_secs_f64();
    state
        .metrics
        .scrape_duration_seconds
        .with_label_values(&[config_name])
        .observe(duration);
    state
        .metrics
        .scrape_requests_total
        .with_label_values(&[config_name, "success"])
        .inc();

    let response_json = json!(result);
    let json_len = serde_json::to_vec(&response_json)
        .map(|v| v.len())
        .unwrap_or(0) as f64;
    state
        .metrics
        .scrape_api_response_json_bytes
        .with_label_values(&[config_name, "fetched"])
        .observe(json_len);

    state
        .moka
        .insert(cache_key, response_json.to_string())
        .await;

    if let Some(ref athena) = state.athena {
        let table = config
            .output
            .as_ref()
            .map(|o| o.table.as_str())
            .unwrap_or("scraped_data");
        let payload_keys: Value = result
            .keys()
            .map(|k| Value::String(k.clone()))
            .collect::<Vec<_>>()
            .into();
        if let Err(e) = athena
            .insert(table)
            .payload(response_json.clone())
            .execute()
            .await
        {
            tracing::warn!("Failed to persist scraped data: {}", e);
            if let Some(ref bus) = state.audit {
                bus.try_log_db(table, "error", Some(e.to_string()), payload_keys.clone());
            }
        } else if let Some(ref bus) = state.audit {
            bus.try_log_db(table, "ok", None, payload_keys);
        }
    }

    HttpResponse::Ok().json(response_json)
}
