//! CLI port of `scripts/scrape_nl_comps.py` — KVK → `nl_companies` + `global_companies_registry`.

use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::{Arc, RwLock};
use std::time::Duration;

use athena_rs::AthenaClient;
use clap::Parser;
use dexter_sentinel::scrape::nl::{run_nl_randomized_loop, run_nl_single_pass, NlScrapeConfig};
use dexter_sentinel::scrape::DexterAthena;

#[derive(Parser, Debug)]
#[command(
    name = "nl-scrape",
    about = "Scrape purely Dutch company schemas from KVK API"
)]
struct Cli {
    /// Output CSV containing parsed `nl_companies` formatted rows
    #[arg(short = 'o', long, default_value = "nl_companies.csv")]
    output: PathBuf,

    /// Specific company name or fragment to search for
    #[arg(short = 'q', long)]
    query: Option<String>,

    /// URL to the specific KVK JSON API
    #[arg(long, default_value = "https://web-api.kvk.nl/zoeken/v3/search")]
    kvk_api_url: String,

    /// URL for the Dexter Athena Gateway
    #[arg(long, default_value = "https://cluster.athena-db.com")]
    athena_url: String,

    /// API key for the Dexter Athena Gateway (also uses ATHENA_API_KEY env)
    #[arg(long, env = "ATHENA_API_KEY")]
    athena_key: Option<String>,

    /// Value of the X-Athena-Client header
    #[arg(long, default_value = "dexter")]
    athena_client: String,

    /// Optional X-User-Id header for the Athena Gateway inserts
    #[arg(long, env = "ATHENA_USER_ID")]
    athena_user_id: Option<String>,

    /// Optional X-Company-Id header for the Athena Gateway inserts
    #[arg(long, env = "ATHENA_COMPANY_ID")]
    athena_company_id: Option<String>,

    /// Optional X-Organization-Id header for the Athena Gateway inserts
    #[arg(long, env = "ATHENA_ORGANIZATION_ID")]
    athena_organization_id: Option<String>,

    /// Print function timing and memory telemetry output (if instrumented)
    #[arg(long, default_value_t = false)]
    telemetry: bool,

    /// Dry run: scrape and save locally but do not upload to Athena
    #[arg(long, default_value_t = false)]
    no_insert: bool,

    /// Max concurrent network inserts targeting Athena
    #[arg(long, default_value_t = 25)]
    max_workers: usize,

    /// Enable recursive continuous scraping by using the incoming dataset to generate random queries
    #[arg(short = 'r', long, default_value_t = false)]
    randomize_query: bool,

    /// Maximum string length when randomly slicing company names for the next query
    #[arg(long, default_value_t = 2)]
    max_query_length: usize,

    /// Load HTTP/proxy settings from Athena `http_request_profiles` (requires insert mode)
    #[arg(long)]
    http_profile: Option<String>,

    #[arg(long)]
    proxy_url: Option<String>,
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    dotenv::dotenv().ok();
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
        )
        .init();

    let cli = Cli::parse();
    let insert = !cli.no_insert;

    let (dexter, http) = if insert {
        let key = cli
            .athena_key
            .filter(|s| !s.is_empty())
            .or_else(|| std::env::var("ATHENA_API_KEY").ok())
            .filter(|s| !s.is_empty())
            .ok_or_else(|| {
                anyhow::anyhow!("set --athena-key or ATHENA_API_KEY when not using --no-insert")
            })?;
        let base_trim = cli.athena_url.trim_end_matches('/').to_string();
        let sdk = Arc::new(AthenaClient::new(&base_trim, &key, &cli.athena_client).await?);
        let audit = dexter_sentinel::audit::spawn_audit_flusher(
            Arc::clone(&sdk),
            None,
            8192,
            100,
            Duration::from_secs(2),
        );
        let prof = if let Some(ref n) = cli.http_profile {
            let cache = Arc::new(RwLock::new(HashMap::new()));
            dexter_sentinel::data::http_profiles::resolve_profile_cached(Some(&sdk), &cache, n)
                .await
        } else {
            None
        };
        if cli.http_profile.is_some() && prof.is_none() {
            anyhow::bail!("--http-profile not found, disabled, or not loaded");
        }
        let audit_label = if cli.proxy_url.is_some() {
            Some("cli_override".to_string())
        } else {
            cli.http_profile.clone()
        };
        let http = dexter_sentinel::request::client::build_client_for_cli(
            prof.as_ref(),
            cli.proxy_url.as_deref(),
            true,
        )?;
        let http_loop = http.clone();
        let dex = DexterAthena::new_with_arc(
            sdk,
            base_trim,
            http,
            key,
            cli.athena_client.clone(),
            cli.athena_user_id.clone(),
            cli.athena_company_id.clone(),
            cli.athena_organization_id.clone(),
            None,
            audit,
            audit_label,
        );
        (Some(Arc::new(dex)), http_loop)
    } else {
        let http = dexter_sentinel::request::client::build_client_for_cli(
            None,
            cli.proxy_url.as_deref(),
            true,
        )?;
        (None, http)
    };

    if cli.telemetry {
        dexter_sentinel::scrape::log::TELEMETRY.store(true, std::sync::atomic::Ordering::Relaxed);
        dexter_sentinel::scrape::log::rate_reset();
        tokio::spawn(async {
            dexter_sentinel::scrape::log::rate_logger_loop_nl().await;
        });
    }

    let cfg = NlScrapeConfig {
        output_csv: cli.output.clone(),
        kvk_api_url: cli.kvk_api_url.clone(),
        insert,
        max_workers: cli.max_workers,
        max_query_len: cli.max_query_length,
    };

    if cli.randomize_query {
        tracing::info!("nl-scrape: randomized loop");
        run_nl_randomized_loop(http, dexter, cfg).await
    } else {
        let q = cli.query.as_deref().unwrap_or("").trim();
        if q.is_empty() {
            anyhow::bail!(
                "missing --query for single run (use --randomize-query for continuous mode)"
            );
        }
        tracing::info!("nl-scrape: single query");
        run_nl_single_pass(&http, dexter, &cfg, q).await
    }
}
