use std::collections::HashMap;
use std::hash::Hash;

use chrono::NaiveDateTime;

use crate::types::*;

/// Parse an AEMO date-time string.
///
/// The primary format across all 6 tables we support is
/// `"YYYY/MM/DD HH:MM:SS"`. Some other AEMO tables (bidding, sub-5s
/// FCAS) use alternate formats with milliseconds or hyphen separators;
/// we accept those as fallbacks for defensiveness even though none of
/// our 6 tables use them in practice.
pub fn parse_aemo_datetime(s: &str) -> Option<NaiveDateTime> {
    const FORMATS: &[&str] = &[
        "%Y/%m/%d %H:%M:%S",
        "%Y/%m/%d %H:%M:%S%.f",
        "%Y-%m-%d %H:%M:%S",
    ];
    for fmt in FORMATS {
        if let Ok(dt) = NaiveDateTime::parse_from_str(s, fmt) {
            return Some(dt);
        }
    }
    None
}

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

    #[test]
    fn parses_standard_format() {
        assert!(parse_aemo_datetime("2024/01/01 00:05:00").is_some());
    }

    #[test]
    fn parses_millisecond_format() {
        assert!(parse_aemo_datetime("2024/01/01 00:05:00.500").is_some());
    }

    #[test]
    fn parses_hyphen_format() {
        assert!(parse_aemo_datetime("2024-01-01 00:05:00").is_some());
    }

    #[test]
    fn rejects_invalid() {
        assert!(parse_aemo_datetime("not a date").is_none());
        assert!(parse_aemo_datetime("").is_none());
    }
}

// Row accessor
//
// `Row` is a zero-copy view over a single AEMO data row. It holds a
// slice of field references (all pointing into the original CSV
// string) together with a column-name → position index built once from
// the `I` header row. Accessors trim whitespace and surrounding quotes
// on demand, so we never allocate a `String` just to look up a field.

struct Row<'a, 'b> {
    fields: &'b [&'a str],
    col_idx: &'b HashMap<&'a str, usize>,
}

impl<'a> Row<'a, '_> {
    fn get(&self, name: &str) -> Option<&'a str> {
        let i = *self.col_idx.get(name)?;
        self.fields.get(i).copied().map(|s| s.trim().trim_matches('"'))
    }

    fn f64(&self, name: &str) -> Option<f64> {
        self.get(name)?.parse().ok()
    }

    fn u8(&self, name: &str) -> u8 {
        self.get(name).and_then(|s| s.parse().ok()).unwrap_or(0)
    }

    fn region(&self) -> Option<Region> {
        self.get("REGIONID")?.parse().ok()
    }

    fn datetime(&self, name: &str) -> Option<NaiveDateTime> {
        parse_aemo_datetime(self.get(name)?)
    }

    fn string(&self, name: &str) -> Option<String> {
        Some(self.get(name)?.to_string())
    }
}

// ── Core scanner ────────────────────────────────────────────────────
//
// Every AEMO CSV we parse follows the same shape:
//   I,TABLE,SUBTYPE,COL1,COL2,...   — column header row
//   D,TABLE,SUBTYPE,VAL1,VAL2,...   — data rows
//
// `scan_rows` walks the file exactly once, maintaining a
// `HashMap<&str, usize>` column index that updates whenever it sees a
// matching `I` row, and calls `visit` on every matching `D` row.
//
// The column-index keys are `&str` slices borrowing directly from the
// CSV — no allocation per header. Data rows are likewise passed as
// `Vec<&str>` which borrow from each line.

fn scan_rows<'a>(
    csv: &'a str,
    table: &str,
    subtypes: &[&str],
    mut visit: impl FnMut(&Row<'a, '_>),
) {
    let mut col_idx: HashMap<&'a str, usize> = HashMap::new();

    for line in csv.lines() {
        let fields: Vec<&'a str> = line.split(',').collect();
        if fields.len() < 4 {
            continue;
        }

        let tag = fields[0].trim();
        let tbl = fields[1].trim();
        let sub = fields[2].trim();
        let matches = tbl == table && subtypes.contains(&sub);
        if !matches {
            continue;
        }

        if tag == "I" {
            col_idx.clear();
            for (i, f) in fields.iter().enumerate() {
                col_idx.insert(f.trim().trim_matches('"'), i);
            }
            continue;
        }

        if tag == "D" && !col_idx.is_empty() {
            let row = Row { fields: &fields, col_idx: &col_idx };
            visit(&row);
        }
    }
}

// ── Generic parse helpers ──────────────────────────────────────────

/// Scan an AEMO CSV, calling `build` on each matching data row and
/// collecting the `Some` results into a `Vec`. No deduplication — use
/// this for tables without an `INTERVENTION` column or where
/// duplicates are impossible.
fn parse_typed<T>(
    csv: &str,
    table: &str,
    subtypes: &[&str],
    build: impl Fn(&Row) -> Option<T>,
) -> Vec<T> {
    let mut out = Vec::new();
    scan_rows(csv, table, subtypes, |row| {
        if let Some(t) = build(row) {
            out.push(t);
        }
    });
    out
}

/// Scan an AEMO CSV with `INTERVENTION`-aware deduplication.
///
/// For each matching data row, `build` constructs the typed record and
/// `key` extracts its primary key. When multiple rows share the same
/// key, the one with the highest `INTERVENTION` value wins — this
/// implements the convention that AEMO publishes both a counterfactual
/// (`INTERVENTION=0`) and the actual cleared values (`INTERVENTION=1`)
/// during market interventions, and consumers want the latter.
fn parse_typed_deduped<T, K: Eq + Hash>(
    csv: &str,
    table: &str,
    subtypes: &[&str],
    key: impl Fn(&T) -> K,
    build: impl Fn(&Row) -> Option<T>,
) -> Vec<T> {
    let mut best: HashMap<K, (i32, T)> = HashMap::new();
    scan_rows(csv, table, subtypes, |row| {
        let Some(record) = build(row) else { return };
        let iv = row.get("INTERVENTION").and_then(|s| s.parse().ok()).unwrap_or(0);
        let k = key(&record);
        match best.get(&k) {
            Some((existing, _)) if *existing >= iv => {}
            _ => {
                best.insert(k, (iv, record));
            }
        }
    });
    best.into_values().map(|(_, t)| t).collect()
}

const PRICE_RANGE: std::ops::RangeInclusive<f64> = -1000.0..=17500.0;

fn valid_price(p: f64) -> bool {
    p.is_finite() && PRICE_RANGE.contains(&p)
}

// ── Per-table parsers ──────────────────────────────────────────────

pub fn parse_dispatch_prices(csv: &str) -> Vec<DispatchPrice> {
    parse_typed_deduped(
        csv,
        "DISPATCH",
        &["PRICE"],
        |d: &DispatchPrice| (d.settlement_date, d.region),
        |row| {
            let price = row.f64("RRP")?;
            if !valid_price(price) {
                return None;
            }
            Some(DispatchPrice {
                region: row.region()?,
                settlement_date: row.datetime("SETTLEMENTDATE")?,
                price,
                apc_flag: row.u8("APCFLAG"),
                market_suspended_flag: row.u8("MARKETSUSPENDEDFLAG"),
            })
        },
    )
}

pub fn parse_trading_prices(csv: &str) -> Vec<TradingPrice> {
    // TRADINGPRICE has no INTERVENTION column and no duplicate rows,
    // so we skip dedup entirely for speed.
    parse_typed(csv, "TRADING", &["PRICE"], |row| {
        let price = row.f64("RRP")?;
        if !valid_price(price) {
            return None;
        }
        Some(TradingPrice {
            region: row.region()?,
            settlement_date: row.datetime("SETTLEMENTDATE")?,
            price,
            invalid: row.u8("INVALIDFLAG") != 0,
        })
    })
}

pub fn parse_region_demand(csv: &str) -> Vec<RegionDemand> {
    parse_typed_deduped(
        csv,
        "DISPATCH",
        &["REGIONSUM"],
        |d: &RegionDemand| (d.settlement_date, d.region),
        |row| {
            Some(RegionDemand {
                region: row.region()?,
                settlement_date: row.datetime("SETTLEMENTDATE")?,
                total_demand: row.f64("TOTALDEMAND")?,
                available_generation: row.f64("AVAILABLEGENERATION")?,
                net_interchange: row.f64("NETINTERCHANGE")?,
            })
        },
    )
}

pub fn parse_interconnector_flows(csv: &str) -> Vec<InterconnectorFlow> {
    parse_typed_deduped(
        csv,
        "DISPATCH",
        &["INTERCONNECTORRES"],
        |f: &InterconnectorFlow| (f.settlement_date, f.interconnector_id.clone()),
        |row| {
            Some(InterconnectorFlow {
                interconnector_id: row.string("INTERCONNECTORID")?,
                settlement_date: row.datetime("SETTLEMENTDATE")?,
                mw_flow: row.f64("MWFLOW")?,
                metered_mw_flow: row.f64("METEREDMWFLOW")?,
                mw_losses: row.f64("MWLOSSES")?,
            })
        },
    )
}

pub fn parse_predispatch_prices(csv: &str) -> Vec<PredispatchPrice> {
    // PREDISPATCH REGION_PRICES also carries INTERVENTION — during
    // market interventions the forecast is re-run with INTERVENTION=1,
    // duplicating every (DATETIME, REGIONID) pair.
    parse_typed_deduped(
        csv,
        "PREDISPATCH",
        &["PRICE", "REGION_PRICES"],
        |f: &PredispatchPrice| (f.forecast_time, f.region),
        |row| {
            let price = row.f64("RRP")?;
            if !valid_price(price) {
                return None;
            }
            Some(PredispatchPrice {
                region: row.region()?,
                forecast_time: row.datetime("DATETIME")?,
                price,
            })
        },
    )
}

pub fn parse_unit_scada(csv: &str) -> Vec<UnitScada> {
    parse_typed(csv, "DISPATCH", &["UNIT_SCADA", "SCADA_VALUE"], |row| {
        Some(UnitScada {
            duid: row.string("DUID")?,
            settlement_date: row.datetime("SETTLEMENTDATE")?,
            scada_value: row.f64("SCADAVALUE")?,
        })
    })
}

// Parser unit tests (crafted CSV fixtures, no network)
#[cfg(test)]
mod parser_tests {
    use super::*;

    // A minimal DISPATCHPRICE CSV with one (settlement, region) duplicated
    // under INTERVENTION=0 and =1. Dedup must keep only the =1 row.
    #[test]
    fn dispatch_prices_dedupe_intervention() {
        let csv = "\
C,NEMP.WORLD,DISPATCHPRICE,...
I,DISPATCH,PRICE,4,SETTLEMENTDATE,RUNNO,REGIONID,DISPATCHINTERVAL,INTERVENTION,RRP,EEP,ROP,APCFLAG,MARKETSUSPENDEDFLAG
D,DISPATCH,PRICE,4,\"2022/06/15 10:00:00\",1,NSW1,1,0,99.00,0,0,0,1
D,DISPATCH,PRICE,4,\"2022/06/15 10:00:00\",1,NSW1,1,1,123.45,0,0,1,1
D,DISPATCH,PRICE,4,\"2022/06/15 10:00:00\",1,QLD1,1,0,80.00,0,0,0,0
";
        let prices = parse_dispatch_prices(csv);
        assert_eq!(prices.len(), 2, "Expected 2 rows after dedup, got {}", prices.len());
        let nsw = prices.iter().find(|p| p.region == Region::NSW1).unwrap();
        assert_eq!(nsw.price, 123.45, "Expected INTERVENTION=1 price (123.45)");
        assert_eq!(nsw.market_suspended_flag, 1);
    }

    // Trading price has no INTERVENTION column; the INVALIDFLAG column must
    // surface on the struct.
    #[test]
    fn trading_prices_surface_invalidflag() {
        let csv = "\
C,NEMP.WORLD,TRADINGPRICE,...
I,TRADING,PRICE,2,SETTLEMENTDATE,RUNNO,REGIONID,PERIODID,RRP,EEP,INVALIDFLAG,LASTCHANGED
D,TRADING,PRICE,2,\"2024/01/01 00:05:00\",1,NSW1,1,50.0,0,0,\"2024/01/01 00:05:00\"
D,TRADING,PRICE,2,\"2024/01/01 00:10:00\",1,NSW1,2,60.0,0,1,\"2024/01/01 00:10:00\"
";
        let prices = parse_trading_prices(csv);
        assert_eq!(prices.len(), 2);
        assert!(!prices[0].invalid || !prices[1].invalid);
        assert!(prices.iter().any(|p| p.invalid));
    }

    // Predispatch must dedupe on (DATETIME, REGIONID).
    #[test]
    fn predispatch_prices_dedupe_intervention() {
        let csv = "\
C,NEMP.WORLD,PREDISPATCHPRICE,...
I,PREDISPATCH,REGION_PRICES,2,PREDISPATCHSEQNO,RUNNO,REGIONID,PERIODID,INTERVENTION,RRP,EEP,RRP1,EEP1,RRP2,EEP2,RRP3,EEP3,RRP4,EEP4,RRP5,EEP5,RRP6,EEP6,RRP7,EEP7,RRP8,EEP8,LASTCHANGED,DATETIME
D,PREDISPATCH,REGION_PRICES,2,2022061500,1,NSW1,1,0,75.00,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,\"2022/06/15 10:00:00\",\"2022/06/15 10:30:00\"
D,PREDISPATCH,REGION_PRICES,2,2022061500,1,NSW1,1,1,95.50,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,\"2022/06/15 10:00:00\",\"2022/06/15 10:30:00\"
";
        let forecasts = parse_predispatch_prices(csv);
        assert_eq!(forecasts.len(), 1, "Expected 1 row after dedup, got {}", forecasts.len());
        assert_eq!(forecasts[0].price, 95.50, "Expected INTERVENTION=1 price");
    }

    // Interconnector dedup keyed by (settlement, interconnector_id), and
    // METEREDMWFLOW must differ from MWFLOW.
    #[test]
    fn interconnector_dedupe_and_metered_flow() {
        let csv = "\
C,NEMP.WORLD,DISPATCHINTERCONNECTORRES,...
I,DISPATCH,INTERCONNECTORRES,3,SETTLEMENTDATE,RUNNO,INTERCONNECTORID,DISPATCHINTERVAL,INTERVENTION,METEREDMWFLOW,MWFLOW,MWLOSSES,MARGINALVALUE,VIOLATIONDEGREE
D,DISPATCH,INTERCONNECTORRES,3,\"2022/06/15 10:00:00\",1,VIC1-NSW1,1,0,100.5,95.0,5.0,0,0
D,DISPATCH,INTERCONNECTORRES,3,\"2022/06/15 10:00:00\",1,VIC1-NSW1,1,1,105.2,100.0,5.2,0,0
";
        let flows = parse_interconnector_flows(csv);
        assert_eq!(flows.len(), 1);
        assert_eq!(flows[0].mw_flow, 100.0);
        assert_eq!(flows[0].metered_mw_flow, 105.2);
    }

    // Region demand must pick up TOTALDEMAND, AVAILABLEGENERATION,
    // NETINTERCHANGE correctly.
    #[test]
    fn region_demand_fields() {
        let csv = "\
I,DISPATCH,REGIONSUM,7,SETTLEMENTDATE,RUNNO,REGIONID,INTERVENTION,TOTALDEMAND,AVAILABLEGENERATION,AVAILABLELOAD,DEMANDFORECAST,NETINTERCHANGE
D,DISPATCH,REGIONSUM,7,\"2024/01/01 00:05:00\",1,NSW1,0,9000.0,13000.0,0.0,9100.0,-800.0
";
        let demand = parse_region_demand(csv);
        assert_eq!(demand.len(), 1);
        assert_eq!(demand[0].total_demand, 9000.0);
        assert_eq!(demand[0].available_generation, 13000.0);
        assert_eq!(demand[0].net_interchange, -800.0);
    }

    // SCADA must accept negative values (loads, batteries charging).
    #[test]
    fn scada_accepts_negative_values() {
        let csv = "\
I,DISPATCH,UNIT_SCADA,1,SETTLEMENTDATE,DUID,SCADAVALUE
D,DISPATCH,UNIT_SCADA,1,\"2024/01/01 00:05:00\",BBATTERY,-45.3
D,DISPATCH,UNIT_SCADA,1,\"2024/01/01 00:05:00\",GENUNIT1,250.1
";
        let scada = parse_unit_scada(csv);
        assert_eq!(scada.len(), 2);
        let batt = scada.iter().find(|s| s.duid == "BBATTERY").unwrap();
        assert_eq!(batt.scada_value, -45.3);
    }

    // Price range filter drops obvious parse errors (column shift puts a
    // non-price value into the RRP field).
    #[test]
    fn dispatch_filters_out_of_range_price() {
        let csv = "\
I,DISPATCH,PRICE,4,SETTLEMENTDATE,RUNNO,REGIONID,DISPATCHINTERVAL,INTERVENTION,RRP,EEP,ROP,APCFLAG,MARKETSUSPENDEDFLAG
D,DISPATCH,PRICE,4,\"2024/01/01 00:05:00\",1,NSW1,1,0,50.0,0,0,0,0
D,DISPATCH,PRICE,4,\"2024/01/01 00:05:00\",1,QLD1,1,0,99999.0,0,0,0,0
";
        let prices = parse_dispatch_prices(csv);
        assert_eq!(prices.len(), 1);
        assert_eq!(prices[0].region, Region::NSW1);
    }
}
