use std::{
    collections::{BTreeMap, BTreeSet},
    fs::File,
    path::Path,
};

use anyhow::{Context, Result};
use chrono::{DateTime, NaiveDate, NaiveDateTime};
use parquet::file::reader::{FileReader, SerializedFileReader};
use serde::{Serialize, Deserialize};

use crate::dataset::{FileFormat, detect_format};

const MAX_STORED_VALUE_LEN: usize = 128;
const MAX_EXAMPLES: usize = 100;
const MAX_CLASSIFY_LEN: usize = 128;
const PROGRESS_EVERY_ROWS: usize = 100_000;

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DatasetProfile {
    pub path: String,
    pub format: String,
    pub rows_profiled: usize,
    pub columns: Vec<ColumnProfile>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ColumnProfile {
    pub name: String,
    pub inferred_type: InferredType,
    pub sample_count: usize,
    pub null_count: usize,
    pub null_percent: f64,
    pub unique_count: usize,
    pub unique_count_capped: bool,
    pub invalid_count: usize,
    pub invalid_examples: Vec<String>,
    pub min_len: usize,
    pub max_len: usize,
    pub numeric_min: Option<f64>,
    pub numeric_max: Option<f64>,
    pub top_values: Vec<ValueCount>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum InferredType {
    Empty,
    Boolean,
    Integer,
    Float,
    Date,
    DateTime,
    String,
}

impl InferredType {
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Empty => "empty",
            Self::Boolean => "boolean",
            Self::Integer => "integer",
            Self::Float => "float",
            Self::Date => "date",
            Self::DateTime => "datetime",
            Self::String => "string",
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ValueCount {
    pub value: String,
    pub count: usize,
}

#[derive(Debug, Default, Clone)]
struct TypeCounts {
    boolean: usize,
    integer: usize,
    float: usize,
    date: usize,
    datetime: usize,
    string: usize,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ValueKind {
    Boolean,
    Integer,
    Float,
    Date,
    DateTime,
    String,
}

#[derive(Debug, Clone)]
struct ColumnAccumulator {
    name: String,
    sample_count: usize,
    null_count: usize,
    unique: BTreeSet<String>,
    unique_capped: bool,
    frequencies: BTreeMap<String, usize>,
    kinds: TypeCounts,
    examples: Vec<(String, ValueKind)>,
    min_len: usize,
    max_len: usize,
    numeric_min: Option<f64>,
    numeric_max: Option<f64>,
}

impl ColumnAccumulator {
    fn new(name: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            sample_count: 0,
            null_count: 0,
            unique: BTreeSet::new(),
            unique_capped: false,
            frequencies: BTreeMap::new(),
            kinds: TypeCounts::default(),
            examples: Vec::new(),
            min_len: usize::MAX,
            max_len: 0,
            numeric_min: None,
            numeric_max: None,
        }
    }

    fn ingest(&mut self, value: Option<&str>) {
        self.sample_count += 1;
        let value = value.unwrap_or("").trim();
        self.min_len = self.min_len.min(value.len());
        self.max_len = self.max_len.max(value.len());

        if is_null(value) {
            self.null_count += 1;
            return;
        }

        let stored_value = stored_value(value);
        self.unique.insert(stored_value.clone());
        *self.frequencies.entry(stored_value.clone()).or_insert(0) += 1;

        let kind = classify_value(value);
        self.kinds.add(kind);

        if self.examples.len() < MAX_EXAMPLES
            && !self
                .examples
                .iter()
                .any(|(example, _)| example == &stored_value)
        {
            self.examples.push((stored_value, kind));
        }

        if matches!(kind, ValueKind::Integer | ValueKind::Float)
            && let Ok(number) = value.parse::<f64>()
            && number.is_finite()
        {
            self.numeric_min = Some(self.numeric_min.map_or(number, |min| min.min(number)));
            self.numeric_max = Some(self.numeric_max.map_or(number, |max| max.max(number)));
        }
    }

    fn finish(self) -> ColumnProfile {
        let non_null_count = self.sample_count.saturating_sub(self.null_count);
        let inferred_type = infer_type(&self.kinds, non_null_count);
        let invalid_count = invalid_count(&self.kinds, inferred_type);
        let invalid_examples = invalid_examples(&self.examples, inferred_type);

        ColumnProfile {
            name: self.name,
            inferred_type,
            sample_count: self.sample_count,
            null_count: self.null_count,
            null_percent: percentage(self.null_count, self.sample_count),
            unique_count: self.unique.len(),
            unique_count_capped: self.unique_capped,
            invalid_count,
            invalid_examples,
            min_len: if self.min_len == usize::MAX {
                0
            } else {
                self.min_len
            },
            max_len: self.max_len,
            numeric_min: self.numeric_min,
            numeric_max: self.numeric_max,
            top_values: top_values(self.frequencies),
        }
    }
}

impl TypeCounts {
    fn add(&mut self, kind: ValueKind) {
        match kind {
            ValueKind::Boolean => self.boolean += 1,
            ValueKind::Integer => self.integer += 1,
            ValueKind::Float => self.float += 1,
            ValueKind::Date => self.date += 1,
            ValueKind::DateTime => self.datetime += 1,
            ValueKind::String => self.string += 1,
        }
    }
}

pub fn profile_path(path: &Path, limit: Option<usize>) -> Result<DatasetProfile> {
    profile_path_with_progress(path, limit, true)
}

pub fn profile_path_silent(path: &Path, limit: Option<usize>) -> Result<DatasetProfile> {
    profile_path_with_progress(path, limit, false)
}

pub fn profile_path_streaming<F>(
    path: &Path,
    limit: Option<usize>,
    mut progress_cb: F,
) -> Result<DatasetProfile>
where
    F: FnMut(&DatasetProfile),
{
    let format = detect_format(path)?;
    match format {
        FileFormat::Csv => profile_csv_streaming(path, limit, &mut progress_cb),
        FileFormat::Parquet => profile_parquet_streaming(path, limit, &mut progress_cb),
    }
}

fn should_report_progress(rows: usize) -> bool {
    if rows <= 10_000 {
        rows % 1_000 == 0
    } else {
        rows % 10_000 == 0
    }
}

fn profile_path_with_progress(
    path: &Path,
    limit: Option<usize>,
    show_progress: bool,
) -> Result<DatasetProfile> {
    let format = detect_format(path)?;
    match format {
        FileFormat::Csv => profile_csv(path, limit, show_progress),
        FileFormat::Parquet => profile_parquet(path, limit, show_progress),
    }
}

pub fn profile_column_iter<'a, I>(name: &str, rows: I, col: usize) -> ColumnProfile
where
    I: IntoIterator<Item = &'a Vec<String>>,
{
    let mut accumulator = ColumnAccumulator::new(name);
    for row in rows {
        accumulator.ingest(row.get(col).map(String::as_str));
    }
    accumulator.finish()
}

fn profile_csv(path: &Path, limit: Option<usize>, show_progress: bool) -> Result<DatasetProfile> {
    let mut reader = csv::Reader::from_path(path)
        .with_context(|| format!("failed to open CSV: {}", path.display()))?;
    let headers = reader
        .headers()
        .with_context(|| format!("failed to read CSV header: {}", path.display()))?
        .iter()
        .map(ToOwned::to_owned)
        .collect::<Vec<_>>();
    let mut columns = headers
        .iter()
        .map(ColumnAccumulator::new)
        .collect::<Vec<_>>();
    let mut rows_profiled = 0;
    if show_progress {
        eprintln!(
            "profiling csv: {} cols, {}",
            headers.len(),
            limit_text(limit, None)
        );
    }

    for (idx, record) in reader.records().enumerate() {
        if limit.is_some_and(|limit| idx >= limit) {
            break;
        }

        let record =
            record.with_context(|| format!("failed to read CSV row: {}", path.display()))?;
        for (col_idx, column) in columns.iter_mut().enumerate() {
            column.ingest(record.get(col_idx));
        }
        rows_profiled += 1;
        if show_progress {
            progress(rows_profiled, limit);
        }
    }

    Ok(DatasetProfile {
        path: path.display().to_string(),
        format: "csv".to_string(),
        rows_profiled,
        columns: columns.into_iter().map(ColumnAccumulator::finish).collect(),
    })
}

fn profile_parquet(
    path: &Path,
    limit: Option<usize>,
    show_progress: bool,
) -> Result<DatasetProfile> {
    let file =
        File::open(path).with_context(|| format!("failed to open Parquet: {}", path.display()))?;
    let reader = SerializedFileReader::new(file)
        .with_context(|| format!("failed to read Parquet metadata: {}", path.display()))?;
    let total_rows = usize::try_from(reader.metadata().file_metadata().num_rows()).ok();
    let headers = reader
        .metadata()
        .file_metadata()
        .schema_descr()
        .columns()
        .iter()
        .map(|col| col.name().to_string())
        .collect::<Vec<_>>();
    let mut columns = headers
        .iter()
        .map(ColumnAccumulator::new)
        .collect::<Vec<_>>();
    let mut row_iter = reader
        .get_row_iter(None)
        .with_context(|| format!("failed to read Parquet rows: {}", path.display()))?;
    let mut rows_profiled = 0;
    if show_progress {
        eprintln!(
            "profiling parquet: {} cols, {}",
            headers.len(),
            limit_text(limit, total_rows)
        );
    }

    for (idx, row) in (&mut row_iter).enumerate() {
        if limit.is_some_and(|limit| idx >= limit) {
            break;
        }

        let row =
            row.with_context(|| format!("failed to decode Parquet row: {}", path.display()))?;
        let values = row
            .get_column_iter()
            .map(|(_, field)| field.to_string())
            .collect::<Vec<_>>();
        for (col_idx, column) in columns.iter_mut().enumerate() {
            column.ingest(values.get(col_idx).map(String::as_str));
        }
        rows_profiled += 1;
        if show_progress {
            progress(rows_profiled, limit.or(total_rows));
        }
    }

    Ok(DatasetProfile {
        path: path.display().to_string(),
        format: "parquet".to_string(),
        rows_profiled,
        columns: columns.into_iter().map(ColumnAccumulator::finish).collect(),
    })
}

pub fn print_text(profile: &DatasetProfile) {
    println!("{} ({})", profile.path, profile.format);
    println!(
        "rows profiled: {} | columns: {}",
        profile.rows_profiled,
        profile.columns.len()
    );
    println!();
    println!(
        "{:<24} {:<10} {:>8} {:>8} {:>9} {:>8}",
        "column", "type", "nulls", "null%", "unique", "invalid"
    );
    println!("{}", "-".repeat(76));

    for column in &profile.columns {
        println!(
            "{:<24} {:<10} {:>8} {:>7.1}% {:>9} {:>8}",
            truncate_for_text(&column.name, 24),
            column.inferred_type.as_str(),
            column.null_count,
            column.null_percent,
            unique_count_text(column),
            column.invalid_count
        );
    }
}

fn infer_type(counts: &TypeCounts, non_null_count: usize) -> InferredType {
    if non_null_count == 0 {
        return InferredType::Empty;
    }

    let float_compatible = counts.integer + counts.float;
    let datetime_compatible = counts.date + counts.datetime;

    if counts.boolean == non_null_count {
        InferredType::Boolean
    } else if counts.integer == non_null_count {
        InferredType::Integer
    } else if float_compatible == non_null_count {
        InferredType::Float
    } else if counts.date == non_null_count {
        InferredType::Date
    } else if datetime_compatible == non_null_count {
        InferredType::DateTime
    } else if counts.boolean * 100 >= non_null_count * 80 {
        InferredType::Boolean
    } else if counts.integer * 100 >= non_null_count * 80 && counts.float == 0 {
        InferredType::Integer
    } else if float_compatible * 100 >= non_null_count * 80 {
        InferredType::Float
    } else if counts.date * 100 >= non_null_count * 80 && counts.datetime == 0 {
        InferredType::Date
    } else if datetime_compatible * 100 >= non_null_count * 80 {
        InferredType::DateTime
    } else {
        InferredType::String
    }
}

fn invalid_examples(values: &[(String, ValueKind)], inferred_type: InferredType) -> Vec<String> {
    if inferred_type == InferredType::String || inferred_type == InferredType::Empty {
        return Vec::new();
    }

    let mut examples = Vec::new();
    for (value, kind) in values {
        if !is_compatible(*kind, inferred_type) && !examples.contains(value) {
            examples.push(value.clone());
        }

        if examples.len() >= 5 {
            break;
        }
    }

    examples
}

fn invalid_count(counts: &TypeCounts, inferred_type: InferredType) -> usize {
    counts.boolean + counts.integer + counts.float + counts.date + counts.datetime + counts.string
        - compatible_count(counts, inferred_type)
}

fn compatible_count(counts: &TypeCounts, inferred_type: InferredType) -> usize {
    match inferred_type {
        InferredType::Boolean => counts.boolean,
        InferredType::Integer => counts.integer,
        InferredType::Float => counts.integer + counts.float,
        InferredType::Date => counts.date,
        InferredType::DateTime => counts.date + counts.datetime,
        InferredType::String | InferredType::Empty => {
            counts.boolean
                + counts.integer
                + counts.float
                + counts.date
                + counts.datetime
                + counts.string
        }
    }
}

fn is_compatible(kind: ValueKind, inferred_type: InferredType) -> bool {
    match inferred_type {
        InferredType::Boolean => kind == ValueKind::Boolean,
        InferredType::Integer => kind == ValueKind::Integer,
        InferredType::Float => matches!(kind, ValueKind::Integer | ValueKind::Float),
        InferredType::Date => kind == ValueKind::Date,
        InferredType::DateTime => matches!(kind, ValueKind::Date | ValueKind::DateTime),
        InferredType::String | InferredType::Empty => true,
    }
}

fn classify_value(value: &str) -> ValueKind {
    if value.len() > MAX_CLASSIFY_LEN {
        return ValueKind::String;
    }

    if parse_bool(value) {
        ValueKind::Boolean
    } else if value.parse::<i64>().is_ok() {
        ValueKind::Integer
    } else if value.parse::<f64>().is_ok_and(f64::is_finite) {
        ValueKind::Float
    } else if parse_datetime(value) {
        ValueKind::DateTime
    } else if parse_date(value) {
        ValueKind::Date
    } else {
        ValueKind::String
    }
}

fn parse_bool(value: &str) -> bool {
    matches!(
        value.to_ascii_lowercase().as_str(),
        "true" | "false" | "yes" | "no" | "y" | "n"
    )
}

fn parse_date(value: &str) -> bool {
    ["%Y-%m-%d", "%Y/%m/%d", "%m/%d/%Y", "%d-%m-%Y"]
        .iter()
        .any(|format| NaiveDate::parse_from_str(value, format).is_ok())
}

fn parse_datetime(value: &str) -> bool {
    DateTime::parse_from_rfc3339(value).is_ok()
        || [
            "%Y-%m-%d %H:%M:%S",
            "%Y-%m-%d %H:%M",
            "%Y/%m/%d %H:%M:%S",
            "%Y/%m/%d %H:%M",
        ]
        .iter()
        .any(|format| NaiveDateTime::parse_from_str(value, format).is_ok())
}

fn is_null(value: &str) -> bool {
    value.is_empty()
        || matches!(
            value.to_ascii_lowercase().as_str(),
            "null" | "nil" | "none" | "na" | "n/a"
        )
}

fn top_values(frequencies: BTreeMap<String, usize>) -> Vec<ValueCount> {
    let mut values = frequencies
        .into_iter()
        .map(|(value, count)| ValueCount { value, count })
        .collect::<Vec<_>>();
    values.sort_by(|left, right| {
        right
            .count
            .cmp(&left.count)
            .then_with(|| left.value.cmp(&right.value))
    });
    values.truncate(5);
    values
}

fn stored_value(value: &str) -> String {
    truncate_for_text(value, MAX_STORED_VALUE_LEN)
}

fn unique_count_text(column: &ColumnProfile) -> String {
    if column.unique_count_capped {
        format!("{}+", column.unique_count)
    } else {
        column.unique_count.to_string()
    }
}

fn limit_text(limit: Option<usize>, total_rows: Option<usize>) -> String {
    match (limit, total_rows) {
        (Some(limit), Some(total)) => {
            format!("profiling up to {} of {} rows", limit.min(total), total)
        }
        (Some(limit), None) => format!("profiling up to {limit} rows"),
        (None, Some(total)) => format!("profiling all {total} rows"),
        (None, None) => "profiling all rows".to_string(),
    }
}

fn progress(rows_profiled: usize, total_rows: Option<usize>) {
    if rows_profiled % PROGRESS_EVERY_ROWS != 0 {
        return;
    }

    match total_rows {
        Some(total) => eprintln!("profiled {rows_profiled}/{total} rows"),
        None => eprintln!("profiled {rows_profiled} rows"),
    }
}

fn percentage(part: usize, total: usize) -> f64 {
    if total == 0 {
        0.0
    } else {
        part as f64 * 100.0 / total as f64
    }
}

fn truncate_for_text(value: &str, max: usize) -> String {
    if value.chars().count() <= max {
        return value.to_string();
    }

    let mut out = value
        .chars()
        .take(max.saturating_sub(3))
        .collect::<String>();
    out.push_str("...");
    out
}

fn profile_csv_streaming<F>(
    path: &Path,
    limit: Option<usize>,
    progress_cb: &mut F,
) -> Result<DatasetProfile>
where
    F: FnMut(&DatasetProfile),
{
    let mut reader = csv::Reader::from_path(path)
        .with_context(|| format!("failed to open CSV: {}", path.display()))?;
    let headers = reader
        .headers()
        .with_context(|| format!("failed to read CSV header: {}", path.display()))?
        .iter()
        .map(ToOwned::to_owned)
        .collect::<Vec<_>>();
    let mut columns = headers
        .iter()
        .map(ColumnAccumulator::new)
        .collect::<Vec<_>>();
    let mut rows_profiled = 0;

    for (idx, record) in reader.records().enumerate() {
        if limit.is_some_and(|limit| idx >= limit) {
            break;
        }

        let record =
            record.with_context(|| format!("failed to read CSV row: {}", path.display()))?;
        for (col_idx, column) in columns.iter_mut().enumerate() {
            column.ingest(record.get(col_idx));
        }
        rows_profiled += 1;

        if should_report_progress(rows_profiled) {
            let partial = DatasetProfile {
                path: path.display().to_string(),
                format: "csv".to_string(),
                rows_profiled,
                columns: columns.iter().map(|col| col.clone().finish()).collect(),
            };
            progress_cb(&partial);
        }
    }

    Ok(DatasetProfile {
        path: path.display().to_string(),
        format: "csv".to_string(),
        rows_profiled,
        columns: columns.into_iter().map(ColumnAccumulator::finish).collect(),
    })
}

fn profile_parquet_streaming<F>(
    path: &Path,
    limit: Option<usize>,
    progress_cb: &mut F,
) -> Result<DatasetProfile>
where
    F: FnMut(&DatasetProfile),
{
    let file =
        File::open(path).with_context(|| format!("failed to open Parquet: {}", path.display()))?;
    let reader = SerializedFileReader::new(file)
        .with_context(|| format!("failed to read Parquet metadata: {}", path.display()))?;
    let headers = reader
        .metadata()
        .file_metadata()
        .schema_descr()
        .columns()
        .iter()
        .map(|col| col.name().to_string())
        .collect::<Vec<_>>();
    let mut columns = headers
        .iter()
        .map(ColumnAccumulator::new)
        .collect::<Vec<_>>();
    let mut row_iter = reader
        .get_row_iter(None)
        .with_context(|| format!("failed to read Parquet rows: {}", path.display()))?;
    let mut rows_profiled = 0;

    for (idx, row) in (&mut row_iter).enumerate() {
        if limit.is_some_and(|limit| idx >= limit) {
            break;
        }

        let row =
            row.with_context(|| format!("failed to decode Parquet row: {}", path.display()))?;
        let values = row
            .get_column_iter()
            .map(|(_, field)| field.to_string())
            .collect::<Vec<_>>();
        for (col_idx, column) in columns.iter_mut().enumerate() {
            column.ingest(values.get(col_idx).map(String::as_str));
        }
        rows_profiled += 1;

        if should_report_progress(rows_profiled) {
            let partial = DatasetProfile {
                path: path.display().to_string(),
                format: "parquet".to_string(),
                rows_profiled,
                columns: columns.iter().map(|col| col.clone().finish()).collect(),
            };
            progress_cb(&partial);
        }
    }

    Ok(DatasetProfile {
        path: path.display().to_string(),
        format: "parquet".to_string(),
        rows_profiled,
        columns: columns.into_iter().map(ColumnAccumulator::finish).collect(),
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::{fs, time::SystemTime};

    fn rows(values: &[&str]) -> Vec<Vec<String>> {
        values.iter().map(|value| vec![value.to_string()]).collect()
    }

    #[test]
    fn infers_integer_column() {
        let rows = rows(&["1", "2", "3"]);
        let profile = profile_column_iter("id", rows.iter(), 0);
        assert_eq!(profile.inferred_type, InferredType::Integer);
        assert_eq!(profile.numeric_min, Some(1.0));
        assert_eq!(profile.numeric_max, Some(3.0));
    }

    #[test]
    fn infers_float_from_integer_and_float() {
        let rows = rows(&["1", "2.5", "3"]);
        let profile = profile_column_iter("amount", rows.iter(), 0);
        assert_eq!(profile.inferred_type, InferredType::Float);
    }

    #[test]
    fn reports_invalid_examples_for_dominant_type() {
        let rows = rows(&["1", "2", "3", "bad", "4"]);
        let profile = profile_column_iter("mostly_int", rows.iter(), 0);
        assert_eq!(profile.inferred_type, InferredType::Integer);
        assert_eq!(profile.invalid_count, 1);
        assert_eq!(profile.invalid_examples, vec!["bad"]);
    }

    #[test]
    fn tracks_top_values() {
        let rows = rows(&["paid", "new", "paid"]);
        let profile = profile_column_iter("status", rows.iter(), 0);
        assert_eq!(profile.top_values[0].value, "paid");
        assert_eq!(profile.top_values[0].count, 2);
    }

    #[test]
    fn profile_path_streams_csv_with_limit() {
        let stamp = SystemTime::now()
            .duration_since(SystemTime::UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let path = std::env::temp_dir().join(format!("dap-profile-{stamp}.csv"));
        fs::write(
            &path,
            "id,status,amount\n1,new,10.5\n2,paid,20\n3,paid,30\n",
        )
        .unwrap();

        let profile = profile_path(&path, Some(2)).unwrap();
        fs::remove_file(&path).unwrap();

        assert_eq!(profile.rows_profiled, 2);
        assert_eq!(profile.columns[0].inferred_type, InferredType::Integer);
        assert_eq!(profile.columns[1].top_values[0].value, "new");
        assert_eq!(profile.columns[2].inferred_type, InferredType::Float);
    }

    #[test]
    fn profile_path_streaming_sends_progress() {
        let stamp = SystemTime::now()
            .duration_since(SystemTime::UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let path = std::env::temp_dir().join(format!("dap-profile-stream-{stamp}.csv"));
        let mut contents = String::from("id,val\n");
        for i in 1..=1005 {
            contents.push_str(&format!("{i},val-{i}\n"));
        }
        fs::write(&path, contents).unwrap();

        let mut progress_counts = Vec::new();
        let final_profile = profile_path_streaming(&path, None, |partial| {
            progress_counts.push(partial.rows_profiled);
        })
        .unwrap();
        fs::remove_file(&path).unwrap();

        assert_eq!(final_profile.rows_profiled, 1005);
        // We should have received progress at row 1000, since it's a multiple of 1000 and <= 10000.
        assert_eq!(progress_counts, vec![1000]);
    }
}
