//! CLI port of `scripts/scrape_global_registry.py` — KVK / Northdata / EGRUL / registries.io → Athena
//! `global_companies_registry` (+ `ee_companies` when EE, `ru_companies` when RU,
//! `pl_companies` / `cz_companies` when using registries.io for PL / CZ).

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

use athena_rs::AthenaClient;
use clap::{Parser, ValueEnum};
use dexter_sentinel::scrape::global::{
    run_randomized_registry_loop, run_single_registry_pass, GlobalRegistryPipelineConfig,
    RegistryEngine, VatIdSearchEngine,
};
use dexter_sentinel::scrape::DexterAthena;
use dexter_sentinel::audit::AuditBus;

#[derive(Debug, Clone, Copy, ValueEnum)]
enum SearchEngine {
    Kvk,
    Northdata,
    /// Russian Unified State Register (ФНС egrul.nalog.ru search API)
    Egrul,
    /// registries.io JSON API — use with `--country-code PL` or `CZ`
    RegistriesIo,
}

#[derive(Debug, Clone, ValueEnum)]
enum VatCli {
    None,
    Bizzy,
}

#[derive(Parser, Debug)]
#[command(
    name = "dexter-scrape",
    about = "Global company registry scraper (defaults: NL→KVK, else Northdata; use egrul for RU) with optional Athena upload"
)]
struct Cli {
    /// Two-letter ISO country code (e.g., NL, EE, RU, PL)
    #[arg(short = 'c', long)]
    country_code: String,

    /// Search engine (default: kvk for NL, northdata for other countries; use egrul for RU FNS)
    #[arg(short = 's', long)]
    search_engine: Option<SearchEngine>,

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

    /// Path to output the raw scraped rows (CSV format)
    #[arg(short = 'o', long, default_value = "global_registry_scrape.csv")]
    output: PathBuf,

    /// URL for the Dexter Athena Gateway
    #[arg(long, default_value = "https://mirror3.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,

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

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

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

    /// Path to file containing English/Dutch words to use as fallback random queries
    #[arg(long)]
    randomize_words_file: Option<PathBuf>,

    /// Initial CSV file to seed the randomize query algorithm with names
    #[arg(long)]
    randomize_seed_csv: Option<PathBuf>,

    /// Skip all CSV query seeds (same as --output and scripts/nl_companies.csv); words + last results only
    #[arg(long, default_value_t = false)]
    randomize_no_seed_csv: bool,

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

    /// URL prefix to use when querying Northdata
    #[arg(long, default_value = "https://www.northdata.com")]
    northdata_base_url: String,

    /// Base URL for FNS EGRUL/EGRIP search (used with `--search-engine egrul` / country RU)
    #[arg(long, default_value = "https://egrul.nalog.ru")]
    egrul_base_url: String,

    /// Base URL for registries.io (used with `--search-engine registries-io`)
    #[arg(long, default_value = "https://registries.io")]
    registries_io_base_url: String,

    /// Page size for registries.io `GET /api/companies`
    #[arg(long, default_value_t = 50)]
    registries_io_page_limit: u32,

    /// Path to a file holding NORTHDATA_COOKIE (used when scraping Northdata)
    #[arg(long)]
    northdata_cookie_file: Option<PathBuf>,

    /// Do not overwrite output CSV across iterations (append rows instead)
    #[arg(short = 'a', long, default_value_t = false)]
    append_csv: bool,

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

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

    /// Try to enrich missing VAT identifiers by querying Bizzy (NL only)
    #[arg(long, value_enum, default_value_t = VatCli::None)]
    vat_id_search_engine: VatCli,

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

    /// Override `proxy_url` for this run (audit label `cli_override`)
    #[arg(long)]
    proxy_url: Option<String>,
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    dotenv::dotenv().ok();
    dexter_sentinel::scrape::log::init_scrape_cli_subscriber();

    let cli = Cli::parse();

    let cc = cli.country_code.trim().to_uppercase();
    if cc.len() != 2 || !cc.chars().all(|c| c.is_ascii_alphabetic()) {
        anyhow::bail!("--country-code must be ISO 3166-1 alpha-2 (e.g. NL)");
    }

    let engine = match cli.search_engine {
        Some(SearchEngine::Kvk) => RegistryEngine::Kvk,
        Some(SearchEngine::Northdata) => RegistryEngine::Northdata,
        Some(SearchEngine::Egrul) => RegistryEngine::Egrul,
        Some(SearchEngine::RegistriesIo) => RegistryEngine::RegistriesIo,
        None => {
            if cc == "NL" {
                RegistryEngine::Kvk
            } else {
                RegistryEngine::Northdata
            }
        }
    };

    if matches!(engine, RegistryEngine::Kvk) && cc != "NL" {
        anyhow::bail!("KVK search requires --country-code NL (or omit --search-engine with NL)");
    }
    if matches!(engine, RegistryEngine::Egrul) && cc != "RU" {
        anyhow::bail!("--search-engine egrul requires --country-code RU");
    }
    if matches!(engine, RegistryEngine::RegistriesIo) && cc != "PL" && cc != "CZ" {
        anyhow::bail!(
            "--search-engine registries-io requires --country-code PL or CZ"
        );
    }

    let vat: VatIdSearchEngine = match cli.vat_id_search_engine {
        VatCli::None => VatIdSearchEngine::None,
        VatCli::Bizzy => VatIdSearchEngine::Bizzy,
    };
    if matches!(vat, VatIdSearchEngine::Bizzy) {
        if cc != "NL" {
            anyhow::bail!("--vat-id-search-engine bizzy requires --country-code NL");
        }
        if !matches!(engine, RegistryEngine::Kvk) {
            anyhow::bail!("--vat-id-search-engine bizzy requires --search-engine kvk");
        }
    }

    let insert: bool = !cli.no_insert;
    let (dexter, http) = if insert {
        let athena_key: String = 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: String = cli.athena_url.trim_end_matches('/').to_string();
        let sdk: Arc<AthenaClient> = Arc::new(
            AthenaClient::new(&base_trim, &athena_key, &cli.athena_client).await?,
        );
        let audit: AuditBus = dexter_sentinel::audit::spawn_audit_flusher(
            Arc::clone(&sdk),
            None,
            8192,
            100,
            Duration::from_secs(2),
        );
        let prof: Option<dexter_sentinel::http_profile::HttpClientProfile> = 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: Option<String> = 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,
            athena_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_global().await;
        });
    }

    let mut northdata_cookie = std::env::var("NORTHDATA_COOKIE").ok();
    if let Some(ref path) = cli.northdata_cookie_file {
        northdata_cookie = Some(std::fs::read_to_string(path)?.trim().to_string());
    }

    let words_file = cli
        .randomize_words_file
        .unwrap_or_else(|| PathBuf::from("scripts/words.txt"));

    let cfg = GlobalRegistryPipelineConfig {
        engine,
        country_code: cc.clone(),
        output_csv: cli.output.clone(),
        kvk_api_url: cli.kvk_api_url.clone(),
        northdata_base_url: cli.northdata_base_url.clone(),
        egrul_base_url: cli.egrul_base_url.clone(),
        registries_io_base_url: cli.registries_io_base_url.clone(),
        registries_io_page_limit: cli.registries_io_page_limit,
        northdata_cookie,
        min_query_len: cli.min_query_length,
        max_query_len: cli.max_query_length,
        query_seed: cli.query.clone().unwrap_or_default(),
        words_file,
        seed_csv_path: cli.randomize_seed_csv.clone(),
        randomize_no_seed_csv: cli.randomize_no_seed_csv,
        insert,
        max_workers: cli.max_workers,
        vat_id_search_engine: vat,
    };

    if cli.randomize_query {
        tracing::info!(
            "dexter-scrape: randomized loop ({:?}) → global_companies_registry",
            engine
        );
        run_randomized_registry_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!("dexter-scrape: single query ({:?})", engine);
        run_single_registry_pass(&http, dexter, &cfg, q, cli.append_csv).await
    }
}
