use std::collections::{BTreeMap, HashMap};
use std::fs::{self, File};
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::process::Command;
use std::time::Duration;

use anyhow::{Context, Result, anyhow, bail};
use calamine::{Data, Reader, open_workbook_auto};
use chrono::{DateTime, NaiveDateTime, Utc};
use clap::Parser;
use filetime::{FileTime, set_file_times};
use fs2::FileExt;
use regex::Regex;
use reqwest::blocking::Client;
use serde::Deserialize;
use serde_json::{Map, Value, json};
use sha1::{Digest, Sha1};

const DATASETS: &[&str] = &[
    "documents",
    "postings",
    "business_events",
    "document_changes",
    "companies",
    "reglog",
    "audit",
    "host",
];

#[derive(Debug, Parser)]
#[command(about = "Rust AW-rus file-1C ingest cycle for ClickHouse")]
struct Cli {
    #[arg(long, env = "AW_1C_ROOT")]
    root: Option<PathBuf>,

    #[arg(long)]
    config: Option<PathBuf>,

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

    #[arg(long, default_value_t = 30, env = "AW_1C_INGEST_TIMEOUT_SECONDS")]
    timeout_seconds: u64,

    #[arg(long)]
    dry_run: bool,

    #[arg(long)]
    skip_post_refresh: bool,

    #[arg(long)]
    skip_briefs: bool,

    #[arg(long)]
    json: bool,
}

#[derive(Debug, Deserialize)]
struct RawConfig {
    clickhouse: ClickHouseConfig,
    landing: BTreeMap<String, PathBuf>,
    #[serde(default)]
    formats: BTreeMap<String, String>,
    archive_dir: Option<PathBuf>,
    #[serde(default)]
    delete_after_load: bool,
    #[serde(default = "default_min_file_age_seconds")]
    min_file_age_seconds: u64,
}

#[derive(Debug, Deserialize)]
struct ClickHouseConfig {
    host: String,
    #[serde(default = "default_clickhouse_port")]
    port: u16,
    #[serde(default = "default_clickhouse_user")]
    username: String,
    #[serde(default)]
    password: String,
    #[serde(default = "default_clickhouse_database")]
    database: String,
}

#[derive(Debug, Default)]
struct RunSummary {
    files_seen: usize,
    files_loaded: usize,
    rows_loaded: usize,
    files_skipped_young: usize,
    files_archived: usize,
    sql_statements: usize,
}

#[derive(Debug, Clone)]
struct DocumentMeta {
    company_entity_key: String,
    organization: String,
    department: String,
    document_id: String,
    document_number: String,
    document_type: String,
    operation_type: String,
    user: String,
    counterparty: String,
}

type DocumentIndex = HashMap<(String, String), DocumentMeta>;

fn main() {
    let code = match run() {
        Ok(code) => code,
        Err(err) => {
            eprintln!("{err:#}");
            1
        }
    };
    std::process::exit(code);
}

fn run() -> Result<i32> {
    let cli = Cli::parse();
    let root = cli
        .root
        .clone()
        .or_else(|| std::env::var_os("AW_1C_ROOT").map(PathBuf::from))
        .unwrap_or_else(|| PathBuf::from("/opt/activitywatch/clickhouse-1c"));
    let config_path = cli
        .config
        .clone()
        .unwrap_or_else(|| root.join("etl/config.yml"));
    let lock_path = root.join(".ingest.lock");
    let lock_file =
        File::create(&lock_path).with_context(|| format!("open lock {}", lock_path.display()))?;
    if lock_file.try_lock_exclusive().is_err() {
        println!("ingest cycle already running");
        return Ok(0);
    }

    let config = load_config(&config_path)?;
    let client = ClickHouseClient::new(&config.clickhouse, cli.timeout_seconds)?;
    let mut summary = RunSummary::default();

    build_business_exports(&config, cli.dry_run, &mut summary)?;
    load_datasets(
        &config,
        &client,
        cli.dataset.as_deref(),
        cli.dry_run,
        &mut summary,
    )?;
    load_registry(&config, &client, &root, cli.dry_run, &mut summary)?;

    if !cli.dry_run {
        run_sql_file(
            &client,
            &root.join("detections/build_entity_timeline.sql"),
            &mut summary,
        )?;
        run_sql_file(
            &client,
            &root.join("clickhouse/init/04_company_intelligence.sql"),
            &mut summary,
        )?;
        if !cli.skip_post_refresh {
            run_optional_script(&root.join("ops/run_company_registry_bindings_refresh.sh"))?;
            run_optional_script(&root.join("ops/run_company_intelligence_refresh.sh"))?;
        }
        run_sql_file(
            &client,
            &root.join("detections/insert_detections.sql"),
            &mut summary,
        )?;
        run_sql_file(
            &client,
            &root.join("detections/open_cases_from_detections.sql"),
            &mut summary,
        )?;
        if !cli.skip_briefs {
            let _ = run_optional_script(&root.join("ops/run_manager_brief.sh"));
            let _ = run_optional_script(&root.join("ops/run_recovery_brief.sh"));
        }
        let _ = run_optional_script(&root.join("ops/check_ingest_freshness.sh"));
    }

    if cli.json {
        println!(
            "{}",
            serde_json::to_string_pretty(&json!({
                "ok": true,
                "dry_run": cli.dry_run,
                "files_seen": summary.files_seen,
                "files_loaded": summary.files_loaded,
                "rows_loaded": summary.rows_loaded,
                "files_skipped_young": summary.files_skipped_young,
                "files_archived": summary.files_archived,
                "sql_statements": summary.sql_statements,
            }))?
        );
    } else {
        println!(
            "aw-1c-ingest-rust ok dry_run={} files_loaded={} rows_loaded={} skipped_young={} sql_statements={}",
            cli.dry_run,
            summary.files_loaded,
            summary.rows_loaded,
            summary.files_skipped_young,
            summary.sql_statements
        );
    }
    Ok(0)
}

fn default_clickhouse_port() -> u16 {
    8123
}

fn default_clickhouse_user() -> String {
    "default".to_string()
}

fn default_clickhouse_database() -> String {
    "analytics_1c".to_string()
}

fn default_min_file_age_seconds() -> u64 {
    180
}

fn load_config(path: &Path) -> Result<RawConfig> {
    let text = fs::read_to_string(path).with_context(|| format!("read {}", path.display()))?;
    serde_yaml::from_str(&text).with_context(|| format!("parse {}", path.display()))
}

struct ClickHouseClient {
    client: Client,
    base_url: String,
    username: String,
    password: String,
    database: String,
}

impl ClickHouseClient {
    fn new(config: &ClickHouseConfig, timeout_seconds: u64) -> Result<Self> {
        let host = config.host.trim_end_matches('/');
        let base_url = if host.starts_with("http://") || host.starts_with("https://") {
            host.to_string()
        } else {
            format!("http://{}:{}", host, config.port)
        };
        let client = Client::builder()
            .timeout(Duration::from_secs(timeout_seconds))
            .build()
            .context("build ClickHouse HTTP client")?;
        Ok(Self {
            client,
            base_url,
            username: config.username.clone(),
            password: config.password.clone(),
            database: config.database.clone(),
        })
    }

    fn execute(&self, sql: &str) -> Result<()> {
        if sql.trim().is_empty() {
            return Ok(());
        }
        let mut req = self
            .client
            .post(&self.base_url)
            .query(&[("database", self.database.as_str())])
            .body(sql.to_string());
        if !self.username.trim().is_empty() {
            req = req.basic_auth(self.username.clone(), Some(self.password.clone()));
        }
        let response = req.send().context("ClickHouse request")?;
        let status = response.status();
        let body = response.text().unwrap_or_default();
        if !status.is_success() {
            bail!("ClickHouse HTTP {status}: {}", body.trim());
        }
        Ok(())
    }

    fn insert_json_each_row(&self, table: &str, columns: &[&str], rows: &[Value]) -> Result<()> {
        if rows.is_empty() {
            return Ok(());
        }
        let mut payload = String::new();
        payload.push_str(&format!(
            "INSERT INTO {} ({}) FORMAT JSONEachRow\n",
            table,
            columns.join(", ")
        ));
        for row in rows {
            payload.push_str(&serde_json::to_string(row)?);
            payload.push('\n');
        }
        self.execute(&payload)
    }
}

fn dataset_format(config: &RawConfig, dataset: &str) -> String {
    config
        .formats
        .get(dataset)
        .or_else(|| config.formats.get("default"))
        .cloned()
        .unwrap_or_else(|| "jsonl".to_string())
}

fn ready_files(config: &RawConfig, dataset: &str) -> Result<Vec<PathBuf>> {
    let Some(root) = config.landing.get(dataset) else {
        return Ok(Vec::new());
    };
    if !root.exists() {
        return Ok(Vec::new());
    }
    let mut out = Vec::new();
    for entry in fs::read_dir(root).with_context(|| format!("read {}", root.display()))? {
        let path = entry?.path();
        if path.is_file() {
            out.push(path);
        }
    }
    out.sort();
    Ok(out)
}

fn file_age_seconds(path: &Path) -> Result<u64> {
    let meta = fs::metadata(path).with_context(|| format!("stat {}", path.display()))?;
    let modified = meta
        .modified()
        .with_context(|| format!("mtime {}", path.display()))?;
    Ok(modified.elapsed().unwrap_or_default().as_secs())
}

fn file_is_ready(path: &Path, config: &RawConfig) -> Result<bool> {
    Ok(file_age_seconds(path)? >= config.min_file_age_seconds)
}

fn read_rows(path: &Path, fmt: &str) -> Result<Vec<Map<String, Value>>> {
    match fmt {
        "jsonl" => {
            let text = read_text_lossy(path)?;
            text.lines()
                .filter(|line| !line.trim().is_empty())
                .map(|line| parse_object_line(line, path))
                .collect()
        }
        "json" => {
            let value: Value = serde_json::from_str(&read_text_lossy(path)?)
                .with_context(|| format!("parse json {}", path.display()))?;
            match value {
                Value::Array(items) => items
                    .into_iter()
                    .map(|item| match item {
                        Value::Object(map) => Ok(map),
                        _ => bail!("json item is not object in {}", path.display()),
                    })
                    .collect(),
                Value::Object(map) => Ok(vec![map]),
                _ => bail!("json root is not object/list in {}", path.display()),
            }
        }
        "csv" => {
            let mut reader = csv::Reader::from_path(path)
                .with_context(|| format!("open csv {}", path.display()))?;
            let headers = reader.headers()?.clone();
            let mut rows = Vec::new();
            for record in reader.records() {
                let record = record?;
                let mut map = Map::new();
                for (idx, value) in record.iter().enumerate() {
                    if let Some(header) = headers.get(idx) {
                        map.insert(header.to_string(), Value::String(value.to_string()));
                    }
                }
                rows.push(map);
            }
            Ok(rows)
        }
        other => bail!("unsupported format {other} for {}", path.display()),
    }
}

fn read_text_lossy(path: &Path) -> Result<String> {
    let mut bytes = Vec::new();
    File::open(path)
        .with_context(|| format!("open {}", path.display()))?
        .read_to_end(&mut bytes)?;
    if bytes.starts_with(&[0xEF, 0xBB, 0xBF]) {
        bytes.drain(0..3);
    }
    String::from_utf8(bytes).with_context(|| format!("decode utf-8 {}", path.display()))
}

fn parse_object_line(line: &str, path: &Path) -> Result<Map<String, Value>> {
    match serde_json::from_str::<Value>(line)
        .with_context(|| format!("parse jsonl {}", path.display()))?
    {
        Value::Object(map) => Ok(map),
        _ => bail!("jsonl row is not object in {}", path.display()),
    }
}

fn load_datasets(
    config: &RawConfig,
    client: &ClickHouseClient,
    only_dataset: Option<&str>,
    dry_run: bool,
    summary: &mut RunSummary,
) -> Result<()> {
    let datasets: Vec<&str> = match only_dataset {
        Some(dataset) => vec![dataset],
        None => DATASETS.to_vec(),
    };
    for dataset in datasets {
        let fmt = dataset_format(config, dataset);
        for path in ready_files(config, dataset)? {
            summary.files_seen += 1;
            if !file_is_ready(&path, config)? {
                summary.files_skipped_young += 1;
                println!(
                    "skip {dataset}: {} age={}s < min_file_age_seconds={}",
                    path.file_name().and_then(|v| v.to_str()).unwrap_or("-"),
                    file_age_seconds(&path)?,
                    config.min_file_age_seconds
                );
                continue;
            }
            let rows = read_rows(&path, &fmt)?;
            if rows.is_empty() {
                archive_or_delete(config, dataset, &path, dry_run, summary)?;
                continue;
            }
            if !dry_run {
                let source_file = file_name(&path);
                let raw_rows: Vec<Value> = rows
                    .iter()
                    .map(|row| {
                        json!({
                            "source_file": source_file,
                            "payload": serde_json::to_string(row).unwrap_or_else(|_| "{}".to_string()),
                        })
                    })
                    .collect();
                client.insert_json_each_row(
                    raw_table(dataset)?,
                    &["source_file", "payload"],
                    &raw_rows,
                )?;
                let core_rows: Vec<Value> = rows
                    .iter()
                    .map(|row| map_core_row(dataset, &source_file, row))
                    .collect::<Result<Vec<_>>>()?;
                client.insert_json_each_row(
                    core_table(dataset)?,
                    core_columns(dataset)?,
                    &core_rows,
                )?;
            }
            summary.files_loaded += 1;
            summary.rows_loaded += rows.len();
            archive_or_delete(config, dataset, &path, dry_run, summary)?;
            println!("loaded {dataset}: {} rows={}", file_name(&path), rows.len());
        }
    }
    Ok(())
}

fn raw_table(dataset: &str) -> Result<&'static str> {
    Ok(match dataset {
        "documents" => "raw_1c_documents",
        "postings" => "raw_1c_postings",
        "business_events" => "raw_1c_business_events",
        "document_changes" => "raw_1c_document_changes",
        "companies" => "raw_1c_companies",
        "reglog" => "raw_reglog",
        "audit" => "raw_audit",
        "host" => "raw_host_metrics",
        _ => bail!("unknown dataset {dataset}"),
    })
}

fn core_table(dataset: &str) -> Result<&'static str> {
    Ok(match dataset {
        "documents" => "documents",
        "postings" => "postings",
        "business_events" => "business_events",
        "document_changes" => "document_change_events",
        "companies" => "companies",
        "reglog" => "reglog_events",
        "audit" => "audit_events",
        "host" => "host_events",
        _ => bail!("unknown dataset {dataset}"),
    })
}

fn core_columns(dataset: &str) -> Result<&'static [&'static str]> {
    Ok(match dataset {
        "documents" => &[
            "ts",
            "infobase",
            "organization",
            "department",
            "doc_type",
            "doc_id",
            "doc_number",
            "author",
            "counterparty",
            "operation_type",
            "amount",
            "status",
            "posted",
            "source_file",
        ],
        "postings" => &[
            "ts",
            "infobase",
            "registrar",
            "operation_type",
            "account_dt",
            "account_ct",
            "amount",
            "source_file",
        ],
        "business_events" => &[
            "ts",
            "event_id",
            "infobase",
            "company_entity_key",
            "organization",
            "department",
            "document_id",
            "document_number",
            "document_type",
            "registrar",
            "operation_type",
            "event_kind",
            "user",
            "counterparty",
            "counterparty_inn",
            "debit_account",
            "credit_account",
            "amount",
            "currency",
            "line_no",
            "evidence_ref",
            "source_file",
        ],
        "document_changes" => &[
            "ts",
            "change_id",
            "infobase",
            "company_entity_key",
            "organization",
            "document_id",
            "document_number",
            "document_type",
            "change_kind",
            "field_name",
            "user",
            "before_value",
            "after_value",
            "risk_tag",
            "evidence_ref",
            "source_file",
        ],
        "companies" => &[
            "ts",
            "infobase",
            "company_name",
            "organization",
            "owner_user",
            "base_id",
            "base_path",
            "status",
            "db_size_bytes",
            "reglog_size_bytes",
            "active_locks",
            "temp_db_present",
            "scheduler_touched",
            "activity_score",
            "source_file",
        ],
        "reglog" => &[
            "ts",
            "infobase",
            "user",
            "host",
            "app",
            "event_name",
            "level",
            "duration_ms",
            "message",
            "source_file",
        ],
        "audit" => &[
            "ts",
            "infobase",
            "user",
            "object_type",
            "object_id",
            "action",
            "before_hash",
            "after_hash",
            "risk_tag",
            "source_file",
        ],
        "host" => &[
            "ts",
            "host",
            "cpu_pct",
            "ram_pct",
            "disk_free_gb",
            "disk_latency_ms",
            "smb_errors",
            "rdp_sessions",
            "backup_ok",
            "source_file",
        ],
        _ => bail!("unknown dataset {dataset}"),
    })
}

fn map_core_row(dataset: &str, source_file: &str, row: &Map<String, Value>) -> Result<Value> {
    let value = match dataset {
        "documents" => json!({
            "ts": ts_value(row, &["ts", "posted_at", "created_at"]),
            "infobase": s(row, "infobase"),
            "organization": s(row, "organization"),
            "department": s(row, "department"),
            "doc_type": s(row, "doc_type"),
            "doc_id": s(row, "doc_id"),
            "doc_number": s(row, "doc_number"),
            "author": s(row, "author"),
            "counterparty": s(row, "counterparty"),
            "operation_type": s(row, "operation_type"),
            "amount": f(row, "amount"),
            "status": s(row, "status"),
            "posted": u(row, "posted"),
            "source_file": source_file,
        }),
        "postings" => json!({
            "ts": ts_value(row, &["ts"]),
            "infobase": s(row, "infobase"),
            "registrar": s(row, "registrar"),
            "operation_type": s(row, "operation_type"),
            "account_dt": s(row, "account_dt"),
            "account_ct": s(row, "account_ct"),
            "amount": f(row, "amount"),
            "source_file": source_file,
        }),
        "business_events" => json!({
            "ts": ts_value(row, &["ts", "event_time"]),
            "event_id": s(row, "event_id"),
            "infobase": s(row, "infobase"),
            "company_entity_key": s(row, "company_entity_key"),
            "organization": s(row, "organization"),
            "department": s(row, "department"),
            "document_id": s_alt(row, &["document_id", "doc_id"]),
            "document_number": s_alt(row, &["document_number", "doc_number"]),
            "document_type": s_alt(row, &["document_type", "doc_type"]),
            "registrar": s(row, "registrar"),
            "operation_type": s(row, "operation_type"),
            "event_kind": s(row, "event_kind"),
            "user": s_alt(row, &["user", "author"]),
            "counterparty": s(row, "counterparty"),
            "counterparty_inn": s(row, "counterparty_inn"),
            "debit_account": s_alt(row, &["debit_account", "account_dt"]),
            "credit_account": s_alt(row, &["credit_account", "account_ct"]),
            "amount": f(row, "amount"),
            "currency": s_default(row, "currency", "RUB"),
            "line_no": u(row, "line_no"),
            "evidence_ref": s(row, "evidence_ref"),
            "source_file": source_file,
        }),
        "document_changes" => json!({
            "ts": ts_value(row, &["ts", "change_time"]),
            "change_id": s(row, "change_id"),
            "infobase": s(row, "infobase"),
            "company_entity_key": s(row, "company_entity_key"),
            "organization": s(row, "organization"),
            "document_id": s_alt(row, &["document_id", "doc_id"]),
            "document_number": s_alt(row, &["document_number", "doc_number"]),
            "document_type": s_alt(row, &["document_type", "doc_type"]),
            "change_kind": s(row, "change_kind"),
            "field_name": s(row, "field_name"),
            "user": s_alt(row, &["user", "author"]),
            "before_value": s(row, "before_value"),
            "after_value": s(row, "after_value"),
            "risk_tag": s(row, "risk_tag"),
            "evidence_ref": s(row, "evidence_ref"),
            "source_file": source_file,
        }),
        "companies" => json!({
            "ts": ts_value(row, &["ts"]),
            "infobase": s(row, "infobase"),
            "company_name": s_alt(row, &["company_name", "counterparty", "infobase"]),
            "organization": s(row, "organization"),
            "owner_user": s_alt(row, &["owner_user", "author"]),
            "base_id": s_alt(row, &["base_id", "doc_id"]),
            "base_path": s(row, "base_path"),
            "status": s(row, "status"),
            "db_size_bytes": u(row, "db_size_bytes"),
            "reglog_size_bytes": u(row, "reglog_size_bytes"),
            "active_locks": u(row, "active_locks"),
            "temp_db_present": u(row, "temp_db_present"),
            "scheduler_touched": u(row, "scheduler_touched"),
            "activity_score": f_alt(row, &["activity_score", "amount"]),
            "source_file": source_file,
        }),
        "reglog" => json!({
            "ts": ts_value(row, &["ts"]),
            "infobase": s(row, "infobase"),
            "user": s(row, "user"),
            "host": s(row, "host"),
            "app": s(row, "app"),
            "event_name": s(row, "event_name"),
            "level": s_default(row, "level", "info"),
            "duration_ms": u(row, "duration_ms"),
            "message": s(row, "message"),
            "source_file": source_file,
        }),
        "audit" => json!({
            "ts": ts_value(row, &["ts"]),
            "infobase": s(row, "infobase"),
            "user": s(row, "user"),
            "object_type": s(row, "object_type"),
            "object_id": s(row, "object_id"),
            "action": s(row, "action"),
            "before_hash": s(row, "before_hash"),
            "after_hash": s(row, "after_hash"),
            "risk_tag": s(row, "risk_tag"),
            "source_file": source_file,
        }),
        "host" => json!({
            "ts": ts_value(row, &["ts"]),
            "host": s(row, "host"),
            "cpu_pct": f(row, "cpu_pct"),
            "ram_pct": f(row, "ram_pct"),
            "disk_free_gb": f(row, "disk_free_gb"),
            "disk_latency_ms": f(row, "disk_latency_ms"),
            "smb_errors": u(row, "smb_errors"),
            "rdp_sessions": u(row, "rdp_sessions"),
            "backup_ok": u(row, "backup_ok"),
            "source_file": source_file,
        }),
        _ => bail!("unknown dataset {dataset}"),
    };
    Ok(value)
}

fn build_business_exports(
    config: &RawConfig,
    dry_run: bool,
    summary: &mut RunSummary,
) -> Result<()> {
    let company_index = load_company_index(config)?;
    let (by_id, by_number) = build_document_index(config, &company_index)?;

    for path in ready_files(config, "documents")? {
        if !file_is_ready(&path, config)? {
            continue;
        }
        let rows = read_rows(&path, &dataset_format(config, "documents"))?;
        let events = build_document_events(&rows, &file_name(&path), &company_index);
        let out = output_path(
            config,
            "business_events",
            &path,
            "business-events-documents",
        )?;
        if !dry_run {
            write_generated_jsonl(&out, &events, config.min_file_age_seconds)?;
        }
        summary.files_seen += 1;
    }
    for path in ready_files(config, "postings")? {
        if !file_is_ready(&path, config)? {
            continue;
        }
        let rows = read_rows(&path, &dataset_format(config, "postings"))?;
        let events =
            build_posting_events(&rows, &file_name(&path), &company_index, &by_id, &by_number);
        let out = output_path(config, "business_events", &path, "business-events-postings")?;
        if !dry_run {
            write_generated_jsonl(&out, &events, config.min_file_age_seconds)?;
        }
        summary.files_seen += 1;
    }
    for path in ready_files(config, "audit")? {
        if !file_is_ready(&path, config)? {
            continue;
        }
        let rows = read_rows(&path, &dataset_format(config, "audit"))?;
        let events =
            build_document_changes(&rows, &file_name(&path), &company_index, &by_id, &by_number);
        let out = output_path(config, "document_changes", &path, "document-changes-audit")?;
        if !dry_run {
            write_generated_jsonl(&out, &events, config.min_file_age_seconds)?;
        }
        summary.files_seen += 1;
    }
    Ok(())
}

fn load_company_index(config: &RawConfig) -> Result<HashMap<String, String>> {
    let mut latest: HashMap<String, (String, String)> = HashMap::new();
    for path in ready_files(config, "companies")? {
        if !file_is_ready(&path, config)? {
            continue;
        }
        for row in read_rows(&path, &dataset_format(config, "companies"))? {
            let infobase = s(&row, "infobase");
            if infobase.is_empty() {
                continue;
            }
            let key =
                canonical_company_entity_key(&s(&row, "base_id"), &s(&row, "base_path"), &infobase);
            let ts = ts_value(&row, &["ts"]);
            match latest.get(&infobase) {
                Some((current_ts, _)) if current_ts > &ts => {}
                _ => {
                    latest.insert(infobase, (ts, key));
                }
            }
        }
    }
    Ok(latest.into_iter().map(|(k, (_, v))| (k, v)).collect())
}

fn build_document_index(
    config: &RawConfig,
    company_index: &HashMap<String, String>,
) -> Result<(DocumentIndex, DocumentIndex)> {
    let mut by_id = HashMap::new();
    let mut by_number = HashMap::new();
    for path in ready_files(config, "documents")? {
        if !file_is_ready(&path, config)? {
            continue;
        }
        for row in read_rows(&path, &dataset_format(config, "documents"))? {
            let infobase = s(&row, "infobase");
            if infobase.is_empty() {
                continue;
            }
            let document_id = s_alt(&row, &["doc_id", "document_id"]);
            let document_number = s_alt(&row, &["doc_number", "document_number"]);
            let meta = DocumentMeta {
                company_entity_key: derive_company_entity_key(&row, company_index),
                organization: s(&row, "organization"),
                department: s(&row, "department"),
                document_id: document_id.clone(),
                document_number: document_number.clone(),
                document_type: s_alt(&row, &["doc_type", "document_type"]),
                operation_type: s(&row, "operation_type"),
                user: s_alt(&row, &["author", "user"]),
                counterparty: s(&row, "counterparty"),
            };
            if !document_id.is_empty() {
                by_id.insert((infobase.clone(), document_id), meta.clone());
            }
            if !document_number.is_empty() {
                by_number.insert((infobase, document_number), meta);
            }
        }
    }
    Ok((by_id, by_number))
}

fn build_document_events(
    rows: &[Map<String, Value>],
    source_file: &str,
    company_index: &HashMap<String, String>,
) -> Vec<Value> {
    rows.iter()
        .map(|row| {
            let ts = ts_value(row, &["ts", "posted_at", "created_at"]);
            let infobase = s(row, "infobase");
            let document_id = s_alt(row, &["doc_id", "document_id"]);
            let document_number = s_alt(row, &["doc_number", "document_number"]);
            json!({
                "ts": ts,
                "event_id": stable_id("document_snapshot", &[source_file, &infobase, &document_id, &document_number, &ts]),
                "infobase": infobase,
                "company_entity_key": derive_company_entity_key(row, company_index),
                "organization": s(row, "organization"),
                "department": s(row, "department"),
                "document_id": document_id,
                "document_number": document_number,
                "document_type": s_alt(row, &["doc_type", "document_type"]),
                "registrar": s_alt(row, &["doc_id", "doc_number", "document_id", "document_number"]),
                "operation_type": s(row, "operation_type"),
                "event_kind": "document_snapshot",
                "user": s_alt(row, &["author", "user"]),
                "counterparty": s(row, "counterparty"),
                "counterparty_inn": s(row, "counterparty_inn"),
                "debit_account": "",
                "credit_account": "",
                "amount": f(row, "amount"),
                "currency": s_default(row, "currency", "RUB"),
                "line_no": 0,
                "evidence_ref": format!("document:{}", s_alt(row, &["doc_id", "doc_number", "document_id", "document_number"])),
            })
        })
        .collect()
}

fn build_posting_events(
    rows: &[Map<String, Value>],
    source_file: &str,
    company_index: &HashMap<String, String>,
    by_id: &HashMap<(String, String), DocumentMeta>,
    by_number: &HashMap<(String, String), DocumentMeta>,
) -> Vec<Value> {
    rows.iter()
        .enumerate()
        .map(|(idx, row)| {
            let ts = ts_value(row, &["ts"]);
            let infobase = s(row, "infobase");
            let registrar = s(row, "registrar");
            let meta = lookup_document_meta(row, by_id, by_number);
            let line_no = u_alt(row, &["line_no"]).unwrap_or((idx + 1) as u64);
            json!({
                "ts": ts,
                "event_id": stable_id("posting", &[source_file, &infobase, &registrar, &line_no.to_string(), &ts]),
                "infobase": infobase,
                "company_entity_key": meta.map(|m| m.company_entity_key.clone()).unwrap_or_else(|| derive_company_entity_key(row, company_index)),
                "organization": meta.map(|m| m.organization.clone()).unwrap_or_else(|| s(row, "organization")),
                "department": meta.map(|m| m.department.clone()).unwrap_or_else(|| s(row, "department")),
                "document_id": meta.map(|m| m.document_id.clone()).unwrap_or_else(|| registrar.clone()),
                "document_number": meta.map(|m| m.document_number.clone()).unwrap_or_else(|| s(row, "document_number")),
                "document_type": meta.map(|m| m.document_type.clone()).unwrap_or_else(|| s(row, "document_type")),
                "registrar": registrar,
                "operation_type": s_default(row, "operation_type", &meta.map(|m| m.operation_type.clone()).unwrap_or_default()),
                "event_kind": "posting",
                "user": meta.map(|m| m.user.clone()).unwrap_or_else(|| s_alt(row, &["user", "author"])),
                "counterparty": meta.map(|m| m.counterparty.clone()).unwrap_or_else(|| s(row, "counterparty")),
                "counterparty_inn": s(row, "counterparty_inn"),
                "debit_account": s_alt(row, &["account_dt", "debit_account"]),
                "credit_account": s_alt(row, &["account_ct", "credit_account"]),
                "amount": f(row, "amount"),
                "currency": s_default(row, "currency", "RUB"),
                "line_no": line_no,
                "evidence_ref": format!("posting:{}:{}", s(row, "registrar"), line_no),
            })
        })
        .collect()
}

fn build_document_changes(
    rows: &[Map<String, Value>],
    source_file: &str,
    company_index: &HashMap<String, String>,
    by_id: &HashMap<(String, String), DocumentMeta>,
    by_number: &HashMap<(String, String), DocumentMeta>,
) -> Vec<Value> {
    rows.iter()
        .map(|row| {
            let ts = ts_value(row, &["ts"]);
            let infobase = s(row, "infobase");
            let object_type = s(row, "object_type");
            let object_id = s(row, "object_id");
            let meta = lookup_document_meta(row, by_id, by_number);
            json!({
                "ts": ts,
                "change_id": stable_id("change", &[source_file, &infobase, &object_type, &object_id, &s(row, "action"), &ts]),
                "infobase": infobase,
                "company_entity_key": meta.map(|m| m.company_entity_key.clone()).unwrap_or_else(|| derive_company_entity_key(row, company_index)),
                "organization": meta.map(|m| m.organization.clone()).unwrap_or_else(|| s(row, "organization")),
                "document_id": meta.map(|m| m.document_id.clone()).unwrap_or_else(|| if object_type == "document" { object_id.clone() } else { s(row, "document_id") }),
                "document_number": meta.map(|m| m.document_number.clone()).unwrap_or_else(|| s(row, "document_number")),
                "document_type": meta.map(|m| m.document_type.clone()).unwrap_or_else(|| s(row, "document_type")),
                "change_kind": s_alt(row, &["change_kind", "action", "object_type"]),
                "field_name": s_alt(row, &["field_name", "object_type"]),
                "user": s_alt(row, &["user", "author"]),
                "before_value": s_alt(row, &["before_value", "before_hash"]),
                "after_value": s_alt(row, &["after_value", "after_hash"]),
                "risk_tag": s(row, "risk_tag"),
                "evidence_ref": format!("audit:{}:{}", object_type, object_id),
            })
        })
        .collect()
}

fn lookup_document_meta<'a>(
    row: &Map<String, Value>,
    by_id: &'a HashMap<(String, String), DocumentMeta>,
    by_number: &'a HashMap<(String, String), DocumentMeta>,
) -> Option<&'a DocumentMeta> {
    let infobase = s(row, "infobase");
    let registrar = s_alt(row, &["registrar", "document_id", "doc_id"]);
    let number = s_alt(row, &["document_number", "doc_number"]);
    by_id
        .get(&(infobase.clone(), registrar))
        .or_else(|| by_number.get(&(infobase, number)))
}

fn output_path(config: &RawConfig, dataset: &str, source: &Path, prefix: &str) -> Result<PathBuf> {
    let root = config
        .landing
        .get(dataset)
        .ok_or_else(|| anyhow!("missing landing root for {dataset}"))?;
    let stem = source
        .file_stem()
        .and_then(|v| v.to_str())
        .unwrap_or("input");
    Ok(root.join(format!("{prefix}-{stem}.jsonl")))
}

fn write_generated_jsonl(path: &Path, rows: &[Value], min_file_age_seconds: u64) -> Result<()> {
    if rows.is_empty() {
        let _ = fs::remove_file(path);
        return Ok(());
    }
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent)?;
    }
    let mut tmp = tempfile::NamedTempFile::new_in(path.parent().unwrap_or_else(|| Path::new(".")))?;
    for row in rows {
        writeln!(tmp, "{}", serde_json::to_string(row)?)?;
    }
    tmp.persist(path)
        .map_err(|err| anyhow!("persist {}: {}", path.display(), err))?;
    let aged = FileTime::from_unix_time(
        Utc::now().timestamp() - (min_file_age_seconds as i64 + 1).max(5),
        0,
    );
    set_file_times(path, aged, aged).with_context(|| format!("set mtime {}", path.display()))?;
    Ok(())
}

fn load_registry(
    config: &RawConfig,
    client: &ClickHouseClient,
    root: &Path,
    dry_run: bool,
    summary: &mut RunSummary,
) -> Result<()> {
    let landing = root.join("landing/registry");
    if !landing.exists() {
        return Ok(());
    }
    for entry in fs::read_dir(&landing)? {
        let path = entry?.path();
        if !path.is_file()
            || path
                .extension()
                .and_then(|v| v.to_str())
                .unwrap_or("")
                .to_lowercase()
                != "xlsx"
        {
            continue;
        }
        summary.files_seen += 1;
        if !file_is_ready(&path, config)? {
            summary.files_skipped_young += 1;
            continue;
        }
        let rows = parse_registry_xlsx(&path)?;
        if !dry_run && !rows.is_empty() {
            let raw_rows: Vec<Value> = rows
                .iter()
                .map(|row| {
                    json!({
                        "source_file": row.get("source_file").and_then(Value::as_str).unwrap_or(""),
                        "source_sheet": row.get("source_sheet").and_then(Value::as_str).unwrap_or(""),
                        "payload": serde_json::to_string(row).unwrap_or_else(|_| "{}".to_string()),
                    })
                })
                .collect();
            client.insert_json_each_row(
                "raw_1c_company_registry",
                &["source_file", "source_sheet", "payload"],
                &raw_rows,
            )?;
            client.insert_json_each_row(
                "company_registry",
                &[
                    "ts",
                    "source_file",
                    "source_sheet",
                    "company_name",
                    "company_key",
                    "assignee_name",
                    "registry_status",
                    "share_text",
                    "key_contour",
                    "inn",
                    "kpp",
                ],
                &rows,
            )?;
        }
        summary.files_loaded += 1;
        summary.rows_loaded += rows.len();
        archive_or_delete(config, "registry", &path, dry_run, summary)?;
        println!("loaded registry: {} rows={}", file_name(&path), rows.len());
    }
    Ok(())
}

fn parse_registry_xlsx(path: &Path) -> Result<Vec<Value>> {
    let mut workbook =
        open_workbook_auto(path).with_context(|| format!("open xlsx {}", path.display()))?;
    let now = Utc::now()
        .naive_utc()
        .format("%Y-%m-%d %H:%M:%S")
        .to_string();
    let mut tax_map: HashMap<String, (String, String)> = HashMap::new();
    if let Ok(range) = workbook.worksheet_range("Лист2") {
        for row in range.rows().skip(2) {
            let company_name = cell_text(row.get(1));
            if company_name.is_empty() {
                continue;
            }
            tax_map.insert(
                normalize_company_key(&company_name),
                (cell_text(row.get(2)), cell_text(row.get(3))),
            );
        }
    }
    let mut out = Vec::new();
    let Ok(range) = workbook.worksheet_range("ОСНОВНОЙ") else {
        return Ok(out);
    };
    let rows: Vec<_> = range.rows().collect();
    if rows.len() < 3 {
        return Ok(out);
    }
    let top_headers: Vec<String> = rows[0].iter().map(|v| cell_text(Some(v))).collect();
    let manager_headers: Vec<String> = rows[1].iter().map(|v| cell_text(Some(v))).collect();
    let mut specs = Vec::<(usize, String, Option<usize>, String, String)>::new();
    let mut current: Option<(usize, String, String)> = None;
    for idx in 1..manager_headers.len() {
        let manager = manager_headers.get(idx).cloned().unwrap_or_default();
        let top = top_headers.get(idx).cloned().unwrap_or_default();
        if !manager.is_empty() {
            current = Some((idx, manager.clone(), "active".to_string()));
            specs.push((idx, manager, None, String::new(), "active".to_string()));
        } else if top.to_lowercase().contains("исключ") {
            current = Some((idx, String::new(), "excluded".to_string()));
            specs.push((
                idx,
                String::new(),
                None,
                String::new(),
                "excluded".to_string(),
            ));
        } else if let Some((col, assignee, status)) = current.take() {
            if let Some(last) = specs.last_mut() {
                if last.0 == col && last.1 == assignee && last.4 == status {
                    last.2 = Some(idx);
                    last.3 = top;
                }
            }
        }
    }
    for row in rows.into_iter().skip(2) {
        for (col, assignee, meta_col, meta_label, status) in &specs {
            let company_name = cell_text(row.get(*col));
            if company_name.is_empty() {
                continue;
            }
            let meta = meta_col
                .map(|idx| cell_text(row.get(idx)))
                .unwrap_or_default();
            let company_key = normalize_company_key(&company_name);
            let (inn, kpp) = tax_map.get(&company_key).cloned().unwrap_or_default();
            let key_contour = if meta.to_uppercase() == "ЕСТЬ"
                && meta_label.to_lowercase().contains("ключ")
            {
                1
            } else {
                0
            };
            out.push(json!({
                "ts": now,
                "source_file": file_name(path),
                "source_sheet": "ОСНОВНОЙ",
                "company_name": company_name,
                "company_key": company_key,
                "assignee_name": assignee,
                "registry_status": status,
                "share_text": if meta_label.to_lowercase().contains("ключ") { "" } else { &meta },
                "key_contour": key_contour,
                "inn": inn,
                "kpp": kpp,
            }));
        }
    }
    Ok(out)
}

fn cell_text(value: Option<&Data>) -> String {
    match value {
        Some(Data::String(v)) => v.trim().to_string(),
        Some(Data::Float(v)) => {
            if v.fract() == 0.0 {
                format!("{}", *v as i64)
            } else {
                v.to_string()
            }
        }
        Some(Data::Int(v)) => v.to_string(),
        Some(Data::Bool(v)) => v.to_string(),
        Some(Data::DateTime(v)) => v.to_string(),
        Some(Data::DateTimeIso(v)) => v.trim().to_string(),
        Some(Data::DurationIso(v)) => v.trim().to_string(),
        Some(Data::Empty) | None => String::new(),
        Some(other) => other.to_string(),
    }
}

fn archive_or_delete(
    config: &RawConfig,
    dataset: &str,
    path: &Path,
    dry_run: bool,
    summary: &mut RunSummary,
) -> Result<()> {
    if dry_run {
        return Ok(());
    }
    if let Some(root) = &config.archive_dir {
        let archive_root = root.join(dataset);
        fs::create_dir_all(&archive_root)
            .with_context(|| format!("create {}", archive_root.display()))?;
        let dest = unique_archive_path(&archive_root.join(file_name(path)));
        fs::rename(path, &dest)
            .or_else(|_| {
                fs::copy(path, &dest)?;
                fs::remove_file(path)
            })
            .with_context(|| format!("archive {} -> {}", path.display(), dest.display()))?;
        summary.files_archived += 1;
    } else if config.delete_after_load {
        fs::remove_file(path).with_context(|| format!("delete {}", path.display()))?;
    }
    Ok(())
}

fn unique_archive_path(path: &Path) -> PathBuf {
    if !path.exists() {
        return path.to_path_buf();
    }
    let parent = path.parent().unwrap_or_else(|| Path::new("."));
    let stem = path.file_stem().and_then(|v| v.to_str()).unwrap_or("file");
    let ext = path.extension().and_then(|v| v.to_str()).unwrap_or("");
    let suffix = Utc::now().format("%Y%m%d%H%M%S");
    if ext.is_empty() {
        parent.join(format!("{stem}.{suffix}"))
    } else {
        parent.join(format!("{stem}.{suffix}.{ext}"))
    }
}

fn run_sql_file(client: &ClickHouseClient, path: &Path, summary: &mut RunSummary) -> Result<()> {
    let sql = fs::read_to_string(path).with_context(|| format!("read SQL {}", path.display()))?;
    for statement in split_sql_statements(&sql) {
        client
            .execute(&statement)
            .with_context(|| format!("execute SQL {}", path.display()))?;
        summary.sql_statements += 1;
    }
    Ok(())
}

fn split_sql_statements(sql: &str) -> Vec<String> {
    let cleaned = sql
        .lines()
        .filter(|line| !line.trim_start().starts_with("--"))
        .collect::<Vec<_>>()
        .join("\n");
    cleaned
        .split(';')
        .map(str::trim)
        .filter(|item| !item.is_empty())
        .map(|item| format!("{item};"))
        .collect()
}

fn run_optional_script(path: &Path) -> Result<()> {
    if !path.exists() {
        return Ok(());
    }
    let status = Command::new(path)
        .status()
        .with_context(|| format!("run {}", path.display()))?;
    if !status.success() {
        bail!("{} exited with {}", path.display(), status);
    }
    Ok(())
}

fn derive_company_entity_key(row: &Map<String, Value>, index: &HashMap<String, String>) -> String {
    let explicit = s(row, "company_entity_key");
    if !explicit.is_empty() {
        return explicit;
    }
    let infobase = s(row, "infobase");
    index
        .get(&infobase)
        .cloned()
        .filter(|v| !v.is_empty())
        .unwrap_or_else(|| {
            canonical_company_entity_key(&s(row, "base_id"), &s(row, "base_path"), &infobase)
        })
}

fn canonical_company_entity_key(base_id: &str, base_path: &str, infobase: &str) -> String {
    if !base_id.trim().is_empty() {
        return format!("baseid:{}", base_id.trim());
    }
    if !base_path.trim().is_empty() {
        let normalized = normalize_base_path(base_path);
        if !normalized.is_empty() {
            return format!("basepath:{normalized}");
        }
    }
    let normalized = normalize_infobase_key(infobase);
    if normalized.is_empty() {
        String::new()
    } else {
        format!("infobase:{normalized}")
    }
}

fn normalize_base_path(value: &str) -> String {
    let re_slashes = Regex::new(r"/+").unwrap();
    let re_bad = Regex::new(r"[^0-9A-ZА-ЯЁ:/._ -]+").unwrap();
    let text = value.to_uppercase().replace('\\', "/");
    collapse_ws(&re_bad.replace_all(&re_slashes.replace_all(&text, "/"), " "))
}

fn normalize_infobase_key(value: &str) -> String {
    let re_year = Regex::new(r"(^|\s)20[0-9]{2}($|\s)").unwrap();
    let re_bad = Regex::new(r"[^0-9A-ZА-ЯЁ]+").unwrap();
    let text = value.to_uppercase();
    collapse_ws(&re_bad.replace_all(&re_year.replace_all(&text, " "), " "))
}

fn normalize_company_key(value: &str) -> String {
    let re_year = Regex::new(r"(^|\s)20\d{2}($|\s)").unwrap();
    let re_bad = Regex::new(r"[^0-9A-ZА-Я]+").unwrap();
    let text = value.to_uppercase().replace('Ё', "Е");
    collapse_ws(&re_bad.replace_all(&re_year.replace_all(&text, " "), " "))
}

fn collapse_ws(value: &str) -> String {
    value.split_whitespace().collect::<Vec<_>>().join(" ")
}

fn stable_id(prefix: &str, parts: &[&str]) -> String {
    let mut hasher = Sha1::new();
    hasher.update(parts.join("|").as_bytes());
    let digest = hasher.finalize();
    format!("{prefix}:{:x}", digest)[..prefix.len() + 1 + 16].to_string()
}

fn ts_value(row: &Map<String, Value>, keys: &[&str]) -> String {
    let raw = keys
        .iter()
        .find_map(|key| row.get(*key))
        .cloned()
        .unwrap_or(Value::Null);
    normalize_ts(&raw)
}

fn normalize_ts(value: &Value) -> String {
    let raw = value_to_string(value);
    if raw.trim().is_empty() {
        return Utc::now()
            .naive_utc()
            .format("%Y-%m-%d %H:%M:%S")
            .to_string();
    }
    if let Ok(ts) = DateTime::parse_from_rfc3339(&raw.replace('Z', "+00:00")) {
        return ts.naive_utc().format("%Y-%m-%d %H:%M:%S").to_string();
    }
    for fmt in [
        "%Y-%m-%d %H:%M:%S",
        "%Y-%m-%dT%H:%M:%S",
        "%d.%m.%Y %H:%M:%S",
    ] {
        if let Ok(ts) = NaiveDateTime::parse_from_str(&raw, fmt) {
            return ts.format("%Y-%m-%d %H:%M:%S").to_string();
        }
    }
    raw
}

fn s(row: &Map<String, Value>, key: &str) -> String {
    row.get(key)
        .map(value_to_string)
        .unwrap_or_default()
        .trim()
        .to_string()
}

fn s_alt(row: &Map<String, Value>, keys: &[&str]) -> String {
    keys.iter()
        .map(|key| s(row, key))
        .find(|value| !value.is_empty())
        .unwrap_or_default()
}

fn s_default(row: &Map<String, Value>, key: &str, default: &str) -> String {
    let value = s(row, key);
    if value.is_empty() {
        default.to_string()
    } else {
        value
    }
}

fn f(row: &Map<String, Value>, key: &str) -> f64 {
    row.get(key).and_then(value_to_f64).unwrap_or(0.0)
}

fn f_alt(row: &Map<String, Value>, keys: &[&str]) -> f64 {
    keys.iter()
        .find_map(|key| row.get(*key).and_then(value_to_f64))
        .unwrap_or(0.0)
}

fn u(row: &Map<String, Value>, key: &str) -> u64 {
    row.get(key).and_then(value_to_u64).unwrap_or(0)
}

fn u_alt(row: &Map<String, Value>, keys: &[&str]) -> Option<u64> {
    keys.iter()
        .find_map(|key| row.get(*key).and_then(value_to_u64))
}

fn value_to_string(value: &Value) -> String {
    match value {
        Value::Null => String::new(),
        Value::String(v) => v.clone(),
        Value::Number(v) => v.to_string(),
        Value::Bool(v) => {
            if *v {
                "1".to_string()
            } else {
                "0".to_string()
            }
        }
        other => other.to_string(),
    }
}

fn value_to_f64(value: &Value) -> Option<f64> {
    match value {
        Value::Number(v) => v.as_f64(),
        Value::String(v) => v.replace(',', ".").parse().ok(),
        Value::Bool(v) => Some(if *v { 1.0 } else { 0.0 }),
        _ => None,
    }
}

fn value_to_u64(value: &Value) -> Option<u64> {
    match value {
        Value::Number(v) => v.as_u64().or_else(|| v.as_f64().map(|v| v.max(0.0) as u64)),
        Value::String(v) => v.parse().ok(),
        Value::Bool(v) => Some(if *v { 1 } else { 0 }),
        _ => None,
    }
}

fn file_name(path: &Path) -> String {
    path.file_name()
        .and_then(|v| v.to_str())
        .unwrap_or("unknown")
        .to_string()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn canonical_company_key_matches_python_shape() {
        assert_eq!(
            canonical_company_entity_key("", "c:\\base\\ООО Тест 2026", ""),
            "basepath:C:/BASE/ООО ТЕСТ 2026"
        );
        assert_eq!(
            canonical_company_entity_key("", "", "НОРД 2026"),
            "infobase:НОРД"
        );
    }

    #[test]
    fn core_document_row_maps_defaults() {
        let mut row = Map::new();
        row.insert("infobase".to_string(), json!("БАЗА"));
        row.insert("amount".to_string(), json!("12.50"));
        let mapped = map_core_row("documents", "docs.jsonl", &row).unwrap();
        assert_eq!(mapped["infobase"], "БАЗА");
        assert_eq!(mapped["amount"], 12.5);
        assert_eq!(mapped["source_file"], "docs.jsonl");
    }

    #[test]
    fn split_sql_ignores_empty_chunks() {
        assert_eq!(split_sql_statements("SELECT 1; ; SELECT 2;").len(), 2);
    }
}