//! Maps Northdata / KVK / NL / EGRUL / registries.io rows to `global_companies_registry` and
//! country tables (`ee_companies`, `ru_companies`, `pl_companies`, `cz_companies`).
//! Mirrors `scripts/scrape_global_registry.py` and NL `map_to_global_company`.

use std::collections::HashMap;

use once_cell::sync::Lazy;
use percent_encoding::{percent_decode_str, utf8_percent_encode, NON_ALPHANUMERIC};
use regex::Regex;
use serde_json::{Map, Number, Value};
use sha2::{Digest, Sha256};
use url::Url;

/// Columns allowed on `public.global_companies_registry` (Python `GLOBAL_REGISTRY_KEYS`).
pub static GLOBAL_REGISTRY_KEYS: &[&str] = &[
    "name",
    "company_number",
    "rights_form",
    "country_code",
    "status",
    "last_update",
    "description",
    "region",
    "rights_form_code",
    "state",
    "name_english",
    "language",
    "hash",
    "address",
    "global_company_id",
    "start_date",
    "legal_persons",
    "contact_data",
    "activities",
    "activity_code_table",
    "establishment_date",
    "establishment_type",
    "entity_type",
    "vat_id",
    "email",
    "iban",
    "phone",
    "postal_code",
    "house_number",
    "province",
    "city",
    "street",
    "country",
    "website",
];

static GLOBAL_KEY_SET: Lazy<std::collections::HashSet<&'static str>> =
    Lazy::new(|| GLOBAL_REGISTRY_KEYS.iter().copied().collect());

static EE_COMPANIES_KEYS: &[&str] = &[
    "registrikood",
    "name",
    "legal_form",
    "status",
    "location",
    "vat_id",
    "register_reference",
    "source_url",
];

static EE_KEY_SET: Lazy<std::collections::HashSet<&'static str>> =
    Lazy::new(|| EE_COMPANIES_KEYS.iter().copied().collect());

static RU_COMPANIES_KEYS: &[&str] = &[
    "inn",
    "ogrn",
    "kpp",
    "name",
    "name_short",
    "legal_form",
    "status",
    "address",
    "reg_date",
    "source_url",
];

static RU_KEY_SET: Lazy<std::collections::HashSet<&'static str>> =
    Lazy::new(|| RU_COMPANIES_KEYS.iter().copied().collect());

/// Columns allowed on `cz_companies` (registries.io search rows).
static CZ_COMPANY_TABLE_KEYS: &[&str] = &[
    "registration_number",
    "name",
    "legal_form",
    "status",
    "address",
    "registration_date",
    "slug",
    "vat_id",
    "extra_registration_numbers",
    "source_url",
];

static CZ_COMPANY_KEY_SET: Lazy<std::collections::HashSet<&'static str>> =
    Lazy::new(|| CZ_COMPANY_TABLE_KEYS.iter().copied().collect());

/// Columns for `pl_companies` including registries.io profile (`source_document`, denormalized sections).
static PL_COMPANY_TABLE_KEYS: &[&str] = &[
    "registration_number",
    "name",
    "legal_form",
    "status",
    "address",
    "registration_date",
    "slug",
    "vat_id",
    "extra_registration_numbers",
    "source_url",
    "registry_uuid",
    "source_document",
    "registered_name",
    "registration_numbers",
    "share_capital_value",
    "share_capital_currency",
    "websites",
    "addresses",
    "activities",
    "authorized_body_name",
    "authorized_body_entities",
    "shareholders_entities",
];

static PL_COMPANY_KEY_SET: Lazy<std::collections::HashSet<&'static str>> =
    Lazy::new(|| PL_COMPANY_TABLE_KEYS.iter().copied().collect());

fn filter_company_row_by_key_set(
    d: &Map<String, Value>,
    set: &std::collections::HashSet<&'static str>,
) -> Map<String, Value> {
    let mut out = Map::new();
    for (k, v) in d {
        if v.is_null() {
            continue;
        }
        if set.contains(k.as_str()) {
            out.insert(k.clone(), v.clone());
        }
    }
    out
}

static RE_EE_SEGMENT: Lazy<Regex> = Lazy::new(|| Regex::new(r"(?i)^RIK\s*(\d{7,9})\s*$").unwrap());
static RE_EE_DIGITS: Lazy<Regex> = Lazy::new(|| Regex::new(r"^\d{7,9}$").unwrap());

pub fn filter_global_payload(d: &Map<String, Value>) -> Map<String, Value> {
    let mut out = Map::new();
    for (k, v) in d {
        if v.is_null() {
            continue;
        }
        if GLOBAL_KEY_SET.contains(k.as_str()) {
            out.insert(k.clone(), v.clone());
        }
    }
    out.remove("global_company_id");
    out
}

pub fn filter_ee_payload(d: &Map<String, Value>) -> Map<String, Value> {
    let mut out = Map::new();
    for (k, v) in d {
        if v.is_null() {
            continue;
        }
        if EE_KEY_SET.contains(k.as_str()) {
            out.insert(k.clone(), v.clone());
        }
    }
    out
}

pub fn filter_ru_payload(d: &Map<String, Value>) -> Map<String, Value> {
    let mut out = Map::new();
    for (k, v) in d {
        if v.is_null() {
            continue;
        }
        if RU_KEY_SET.contains(k.as_str()) {
            out.insert(k.clone(), v.clone());
        }
    }
    out
}

pub fn filter_pl_company_payload(d: &Map<String, Value>) -> Map<String, Value> {
    filter_company_row_by_key_set(d, &PL_COMPANY_KEY_SET)
}

pub fn filter_cz_company_payload(d: &Map<String, Value>) -> Map<String, Value> {
    filter_company_row_by_key_set(d, &CZ_COMPANY_KEY_SET)
}

fn egrul_str(m: &Map<String, Value>, key: &str) -> Option<String> {
    m.get(key)
        .and_then(|v| v.as_str())
        .map(|s| s.trim().to_string())
        .filter(|s| !s.is_empty())
}

/// Map one EGRUL `search-result` row (FNS short keys) to `global_companies_registry` fields.
pub fn map_egrul_api_row_to_global(hit: &Map<String, Value>) -> Option<Map<String, Value>> {
    let ogrn = egrul_str(hit, "o");
    let inn = egrul_str(hit, "i");
    let company_number = ogrn
        .clone()
        .or_else(|| inn.clone())
        .filter(|s| !s.is_empty())?;
    let name_full = egrul_str(hit, "n").or_else(|| egrul_str(hit, "c"))?;
    let name_short = egrul_str(hit, "c");
    if name_full.is_empty() {
        return None;
    }
    let cc = "RU".to_string();
    let hash_input = format!("{}|{}|{}", name_full, company_number, cc);
    let hash_val = format!("{:x}", Sha256::digest(hash_input.as_bytes()));

    let addr = egrul_str(hit, "g");
    let reg_date = egrul_str(hit, "r");
    let kpp = egrul_str(hit, "p");
    let kind = egrul_str(hit, "k");
    let rights_form = match kind.as_deref() {
        Some("ul") => Some(Value::String("ЮЛ".to_string())),
        Some("ip") => Some(Value::String("ИП".to_string())),
        Some(x) if !x.is_empty() => Some(Value::String(x.to_string())),
        _ => None,
    };

    let description = addr
        .as_ref()
        .map(|s| Value::String(s.clone()))
        .unwrap_or(Value::Null);

    let contact_data = kpp
        .as_ref()
        .map(|k| Value::String(format!("KPP {k}")))
        .unwrap_or(Value::Null);

    let mut global_row = Map::new();
    global_row.insert("name".to_string(), Value::String(name_full.clone()));
    global_row.insert(
        "company_number".to_string(),
        Value::String(company_number.clone()),
    );
    global_row.insert(
        "rights_form".to_string(),
        rights_form.clone().unwrap_or(Value::Null),
    );
    global_row.insert("country_code".to_string(), Value::String(cc));
    global_row.insert("status".to_string(), Value::Null);
    global_row.insert("last_update".to_string(), Value::Null);
    global_row.insert("description".to_string(), description.clone());
    global_row.insert("region".to_string(), egrul_str(hit, "rn").map(Value::String).unwrap_or(Value::Null));
    global_row.insert("rights_form_code".to_string(), kind.map(Value::String).unwrap_or(Value::Null));
    global_row.insert("state".to_string(), Value::Null);
    global_row.insert("name_english".to_string(), name_short.map(Value::String).unwrap_or(Value::Null));
    global_row.insert("language".to_string(), Value::String("ru".to_string()));
    global_row.insert("hash".to_string(), Value::String(hash_val));
    global_row.insert("address".to_string(), description);
    global_row.insert("start_date".to_string(), Value::Null);
    global_row.insert("legal_persons".to_string(), Value::Null);
    global_row.insert("contact_data".to_string(), contact_data);
    global_row.insert("activities".to_string(), Value::Null);
    if let Some(cnt) = egrul_str(hit, "cnt") {
        global_row.insert("activity_code_table".to_string(), Value::String(cnt));
    } else {
        global_row.insert("activity_code_table".to_string(), Value::Null);
    }
    global_row.insert(
        "establishment_date".to_string(),
        reg_date.map(Value::String).unwrap_or(Value::Null),
    );
    global_row.insert("establishment_type".to_string(), Value::Null);
    global_row.insert("entity_type".to_string(), Value::Null);
    global_row.insert(
        "vat_id".to_string(),
        inn.as_ref()
            .map(|i| Value::String(i.clone()))
            .unwrap_or(Value::Null),
    );
    global_row.insert("email".to_string(), Value::Null);
    global_row.insert("iban".to_string(), Value::Null);
    global_row.insert("phone".to_string(), Value::Null);
    global_row.insert("postal_code".to_string(), Value::Null);
    global_row.insert("house_number".to_string(), Value::Null);
    global_row.insert("province".to_string(), Value::Null);
    global_row.insert("city".to_string(), Value::Null);
    global_row.insert("street".to_string(), Value::Null);
    global_row.insert(
        "country".to_string(),
        Value::String("Россия".to_string()),
    );
    let src = "https://egrul.nalog.ru/";
    global_row.insert("website".to_string(), Value::String(src.to_string()));

    Some(filter_global_payload(&global_row))
}

fn ru_kpp_from_contact_data(row: &Map<String, Value>) -> Option<String> {
    let s = row.get("contact_data")?.as_str()?;
    s.strip_prefix("KPP ")
        .map(|rest| rest.trim().to_string())
        .filter(|k| !k.is_empty())
}

pub fn map_global_row_to_ru_company(row: &Map<String, Value>) -> Option<Map<String, Value>> {
    let cc = row
        .get("country_code")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .to_uppercase();
    if cc != "RU" {
        return None;
    }
    let cn = row
        .get("company_number")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .trim();
    if cn.is_empty() {
        return None;
    }
    let inn_raw = row.get("vat_id").and_then(|v| v.as_str()).unwrap_or("").trim();
    let all_digit = |s: &str| s.chars().all(|c| c.is_ascii_digit());
    let (ogrn_opt, inn_final) = if all_digit(cn) && (cn.len() == 13 || cn.len() == 15) {
        (Some(cn.to_string()), inn_raw.to_string())
    } else if all_digit(cn) && (cn.len() == 10 || cn.len() == 12) {
        (None, cn.to_string())
    } else {
        (Some(cn.to_string()), inn_raw.to_string())
    };

    let has_ogrn = ogrn_opt.as_ref().is_some_and(|s| !s.is_empty());
    let has_inn = !inn_final.is_empty();
    if !has_ogrn && !has_inn {
        return None;
    }

    let name = row
        .get("name")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .trim();
    if name.is_empty() {
        return None;
    }

    let mut ru: Map<String, Value> = Map::new();
    if !inn_final.is_empty() {
        ru.insert("inn".to_string(), Value::String(inn_final));
    }
    if let Some(ref o) = ogrn_opt {
        if !o.is_empty() {
            ru.insert("ogrn".to_string(), Value::String(o.clone()));
        }
    }
    if let Some(k) = ru_kpp_from_contact_data(row) {
        ru.insert("kpp".to_string(), Value::String(k));
    }
    ru.insert("name".to_string(), Value::String(name.to_string()));
    if let Some(ns) = row
        .get("name_english")
        .and_then(|v| v.as_str())
        .filter(|s| !s.is_empty())
    {
        ru.insert("name_short".to_string(), Value::String(ns.to_string()));
    }
    if let Some(lf) = row
        .get("rights_form")
        .and_then(|v| v.as_str())
        .filter(|s| !s.is_empty())
    {
        ru.insert("legal_form".to_string(), Value::String(lf.to_string()));
    }
    if let Some(v) = row.get("status").cloned() {
        ru.insert("status".to_string(), v);
    }
    if let Some(a) = row
        .get("address")
        .or_else(|| row.get("description"))
        .and_then(|v| v.as_str())
        .filter(|s| !s.is_empty())
    {
        ru.insert("address".to_string(), Value::String(a.to_string()));
    }
    if let Some(d) = row
        .get("establishment_date")
        .and_then(|v| v.as_str())
        .filter(|s| !s.is_empty())
    {
        ru.insert("reg_date".to_string(), Value::String(d.to_string()));
    }
    ru.insert(
        "source_url".to_string(),
        Value::String("https://egrul.nalog.ru/".to_string()),
    );

    Some(filter_ru_payload(&ru))
}

fn map_global_row_to_plcz_company(row: &Map<String, Value>, want_cc: &str) -> Option<Map<String, Value>> {
    let cc = row
        .get("country_code")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .to_uppercase();
    if cc.as_str() != want_cc {
        return None;
    }
    let registration_number = row
        .get("company_number")
        .and_then(|v| v.as_str())
        .map(str::trim)
        .filter(|s| !s.is_empty())?
        .to_string();
    let name = row
        .get("name")
        .and_then(|v| v.as_str())
        .map(str::trim)
        .filter(|s| !s.is_empty())?
        .to_string();

    let mut plcz: Map<String, Value> = Map::new();
    plcz.insert("registration_number".to_string(), Value::String(registration_number));
    plcz.insert("name".to_string(), Value::String(name));
    if let Some(s) = row
        .get("rights_form")
        .and_then(|v| v.as_str())
        .filter(|s| !s.is_empty())
    {
        plcz.insert("legal_form".to_string(), Value::String(s.to_string()));
    }
    if let Some(v) = row.get("status").cloned() {
        plcz.insert("status".to_string(), v);
    }
    if let Some(s) = row
        .get("address")
        .and_then(|v| v.as_str())
        .filter(|s| !s.is_empty())
    {
        plcz.insert("address".to_string(), Value::String(s.to_string()));
    }
    let reg_date = row
        .get("establishment_date")
        .or_else(|| row.get("start_date"))
        .and_then(|v| v.as_str())
        .filter(|s| !s.is_empty());
    if let Some(d) = reg_date {
        plcz.insert("registration_date".to_string(), Value::String(d.to_string()));
    }
    if let Some(s) = row
        .get("registries_slug")
        .and_then(|v| v.as_str())
        .filter(|s| !s.is_empty())
    {
        plcz.insert("slug".to_string(), Value::String(s.to_string()));
    }
    if let Some(v) = row.get("vat_id").cloned() {
        if !v.is_null() {
            plcz.insert("vat_id".to_string(), v);
        }
    }
    if let Some(v) = row.get("registries_extra_registration_numbers").cloned() {
        if !v.is_null() {
            plcz.insert("extra_registration_numbers".to_string(), v);
        }
    }
    if let Some(s) = row
        .get("website")
        .and_then(|v| v.as_str())
        .filter(|s| !s.is_empty())
    {
        plcz.insert("source_url".to_string(), Value::String(s.to_string()));
    }

    Some(plcz)
}

/// Map a global registry row to `pl_companies` when `country_code` is PL (registries.io, etc.).
pub fn map_global_row_to_pl_company(row: &Map<String, Value>) -> Option<Map<String, Value>> {
    let mut pl = map_global_row_to_plcz_company(row, "PL")?;
    if let Some(Value::Object(ref doc)) = row.get("registries_pl_source_document") {
        enrich_pl_row_from_registries_pl_detail(&mut pl, doc);
    }
    Some(filter_pl_company_payload(&pl))
}

/// Map a global registry row to `cz_companies` when `country_code` is CZ.
pub fn map_global_row_to_cz_company(row: &Map<String, Value>) -> Option<Map<String, Value>> {
    let plcz = map_global_row_to_plcz_company(row, "CZ")?;
    Some(filter_cz_company_payload(&plcz))
}

/// Normalize KRS / numeric registry ids (strip leading zeros; trim).
pub(crate) fn normalize_pl_registration_number(s: &str) -> String {
    let t = s.trim();
    if t.is_empty() {
        return String::new();
    }
    if t.chars().all(|c| c.is_ascii_digit()) {
        let stripped = t.trim_start_matches('0');
        if stripped.is_empty() {
            "0".to_string()
        } else {
            stripped.to_string()
        }
    } else {
        t.to_string()
    }
}

fn pl_detail_sections_array<'a>(detail: &'a Map<String, Value>) -> Option<&'a Vec<Value>> {
    detail.get("sections").and_then(|v| v.as_array())
}

fn pl_detail_basic_payload<'a>(detail: &'a Map<String, Value>) -> Option<&'a Map<String, Value>> {
    let arr = pl_detail_sections_array(detail)?;
    for s in arr {
        let o = s.as_object()?;
        let id_ok = o.get("id").and_then(|v| v.as_str()) == Some("pl-basic-info");
        let struct_ok = o.get("structure").and_then(|v| v.as_str()) == Some("BASIC_INFO");
        if id_ok || struct_ok {
            return o.get("payload").and_then(|v| v.as_object());
        }
    }
    None
}

fn pl_detail_section_payload_by_id<'a>(
    detail: &'a Map<String, Value>,
    section_id: &str,
) -> Option<&'a Map<String, Value>> {
    let arr = pl_detail_sections_array(detail)?;
    for s in arr {
        let o = s.as_object()?;
        if o.get("id").and_then(|v| v.as_str()) == Some(section_id) {
            return o.get("payload").and_then(|v| v.as_object());
        }
    }
    None
}

fn pl_detail_primary_registration_number(detail: &Map<String, Value>) -> Option<String> {
    if let Some(s) = detail
        .get("registrationNumber")
        .and_then(|v| v.as_str())
        .map(str::trim)
        .filter(|s| !s.is_empty())
    {
        let n = normalize_pl_registration_number(s);
        if !n.is_empty() {
            return Some(n);
        }
    }
    let basic = pl_detail_basic_payload(detail)?;
    let nums = basic.get("registrationNumbers").and_then(|v| v.as_array())?;
    let mut primary_val: Option<String> = None;
    let mut first_val: Option<String> = None;
    for n in nums {
        let o = n.as_object()?;
        let val = o.get("value").and_then(|v| v.as_str()).map(str::trim).filter(|s| !s.is_empty())?;
        if first_val.is_none() {
            first_val = Some(normalize_pl_registration_number(val));
        }
        let is_primary = o.get("isPrimary").and_then(|v| v.as_bool()) == Some(true);
        let label = o.get("label").and_then(|v| v.as_str()).unwrap_or("").to_lowercase();
        if is_primary || label.contains("krs") {
            primary_val = Some(normalize_pl_registration_number(val));
            break;
        }
    }
    primary_val.or(first_val).filter(|s| !s.is_empty())
}

fn pl_detail_flat_reg_values_from_basic(basic: &Map<String, Value>) -> Vec<String> {
    let Some(nums) = basic.get("registrationNumbers").and_then(|v| v.as_array()) else {
        return vec![];
    };
    nums.iter()
        .filter_map(|n| {
            n.as_object()?
                .get("value")
                .and_then(|v| v.as_str())
                .map(str::trim)
                .filter(|s| !s.is_empty())
                .map(|s| s.to_string())
        })
        .collect()
}

fn pl_detail_address_line(detail: &Map<String, Value>) -> Option<String> {
    let pay = pl_detail_section_payload_by_id(detail, "pl-addresses")?;
    let arr = pay.get("addresses").and_then(|v| v.as_array())?;
    let first = arr.first()?.as_object()?;
    if let Some(line) = first
        .get("addressInOneLine")
        .and_then(|v| v.as_str())
        .map(str::trim)
        .filter(|s| !s.is_empty())
    {
        return Some(line.to_string());
    }
    None
}

fn pl_detail_postal_city_street(detail: &Map<String, Value>) -> (Option<String>, Option<String>, Option<String>) {
    let pay = pl_detail_section_payload_by_id(detail, "pl-addresses");
    let Some(pay) = pay else {
        return (None, None, None);
    };
    let Some(arr) = pay.get("addresses").and_then(|v| v.as_array()) else {
        return (None, None, None);
    };
    let Some(first) = arr.first().and_then(|v| v.as_object()) else {
        return (None, None, None);
    };
    let pc = first
        .get("postalCode")
        .and_then(|v| v.as_str())
        .map(str::to_string);
    let city = first
        .get("addressLocality")
        .and_then(|v| v.as_str())
        .map(str::to_string);
    let street = first
        .get("streetName")
        .and_then(|v| v.as_str())
        .map(str::to_string);
    (pc, city, street)
}

fn pl_detail_activities_text(detail: &Map<String, Value>) -> Option<String> {
    let pay = pl_detail_section_payload_by_id(detail, "pl-activities")?;
    let arr = pay.get("activities").and_then(|v| v.as_array())?;
    let parts: Vec<String> = arr
        .iter()
        .filter_map(|a| {
            a.as_object()?
                .get("value")
                .and_then(|v| v.as_str())
                .map(str::trim)
                .filter(|s| !s.is_empty())
                .map(|s| s.to_string())
        })
        .collect();
    if parts.is_empty() {
        None
    } else {
        Some(parts.join("; "))
    }
}

fn pl_detail_vat_from_objects(detail: &Map<String, Value>, primary: &str) -> Option<String> {
    let basic = pl_detail_basic_payload(detail)?;
    let nums = basic.get("registrationNumbers").and_then(|v| v.as_array())?;
    let primary_n = normalize_pl_registration_number(primary);
    let is_digits = |s: &str| s.chars().all(|c| c.is_ascii_digit());
    for n in nums {
        let o = n.as_object()?;
        let val = o.get("value").and_then(|v| v.as_str()).map(str::trim).filter(|s| !s.is_empty())?;
        let label = o.get("label").and_then(|v| v.as_str()).unwrap_or("").to_lowercase();
        if normalize_pl_registration_number(val) == primary_n {
            continue;
        }
        if label.contains("nip") || (val.len() == 10 && is_digits(val)) {
            return Some(val.to_string());
        }
    }
    None
}

fn pl_detail_website(detail: &Map<String, Value>) -> Option<String> {
    let basic = pl_detail_basic_payload(detail)?;
    let arr = basic.get("websites").and_then(|v| v.as_array())?;
    arr.iter()
        .filter_map(|w| w.as_str().map(str::trim).filter(|s| !s.is_empty()))
        .next()
        .map(str::to_string)
}

fn pl_detail_legal_persons_summary(detail: &Map<String, Value>) -> Option<String> {
    let pay = pl_detail_section_payload_by_id(detail, "pl-authorized-body")?;
    let arr = pay.get("entities").and_then(|v| v.as_array())?;
    let names: Vec<String> = arr
        .iter()
        .filter_map(|e| {
            e.as_object()?
                .get("name")
                .and_then(|v| v.as_str())
                .map(str::trim)
                .filter(|s| !s.is_empty())
                .map(|s| s.to_string())
        })
        .collect();
    if names.is_empty() {
        None
    } else {
        Some(names.join(", "))
    }
}

fn enrich_pl_row_from_registries_pl_detail(pl: &mut Map<String, Value>, detail: &Map<String, Value>) {
    pl.insert(
        "source_document".to_string(),
        Value::Object(detail.clone()),
    );
    if let Some(u) = detail.get("id").and_then(|v| v.as_str()).filter(|s| !s.is_empty()) {
        pl.insert("registry_uuid".to_string(), Value::String(u.to_string()));
    }
    if let Some(basic) = pl_detail_basic_payload(detail) {
        if let Some(n) = basic
            .get("registeredName")
            .and_then(|v| v.as_str())
            .map(str::trim)
            .filter(|s| !s.is_empty())
        {
            pl.insert("registered_name".to_string(), Value::String(n.to_string()));
        }
        if let Some(arr) = basic.get("registrationNumbers") {
            pl.insert("registration_numbers".to_string(), arr.clone());
        }
        if let Some(cap) = basic.get("issuedShareCapital").and_then(|v| v.as_object()) {
            if let Some(v) = cap.get("value") {
                pl.insert("share_capital_value".to_string(), v.clone());
            }
            if let Some(cur) = cap.get("currency").and_then(|v| v.as_str()) {
                pl.insert(
                    "share_capital_currency".to_string(),
                    Value::String(cur.to_string()),
                );
            }
        }
        if let Some(w) = basic.get("websites") {
            pl.insert("websites".to_string(), w.clone());
        }
    }
    if let Some(pay) = pl_detail_section_payload_by_id(detail, "pl-addresses") {
        if let Some(arr) = pay.get("addresses").and_then(|v| v.as_array()) {
            pl.insert("addresses".to_string(), Value::Array(arr.clone()));
        }
    }
    if let Some(pay) = pl_detail_section_payload_by_id(detail, "pl-activities") {
        if let Some(a) = pay.get("activities") {
            pl.insert("activities".to_string(), a.clone());
        }
    }
    if let Some(pay) = pl_detail_section_payload_by_id(detail, "pl-authorized-body") {
        if let Some(n) = pay.get("name").and_then(|v| v.as_str()) {
            pl.insert(
                "authorized_body_name".to_string(),
                Value::String(n.to_string()),
            );
        }
        if let Some(e) = pay.get("entities") {
            pl.insert("authorized_body_entities".to_string(), e.clone());
        }
    }
    if let Some(pay) = pl_detail_section_payload_by_id(detail, "pl-shareholders") {
        if let Some(e) = pay.get("entities") {
            pl.insert("shareholders_entities".to_string(), e.clone());
        }
    }
}

/// Map registries.io **company detail** JSON (with `sections`) to `global_companies_registry` row shape.
/// Sets `registries_pl_source_document` for `pl_companies` enrichment (stripped on global insert).
pub fn map_registries_io_pl_detail_to_global(
    detail: &Map<String, Value>,
    cli_country: &str,
    public_site_base: &str,
) -> Option<Map<String, Value>> {
    let cc = cli_country.trim().to_uppercase();
    if cc != "PL" {
        return None;
    }
    let name = detail
        .get("name")
        .and_then(|v| v.as_str())
        .map(str::trim)
        .filter(|s| !s.is_empty())?;
    let primary = pl_detail_primary_registration_number(detail)?;
    let slug = detail
        .get("slug")
        .and_then(|v| v.as_str())
        .map(str::trim)
        .filter(|s| !s.is_empty());
    let basic = pl_detail_basic_payload(detail);

    let legal_form = basic
        .and_then(|b| b.get("legalForm").or_else(|| b.get("legal_form")))
        .and_then(|v| v.as_str())
        .map(str::trim)
        .filter(|s| !s.is_empty())
        .map(|s| s.to_string());

    let status = basic
        .and_then(|b| b.get("status"))
        .and_then(|v| v.as_str())
        .map(str::trim)
        .filter(|s| !s.is_empty())
        .map(|s| Value::String(s.to_string()))
        .unwrap_or(Value::Null);

    let reg_date = basic
        .and_then(|b| b.get("registrationDate").or_else(|| b.get("registration_date")))
        .and_then(|v| v.as_str())
        .map(str::trim)
        .filter(|s| !s.is_empty())
        .map(|s| s.to_string());

    let address_line = pl_detail_address_line(detail)
        .or_else(|| {
            detail
                .get("address")
                .and_then(|v| v.as_str())
                .map(str::trim)
                .filter(|s| !s.is_empty())
                .map(|s| s.to_string())
        })
        .map(Value::String)
        .unwrap_or(Value::Null);

    let (postal, city, street) = pl_detail_postal_city_street(detail);
    let postal_val = postal.map(Value::String).unwrap_or(Value::Null);
    let city_val = city.map(Value::String).unwrap_or(Value::Null);
    let street_val = street.map(Value::String).unwrap_or(Value::Null);

    let flat_regs: Vec<String> = basic
        .map(pl_detail_flat_reg_values_from_basic)
        .unwrap_or_default();
    let vat = pl_detail_vat_from_objects(detail, &primary)
        .or_else(|| infer_registries_vat_id(&flat_regs, &primary, "PL"))
        .map(Value::String)
        .unwrap_or(Value::Null);

    let activities_txt = pl_detail_activities_text(detail);
    let activities_val = activities_txt
        .as_ref()
        .map(|s| Value::String(s.clone()))
        .unwrap_or(Value::Null);

    let description = basic
        .and_then(|b| b.get("issuedShareCapital").and_then(|v| v.as_object()))
        .and_then(|cap| {
            let val = cap.get("value").cloned().unwrap_or(Value::Null);
            let cur = cap
                .get("currency")
                .and_then(|v| v.as_str())
                .unwrap_or("");
            if val.is_null() {
                None
            } else {
                Some(Value::String(format!("{val} {cur}").trim().to_string()))
            }
        })
        .unwrap_or(Value::Null);

    let legal_persons = pl_detail_legal_persons_summary(detail)
        .map(Value::String)
        .unwrap_or(Value::Null);

    let website_str = pl_detail_website(detail).unwrap_or_else(|| {
        registries_public_company_url(
            public_site_base,
            "pl",
            slug,
            primary.as_str(),
        )
    });

    let hash_input = format!("{}|{}|{}", name, primary, cc);
    let hash_val = format!("{:x}", Sha256::digest(hash_input.as_bytes()));

    let mut global_row = Map::new();
    global_row.insert("name".to_string(), Value::String(name.to_string()));
    global_row.insert(
        "company_number".to_string(),
        Value::String(primary.clone()),
    );
    global_row.insert(
        "rights_form".to_string(),
        legal_form.map(Value::String).unwrap_or(Value::Null),
    );
    global_row.insert("country_code".to_string(), Value::String(cc.clone()));
    global_row.insert("status".to_string(), status);
    global_row.insert("last_update".to_string(), Value::Null);
    global_row.insert("description".to_string(), description);
    global_row.insert("region".to_string(), Value::Null);
    global_row.insert("rights_form_code".to_string(), Value::Null);
    global_row.insert("state".to_string(), Value::Null);
    global_row.insert("name_english".to_string(), Value::Null);
    global_row.insert(
        "language".to_string(),
        Value::String("pl".to_string()),
    );
    global_row.insert("hash".to_string(), Value::String(hash_val));
    global_row.insert("address".to_string(), address_line.clone());
    global_row.insert("start_date".to_string(), Value::Null);
    global_row.insert("legal_persons".to_string(), legal_persons);
    global_row.insert("contact_data".to_string(), Value::Null);
    global_row.insert("activities".to_string(), activities_val);
    global_row.insert("activity_code_table".to_string(), Value::Null);
    global_row.insert(
        "establishment_date".to_string(),
        reg_date.map(Value::String).unwrap_or(Value::Null),
    );
    global_row.insert("establishment_type".to_string(), Value::Null);
    global_row.insert("entity_type".to_string(), Value::Null);
    global_row.insert("vat_id".to_string(), vat);
    global_row.insert("email".to_string(), Value::Null);
    global_row.insert("iban".to_string(), Value::Null);
    global_row.insert("phone".to_string(), Value::Null);
    global_row.insert("postal_code".to_string(), postal_val);
    global_row.insert("house_number".to_string(), Value::Null);
    global_row.insert("province".to_string(), Value::Null);
    global_row.insert("city".to_string(), city_val);
    global_row.insert("street".to_string(), street_val);
    global_row.insert("country".to_string(), Value::Null);
    global_row.insert(
        "website".to_string(),
        Value::String(website_str),
    );

    let mut filtered = filter_global_payload(&global_row);
    if let Some(s) = slug {
        filtered.insert("registries_slug".to_string(), Value::String(s.to_string()));
    }
    if !flat_regs.is_empty() {
        filtered.insert(
            "registries_extra_registration_numbers".to_string(),
            Value::Array(
                flat_regs
                    .iter()
                    .map(|s| Value::String(s.clone()))
                    .collect(),
            ),
        );
    }
    filtered.insert(
        "registries_pl_source_document".to_string(),
        Value::Object(detail.clone()),
    );
    Some(filtered)
}

pub fn parse_ee_registrikood_from_segment(segment: &str) -> Option<String> {
    let s = segment.trim();
    if s.is_empty() {
        return None;
    }
    if let Some(c) = RE_EE_SEGMENT.captures(s) {
        return Some(c.get(1)?.as_str().to_string());
    }
    if RE_EE_DIGITS.is_match(s) {
        return Some(s.to_string());
    }
    None
}

static RE_EE_OU: Lazy<Regex> = Lazy::new(|| Regex::new(r"(?i)\bOÜ\b|Osaühing").unwrap());
static RE_EE_AS: Lazy<Regex> = Lazy::new(|| Regex::new(r"(?i)\bAS\b|Aktsiaselts").unwrap());
static RE_EE_FIE: Lazy<Regex> = Lazy::new(|| Regex::new(r"(?i)\bFIE\b").unwrap());
static RE_EE_MTU: Lazy<Regex> = Lazy::new(|| Regex::new(r"(?i)\bMTÜ\b").unwrap());
static RE_EE_UU: Lazy<Regex> = Lazy::new(|| Regex::new(r"(?i)\bUÜ\b").unwrap());
static RE_EE_TU: Lazy<Regex> = Lazy::new(|| Regex::new(r"(?i)\bTÜ\b|Täisühing").unwrap());

pub fn infer_ee_legal_form(name: &str) -> Option<String> {
    if name.is_empty() {
        return None;
    }
    if RE_EE_OU.is_match(name) {
        return Some("OÜ".to_string());
    }
    if RE_EE_AS.is_match(name) {
        return Some("AS".to_string());
    }
    if RE_EE_FIE.is_match(name) {
        return Some("FIE".to_string());
    }
    if RE_EE_MTU.is_match(name) {
        return Some("MTÜ".to_string());
    }
    if RE_EE_UU.is_match(name) {
        return Some("UÜ".to_string());
    }
    if RE_EE_TU.is_match(name) {
        return Some("TÜ".to_string());
    }
    None
}

pub fn map_global_row_to_ee_company(row: &Map<String, Value>) -> Option<Map<String, Value>> {
    let cc = row
        .get("country_code")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .to_uppercase();
    if cc != "EE" {
        return None;
    }
    let segment = row
        .get("company_number")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .trim()
        .to_string();
    let rk = parse_ee_registrikood_from_segment(&segment)?;
    let name = row
        .get("name")
        .and_then(|v| v.as_str())
        .unwrap_or("")
        .trim();
    if name.is_empty() {
        return None;
    }
    let mut ee: Map<String, Value> = Map::new();
    ee.insert("registrikood".to_string(), Value::String(rk));
    ee.insert("name".to_string(), Value::String(name.to_string()));
    if let Some(lf) = infer_ee_legal_form(name) {
        ee.insert("legal_form".to_string(), Value::String(lf));
    }
    if let Some(v) = row.get("status").cloned() {
        ee.insert("status".to_string(), v);
    }
    let location = row
        .get("description")
        .or_else(|| row.get("address"))
        .and_then(|v| v.as_str())
        .map(|s| s.to_string());
    if let Some(s) = location {
        ee.insert("location".to_string(), Value::String(s));
    }
    if let Some(v) = row.get("vat_id").cloned() {
        ee.insert("vat_id".to_string(), v);
    }
    if !segment.is_empty() {
        ee.insert(
            "register_reference".to_string(),
            Value::String(segment.clone()),
        );
    }
    if let Some(v) = row.get("website").cloned() {
        ee.insert("source_url".to_string(), v);
    }
    Some(filter_ee_payload(&ee))
}

pub fn entity_type_from_dutch_rights_form(rights_form: Option<&str>) -> Option<&'static str> {
    let rf = rights_form?;
    let mapping: HashMap<&str, &str> = [
        ("Eenmanszaak", "EMZ"),
        ("Vennootschap Onder Firma", "VOF"),
        ("Maatschap", "MS"),
        ("Stichting", "ST"),
        ("Besloten Vennootschap", "BV"),
    ]
    .into_iter()
    .collect();
    mapping.get(rf).copied()
}

pub fn northdata_register_label(segment: &str) -> Option<&'static str> {
    let s = segment.trim();
    if s.is_empty() {
        return None;
    }
    let pairs: &[(&str, &str)] = &[
        ("Companies House ", "Companies House"),
        ("Siren ", "Siren"),
        ("RIK ", "RIK"),
        ("CRO ", "CRO"),
        ("GEMI ", "GEMI"),
        ("ON ", "ON"),
        ("HRB ", "HRB"),
        ("UID ", "UID"),
        ("MCIT ", "MCIT"),
        ("Handelsregister ", "Handelsregister"),
    ];
    for (prefix, label) in pairs {
        if s.starts_with(prefix) {
            return Some(label);
        }
    }
    if s.to_uppercase().starts_with("KRS") {
        return Some("KRS");
    }
    None
}

pub fn northdata_company_number_from_url(url: &str) -> Option<String> {
    let path = Url::parse(url).ok()?.path().trim_matches('/').to_string();
    if path.is_empty() {
        return None;
    }
    let last = path.split('/').next_back()?;
    let decoded = percent_decode_str(last)
        .decode_utf8()
        .ok()?
        .trim()
        .to_string();
    if decoded.is_empty() {
        None
    } else {
        Some(decoded)
    }
}

pub fn northdata_status_from_title(title: Option<&str>) -> Option<&'static str> {
    let t = title?;
    if t.contains('\u{271d}') || t.contains("✝") {
        return Some("dissolved");
    }
    None
}

pub fn map_northdata_hit_to_global(
    hit: &Map<String, Value>,
    country_code: &str,
) -> Option<Map<String, Value>> {
    if hit.get("type").and_then(|v| v.as_str()) != Some("company") {
        return None;
    }
    let name = hit
        .get("name")
        .or_else(|| hit.get("title"))
        .and_then(|v| v.as_str())?;
    let url = hit.get("url").and_then(|v| v.as_str())?;
    if name.is_empty() || url.is_empty() {
        return None;
    }
    let segment = northdata_company_number_from_url(url)?;
    let cc = country_code.to_uppercase();
    let company_number = segment.clone();
    let hash_input = format!("{}|{}|{}", name, company_number, cc);
    let hash_val = format!("{:x}", Sha256::digest(hash_input.as_bytes()));

    let description = hit.get("description").cloned().unwrap_or(Value::Null);
    let title = hit.get("title").and_then(|v| v.as_str());
    let status = northdata_status_from_title(title).map(|s| Value::String(s.to_string()));
    let register = northdata_register_label(&segment).map(|s| Value::String(s.to_string()));

    let mut global_row = Map::new();
    global_row.insert("name".to_string(), Value::String(name.to_string()));
    global_row.insert("company_number".to_string(), Value::String(company_number));
    global_row.insert("rights_form".to_string(), Value::Null);
    global_row.insert("country_code".to_string(), Value::String(cc));
    if let Some(ref s) = status {
        global_row.insert("status".to_string(), s.clone());
    } else {
        global_row.insert("status".to_string(), Value::Null);
    }
    global_row.insert("last_update".to_string(), Value::Null);
    global_row.insert("description".to_string(), description.clone());
    global_row.insert("region".to_string(), Value::Null);
    global_row.insert("rights_form_code".to_string(), Value::Null);
    global_row.insert("state".to_string(), Value::Null);
    global_row.insert("name_english".to_string(), Value::Null);
    global_row.insert("language".to_string(), Value::Null);
    global_row.insert("hash".to_string(), Value::String(hash_val));
    global_row.insert("address".to_string(), description);
    global_row.insert("start_date".to_string(), Value::Null);
    global_row.insert("legal_persons".to_string(), Value::Null);
    global_row.insert("contact_data".to_string(), Value::Null);
    global_row.insert("activities".to_string(), Value::Null);
    global_row.insert(
        "activity_code_table".to_string(),
        register.unwrap_or(Value::Null),
    );
    global_row.insert("establishment_date".to_string(), Value::Null);
    global_row.insert("establishment_type".to_string(), Value::Null);
    global_row.insert("entity_type".to_string(), Value::Null);
    global_row.insert("vat_id".to_string(), Value::Null);
    global_row.insert("email".to_string(), Value::Null);
    global_row.insert("iban".to_string(), Value::Null);
    global_row.insert("phone".to_string(), Value::Null);
    global_row.insert("postal_code".to_string(), Value::Null);
    global_row.insert("house_number".to_string(), Value::Null);
    global_row.insert("province".to_string(), Value::Null);
    global_row.insert("city".to_string(), Value::Null);
    global_row.insert("street".to_string(), Value::Null);
    global_row.insert("country".to_string(), Value::Null);
    global_row.insert("website".to_string(), Value::String(url.to_string()));

    Some(filter_global_payload(&global_row))
}

pub fn map_kvk_company_to_global(
    company: &Map<String, Value>,
    country_code: &str,
) -> Option<Map<String, Value>> {
    let kvk = 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.parse().ok()))
    });
    let name = company.get("name").and_then(|v| v.as_str())?;
    let kvk = kvk?;
    if name.is_empty() {
        return None;
    }

    let cc = country_code.to_uppercase();
    let street = company.get("street").and_then(|v| v.as_str());
    let house_number = company.get("house_number");
    let postal_code = company.get("postal_code").and_then(|v| v.as_str());
    let city = company.get("city").and_then(|v| v.as_str());
    let province = company.get("province").and_then(|v| v.as_str());
    let country = company
        .get("country")
        .and_then(|v| v.as_str())
        .unwrap_or("Nederland");

    let mut address_parts: Vec<String> = vec![];
    if let Some(s) = street {
        address_parts.push(s.to_string());
    }
    if let Some(hn) = house_number {
        if !hn.is_null() {
            address_parts.push(hn.to_string().trim_matches('"').to_string());
        }
    }
    if let Some(s) = postal_code {
        address_parts.push(s.to_string());
    }
    if let Some(s) = city {
        address_parts.push(s.to_string());
    }
    if let Some(s) = province {
        address_parts.push(s.to_string());
    }
    address_parts.push(country.to_string());
    let address = if address_parts.is_empty() {
        None
    } else {
        Some(address_parts.join(", "))
    };

    let company_number = kvk.to_string();
    let hash_input = format!("{}|{}|{}", name, company_number, cc);
    let hash_val = format!("{:x}", Sha256::digest(hash_input.as_bytes()));

    let active = company.get("active").and_then(|v| v.as_bool());
    let status = match active {
        Some(true) => Some("active"),
        Some(false) => Some("inactive"),
        None => None,
    };

    let activities = company.get("activities").cloned();
    let activity_code_table = company
        .get("main_sbi_code")
        .map(|_| Value::String("SBI".to_string()));
    let establishment_type = match company.get("establishment").and_then(|v| v.as_bool()) {
        Some(true) => Some("main"),
        Some(false) => Some("branch"),
        None => None,
    };

    let hn_out = house_number.and_then(|v| {
        if v.is_null() {
            None
        } else {
            Some(Value::String(v.to_string().trim_matches('"').to_string()))
        }
    });

    let mut global_row = Map::new();
    global_row.insert("name".to_string(), Value::String(name.to_string()));
    global_row.insert("company_number".to_string(), Value::String(company_number));
    global_row.insert(
        "rights_form".to_string(),
        company.get("rights_form").cloned().unwrap_or(Value::Null),
    );
    global_row.insert("country_code".to_string(), Value::String(cc));
    global_row.insert(
        "status".to_string(),
        status
            .map(|s| Value::String(s.to_string()))
            .unwrap_or(Value::Null),
    );
    global_row.insert("last_update".to_string(), Value::Null);
    global_row.insert(
        "description".to_string(),
        company
            .get("main_activity_description")
            .or_else(|| company.get("main_activity"))
            .cloned()
            .unwrap_or(Value::Null),
    );
    global_row.insert(
        "region".to_string(),
        province
            .map(|s| Value::String(s.to_string()))
            .unwrap_or(Value::Null),
    );
    global_row.insert(
        "rights_form_code".to_string(),
        company
            .get("rights_form_code")
            .cloned()
            .unwrap_or(Value::Null),
    );
    global_row.insert(
        "state".to_string(),
        province
            .map(|s| Value::String(s.to_string()))
            .unwrap_or(Value::Null),
    );
    global_row.insert("name_english".to_string(), Value::Null);
    global_row.insert("language".to_string(), Value::String("nl".to_string()));
    global_row.insert("hash".to_string(), Value::String(hash_val));
    global_row.insert(
        "address".to_string(),
        address.map(|s| Value::String(s)).unwrap_or(Value::Null),
    );
    global_row.insert("start_date".to_string(), Value::Null);
    global_row.insert("legal_persons".to_string(), Value::Null);
    global_row.insert("contact_data".to_string(), Value::Null);
    global_row.insert("activities".to_string(), activities.unwrap_or(Value::Null));
    global_row.insert(
        "activity_code_table".to_string(),
        activity_code_table.unwrap_or(Value::Null),
    );
    global_row.insert("establishment_date".to_string(), Value::Null);
    global_row.insert(
        "establishment_type".to_string(),
        establishment_type
            .map(|s| Value::String(s.to_string()))
            .unwrap_or(Value::Null),
    );
    global_row.insert("entity_type".to_string(), Value::Null);
    global_row.insert(
        "vat_id".to_string(),
        company.get("vat_id").cloned().unwrap_or(Value::Null),
    );
    global_row.insert("email".to_string(), Value::Null);
    global_row.insert("iban".to_string(), Value::Null);
    global_row.insert("phone".to_string(), Value::Null);
    global_row.insert(
        "postal_code".to_string(),
        postal_code
            .map(|s| Value::String(s.to_string()))
            .unwrap_or(Value::Null),
    );
    global_row.insert("house_number".to_string(), hn_out.unwrap_or(Value::Null));
    global_row.insert(
        "province".to_string(),
        province
            .map(|s| Value::String(s.to_string()))
            .unwrap_or(Value::Null),
    );
    global_row.insert(
        "city".to_string(),
        city.map(|s| Value::String(s.to_string()))
            .unwrap_or(Value::Null),
    );
    global_row.insert(
        "street".to_string(),
        street
            .map(|s| Value::String(s.to_string()))
            .unwrap_or(Value::Null),
    );
    global_row.insert("country".to_string(), Value::String(country.to_string()));
    global_row.insert("website".to_string(), Value::Null);

    Some(filter_global_payload(&global_row))
}

/// NL script `map_to_global_company` (includes `time` unix field).
pub fn map_nl_company_to_global_script(company: &Map<String, Value>) -> Map<String, Value> {
    let street = company.get("street").and_then(|v| v.as_str());
    let house_number = company.get("house_number");
    let postal_code = company.get("postal_code").and_then(|v| v.as_str());
    let city = company.get("city").and_then(|v| v.as_str());
    let province = company.get("province").and_then(|v| v.as_str());
    let country = company
        .get("country")
        .and_then(|v| v.as_str())
        .unwrap_or("NL");

    let mut address_parts: Vec<String> = vec![];
    if let Some(s) = street {
        address_parts.push(s.to_string());
    }
    if let Some(hn) = house_number {
        if !hn.is_null() {
            address_parts.push(hn.to_string());
        }
    }
    if let Some(s) = postal_code {
        address_parts.push(s.to_string());
    }
    if let Some(s) = city {
        address_parts.push(s.to_string());
    }
    if let Some(s) = province {
        address_parts.push(s.to_string());
    }
    address_parts.push(country.to_string());
    let address = if address_parts.is_empty() {
        None
    } else {
        Some(address_parts.join(", "))
    };

    let name_s = company.get("name").and_then(|v| v.as_str()).unwrap_or("");
    let kvk = company.get("kvk_number");
    let country_for_hash = company
        .get("country")
        .and_then(|v| v.as_str())
        .unwrap_or("");
    let hash_input = format!(
        "{}|{}|{}",
        name_s,
        kvk.map(|v| v.to_string()).unwrap_or_default(),
        country_for_hash
    );
    let hash_val = format!("{:x}", Sha256::digest(hash_input.as_bytes()));

    let active = company.get("active").and_then(|v| v.as_bool());
    let status = match active {
        Some(true) => Some("active"),
        Some(false) => Some("inactive"),
        None => None,
    };

    let activity_code_table = company
        .get("main_sbi_code")
        .map(|_| Value::String("SBI".to_string()));
    let establishment_type = match company.get("establishment").and_then(|v| v.as_bool()) {
        Some(true) => Some("main"),
        Some(false) => Some("branch"),
        None => None,
    };

    let company_number = kvk
        .and_then(|v| {
            if v.is_null() {
                None
            } else {
                Some(v.to_string())
            }
        })
        .filter(|s| !s.is_empty());

    let description = company
        .get("main_activity_description")
        .or_else(|| company.get("main_activity"))
        .cloned();

    let unix_ts = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs() as i64)
        .unwrap_or(0);

    let mut global_company = Map::new();
    global_company.insert(
        "name".to_string(),
        company.get("name").cloned().unwrap_or(Value::Null),
    );
    global_company.insert(
        "company_number".to_string(),
        company_number.map(Value::String).unwrap_or(Value::Null),
    );
    global_company.insert(
        "rights_form".to_string(),
        company.get("rights_form").cloned().unwrap_or(Value::Null),
    );
    global_company.insert("country_code".to_string(), Value::String("NL".to_string()));
    global_company.insert(
        "status".to_string(),
        status
            .map(|s| Value::String(s.to_string()))
            .unwrap_or(Value::Null),
    );
    global_company.insert("last_update".to_string(), Value::Null);
    global_company.insert(
        "description".to_string(),
        description.unwrap_or(Value::Null),
    );
    global_company.insert(
        "region".to_string(),
        province
            .map(|s| Value::String(s.to_string()))
            .unwrap_or(Value::Null),
    );
    global_company.insert(
        "rights_form_code".to_string(),
        company
            .get("rights_form_code")
            .cloned()
            .unwrap_or(Value::Null),
    );
    global_company.insert(
        "state".to_string(),
        province
            .map(|s| Value::String(s.to_string()))
            .unwrap_or(Value::Null),
    );
    global_company.insert("name_english".to_string(), Value::Null);
    global_company.insert("language".to_string(), Value::String("nl".to_string()));
    global_company.insert("hash".to_string(), Value::String(hash_val));
    global_company.insert(
        "address".to_string(),
        address.map(|s| Value::String(s)).unwrap_or(Value::Null),
    );
    global_company.insert("start_date".to_string(), Value::Null);
    global_company.insert("legal_persons".to_string(), Value::Null);
    global_company.insert("contact_data".to_string(), Value::Null);
    global_company.insert(
        "activities".to_string(),
        company.get("activities").cloned().unwrap_or(Value::Null),
    );
    global_company.insert(
        "activity_code_table".to_string(),
        activity_code_table.unwrap_or(Value::Null),
    );
    global_company.insert("establishment_date".to_string(), Value::Null);
    global_company.insert(
        "establishment_type".to_string(),
        establishment_type
            .map(|s| Value::String(s.to_string()))
            .unwrap_or(Value::Null),
    );
    global_company.insert("entity_type".to_string(), Value::Null);
    global_company.insert(
        "postal_code".to_string(),
        postal_code
            .map(|s| Value::String(s.to_string()))
            .unwrap_or(Value::Null),
    );
    global_company.insert(
        "house_number".to_string(),
        house_number.cloned().unwrap_or(Value::Null),
    );
    global_company.insert(
        "street".to_string(),
        street
            .map(|s| Value::String(s.to_string()))
            .unwrap_or(Value::Null),
    );
    global_company.insert(
        "province".to_string(),
        province
            .map(|s| Value::String(s.to_string()))
            .unwrap_or(Value::Null),
    );
    global_company.insert(
        "city".to_string(),
        city.map(|s| Value::String(s.to_string()))
            .unwrap_or(Value::Null),
    );
    global_company.insert("country".to_string(), Value::String(country.to_string()));
    global_company.insert("time".to_string(), Value::Number(Number::from(unix_ts)));
    global_company.remove("id");
    global_company.remove("created_at");
    global_company
}

fn infer_registries_vat_id(nums: &[String], primary: &str, country: &str) -> Option<String> {
    let primary_trim = primary.trim();
    let is_digits = |s: &str| s.chars().all(|c| c.is_ascii_digit());
    let trimmed: Vec<&str> = nums
        .iter()
        .map(|s| s.trim())
        .filter(|s| !s.is_empty())
        .collect();
    match country {
        "PL" => trimmed.iter().find_map(|&n| {
            (n.len() == 10 && is_digits(n) && n != primary_trim).then(|| n.to_string())
        }),
        "CZ" => trimmed
            .iter()
            .find_map(|&n| {
                ((8..=10).contains(&n.len()) && is_digits(n) && n != primary_trim)
                    .then(|| n.to_string())
            })
            .or_else(|| {
                trimmed
                    .iter()
                    .find_map(|&n| n.to_uppercase().starts_with("CZ").then(|| n.to_string()))
            }),
        _ => None,
    }
}

fn registries_public_company_url(
    base: &str,
    country_lower: &str,
    slug: Option<&str>,
    fallback_query: &str,
) -> String {
    let root = base.trim_end_matches('/');
    if let Some(s) = slug.filter(|x| !x.is_empty()) {
        return format!("{root}/{country_lower}/{s}");
    }
    let q = utf8_percent_encode(fallback_query, NON_ALPHANUMERIC);
    format!("{root}/search?country={country_lower}&query={q}")
}

/// Map one registries.io `data[]` company to a `global_companies_registry` row. Adds
/// `registries_slug` and `registries_extra_registration_numbers` for `pl_companies` / `cz_companies`
/// mapping; those keys are omitted when inserting into `global_companies_registry`.
pub fn map_registries_io_company_to_global(
    company: &crate::scrape::registries_io::RegistriesIoCompany,
    cli_country: &str,
    public_site_base: &str,
) -> Option<Map<String, Value>> {
    let cc = cli_country.trim().to_uppercase();
    if cc != "PL" && cc != "CZ" {
        return None;
    }
    let mut primary = crate::scrape::registries_io::primary_registration_id(company)?;
    if cc == "PL" {
        primary = normalize_pl_registration_number(&primary);
    }
    if primary.is_empty() {
        return None;
    }
    let name = company.name.trim();
    if name.is_empty() {
        return None;
    }
    if !company.country.trim().is_empty()
        && company.country.trim().to_uppercase() != cc
    {
        tracing::debug!(
            registries_io_country = %company.country,
            cli_country = %cc,
            "registries.io row country differs from CLI; using CLI country"
        );
    }
    let nums: Vec<String> = company.registration_numbers.clone().unwrap_or_default();
    let vat = infer_registries_vat_id(&nums, &primary, cc.as_str())
        .map(Value::String)
        .unwrap_or(Value::Null);
    let hash_input = format!("{}|{}|{}", name, primary, cc);
    let hash_val = format!("{:x}", Sha256::digest(hash_input.as_bytes()));
    let website = registries_public_company_url(
        public_site_base,
        &cc.to_lowercase(),
        company.slug.as_deref(),
        &primary,
    );
    let addr = company
        .address
        .as_ref()
        .map(|s| s.trim().to_string())
        .filter(|s| !s.is_empty());
    let address_val = addr
        .as_ref()
        .map(|s| Value::String(s.clone()))
        .unwrap_or(Value::Null);

    let mut global_row = Map::new();
    global_row.insert("name".to_string(), Value::String(name.to_string()));
    global_row.insert(
        "company_number".to_string(),
        Value::String(primary.clone()),
    );
    global_row.insert(
        "rights_form".to_string(),
        company
            .legal_form
            .as_ref()
            .filter(|s| !s.trim().is_empty())
            .map(|s| Value::String(s.trim().to_string()))
            .unwrap_or(Value::Null),
    );
    global_row.insert("country_code".to_string(), Value::String(cc.clone()));
    global_row.insert(
        "status".to_string(),
        company
            .status
            .as_ref()
            .filter(|s| !s.trim().is_empty())
            .map(|s| Value::String(s.trim().to_string()))
            .unwrap_or(Value::Null),
    );
    global_row.insert("last_update".to_string(), Value::Null);
    global_row.insert("description".to_string(), Value::Null);
    global_row.insert("region".to_string(), Value::Null);
    global_row.insert("rights_form_code".to_string(), Value::Null);
    global_row.insert("state".to_string(), Value::Null);
    global_row.insert("name_english".to_string(), Value::Null);
    global_row.insert(
        "language".to_string(),
        Value::String(cc.to_lowercase()),
    );
    global_row.insert("hash".to_string(), Value::String(hash_val));
    global_row.insert("address".to_string(), address_val);
    global_row.insert("start_date".to_string(), Value::Null);
    global_row.insert("legal_persons".to_string(), Value::Null);
    global_row.insert("contact_data".to_string(), Value::Null);
    global_row.insert("activities".to_string(), Value::Null);
    global_row.insert("activity_code_table".to_string(), Value::Null);
    global_row.insert(
        "establishment_date".to_string(),
        company
            .registration_date
            .as_ref()
            .filter(|s| !s.trim().is_empty())
            .map(|s| Value::String(s.trim().to_string()))
            .unwrap_or(Value::Null),
    );
    global_row.insert("establishment_type".to_string(), Value::Null);
    global_row.insert("entity_type".to_string(), Value::Null);
    global_row.insert("vat_id".to_string(), vat);
    global_row.insert("email".to_string(), Value::Null);
    global_row.insert("iban".to_string(), Value::Null);
    global_row.insert("phone".to_string(), Value::Null);
    global_row.insert("postal_code".to_string(), Value::Null);
    global_row.insert("house_number".to_string(), Value::Null);
    global_row.insert("province".to_string(), Value::Null);
    global_row.insert("city".to_string(), Value::Null);
    global_row.insert("street".to_string(), Value::Null);
    global_row.insert("country".to_string(), Value::Null);
    global_row.insert("website".to_string(), Value::String(website));

    let mut filtered = filter_global_payload(&global_row);
    if let Some(s) = company
        .slug
        .as_ref()
        .map(|x| x.trim())
        .filter(|s| !s.is_empty())
    {
        filtered.insert(
            "registries_slug".to_string(),
            Value::String(s.to_string()),
        );
    }
    if !nums.is_empty() {
        filtered.insert(
            "registries_extra_registration_numbers".to_string(),
            Value::Array(nums.iter().map(|s| Value::String(s.clone())).collect()),
        );
    }
    Some(filtered)
}
