dataprof-runtime 0.13.1

use std::sync::OnceLock;

use dataprof_core::{
    AnalysisOptions, ColumnProfile, DataSource, ExecutionMetadata, KeyColumnSource,
    KeyUniquenessUnits, Locale, NullTokenSet, OutlierUnits, PatternMatchCount, QualityScoreWeights,
    RangeRuleCount, SemanticHintBinding, ShapeMomentEstimator, TextLengthUnit,
};
use dataprof_metrics::{
    AccuracyMetrics, CompletenessMetrics, ConsistencyMetrics, MetricConfidence, PrecisionMetrics,
    QualityAssessment, QualityMetrics, TimelinessMetrics, ValidityMetrics,
};

/// Version of the serialized `ProfileReport` schema written by this build.
///
/// This is intentionally independent of the package version: the document
/// format only changes when the schema itself changes, not on every release.
///
/// Compatibility policy for readers:
/// - Documents without a `schema_version` field are legacy pre-0.10 reports
///   and deserialize with `schema_version == 0`.
/// - Unknown *additive* fields written by a newer dataprof are ignored, so a
///   reader accepts any document whose `schema_version` is at most this
///   constant.
/// - A document with a `schema_version` greater than this constant fails to
///   deserialize with an explicit error instead of being partially decoded
///   into a plausible-but-wrong report.
pub const REPORT_SCHEMA_VERSION: u32 = 1;

/// Why a report does or does not carry a quality assessment.
///
/// `quality` alone cannot answer that. A run that never asked for quality
/// metrics and a run whose quality computation failed both leave it `None`,
/// so absence itself became the plausible value that hid the failure. This
/// names the difference: a consumer deciding on a report can tell "you did
/// not ask for this" from "this broke" from "there was nothing to measure".
#[derive(
    Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, schemars::JsonSchema,
)]
#[serde(tag = "state", rename_all = "snake_case")]
pub enum QualityAnalysisStatus {
    /// Computed: `quality` carries the assessment.
    Computed,
    /// Not requested for this run — the quality pack was deselected.
    NotRequested,
    /// Requested, but no quality sample was supplied to the assembler.
    /// An empty sample is analyzed and reports `Computed`, even without columns.
    NoData,
    /// Requested, but every requested dimension measures whole rows and the
    /// run profiled a subset of columns. Completeness and uniqueness mean
    /// something else after projection, and the report cannot label only
    /// their row-level fields as projected, so they are withheld rather than
    /// published under full-row names.
    WithheldByProjection,
    /// Requested and attempted; the computation failed. `quality` is absent
    /// because the computation broke, not because nothing was asked for.
    Failed {
        /// The error the metrics calculator reported.
        error: String,
    },
    /// Written by a release that did not record this. Only reachable by
    /// deserializing a document from before the field existed that carries no
    /// quality assessment; a stored assessment is read back as `Computed`.
    Unrecorded,
}

/// How the measurements in a report were defined.
///
/// `schema_version` answers whether a document validates. It cannot answer
/// whether two valid documents measured the same way: text lengths counted
/// UTF-8 bytes through 0.11 and count Unicode scalar values since, with no
/// change to the document's shape. This records each intentionally changed
/// definition by name, so a comparison can check exactly the ones both
/// reports declare.
///
/// Every field is optional. A field missing from a stored report means the
/// release that wrote it did not record that definition: unknown, never the
/// current one. Later releases add fields, which older readers ignore.
#[derive(
    Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, schemars::JsonSchema,
)]
#[non_exhaustive]
pub struct MetricSemantics {
    /// Unit of `min_length`, `max_length` and `avg_length` on text columns.
    ///
    /// Absent means not recorded; an explicit null is malformed, so neither
    /// the reader nor the schema accepts one.
    #[serde(
        default,
        skip_serializing_if = "Option::is_none",
        deserialize_with = "present_value"
    )]
    #[schemars(with = "TextLengthUnit")]
    pub text_length_unit: Option<TextLengthUnit>,
    /// The text values counted as null, which every `null_count` and every
    /// completeness figure depend on.
    ///
    /// Absent means not recorded: the release that wrote the report counted
    /// only empty, `null` and `nan`, or recorded nothing at all.
    #[serde(
        default,
        skip_serializing_if = "Option::is_none",
        deserialize_with = "present_value"
    )]
    #[schemars(with = "NullTokenSet")]
    pub null_tokens: Option<NullTokenSet>,
    /// What every pattern's `match_count` and `match_percentage` count.
    ///
    /// Absent means not recorded: through 0.12 they counted every regular
    /// expression match, validated or not.
    #[serde(
        default,
        skip_serializing_if = "Option::is_none",
        deserialize_with = "present_value"
    )]
    #[schemars(with = "PatternMatchCount")]
    pub pattern_match_count: Option<PatternMatchCount>,
    /// The estimator behind every column's `skewness` and `kurtosis`.
    ///
    /// Absent means not recorded: through 0.12 both used a form that matches
    /// neither the population nor the adjusted estimators.
    #[serde(
        default,
        skip_serializing_if = "Option::is_none",
        deserialize_with = "present_value"
    )]
    #[schemars(with = "ShapeMomentEstimator")]
    pub shape_moments: Option<ShapeMomentEstimator>,
    /// What the range and sign rules behind the accuracy score count.
    ///
    /// Absent means not recorded: through 0.12 every rule failure counted,
    /// over the numeric values of every column.
    #[serde(
        default,
        skip_serializing_if = "Option::is_none",
        deserialize_with = "present_value"
    )]
    #[schemars(with = "RangeRuleCount")]
    pub range_rule_count: Option<RangeRuleCount>,
    /// What `key_uniqueness` divides the key's distinct values by.
    ///
    /// Absent means not recorded: through 0.12 it divided by every row,
    /// nulls included.
    #[serde(
        default,
        skip_serializing_if = "Option::is_none",
        deserialize_with = "present_value"
    )]
    #[schemars(with = "KeyUniquenessUnits")]
    pub key_uniqueness_units: Option<KeyUniquenessUnits>,
    /// What `outlier_ratio` is measured over.
    ///
    /// Absent means not recorded: through 0.12 an untested source read 0 and
    /// still assessed accuracy.
    #[serde(
        default,
        skip_serializing_if = "Option::is_none",
        deserialize_with = "present_value"
    )]
    #[schemars(with = "OutlierUnits")]
    pub outlier_units: Option<OutlierUnits>,
    /// Which column `key_uniqueness` may measure.
    ///
    /// Absent means not recorded: through 0.12 the first column with an
    /// id-like name was the key when none was declared.
    #[serde(
        default,
        skip_serializing_if = "Option::is_none",
        deserialize_with = "present_value"
    )]
    #[schemars(with = "KeyColumnSource")]
    pub key_column_source: Option<KeyColumnSource>,
}

/// Decode a field that may be absent (serde's `default` covers that) but,
/// when present, must carry a value: `Option<T>` alone reads `null` as `None`.
pub(crate) fn present_value<'de, D, T>(deserializer: D) -> Result<Option<T>, D::Error>
where
    D: serde::Deserializer<'de>,
    T: serde::Deserialize<'de>,
{
    T::deserialize(deserializer).map(Some)
}

impl MetricSemantics {
    /// The definitions this build measures with.
    pub fn current() -> Self {
        Self {
            text_length_unit: Some(TextLengthUnit::UnicodeScalar),
            null_tokens: Some(NullTokenSet::CommonMarkers),
            pattern_match_count: Some(PatternMatchCount::Validated),
            shape_moments: Some(ShapeMomentEstimator::AdjustedSample),
            range_rule_count: Some(RangeRuleCount::ViolatingValues),
            key_uniqueness_units: Some(KeyUniquenessUnits::NonNullValues),
            outlier_units: Some(OutlierUnits::TestedValues),
            key_column_source: Some(KeyColumnSource::Declared),
        }
    }

    /// Whether every definition this build knows is recorded. A comparison
    /// can only vouch for definitions both reports declare, so an unrecorded
    /// one leaves it unknown. Destructured, so a new field fails to compile
    /// here until it is counted.
    pub(crate) fn is_fully_recorded(&self) -> bool {
        let Self {
            text_length_unit,
            null_tokens,
            pattern_match_count,
            shape_moments,
            range_rule_count,
            key_uniqueness_units,
            outlier_units,
            key_column_source,
        } = self;
        text_length_unit.is_some()
            && null_tokens.is_some()
            && pattern_match_count.is_some()
            && shape_moments.is_some()
            && range_rule_count.is_some()
            && key_uniqueness_units.is_some()
            && outlier_units.is_some()
            && key_column_source.is_some()
    }
}

/// The locale a report's detected patterns were ranked with: `none`, or the
/// ISO 3166-1 alpha-2 code of a [`Locale`] the pattern catalogue supports.
#[derive(
    Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, schemars::JsonSchema,
)]
#[non_exhaustive]
pub enum PatternLocale {
    /// No locale: patterns were ranked without a locale preference.
    #[serde(rename = "none")]
    Unset,
    /// Canada.
    #[serde(rename = "CA")]
    Ca,
    /// Germany.
    #[serde(rename = "DE")]
    De,
    /// France.
    #[serde(rename = "FR")]
    Fr,
    /// United Kingdom.
    #[serde(rename = "GB")]
    Gb,
    /// Italy.
    #[serde(rename = "IT")]
    It,
    /// United States.
    #[serde(rename = "US")]
    Us,
}

impl From<Option<Locale>> for PatternLocale {
    fn from(locale: Option<Locale>) -> Self {
        match locale {
            None => Self::Unset,
            Some(Locale::Ca) => Self::Ca,
            Some(Locale::De) => Self::De,
            Some(Locale::Fr) => Self::Fr,
            Some(Locale::Gb) => Self::Gb,
            Some(Locale::It) => Self::It,
            Some(Locale::Us) => Self::Us,
        }
    }
}

impl PatternLocale {
    /// The locale, or `None` when patterns were ranked without one.
    pub fn locale(self) -> Option<Locale> {
        match self {
            Self::Unset => None,
            Self::Ca => Some(Locale::Ca),
            Self::De => Some(Locale::De),
            Self::Fr => Some(Locale::Fr),
            Self::Gb => Some(Locale::Gb),
            Self::It => Some(Locale::It),
            Self::Us => Some(Locale::Us),
        }
    }
}

/// Which of the source's columns a run profiled.
///
/// A projection that names a column the source lacks is refused before
/// profiling, so a report with [`ColumnSelection::All`] that has no profile
/// for a column shows the source has no such column, and one with
/// [`ColumnSelection::Listed`] shows only that the caller did not select it.
#[derive(
    Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, schemars::JsonSchema,
)]
#[serde(tag = "selection", rename_all = "snake_case")]
#[non_exhaustive]
pub enum ColumnSelection {
    /// Every column of the source.
    All,
    /// Only the named columns.
    #[non_exhaustive]
    Listed {
        /// The names as the caller gave them.
        names: Vec<String>,
    },
}

impl ColumnSelection {
    /// The selection `options` profile with.
    pub fn from_options(options: &AnalysisOptions) -> Self {
        match options.columns() {
            None => Self::All,
            Some(names) => Self::Listed {
                names: names.to_vec(),
            },
        }
    }

    /// Whether the selection includes `column`, which a source that has the
    /// column profiles.
    pub fn includes(&self, column: &str) -> bool {
        match self {
            Self::All => true,
            Self::Listed { names } => names.iter().any(|name| name == column),
        }
    }
}

/// The analysis settings a report was profiled with that change what its
/// scores measure on unchanged data and are recorded nowhere else in it.
///
/// [`MetricSemantics`] records how a release defines its measurements; this
/// records what the caller chose. The score weights live in the quality
/// assessment and the semantic hints in the hint bindings, so they are not
/// repeated here. Later releases add fields, which older readers ignore, and a
/// field missing from a stored report means the release did not record it.
#[derive(
    Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, schemars::JsonSchema,
)]
#[non_exhaustive]
pub struct AnalysisSettings {
    /// The locale detected patterns were ranked with, which the validity
    /// score reads.
    pub pattern_locale: PatternLocale,
    /// The columns the run profiled. A dataset-level score has the column
    /// set as its subject, and a report without a column's profile needs this
    /// to tell a column the source lacks from one that was not selected.
    pub columns: ColumnSelection,
}

impl AnalysisSettings {
    /// The settings `options` profile with.
    pub fn from_options(options: &AnalysisOptions) -> Self {
        Self {
            pattern_locale: options.locale().into(),
            columns: ColumnSelection::from_options(options),
        }
    }
}

/// The dataprof build that produced a report.
///
/// `version` alone cannot tell a development build from the release it
/// follows: both carry the workspace version until the next release bumps it.
/// The commit and the dirty flag tell them apart where the build knows them.
///
/// Absent `commit` and `dirty` mean the build did not record them: a crate
/// built from a package rather than a git checkout, or a build without a
/// usable `git`. An explicit null is malformed, so neither the reader nor the
/// schema accepts one.
#[derive(
    Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize, schemars::JsonSchema,
)]
#[non_exhaustive]
pub struct ReportGenerator {
    /// The dataprof package version, such as `"0.13.0"`.
    pub version: String,
    /// The full git commit hash the build was compiled from.
    #[serde(
        default,
        skip_serializing_if = "Option::is_none",
        deserialize_with = "present_value"
    )]
    #[schemars(with = "String")]
    pub commit: Option<String>,
    /// Whether the compiled sources (`Cargo.toml`, `Cargo.lock` and `crates/`)
    /// had uncommitted changes when the build ran.
    #[serde(
        default,
        skip_serializing_if = "Option::is_none",
        deserialize_with = "present_value"
    )]
    #[schemars(with = "bool")]
    pub dirty: Option<bool>,
}

/// The commit and dirty flag the embedding build recorded, if any.
static BUILD_IDENTITY: OnceLock<(Option<&'static str>, Option<bool>)> = OnceLock::new();

/// Record the build identity of the binary embedding this crate.
///
/// A build script here would rerun on every source change and rebuild every
/// crate that depends on this one, so the leaf crate that owns the build
/// (the Python extension) reads git and records the result once, at load.
/// Only the first call takes effect; reports built before it, and in builds
/// that never call it, record the version alone.
#[doc(hidden)]
pub fn record_build_identity(commit: Option<&'static str>, dirty: Option<bool>) {
    let _ = BUILD_IDENTITY.set((commit, dirty));
}

impl ReportGenerator {
    /// The build running now.
    pub fn current() -> Self {
        let (commit, dirty) = BUILD_IDENTITY.get().copied().unwrap_or((None, None));
        Self {
            version: env!("CARGO_PKG_VERSION").to_string(),
            commit: commit.map(str::to_string),
            dirty,
        }
    }
}

/// Complete profiling report for a data source.
///
/// Contains column-level statistics, execution metadata, and an optional
/// Quality assessment informed by ISO 8000/25012 concepts. This is the primary output of all
/// profiling operations (`Profiler::analyze_file`, `Profiler::analyze_source`,
/// `Profiler::profile_stream`, etc.).
#[derive(Debug, Clone, serde::Serialize, schemars::JsonSchema)]
pub struct ProfileReport {
    /// Version of the serialized report schema (see [`REPORT_SCHEMA_VERSION`]).
    ///
    /// `0` means the document predates schema versioning (a 0.9-era report).
    /// Deserialization rejects versions newer than [`REPORT_SCHEMA_VERSION`].
    #[schemars(schema_with = "schema_version_schema")]
    pub schema_version: u32,
    /// Unique identifier for this report (UUID v4)
    pub id: String,
    /// Timestamp when the report was generated (ISO 8601 / RFC 3339)
    pub timestamp: String,
    /// Data source metadata (file, query, etc.)
    pub data_source: DataSource,
    /// Column-level profiling results
    pub column_profiles: Vec<ColumnProfile>,
    /// Execution metadata (timing, rows processed, truncation info, etc.)
    pub execution: ExecutionMetadata,
    /// Data quality assessment (optional — partial analysis may skip quality)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub quality: Option<QualityAssessment>,
    /// What happened to the quality computation. Always present: it is the
    /// reason `quality` is or is not there, and a report must never leave a
    /// failed computation to a log line. Additive field — documents written
    /// before it deserialize as [`QualityAnalysisStatus::Unrecorded`], or as
    /// `Computed` when they carry an assessment.
    ///
    /// This and `quality` are two halves of one fact:
    /// [`Computed`](QualityAnalysisStatus::Computed) accompanies an assessment
    /// and every other status accompanies its absence. Construction goes
    /// through [`ReportAssembler`](crate::ReportAssembler), which derives one
    /// from the other, and deserialization rejects a document that pairs them
    /// any other way. Assigning either field directly can still break the
    /// pair, and such a report serializes into a document this crate will
    /// refuse to read back.
    pub quality_status: QualityAnalysisStatus,
    /// Per-column evidence of how each semantic hint bound to the data.
    ///
    /// Empty when no hints were supplied. Recorded for provenance: a hint proven
    /// inert over the full data is rejected before a report is returned, so a
    /// successful report only carries bindings that matched something or whose
    /// evidence was sampled. Additive field — older readers ignore it.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub semantic_hint_bindings: Vec<SemanticHintBinding>,
    /// How this report's measurements were defined; see [`MetricSemantics`].
    ///
    /// Every report this build produces records [`MetricSemantics::current`].
    /// `None` is a document written before dataprof recorded it, and its
    /// definitions are unknown. Loading and saving keep the value the
    /// document had rather than the reader's. Additive field.
    #[serde(skip_serializing_if = "Option::is_none")]
    #[schemars(with = "MetricSemantics")]
    pub metric_semantics: Option<MetricSemantics>,
    /// The analysis settings it was profiled with; see [`AnalysisSettings`].
    ///
    /// Every report a profiling run assembles records them. `None` is a
    /// document written before dataprof recorded them, or a report built
    /// directly with [`ProfileReport::new`]: the settings are unknown, not the
    /// defaults. Additive field.
    #[serde(skip_serializing_if = "Option::is_none")]
    #[schemars(with = "AnalysisSettings")]
    pub analysis_settings: Option<AnalysisSettings>,
    /// The dataprof build that produced the report; see [`ReportGenerator`].
    ///
    /// Every report this build produces records [`ReportGenerator::current`].
    /// `None` is a document written before dataprof recorded it. Loading and
    /// saving keep the value the document had rather than the reader's.
    /// Additive field.
    #[serde(skip_serializing_if = "Option::is_none")]
    #[schemars(with = "ReportGenerator")]
    pub generator: Option<ReportGenerator>,
}

fn schema_version_schema(_: &mut schemars::SchemaGenerator) -> schemars::Schema {
    schemars::json_schema!({
        "type": "integer",
        "const": REPORT_SCHEMA_VERSION,
        "minimum": 0
    })
}

/// The canonical persisted report and the legacy Python compatibility summary.
///
/// New JSON saves in both languages use `ProfileReport`. Python's `to_dict()`
/// remains a convenience projection, produced by `PythonProfileReportDocument`
/// below. Existing flat documents cannot acquire the source/quality provenance
/// they never recorded, so the v1 union remains an intentional compatibility
/// surface for loading and resaving those documents.
#[allow(dead_code)]
#[derive(serde::Serialize, schemars::JsonSchema)]
#[serde(untagged)]
#[schemars(title = "ProfileReport")]
enum SerializedProfileReport {
    Rust(Box<ProfileReport>),
    Python(Box<PythonProfileReportDocument>),
}

#[derive(serde::Serialize, serde::Deserialize, schemars::JsonSchema)]
struct PythonProfileReportDocument {
    #[schemars(schema_with = "schema_version_schema")]
    schema_version: u32,
    source: String,
    source_type: PythonSourceType,
    execution: PythonExecutionDocument,
    columns: Vec<PythonColumnDocument>,
    quality: Option<PythonQualityDocument>,
    quality_status: QualityAnalysisStatus,
    #[serde(skip_serializing_if = "Vec::is_empty")]
    semantic_hint_bindings: Vec<SemanticHintBinding>,
    #[serde(skip_serializing_if = "Option::is_none")]
    #[schemars(with = "MetricSemantics")]
    metric_semantics: Option<MetricSemantics>,
    #[serde(skip_serializing_if = "Option::is_none")]
    #[schemars(with = "AnalysisSettings")]
    analysis_settings: Option<AnalysisSettings>,
    #[serde(skip_serializing_if = "Option::is_none")]
    #[schemars(with = "ReportGenerator")]
    generator: Option<ReportGenerator>,
}

#[derive(serde::Serialize, serde::Deserialize, schemars::JsonSchema)]
#[serde(rename_all = "lowercase")]
enum PythonSourceType {
    File,
    Query,
    Dataframe,
    Stream,
    Bytes,
}

#[derive(serde::Serialize, serde::Deserialize, schemars::JsonSchema)]
struct PythonExecutionDocument {
    engine: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    recovery_events: Option<Vec<dataprof_core::execution::RecoveryEvent>>,
    rows_processed: usize,
    columns_detected: usize,
    scan_time_ms: u128,
    source_exhausted: bool,
    truncation_reason: Option<String>,
    bytes_consumed: Option<u64>,
    #[serde(serialize_with = "dataprof_core::serde_helpers::round_4_opt")]
    throughput_rows_sec: Option<f64>,
    #[serde(serialize_with = "dataprof_core::serde_helpers::round_2_opt")]
    memory_peak_mb: Option<f64>,
    error_count: usize,
    ragged_row_count: usize,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    unterminated_quote: Option<bool>,
    sampling_applied: bool,
    #[serde(serialize_with = "dataprof_core::serde_helpers::round_4_opt")]
    sampling_ratio: Option<f64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    sampled_row_ranges: Option<Vec<[u64; 2]>>,
}

#[derive(serde::Serialize, serde::Deserialize, schemars::JsonSchema)]
struct PythonColumnDocument {
    name: String,
    data_type: PythonDataType,
    total_count: usize,
    null_count: usize,
    #[serde(serialize_with = "dataprof_core::serde_helpers::round_2_opt")]
    null_percentage: Option<f64>,
    unique_count: Option<usize>,
    unique_count_is_approximate: Option<bool>,
    #[serde(serialize_with = "dataprof_core::serde_helpers::round_4_opt")]
    uniqueness_ratio: Option<f64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    invalid_count: Option<usize>,
    /// Counts per lexical class. The Python dialect writes them as a plain
    /// mapping, but it is the same four-key object the Rust dialect writes, so
    /// it is described by the same definition: a reader that accepted a partial
    /// mapping here would validate a document the Python loader then discards
    /// as incomplete rather than inventing the missing counts.
    #[serde(skip_serializing_if = "Option::is_none")]
    type_homogeneity: Option<dataprof_core::TypeHomogeneity>,
    #[serde(skip_serializing_if = "Option::is_none")]
    locale_number_count: Option<usize>,
    #[serde(skip_serializing_if = "Option::is_none")]
    stats: Option<PythonColumnStatsDocument>,
    #[serde(skip_serializing_if = "Option::is_none")]
    patterns: Option<Vec<PythonPatternDocument>>,
}

#[derive(serde::Serialize, serde::Deserialize, schemars::JsonSchema)]
#[serde(rename_all = "lowercase")]
enum PythonDataType {
    String,
    Identifier,
    Integer,
    Float,
    Date,
    Boolean,
    Nested,
}

#[derive(serde::Serialize, serde::Deserialize, schemars::JsonSchema)]
struct PythonColumnStatsDocument {
    #[serde(skip_serializing_if = "Option::is_none")]
    min: Option<f64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    max: Option<f64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    mean: Option<f64>,
    // Outer None means this statistic does not apply to the column. An inner
    // None is a numeric spread that overflowed: keep its explicit JSON null.
    #[serde(
        default,
        deserialize_with = "deserialize_present_nullable_stat",
        skip_serializing_if = "Option::is_none"
    )]
    #[schemars(with = "Option<f64>")]
    std_dev: Option<Option<f64>>,
    #[serde(
        default,
        deserialize_with = "deserialize_present_nullable_stat",
        skip_serializing_if = "Option::is_none"
    )]
    #[schemars(with = "Option<f64>")]
    variance: Option<Option<f64>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    median: Option<f64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    mode: Option<f64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    skewness: Option<f64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    kurtosis: Option<f64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    coefficient_of_variation: Option<f64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    quartiles: Option<dataprof_core::Quartiles>,
    #[serde(skip_serializing_if = "Option::is_none")]
    is_approximate: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    outlier_count: Option<usize>,
    /// Shortest value, in Unicode scalar values. Not UTF-8 bytes and not
    /// grapheme clusters: ASCII text is unaffected by that distinction, while a
    /// combining sequence counts each scalar, so the decomposed and precomposed
    /// spellings of the same word report different lengths.
    #[serde(skip_serializing_if = "Option::is_none")]
    min_length: Option<usize>,
    /// Longest value, in Unicode scalar values. See `min_length`.
    #[serde(skip_serializing_if = "Option::is_none")]
    max_length: Option<usize>,
    /// Mean length, in Unicode scalar values. See `min_length`.
    #[serde(skip_serializing_if = "Option::is_none")]
    avg_length: Option<f64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    true_count: Option<usize>,
    #[serde(skip_serializing_if = "Option::is_none")]
    false_count: Option<usize>,
    #[serde(skip_serializing_if = "Option::is_none")]
    true_ratio: Option<f64>,
}

fn deserialize_present_nullable_stat<'de, D>(
    deserializer: D,
) -> Result<Option<Option<f64>>, D::Error>
where
    D: serde::Deserializer<'de>,
{
    serde::Deserialize::deserialize(deserializer).map(Some)
}

#[derive(serde::Serialize, serde::Deserialize, schemars::JsonSchema)]
struct PythonPatternDocument {
    name: String,
    regex: String,
    match_count: usize,
    #[serde(serialize_with = "dataprof_core::serde_helpers::round_2")]
    match_percentage: f64,
    category: dataprof_core::PatternCategory,
    #[serde(serialize_with = "dataprof_core::serde_helpers::round_4")]
    confidence: f64,
}

#[derive(serde::Serialize, serde::Deserialize, schemars::JsonSchema)]
struct PythonQualityDocument {
    /// Null when no dimension was assessable; `assessed_dimensions` is then
    /// empty and every entry in `dimension_scores` is null.
    overall_score: Option<f64>,
    assessed_dimensions: Vec<PythonQualityDimension>,
    dimension_scores: std::collections::BTreeMap<PythonQualityDimension, Option<f64>>,
    low_sample_warning: bool,
    /// Metric components computed from a retained quality sample rather than
    /// from every scanned row. Additive field: a document written before it
    /// existed omits it, and its absence is preserved on reload because
    /// unknown coverage is not the same statement as "nothing was sampled".
    /// Uniqueness appears here as `key_uniqueness` and `duplicate_rows`.
    #[serde(skip_serializing_if = "Option::is_none")]
    sampled_dimensions: Option<Vec<String>>,
    /// Where the scores over every scanned row lie, when some were computed over a
    /// retained sample. Additive field, absent when every score is exact and
    /// on documents written before it existed.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    score_bounds: Option<dataprof_metrics::ScoreBounds>,
    /// Weights behind `overall_score`, always written as the canonical
    /// document writes them, default weights included (#749). Summaries
    /// written before 0.13 omitted default weights, and before the field
    /// existed dropped custom ones too, so its absence there does not show
    /// which weights applied.
    #[serde(default)]
    score_weights: QualityScoreWeights,
    #[serde(skip_serializing_if = "Option::is_none")]
    completeness: Option<CompletenessMetrics>,
    #[serde(skip_serializing_if = "Option::is_none")]
    consistency: Option<ConsistencyMetrics>,
    #[serde(skip_serializing_if = "Option::is_none")]
    uniqueness: Option<PythonUniquenessDocument>,
    #[serde(skip_serializing_if = "Option::is_none")]
    accuracy: Option<AccuracyMetrics>,
    #[serde(skip_serializing_if = "Option::is_none")]
    timeliness: Option<TimelinessMetrics>,
    #[serde(skip_serializing_if = "Option::is_none")]
    validity: Option<ValidityMetrics>,
    #[serde(skip_serializing_if = "Option::is_none")]
    precision: Option<PrecisionMetrics>,
}

#[derive(
    Debug,
    Clone,
    PartialEq,
    Eq,
    PartialOrd,
    Ord,
    Hash,
    serde::Serialize,
    serde::Deserialize,
    schemars::JsonSchema,
)]
#[serde(rename_all = "lowercase")]
enum PythonQualityDimension {
    Completeness,
    Consistency,
    Uniqueness,
    Accuracy,
    Timeliness,
    Validity,
    Precision,
}

#[derive(serde::Serialize, serde::Deserialize, schemars::JsonSchema)]
struct PythonUniquenessDocument {
    duplicate_rows: usize,
    key_uniqueness: f64,
    high_cardinality_warning: bool,
    rows_checked: usize,
    key_column: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    key_values_checked: Option<usize>,
    duplicate_rows_approximate: bool,
}

impl PythonProfileReportDocument {
    fn from_report(report: &ProfileReport) -> serde_json::Result<Self> {
        use dataprof_core::{ColumnStats, DataType, TruncationReason};

        // Reuse the serialized metric blocks: their Serde implementations own
        // rounding and omission, rather than repeating those rules in Python.
        let columns = report
            .column_profiles
            .iter()
            .map(|column| {
                let stats = match &column.stats {
                    ColumnStats::Numeric(stats) => Some(serde_json::to_value(stats)?),
                    ColumnStats::Text(stats) if column.total_count > 0 => {
                        Some(serde_json::to_value(stats)?)
                    }
                    ColumnStats::Boolean(stats) => Some(serde_json::to_value(stats)?),
                    _ => None,
                }
                .map(serde_json::from_value)
                .transpose()?;
                Ok(PythonColumnDocument {
                    name: column.name.clone(),
                    data_type: match column.data_type {
                        DataType::String => PythonDataType::String,
                        DataType::Identifier => PythonDataType::Identifier,
                        DataType::Integer => PythonDataType::Integer,
                        DataType::Float => PythonDataType::Float,
                        DataType::Date => PythonDataType::Date,
                        DataType::Boolean => PythonDataType::Boolean,
                        DataType::Nested => PythonDataType::Nested,
                    },
                    total_count: column.total_count,
                    null_count: column.null_count,
                    null_percentage: (column.total_count > 0)
                        .then(|| column.null_count as f64 / column.total_count as f64 * 100.0),
                    unique_count: column.unique_count,
                    unique_count_is_approximate: column.unique_count_is_approximate,
                    uniqueness_ratio: column
                        .unique_count
                        .filter(|_| column.total_count > 0)
                        .map(|count| count as f64 / column.total_count as f64),
                    invalid_count: column.invalid_count,
                    type_homogeneity: column.type_homogeneity,
                    locale_number_count: column.locale_number_count,
                    stats,
                    patterns: column.patterns.as_ref().map(|patterns| {
                        patterns
                            .iter()
                            .map(|pattern| PythonPatternDocument {
                                name: pattern.name.clone(),
                                regex: pattern.regex.clone(),
                                match_count: pattern.match_count,
                                match_percentage: pattern.match_percentage,
                                category: pattern.category.clone(),
                                confidence: pattern.confidence,
                            })
                            .collect()
                    }),
                })
            })
            .collect::<serde_json::Result<Vec<_>>>()?;

        let mut execution = serde_json::to_value(&report.execution)?;
        execution["truncation_reason"] =
            serde_json::to_value(report.execution.truncation_reason.as_ref().map(|reason| {
                match reason {
                    TruncationReason::MaxRows(n) => format!("max_rows({n})"),
                    TruncationReason::MaxBytes(n) => format!("max_bytes({n})"),
                    TruncationReason::MemoryPressure => "memory_pressure".to_string(),
                    TruncationReason::StopCondition(s) => format!("stop_condition({s})"),
                    TruncationReason::StreamClosed => "stream_closed".to_string(),
                    TruncationReason::Timeout => "timeout".to_string(),
                }
            }))?;

        let quality = report
            .quality
            .as_ref()
            .map(|assessment| {
                let metrics = &assessment.metrics;
                let mut document = serde_json::to_value(metrics)?;
                let scores = serde_json::to_value(assessment.scores())?;
                document["overall_score"] = scores["overall_score"].clone();
                // A document from a later v1 release may retain scores for a
                // dimension this build does not know. The canonical document
                // keeps them; the summary lists only the dimensions it types.
                let known: Vec<String> = dataprof_core::QualityDimension::all()
                    .into_iter()
                    .map(|dimension| dimension.to_string())
                    .collect();
                document["dimension_scores"] = scores["dimension_scores"]
                    .as_object()
                    .expect("dimension scores serialize as an object")
                    .iter()
                    .filter(|(name, _)| known.contains(name))
                    .map(|(name, score)| (name.clone(), score.clone()))
                    .collect();
                document["assessed_dimensions"] =
                    serde_json::to_value(metrics.assessed_dimensions())?;
                document["low_sample_warning"] =
                    serde_json::Value::Bool(metrics.low_sample_warning);
                if let Some(sampled) = assessment.sampled_dimensions() {
                    document["sampled_dimensions"] = serde_json::to_value(sampled)?;
                }
                if let Some(bounds) = assessment.score_bounds() {
                    document["score_bounds"] = serde_json::to_value(bounds)?;
                }
                // The summary has always included this boolean, even when false.
                if let Some(uniqueness) = document.get_mut("uniqueness") {
                    uniqueness["duplicate_rows_approximate"] = serde_json::Value::Bool(
                        metrics
                            .uniqueness
                            .as_ref()
                            .expect("serialized uniqueness is present")
                            .duplicate_rows_approximate,
                    );
                }
                serde_json::from_value(document)
            })
            .transpose()?;

        Ok(Self {
            schema_version: report.schema_version,
            source: report.data_source.identifier(),
            source_type: match report.data_source {
                DataSource::File { .. } => PythonSourceType::File,
                DataSource::Query { .. } => PythonSourceType::Query,
                DataSource::DataFrame { .. } => PythonSourceType::Dataframe,
                DataSource::Stream { .. } => PythonSourceType::Stream,
                DataSource::Bytes { .. } => PythonSourceType::Bytes,
            },
            execution: serde_json::from_value(execution)?,
            columns,
            quality,
            quality_status: report.quality_status.clone(),
            semantic_hint_bindings: report.semantic_hint_bindings.clone(),
            metric_semantics: report.metric_semantics.clone(),
            analysis_settings: report.analysis_settings.clone(),
            generator: report.generator.clone(),
        })
    }
}

/// Generate the JSON Schema 2020-12 document for the current serialized report.
///
/// This is primarily used by the repository's schema generator and drift tests.
/// The committed, versioned document under `docs/schema/` is the public
/// interoperability contract.
#[doc(hidden)]
pub fn profile_report_schema_document() -> serde_json::Value {
    let settings = schemars::generate::SchemaSettings::draft2020_12().for_serialize();
    let schema = settings
        .into_generator()
        .into_root_schema_for::<SerializedProfileReport>();
    let mut document =
        serde_json::to_value(schema).expect("a Schemars schema must serialize to a JSON document");
    let object = document
        .as_object_mut()
        .expect("a root Schemars schema must be a JSON object");
    object.insert(
        "$id".to_string(),
        serde_json::Value::String(format!(
            "https://andreabozzo.github.io/dataprof/schema/profile-report.v{REPORT_SCHEMA_VERSION}.schema.json"
        )),
    );
    make_compatibility_defaults_optional(&mut document);
    allow_additive_properties(&mut document);
    canonicalize_key_order(&mut document);
    document
}

/// Sort every object's keys so the committed schema has one canonical byte
/// layout.
///
/// Key order carries no meaning in JSON Schema, but the committed document is a
/// reviewed artifact: it must not churn because `serde_json` switched its map
/// backing (`preserve_order`) or because Schemars changed the order in which it
/// builds a schema. Arrays such as `required` and `oneOf` keep their order,
/// which *is* meaningful.
fn canonicalize_key_order(value: &mut serde_json::Value) {
    match value {
        serde_json::Value::Object(object) => {
            for child in object.values_mut() {
                canonicalize_key_order(child);
            }
            let mut sorted: Vec<(String, serde_json::Value)> =
                std::mem::take(object).into_iter().collect();
            sorted.sort_by(|(left, _), (right, _)| left.cmp(right));
            object.extend(sorted);
        }
        serde_json::Value::Array(items) => {
            for item in items {
                canonicalize_key_order(item);
            }
        }
        _ => {}
    }
}

/// Drop fields that this build always writes but a reader supplies a default
/// for, from the schema's `required` lists.
///
/// A document written under an earlier v1 build does not carry them. Declaring
/// them required would fail validation for documents the readers accept, which
/// is the opposite of what an additive widening means.
fn make_compatibility_defaults_optional(document: &mut serde_json::Value) {
    for (pointer, field) in [
        ("/$defs/ExecutionMetadata/required", "ragged_row_count"),
        ("/$defs/ProfileReport/required", "quality_status"),
        (
            "/$defs/PythonProfileReportDocument/required",
            "quality_status",
        ),
        (
            "/$defs/PythonQualityDocument/required",
            "sampled_dimensions",
        ),
        ("/$defs/QualityAssessment/required", "scores"),
        // Written always since 0.13 (#749), omitted at the defaults before.
        ("/$defs/QualityMetrics/required", "score_weights"),
        ("/$defs/PythonQualityDocument/required", "score_weights"),
        // Written only when some score was sampled.
        ("/$defs/QualityAssessment/required", "score_bounds"),
        // Written always since 0.13 (#1005), absent before.
        ("/$defs/AccuracyMetrics/required", "rule_values_checked"),
        ("/$defs/AccuracyMetrics/required", "rule_violating_values"),
    ] {
        if let Some(required) = document
            .pointer_mut(pointer)
            .and_then(serde_json::Value::as_array_mut)
        {
            required.retain(|declared| declared.as_str() != Some(field));
        }
    }
    // A definition left with nothing required carries no `required` list.
    if let Some(definitions) = document
        .pointer_mut("/$defs")
        .and_then(serde_json::Value::as_object_mut)
    {
        for definition in definitions.values_mut() {
            if let Some(object) = definition.as_object_mut()
                && object
                    .get("required")
                    .and_then(serde_json::Value::as_array)
                    .is_some_and(Vec::is_empty)
            {
                object.remove("required");
            }
        }
    }
}

fn allow_additive_properties(value: &mut serde_json::Value) {
    match value {
        serde_json::Value::Object(object) => {
            if object.get("additionalProperties") == Some(&serde_json::Value::Bool(false)) {
                object.remove("additionalProperties");
            }
            for child in object.values_mut() {
                allow_additive_properties(child);
            }
        }
        serde_json::Value::Array(items) => {
            for item in items {
                allow_additive_properties(item);
            }
        }
        _ => {}
    }
}

impl ProfileReport {
    /// Serialize the canonical persisted document with deterministic object-key
    /// order. Array order (columns, patterns, recovery events) is preserved.
    pub fn to_json(&self) -> serde_json::Result<String> {
        let mut document = serde_json::to_value(self)?;
        canonicalize_key_order(&mut document);
        serde_json::to_string_pretty(&document)
    }

    /// Serialize Python's historical summary projection. It deliberately omits
    /// some runtime metadata; use [`Self::to_json`] for lossless persistence.
    #[doc(hidden)]
    pub fn summary_json(&self) -> serde_json::Result<String> {
        let summary = PythonProfileReportDocument::from_report(self)?;
        serde_json::to_string(&summary)
    }

    /// Create a new ProfileReport with auto-generated id and timestamp
    pub fn new(
        data_source: DataSource,
        column_profiles: Vec<ColumnProfile>,
        execution: ExecutionMetadata,
        quality: Option<QualityAssessment>,
    ) -> Self {
        Self {
            schema_version: REPORT_SCHEMA_VERSION,
            id: uuid::Uuid::new_v4().to_string(),
            timestamp: chrono::Utc::now().to_rfc3339(),
            data_source,
            column_profiles,
            execution,
            quality_status: match quality {
                Some(_) => QualityAnalysisStatus::Computed,
                // A caller constructing a report directly did not compute
                // quality; `ReportAssembler` overrides this with the reason it
                // actually observed.
                None => QualityAnalysisStatus::NotRequested,
            },
            quality,
            semantic_hint_bindings: Vec::new(),
            metric_semantics: Some(MetricSemantics::current()),
            // Only the assembler knows the options a report was profiled
            // with; it records them through `with_analysis_settings`.
            analysis_settings: None,
            generator: Some(ReportGenerator::current()),
        }
    }

    /// Record the analysis settings the report was profiled with.
    pub(crate) fn with_analysis_settings(mut self, settings: Option<AnalysisSettings>) -> Self {
        self.analysis_settings = settings;
        self
    }

    /// Record what happened to the quality computation.
    ///
    /// Deliberately not public. `quality_status` and `quality` are two halves
    /// of one fact, and [`check_quality_pairing`](Self::check_quality_pairing)
    /// refuses to decode a document that pairs them wrongly — so a public
    /// setter would let a caller build a report that serializes and then
    /// cannot be read back by this same crate. Every path assigns the status
    /// through [`ReportAssembler`](crate::ReportAssembler), which derives it
    /// from the computation that produced the assessment.
    pub(crate) fn with_quality_status(mut self, status: QualityAnalysisStatus) -> Self {
        self.quality_status = status;
        self
    }

    /// Attach per-column semantic-hint binding evidence.
    pub fn with_semantic_hint_bindings(mut self, bindings: Vec<SemanticHintBinding>) -> Self {
        self.semantic_hint_bindings = bindings;
        self
    }

    /// Override the auto-generated ID (useful for deterministic caching/testing)
    pub fn with_id(mut self, id: impl Into<String>) -> Self {
        self.id = id.into();
        self
    }

    /// Override the auto-generated timestamp
    pub fn with_timestamp(mut self, timestamp: impl Into<String>) -> Self {
        self.timestamp = timestamp.into();
        self
    }

    /// Calculate the overall quality score (weighted across the assessed
    /// dimensions). Returns `None` if quality metrics were not computed, or
    /// if no dimension had anything to assess (e.g. an empty dataset) —
    /// absence of evidence is neither a perfect score nor a zero.
    pub fn quality_score(&self) -> Option<f64> {
        self.quality.as_ref().and_then(|q| q.score())
    }

    /// Get the data source identifier (for backwards compatibility)
    pub fn source_identifier(&self) -> String {
        self.data_source.identifier()
    }

    /// Reject a document whose `quality_status` contradicts its `quality`.
    ///
    /// `Computed` and an assessment imply each other: no writer can produce one
    /// without the other, because [`ReportAssembler`](crate::ReportAssembler)
    /// derives the status from the computation that produced the assessment. A
    /// document that pairs them any other way is internally inconsistent, and
    /// loading it would hand a consumer a report that states a verdict its own
    /// contents contradict — a gate could act on an assessment the report says
    /// failed. Failing the decode is the answer the rest of this crate gives to
    /// a malformed document, rather than repairing it into something plausible.
    fn check_quality_pairing(self) -> Result<Self, String> {
        let assessed = self.quality.is_some();
        let claims_computed = matches!(self.quality_status, QualityAnalysisStatus::Computed);
        if claims_computed == assessed {
            return Ok(self);
        }
        Err(if assessed {
            format!(
                "report carries a quality assessment but reports quality_status \
                 {:?}; only `computed` may accompany an assessment",
                self.quality_status
            )
        } else {
            "report reports quality_status `computed` but carries no quality assessment".to_string()
        })
    }
}

/// Mirror of [`ProfileReport`] carrying the field-level deserialization
/// rules (legacy aliases, quality compat). Kept private: the public entry
/// point is the manual [`serde::Deserialize`] impl below, which gates on
/// `schema_version` before any of these fields decode.
#[derive(serde::Deserialize)]
struct ProfileReportFields {
    #[serde(default)]
    schema_version: u32,
    id: String,
    timestamp: String,
    data_source: DataSource,
    column_profiles: Vec<ColumnProfile>,
    #[serde(alias = "scan_info")]
    execution: ExecutionMetadata,
    #[serde(
        alias = "data_quality_metrics",
        default,
        deserialize_with = "deserialize_quality_compat"
    )]
    quality: Option<QualityAssessment>,
    #[serde(default)]
    quality_status: Option<QualityAnalysisStatus>,
    #[serde(default)]
    semantic_hint_bindings: Vec<SemanticHintBinding>,
    #[serde(default)]
    metric_semantics: Option<MetricSemantics>,
    #[serde(default)]
    analysis_settings: Option<AnalysisSettings>,
    #[serde(default)]
    generator: Option<ReportGenerator>,
}

impl From<ProfileReportFields> for ProfileReport {
    fn from(fields: ProfileReportFields) -> Self {
        Self {
            schema_version: fields.schema_version,
            id: fields.id,
            timestamp: fields.timestamp,
            data_source: fields.data_source,
            column_profiles: fields.column_profiles,
            execution: fields.execution,
            // A document written before the field: a stored assessment proves
            // the computation ran, and nothing else about it is knowable.
            quality_status: fields.quality_status.unwrap_or({
                if fields.quality.is_some() {
                    QualityAnalysisStatus::Computed
                } else {
                    QualityAnalysisStatus::Unrecorded
                }
            }),
            quality: fields.quality,
            semantic_hint_bindings: fields.semantic_hint_bindings,
            metric_semantics: fields.metric_semantics,
            analysis_settings: fields.analysis_settings,
            generator: fields.generator,
        }
    }
}

/// Manual deserialization so the `schema_version` gate runs before any other
/// field decodes. A derived deserializer visits fields in document order, so
/// an unsupported future document that also changed structure could fail with
/// a confusing structural error instead of the actionable version error; by
/// buffering the document first, the version check always wins.
impl<'de> serde::Deserialize<'de> for ProfileReport {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: serde::Deserializer<'de>,
    {
        use serde::de::Error;

        let value = serde_json::Value::deserialize(deserializer)?;
        match value.get("schema_version") {
            // Absent field: legacy pre-0.10 document, defaults to version 0.
            None => {}
            Some(serde_json::Value::Number(n)) if n.as_u64().is_some() => {
                let version = n.as_u64().unwrap_or_default();
                if version > u64::from(REPORT_SCHEMA_VERSION) {
                    return Err(D::Error::custom(format!(
                        "report schema version {version} is newer than the latest supported \
                         version {REPORT_SCHEMA_VERSION}; upgrade dataprof to read this report"
                    )));
                }
            }
            // An explicit null or non-integer is malformed, not legacy.
            Some(other) => {
                return Err(D::Error::custom(format!(
                    "report schema_version must be a non-negative integer, got {other}"
                )));
            }
        }
        // `quality_status` deserializes through `Option`, which cannot tell an
        // absent field from an explicit null. Only the absent one is legacy;
        // a null is a malformed current document, and the fallback would
        // silently repair it into a plausible reason it never recorded.
        if value.get("quality_status") == Some(&serde_json::Value::Null) {
            return Err(D::Error::custom(
                "report quality_status must be an object; an explicit null is malformed, \
                 not a document written before the field existed",
            ));
        }
        // Same rule: absent is "not recorded", which reads back as unknown;
        // a null would be a current writer's malformed output.
        for key in ["metric_semantics", "analysis_settings", "generator"] {
            if value.get(key) == Some(&serde_json::Value::Null) {
                return Err(D::Error::custom(format!(
                    "report {key} must be an object; an explicit null is malformed, \
                     not a document written before the field existed"
                )));
            }
        }
        let report = ProfileReportFields::deserialize(value)
            .map(ProfileReport::from)
            .map_err(D::Error::custom)?;
        report.check_quality_pairing().map_err(D::Error::custom)
    }
}

/// Custom deserializer that handles both legacy `DataQualityMetrics` (flat)
/// and new `QualityAssessment` (wrapped with confidence) JSON formats.
fn deserialize_quality_compat<'de, D>(
    deserializer: D,
) -> Result<Option<QualityAssessment>, D::Error>
where
    D: serde::Deserializer<'de>,
{
    use serde::Deserialize;

    let value: Option<serde_json::Value> = Option::deserialize(deserializer)?;
    match value {
        None => Ok(None),
        Some(mut v) => {
            // Either wrapper key marks the canonical shape. Read as flat
            // metrics, a canonical object matches no field and decodes as an
            // empty assessment, dropping every metric without an error.
            if v.get("metrics").is_some() || v.get("confidence").is_some() {
                if let Some(object) = v.as_object_mut() {
                    // Metrics whose provenance was lost are kept, and the
                    // provenance stays unknown, as for a flat legacy object.
                    object
                        .entry("confidence")
                        .or_insert_with(|| serde_json::json!("Unrecorded"));
                }
                let assessment: QualityAssessment =
                    serde_json::from_value(v).map_err(serde::de::Error::custom)?;
                Ok(Some(assessment))
            } else {
                let metrics: QualityMetrics =
                    serde_json::from_value(v).map_err(serde::de::Error::custom)?;
                // A flat legacy quality object says nothing about how its
                // numbers were obtained. Recording `Exact` here would invent
                // that, and a gate asked about the whole source would then
                // rest a verdict on numbers that may have come from a sample.
                Ok(Some(QualityAssessment::new(
                    metrics,
                    MetricConfidence::Unrecorded,
                )))
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use dataprof_core::FileFormat;
    use dataprof_metrics::MetricConfidence;
    use serde_json::json;

    fn report_without_quality() -> ProfileReport {
        ProfileReport::new(
            DataSource::File {
                path: "test.csv".to_string(),
                format: FileFormat::Csv,
                size_bytes: 1024,
                modified_at: None,
                parquet_metadata: None,
            },
            vec![],
            ExecutionMetadata::new(100, 5, 50),
            None,
        )
    }

    /// The `state` strings are the wire vocabulary: consumers branch on them,
    /// the Python binding repeats them by hand, and the committed schema
    /// enumerates them. Pin them here so a variant rename has to be deliberate.
    #[test]
    fn every_status_has_a_stable_wire_name() {
        let cases = [
            (QualityAnalysisStatus::Computed, "computed"),
            (QualityAnalysisStatus::NotRequested, "not_requested"),
            (QualityAnalysisStatus::NoData, "no_data"),
            (
                QualityAnalysisStatus::WithheldByProjection,
                "withheld_by_projection",
            ),
            (
                QualityAnalysisStatus::Failed {
                    error: "boom".to_string(),
                },
                "failed",
            ),
            (QualityAnalysisStatus::Unrecorded, "unrecorded"),
        ];

        for (status, name) in cases {
            let value = serde_json::to_value(&status).unwrap();
            assert_eq!(value.get("state"), Some(&json!(name)), "{status:?}");
            let restored: QualityAnalysisStatus = serde_json::from_value(value).unwrap();
            assert_eq!(restored, status);
        }
    }

    /// A document may not claim one thing in `quality_status` and another in
    /// `quality`. Loading one would let a consumer act on an assessment the
    /// report says never completed.
    #[test]
    fn contradictory_status_and_assessment_fail_to_decode() {
        let with_assessment =
            report_without_quality().with_quality_status(QualityAnalysisStatus::Failed {
                error: "boom".to_string(),
            });
        let mut document = serde_json::to_value(&with_assessment).unwrap();
        document.as_object_mut().unwrap().insert(
            "quality".to_string(),
            serde_json::to_value(QualityAssessment::exact(QualityMetrics::empty())).unwrap(),
        );
        let error = serde_json::from_value::<ProfileReport>(document)
            .expect_err("an assessment under a `failed` status is not a readable report");
        assert!(
            error.to_string().contains("only `computed`"),
            "unhelpful error: {error}"
        );

        let mut document = serde_json::to_value(
            report_without_quality().with_quality_status(QualityAnalysisStatus::Computed),
        )
        .unwrap();
        document.as_object_mut().unwrap().remove("quality");
        let error = serde_json::from_value::<ProfileReport>(document)
            .expect_err("`computed` without an assessment is not a readable report");
        assert!(
            error.to_string().contains("carries no quality assessment"),
            "unhelpful error: {error}"
        );
    }

    /// An absent field is a document written before it existed. An explicit
    /// null is a malformed current one, and the legacy fallback would repair it
    /// into a reason the writer never recorded.
    #[test]
    fn an_explicit_null_status_is_malformed_not_legacy() {
        let mut document = serde_json::to_value(report_without_quality()).unwrap();
        document
            .as_object_mut()
            .unwrap()
            .insert("quality_status".to_string(), serde_json::Value::Null);

        let error = serde_json::from_value::<ProfileReport>(document)
            .expect_err("an explicit null is not a readable status");
        assert!(
            error.to_string().contains("explicit null is malformed"),
            "unhelpful error: {error}"
        );
    }

    fn summary_value(report: &ProfileReport) -> serde_json::Value {
        serde_json::from_str(&report.summary_json().unwrap()).unwrap()
    }

    /// A report built by this crate records how it measured, in both dialects.
    /// A document written before the field reads back as unknown, and saving
    /// it again must not stamp it with the reader's definitions.
    #[test]
    fn metric_semantics_are_recorded_and_absence_survives_a_roundtrip() {
        let report = report_without_quality();
        assert_eq!(report.metric_semantics, Some(MetricSemantics::current()));
        let mut document = serde_json::to_value(&report).unwrap();
        let recorded = json!({
            "text_length_unit": "unicode_scalar",
            "null_tokens": "common_markers",
            "pattern_match_count": "validated",
            "shape_moments": "adjusted_sample",
            "range_rule_count": "violating_values",
            "key_uniqueness_units": "non_null_values",
            "outlier_units": "tested_values",
            "key_column_source": "declared",
        });
        assert_eq!(document["metric_semantics"], recorded);
        assert_eq!(summary_value(&report)["metric_semantics"], recorded);

        document.as_object_mut().unwrap().remove("metric_semantics");
        let legacy: ProfileReport = serde_json::from_value(document).unwrap();
        assert_eq!(legacy.metric_semantics, None);
        assert!(
            serde_json::to_value(&legacy)
                .unwrap()
                .get("metric_semantics")
                .is_none()
        );
        assert!(summary_value(&legacy).get("metric_semantics").is_none());
    }

    #[test]
    fn malformed_metric_semantics_fail_to_decode() {
        for value in [
            serde_json::Value::Null,
            json!("unicode_scalar"),
            json!({"text_length_unit": "utf8_byte"}),
            json!({"text_length_unit": null}),
            json!({"null_tokens": "empty_null_nan"}),
            json!({"null_tokens": null}),
            json!({"pattern_match_count": "regex"}),
            json!({"pattern_match_count": null}),
            json!({"shape_moments": "population"}),
            json!({"shape_moments": null}),
            json!({"range_rule_count": "rule_failures"}),
            json!({"range_rule_count": null}),
            json!({"key_uniqueness_units": "rows"}),
            json!({"key_uniqueness_units": null}),
            json!({"outlier_units": "numeric_values"}),
            json!({"outlier_units": null}),
            json!({"key_column_source": "name_heuristic"}),
            json!({"key_column_source": null}),
        ] {
            let mut document = serde_json::to_value(report_without_quality()).unwrap();
            document["metric_semantics"] = value.clone();
            assert!(
                serde_json::from_value::<ProfileReport>(document).is_err(),
                "{value} decoded"
            );
        }
    }

    /// Later releases add definitions; an older reader keeps the ones it knows.
    #[test]
    fn unknown_metric_definitions_are_ignored() {
        let mut document = serde_json::to_value(report_without_quality()).unwrap();
        document["metric_semantics"]["grapheme_policy"] = json!("extended");
        let restored: ProfileReport = serde_json::from_value(document).unwrap();
        assert_eq!(restored.metric_semantics, Some(MetricSemantics::current()));
    }

    #[test]
    fn quality_status_survives_a_json_roundtrip() {
        let report = report_without_quality().with_quality_status(QualityAnalysisStatus::Failed {
            error: "Metrics calculation failed: no data columns found".to_string(),
        });

        let json = serde_json::to_string(&report).unwrap();
        let restored: ProfileReport = serde_json::from_str(&json).unwrap();

        assert_eq!(restored.quality_status, report.quality_status);
        assert!(restored.quality.is_none());
    }

    /// The serialized document is the contract, so the two states have to be
    /// distinguishable there and not only in the Rust value.
    #[test]
    fn serialized_failure_and_skip_differ() {
        let failed = serde_json::to_value(report_without_quality().with_quality_status(
            QualityAnalysisStatus::Failed {
                error: "boom".to_string(),
            },
        ))
        .unwrap();
        let skipped = serde_json::to_value(
            report_without_quality().with_quality_status(QualityAnalysisStatus::NotRequested),
        )
        .unwrap();

        assert_eq!(
            failed.get("quality_status"),
            Some(&json!({"state": "failed", "error": "boom"}))
        );
        assert_eq!(
            skipped.get("quality_status"),
            Some(&json!({"state": "not_requested"}))
        );
        assert!(failed.get("quality").is_none());
        assert!(skipped.get("quality").is_none());
    }

    /// A document written before the field existed: an assessment proves the
    /// computation ran, and its absence proves nothing.
    #[test]
    fn documents_without_the_field_read_back_honestly() {
        let mut document = serde_json::to_value(report_without_quality()).unwrap();
        document.as_object_mut().unwrap().remove("quality_status");

        let restored: ProfileReport = serde_json::from_value(document.clone()).unwrap();
        assert_eq!(restored.quality_status, QualityAnalysisStatus::Unrecorded);

        document.as_object_mut().unwrap().insert(
            "quality".to_string(),
            serde_json::to_value(QualityAssessment::exact(QualityMetrics::empty())).unwrap(),
        );
        let restored: ProfileReport = serde_json::from_value(document).unwrap();
        assert_eq!(restored.quality_status, QualityAnalysisStatus::Computed);
    }

    #[test]
    fn test_profile_report_json_roundtrip() {
        let report = ProfileReport::new(
            DataSource::File {
                path: "test.csv".to_string(),
                format: FileFormat::Csv,
                size_bytes: 1024,
                modified_at: None,
                parquet_metadata: None,
            },
            vec![],
            ExecutionMetadata::new(100, 5, 50),
            Some(QualityAssessment::exact(QualityMetrics::empty())),
        );

        let json = serde_json::to_string(&report).unwrap();
        let deserialized: ProfileReport = serde_json::from_str(&json).unwrap();

        assert_eq!(deserialized.id, report.id);
        assert_eq!(deserialized.timestamp, report.timestamp);
        assert_eq!(deserialized.source_identifier(), "test.csv");
        assert_eq!(deserialized.execution.rows_processed, 100);
        assert!(deserialized.quality.is_some());
        assert_eq!(deserialized.schema_version, REPORT_SCHEMA_VERSION);
    }

    /// The weights determine the aggregate score, so a summary that drops
    /// custom ones reloads with a score its weights no longer produce.
    #[test]
    fn summary_names_its_score_weights_including_defaults() {
        let summary_quality = |weights| {
            let mut metrics = QualityMetrics::empty();
            metrics.score_weights = weights;
            let report = ProfileReport::new(
                DataSource::File {
                    path: "test.csv".to_string(),
                    format: FileFormat::Csv,
                    size_bytes: 1024,
                    modified_at: None,
                    parquet_metadata: None,
                },
                vec![],
                ExecutionMetadata::new(100, 5, 50),
                Some(QualityAssessment::exact(metrics)),
            );
            let summary: serde_json::Value =
                serde_json::from_str(&report.summary_json().unwrap()).unwrap();
            summary["quality"].clone()
        };

        let custom = QualityScoreWeights {
            completeness: 1.0,
            consistency: 0.0,
            uniqueness: 0.0,
            accuracy: 0.0,
            timeliness: 0.0,
            validity: 0.0,
            precision: 3.0,
        };
        assert_eq!(
            summary_quality(custom)["score_weights"],
            serde_json::to_value(custom).unwrap()
        );
        // Defaults too: a stored score has to name the weights behind it in
        // case a later release changes the defaults (#749).
        assert_eq!(
            summary_quality(QualityScoreWeights::default())["score_weights"],
            serde_json::to_value(QualityScoreWeights::default()).unwrap()
        );
    }

    #[test]
    fn test_serialized_report_carries_schema_version() {
        let report = ProfileReport::new(
            DataSource::File {
                path: "test.csv".to_string(),
                format: FileFormat::Csv,
                size_bytes: 1024,
                modified_at: None,
                parquet_metadata: None,
            },
            vec![],
            ExecutionMetadata::new(100, 5, 50),
            None,
        );

        let value = serde_json::to_value(&report).unwrap();
        assert_eq!(
            value.get("schema_version").and_then(|v| v.as_u64()),
            Some(u64::from(REPORT_SCHEMA_VERSION))
        );
    }

    #[test]
    fn test_profile_report_without_quality() {
        let report = ProfileReport::new(
            DataSource::File {
                path: "test.csv".to_string(),
                format: FileFormat::Csv,
                size_bytes: 1024,
                modified_at: None,
                parquet_metadata: None,
            },
            vec![],
            ExecutionMetadata::new(100, 5, 50),
            None,
        );

        let json = serde_json::to_string(&report).unwrap();
        let deserialized: ProfileReport = serde_json::from_str(&json).unwrap();

        assert!(deserialized.quality.is_none());
        assert_eq!(deserialized.execution.rows_processed, 100);
    }

    #[test]
    fn test_profile_report_deserializes_legacy_quality_metrics() {
        let json = json!({
            "id": "legacy-report",
            "timestamp": "2026-05-22T10:00:00Z",
            "data_source": {
                "type": "file",
                "path": "test.csv",
                "format": "csv",
                "size_bytes": 42
            },
            "column_profiles": [],
            "scan_info": {
                "rows_processed": 10,
                "columns_detected": 2,
                "scan_time_ms": 5,
                "error_count": 0,
                "source_exhausted": true,
                "sampling_applied": false
            },
            "data_quality_metrics": {
                "completeness": {
                    "missing_values_ratio": 0.0,
                    "complete_records_ratio": 100.0,
                    "null_columns": []
                }
            }
        });

        let report: ProfileReport = serde_json::from_value(json).unwrap();

        assert_eq!(report.id, "legacy-report");
        // A document without a schema_version field is a legacy pre-0.10
        // report; it must load and be identifiable as such.
        assert_eq!(report.schema_version, 0);
        assert_eq!(report.execution.rows_processed, 10);
        // Legacy metrics predate the assessability denominators: the facts
        // stay readable, but no score is fabricated from them.
        assert!(report.quality_score().is_none());
        let quality = report
            .quality
            .expect("expected legacy quality to deserialize");
        // Nothing was assessable, so the assessment says so rather than
        // claiming exactness about a score it never produced (#571).
        assert!(matches!(quality.confidence, MetricConfidence::NotAssessed));
        let completeness = quality
            .metrics
            .completeness
            .as_ref()
            .expect("legacy completeness facts should deserialize");
        assert!((completeness.complete_records_ratio - 100.0).abs() < 0.01);
        assert!(quality.metrics.assessed_dimensions().is_empty());
    }

    /// A canonical quality object that lost its `confidence` still holds its
    /// metrics. The flat legacy reader dropped every one of them without an
    /// error; they are kept, and the missing provenance stays unknown.
    #[test]
    fn a_canonical_quality_without_confidence_keeps_its_metrics() {
        let mut document = current_document();
        document["quality"] = json!({
            "metrics": {
                "completeness": {
                    "missing_values_ratio": 0.0,
                    "complete_records_ratio": 100.0,
                    "null_columns": []
                }
            }
        });

        let report: ProfileReport = serde_json::from_value(document).unwrap();

        let quality = report.quality.expect("the assessment survives");
        let completeness = quality
            .metrics
            .completeness
            .as_ref()
            .expect("the metrics survive");
        assert!((completeness.complete_records_ratio - 100.0).abs() < 0.01);
        assert!(!matches!(quality.confidence, MetricConfidence::Exact));
    }

    /// A `confidence` with no `metrics` is a malformed canonical object, not
    /// an empty legacy one.
    #[test]
    fn a_canonical_quality_without_metrics_is_malformed() {
        let mut document = current_document();
        document["quality"] = json!({ "confidence": "Exact" });

        assert!(serde_json::from_value::<ProfileReport>(document).is_err());
    }

    fn current_document() -> serde_json::Value {
        json!({
            "schema_version": REPORT_SCHEMA_VERSION,
            "id": "current-report",
            "timestamp": "2026-07-16T10:00:00Z",
            "data_source": {
                "type": "file",
                "path": "test.csv",
                "format": "csv",
                "size_bytes": 42
            },
            "column_profiles": [],
            "execution": {
                "rows_processed": 10,
                "columns_detected": 2,
                "scan_time_ms": 5,
                "error_count": 0,
                "source_exhausted": true,
                "sampling_applied": false
            }
        })
    }

    #[test]
    fn test_additive_fields_from_newer_writer_are_ignored() {
        // A newer dataprof may add fields within the same schema version;
        // a compatible reader must not break on them.
        let mut json = current_document();
        json["a_future_additive_field"] = json!({"anything": true});
        json["column_profiles"] = json!([]);

        let report: ProfileReport = serde_json::from_value(json).unwrap();
        assert_eq!(report.schema_version, REPORT_SCHEMA_VERSION);
        assert_eq!(report.id, "current-report");
    }

    #[test]
    fn test_unsupported_future_schema_version_fails_explicitly() {
        let mut json = current_document();
        json["schema_version"] = json!(REPORT_SCHEMA_VERSION + 1);

        let err = serde_json::from_value::<ProfileReport>(json).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("schema version") && msg.contains("upgrade dataprof"),
            "expected an actionable schema-version error, got: {msg}"
        );
    }

    #[test]
    fn test_version_error_wins_over_structural_errors() {
        // A future document that also broke structure, with schema_version
        // appearing after the broken field, must still fail with the version
        // error — not a confusing type error from the field encountered first.
        let json_text = format!(
            r#"{{"column_profiles": "not-an-array", "schema_version": {}}}"#,
            REPORT_SCHEMA_VERSION + 1
        );
        let err = serde_json::from_str::<ProfileReport>(&json_text).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("upgrade dataprof"),
            "expected the schema-version error to win, got: {msg}"
        );
    }

    #[test]
    fn test_null_schema_version_is_malformed_not_legacy() {
        let mut json = current_document();
        json["schema_version"] = json!(null);

        let err = serde_json::from_value::<ProfileReport>(json).unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("schema_version must be a non-negative integer"),
            "expected explicit rejection of null schema_version, got: {msg}"
        );
    }

    #[test]
    fn test_non_integer_schema_version_is_rejected() {
        for bad in [json!("1"), json!(1.5), json!(-1), json!(true)] {
            let mut json = current_document();
            json["schema_version"] = bad.clone();
            let err = serde_json::from_value::<ProfileReport>(json).unwrap_err();
            assert!(
                err.to_string().contains("non-negative integer"),
                "expected rejection of {bad}, got: {err}"
            );
        }
    }
}