//! NL `nl_companies` + `global_companies_registry` dual insert — ports `scrape_nl_comps.py`
//! (`process_nl_companies`, randomized loop).

use std::collections::HashSet;
use std::fs::OpenOptions;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};

use rand::seq::SliceRandom;
use serde_json::{Map, Value};
use tokio::sync::Semaphore;

use super::athena::{AthenaInsertStatus, DexterAthena};
use super::global::row_to_gateway_payload;
use super::kvk::fetch_all_kvk_companies;
use super::log::{rate_add_inserted, rate_add_scraped};
use super::mapping::map_nl_company_to_global_script;

/// `(kvk_cache, company_number_cache)` from an existing `nl_companies` CSV.
pub fn load_nl_csv_caches(path: &Path) -> (HashSet<i64>, HashSet<String>) {
    let Ok(f) = std::fs::File::open(path) else {
        return (HashSet::new(), HashSet::new());
    };
    let mut rdr = csv::Reader::from_reader(f);
    let headers: Vec<String> = match rdr.headers() {
        Ok(h) => h.iter().map(String::from).collect(),
        Err(_) => return (HashSet::new(), HashSet::new()),
    };
    let Some(kvk_i) = headers.iter().position(|x| x == "kvk_number") else {
        return (HashSet::new(), HashSet::new());
    };
    let mut kvk_cache = HashSet::new();
    let mut company_numbers = HashSet::new();
    for rec in rdr.records().flatten() {
        if let Some(cell) = rec.get(kvk_i) {
            if let Ok(n) = cell.trim().parse::<i64>() {
                kvk_cache.insert(n);
                company_numbers.insert(n.to_string());
            }
        }
    }
    (kvk_cache, company_numbers)
}

fn cell_to_value(cell: &str) -> Value {
    let t = cell.trim();
    if t.is_empty() {
        return Value::Null;
    }
    if t == "true" {
        return Value::Bool(true);
    }
    if t == "false" {
        return Value::Bool(false);
    }
    if let Ok(i) = t.parse::<i64>() {
        return Value::Number(serde_json::Number::from(i));
    }
    if let Ok(f) = t.parse::<f64>() {
        if let Some(n) = serde_json::Number::from_f64(f) {
            return Value::Number(n);
        }
    }
    Value::String(t.to_string())
}

fn load_companies_from_csv(path: &Path) -> Vec<Map<String, Value>> {
    let Ok(data) = std::fs::read_to_string(path) else {
        return vec![];
    };
    let mut rdr = csv::Reader::from_reader(data.as_bytes());
    let headers: Vec<String> = match rdr.headers() {
        Ok(h) => h.iter().map(String::from).collect(),
        Err(_) => return vec![],
    };
    let mut out = Vec::new();
    for rec in rdr.records().flatten() {
        let mut m = Map::new();
        for (i, h) in headers.iter().enumerate() {
            if let Some(cell) = rec.get(i) {
                m.insert(h.clone(), cell_to_value(cell));
            }
        }
        out.push(m);
    }
    out
}

fn nl_table_payload(company: &Map<String, Value>) -> Value {
    let mut m = Map::new();
    for (k, v) in company {
        if v.is_null() {
            continue;
        }
        if k == "id" {
            continue;
        }
        m.insert(k.clone(), v.clone());
    }
    Value::Object(m)
}

fn kvk_i64(company: &Map<String, Value>) -> Option<i64> {
    company.get("kvk_number").and_then(|v| {
        v.as_i64()
            .or_else(|| v.as_u64().map(|u| u as i64))
            .or_else(|| v.as_str().and_then(|s| s.trim().parse().ok()))
    })
}

async fn insert_nl_company(
    dexter: &DexterAthena,
    company: &Map<String, Value>,
    kvk_cache: &Mutex<HashSet<i64>>,
    batch_max: usize,
) {
    let Some(kvk) = kvk_i64(company) else {
        return;
    };
    let name = company.get("name").and_then(|v| v.as_str()).unwrap_or("");
    if name.is_empty() {
        return;
    }
    if kvk_cache.lock().unwrap().contains(&kvk) {
        return;
    }
    let payload = nl_table_payload(company);
    let (status, detail) = dexter
        .insert_row_with_retry_resilient("nl_companies", payload, batch_max)
        .await;
    match status {
        AthenaInsertStatus::Ok => {
            tracing::info!("Inserted: ({kvk}) {name} [nl_companies]");
            kvk_cache.lock().unwrap().insert(kvk);
            rate_add_inserted(1);
        }
        AthenaInsertStatus::Duplicate => {
            kvk_cache.lock().unwrap().insert(kvk);
        }
        AthenaInsertStatus::Error | AthenaInsertStatus::Exception => {
            tracing::error!("nl_companies insert failed ({kvk}) {name}: {:?}", detail);
        }
    }
}

async fn insert_global_from_nl_row(
    dexter: &DexterAthena,
    company: &Map<String, Value>,
    company_number_cache: &Mutex<HashSet<String>>,
    batch_max: usize,
) {
    let global_row = map_nl_company_to_global_script(company);
    let Some(cn) = global_row
        .get("company_number")
        .and_then(|v| v.as_str())
        .map(str::to_string)
    else {
        return;
    };
    if cn.is_empty() {
        return;
    }
    if company_number_cache.lock().unwrap().contains(&cn) {
        return;
    }
    let payload = row_to_gateway_payload(&global_row);
    let (status, detail) = dexter
        .insert_row_with_retry_resilient("global_companies_registry", payload, batch_max)
        .await;
    match status {
        AthenaInsertStatus::Ok => {
            tracing::info!("Inserted: ({cn}) [global_companies_registry]");
            company_number_cache.lock().unwrap().insert(cn);
            rate_add_inserted(1);
        }
        AthenaInsertStatus::Duplicate => {
            company_number_cache.lock().unwrap().insert(cn);
        }
        AthenaInsertStatus::Error | AthenaInsertStatus::Exception => {
            tracing::error!("global insert failed ({cn}): {:?}", detail);
        }
    }
}

async fn process_nl_company_pair(
    dexter: Arc<DexterAthena>,
    company: Map<String, Value>,
    kvk_cache: Arc<Mutex<HashSet<i64>>>,
    company_number_cache: Arc<Mutex<HashSet<String>>>,
    batch_max: usize,
) {
    let c = &company;
    tokio::join!(
        insert_nl_company(&dexter, c, &kvk_cache, batch_max),
        insert_global_from_nl_row(&dexter, c, &company_number_cache, batch_max),
    );
}

/// Dual Athena insert: `nl_companies` + script-shaped `global_companies_registry` row per company.
pub async fn process_nl_companies(
    dexter: Arc<DexterAthena>,
    companies: Vec<Map<String, Value>>,
    insert: bool,
    max_workers: usize,
    kvk_cache: Arc<Mutex<HashSet<i64>>>,
    company_number_cache: Arc<Mutex<HashSet<String>>>,
) {
    if !insert {
        return;
    }
    if companies.is_empty() {
        tracing::warn!("No companies to insert.");
        return;
    }
    let n = dexter
        .coordinator()
        .insert_concurrency_for(max_workers)
        .await;
    let sem = Arc::new(Semaphore::new(n));
    let mut handles = Vec::new();
    for company in companies {
        let d = Arc::clone(&dexter);
        let kc = Arc::clone(&kvk_cache);
        let cc = Arc::clone(&company_number_cache);
        let permit = match sem.clone().acquire_owned().await {
            Ok(p) => p,
            Err(_) => continue,
        };
        handles.push(tokio::spawn(async move {
            let _p = permit;
            process_nl_company_pair(d, company, kc, cc, max_workers).await;
        }));
    }
    for h in handles {
        let _ = h.await;
    }
}

pub fn save_nl_companies_csv(companies: &[Map<String, Value>], path: &Path) -> std::io::Result<()> {
    if companies.is_empty() {
        tracing::warn!("No companies to save.");
        return Ok(());
    }
    let mut keys: Vec<String> = companies[0].keys().cloned().collect();
    keys.sort();
    keys.retain(|k| k != "id");
    let file_exists = path.is_file() && path.metadata().map(|m| m.len() > 0).unwrap_or(false);
    let mut wtr = csv::WriterBuilder::new().has_headers(false).from_writer(
        OpenOptions::new()
            .create(true)
            .append(file_exists)
            .truncate(!file_exists)
            .write(true)
            .open(path)?,
    );
    if !file_exists {
        wtr.write_record(&keys)?;
    }
    for c in companies {
        let vals: Vec<String> = keys
            .iter()
            .map(|k| {
                c.get(k)
                    .map(|v| match v {
                        Value::Null => String::new(),
                        Value::String(s) => s.clone(),
                        other => other.to_string().trim_matches('"').to_string(),
                    })
                    .unwrap_or_default()
            })
            .collect();
        wtr.write_record(&vals)?;
    }
    wtr.flush()?;
    tracing::info!(
        "Saved {} companies to '{}'.",
        companies.len(),
        path.display()
    );
    Ok(())
}

pub struct NlScrapeConfig {
    pub output_csv: PathBuf,
    pub kvk_api_url: String,
    pub insert: bool,
    pub max_workers: usize,
    pub max_query_len: usize,
}

pub async fn run_nl_single_pass(
    http: &reqwest::Client,
    dexter: Option<Arc<DexterAthena>>,
    cfg: &NlScrapeConfig,
    query: &str,
) -> anyhow::Result<()> {
    let q = query.trim();
    if q.is_empty() {
        anyhow::bail!("empty query");
    }
    let (kvk_init, cn_init) = load_nl_csv_caches(&cfg.output_csv);
    let kvk_cache = Arc::new(Mutex::new(kvk_init));
    let company_number_cache = Arc::new(Mutex::new(cn_init));

    let companies = fetch_all_kvk_companies(http, &cfg.kvk_api_url, q).await;
    tracing::info!("Fetched {} companies from KVK.", companies.len());
    rate_add_scraped(companies.len() as u64);

    if let Some(ref d) = dexter {
        process_nl_companies(
            Arc::clone(d),
            companies.clone(),
            cfg.insert,
            cfg.max_workers,
            Arc::clone(&kvk_cache),
            Arc::clone(&company_number_cache),
        )
        .await;
    }

    if !companies.is_empty() {
        save_nl_companies_csv(&companies, &cfg.output_csv)?;
    } else {
        tracing::warn!("No companies to save.");
    }
    Ok(())
}

/// Infinite loop: random name prefix from CSV → KVK fetch → dual insert → append CSV (Python `randomize_query_loop`).
pub async fn run_nl_randomized_loop(
    http: reqwest::Client,
    dexter: Option<Arc<DexterAthena>>,
    cfg: NlScrapeConfig,
) -> anyhow::Result<()> {
    let mut rng = rand::thread_rng();
    let mut seen_queries = HashSet::<String>::new();
    let (kvk_init, cn_init) = load_nl_csv_caches(&cfg.output_csv);
    let kvk_cache = Arc::new(Mutex::new(kvk_init));
    let company_number_cache = Arc::new(Mutex::new(cn_init));
    let mut csv_source = cfg.output_csv.clone();

    loop {
        let companies_csv = load_companies_from_csv(&csv_source);
        let query = if companies_csv.is_empty() {
            tracing::warn!(
                "No companies in '{}'; using query 'fi'.",
                csv_source.display()
            );
            "fi".to_string()
        } else if let Some(company) = companies_csv.choose(&mut rng) {
            let name = company
                .get("name")
                .and_then(|v| v.as_str())
                .unwrap_or("")
                .trim();
            if name.is_empty() {
                "fi".to_string()
            } else {
                let mut q: String = name.chars().take(cfg.max_query_len).collect();
                q = q.trim().to_string();
                if q.is_empty() {
                    "fi".to_string()
                } else {
                    q
                }
            }
        } else {
            "fi".to_string()
        };

        if seen_queries.contains(&query) {
            tracing::info!("Already tried query '{query}', picking another.");
            continue;
        }
        seen_queries.insert(query.clone());

        tracing::info!("Randomized NL query: '{query}'");

        let companies = fetch_all_kvk_companies(&http, &cfg.kvk_api_url, &query).await;
        rate_add_scraped(companies.len() as u64);

        if let Some(ref d) = dexter {
            process_nl_companies(
                Arc::clone(d),
                companies.clone(),
                cfg.insert,
                cfg.max_workers,
                Arc::clone(&kvk_cache),
                Arc::clone(&company_number_cache),
            )
            .await;
        }

        if !companies.is_empty() {
            save_nl_companies_csv(&companies, &cfg.output_csv)?;
            let mut kc = kvk_cache.lock().unwrap();
            let mut cc = company_number_cache.lock().unwrap();
            for c in &companies {
                if let Some(k) = kvk_i64(c) {
                    kc.insert(k);
                    cc.insert(k.to_string());
                }
            }
        } else {
            tracing::warn!("No companies to save.");
        }

        csv_source = cfg.output_csv.clone();
    }
}
