"""OECD SDMX connector — recorded csvfilewithlabels fixtures (house prices RHP quarterly; KEI production index monthly)."""
from __future__ import annotations

from datetime import date

import pytest

from countryatlas.connectors._util import ConnectorError
from countryatlas.connectors.oecd import OECDConnector, _clean_key, _with_version
from countryatlas.models import IndicatorSourceSpec, RawPayload

HP = "OECD.ECO.MPD,DSD_AN_HOUSE_PRICES@DF_HOUSE_PRICES,1.0"
KEI = "OECD.SDD.STES,DSD_KEI@DF_KEI,4.0"


def _raw(fixtures_dir, name: str, dataset: str, code: str, retrieved_at) -> RawPayload:
    body = (fixtures_dir / "oecd" / name).read_bytes()
    return RawPayload(connector="oecd", dataset=dataset, code=code, url="fixture://oecd", retrieved_at=retrieved_at,
                      status_code=200, content_type="text/csv", body=body)


@pytest.fixture
def oecd() -> OECDConnector:
    return OECDConnector()


def test_key_and_version_helpers():
    assert _clean_key("*.Q.RHP.IX") == ".Q.RHP.IX"
    assert _clean_key(".Q.RHP.IX") == ".Q.RHP.IX"
    assert _with_version("OECD.ELS.SPD,DSD_SOCX_AGG@DF_SOCX_AGG") == "OECD.ELS.SPD,DSD_SOCX_AGG@DF_SOCX_AGG,"
    assert _with_version(HP) == HP


def test_quarterly_index(oecd, fixtures_dir, retrieved_at):
    spec = IndicatorSourceSpec(indicator_id="real-house-price-index", connector="oecd", dataset=HP, code="*.Q.RHP.IX",
                               priority=1, frequency="Q")
    rows = oecd.normalize(_raw(fixtures_dir, "house_prices_RHP_sample.csv", HP, "*.Q.RHP.IX", retrieved_at), spec)
    assert {r.country_id for r in rows} == {"CAN", "USA"}
    usa = sorted((r for r in rows if r.country_id == "USA"), key=lambda r: r.period)
    assert usa[0].period == date(2023, 1, 1) and usa[0].frequency == "Q" and usa[0].year == 2023
    assert any(r.period == date(2025, 4, 1) for r in usa), "2025-Q2 → 2025-04-01"
    q2 = next(r for r in usa if r.period == date(2025, 4, 1))
    assert q2.value == pytest.approx(154.108309038445)
    assert q2.unit == "index (2015 = 100)"
    assert q2.metadata["source_unit"] == "IX" and q2.metadata["base_period"] == "2015"
    assert not q2.is_estimate and not q2.is_forecast
    assert oecd.validate(rows).errors == 0


def test_yoy_transform_derives_growth(oecd, fixtures_dir, retrieved_at):
    spec = IndicatorSourceSpec(indicator_id="house-price-growth", connector="oecd", dataset=HP, code="*.Q.RHP.IX",
                               priority=1, frequency="Q", transform="yoy")
    raw = _raw(fixtures_dir, "house_prices_RHP_sample.csv", HP, "*.Q.RHP.IX", retrieved_at)
    idx = {(r.country_id, r.period): r.value for r in oecd.normalize(
        raw, IndicatorSourceSpec(indicator_id="real-house-price-index", connector="oecd", dataset=HP, code="*.Q.RHP.IX"))}
    rows = oecd.normalize(raw, spec)
    assert rows, "growth rows exist once a full year of history is present"
    # first four quarters (2023) have no previous-year value → dropped
    assert min(r.period for r in rows) == date(2024, 1, 1)
    r = next(r for r in rows if r.country_id == "CAN" and r.period == date(2025, 1, 1))
    expected = (idx[("CAN", date(2025, 1, 1))] / idx[("CAN", date(2024, 1, 1))] - 1) * 100
    assert r.value == pytest.approx(expected)
    assert r.unit == "annual %" and r.metadata["derived"].startswith("year-on-year")


def test_monthly_kei_periods(oecd, fixtures_dir, retrieved_at):
    spec = IndicatorSourceSpec(indicator_id="industrial-production-index", connector="oecd", dataset=KEI,
                               code="*.M.PRVM.IX.BTE.Y._Z", priority=1, frequency="M")
    rows = oecd.normalize(_raw(fixtures_dir, "kei_PRVM_sample.csv", KEI, spec.code, retrieved_at), spec)
    can = sorted((r for r in rows if r.country_id == "CAN"), key=lambda r: r.period)
    assert can[0].frequency == "M" and can[0].period.day == 1 and can[0].period.year == 2026
    assert any(r.period == date(2026, 6, 1) for r in can)
    assert next(r for r in can if r.period == date(2026, 6, 1)).value == pytest.approx(109.273561067984)


def test_post_hoc_filter_and_estimate_flag(oecd, fixtures_dir, retrieved_at):
    body = (fixtures_dir / "oecd" / "kei_PRVM_sample.csv").read_text()
    # mark one row as estimate and another as provisional
    lines = body.splitlines()
    lines[1] = lines[1].replace(",A,Normal value,", ",E,Estimated value,", 1)
    lines[2] = lines[2].replace(",A,Normal value,", ",P,Provisional value,", 1)
    raw = RawPayload(connector="oecd", dataset=KEI, code="*.M.PRVM.IX.BTE.Y._Z", url="fixture://oecd",
                     retrieved_at=retrieved_at, status_code=200, content_type="text/csv", body="\n".join(lines).encode())
    spec = IndicatorSourceSpec(indicator_id="industrial-production-index", connector="oecd", dataset=KEI,
                               code="*.M.PRVM.IX.BTE.Y._Z", priority=1, frequency="M", params={"filter": {"ACTIVITY": "BTE"}})
    rows = oecd.normalize(raw, spec)
    assert sum(r.is_estimate for r in rows) == 2
    assert {r.metadata.get("obs_status") for r in rows if r.is_estimate} == {"E", "P"}
    bad = IndicatorSourceSpec(indicator_id="industrial-production-index", connector="oecd", dataset=KEI,
                              code="*.M.PRVM.IX.BTE.Y._Z", params={"filter": {"ACTIVITY": "ZZZ"}})
    assert oecd.normalize(raw, bad) == []


def test_underspecified_key_raises(oecd, fixtures_dir, retrieved_at):
    body = (fixtures_dir / "oecd" / "house_prices_RHP_sample.csv").read_text()
    dup = body + "\n" + body.splitlines()[1].replace(",RHP,Real house price indices,", ",HPI,Nominal house price indices,")
    raw = RawPayload(connector="oecd", dataset=HP, code="*.Q..IX", url="fixture://oecd", retrieved_at=retrieved_at,
                     status_code=200, content_type="text/csv", body=dup.encode())
    spec = IndicatorSourceSpec(indicator_id="real-house-price-index", connector="oecd", dataset=HP, code="*.Q..IX")
    with pytest.raises(ConnectorError, match="under-specified"):
        oecd.normalize(raw, spec)


def test_registry_specs_for_oecd():
    from countryatlas import registry

    specs = registry.source_specs("oecd")
    by_ind = {s.indicator_id: s for s in specs}
    assert {"real-house-price-index", "tax-revenue-pct-gdp", "average-annual-wages", "industrial-production-index",
            "tertiary-attainment-25-64"} <= set(by_ind)
    assert by_ind["industrial-production-index"].frequency == "M"
    assert by_ind["house-price-growth"].transform == "yoy"
    for s in specs:
        assert s.dataset.count(",") in (1, 2), s.dataset
        assert s.code.startswith("*."), s.code


@pytest.mark.live
def test_live_house_prices_two_countries():
    oecd = OECDConnector()
    spec = IndicatorSourceSpec(indicator_id="real-house-price-index", connector="oecd", dataset=HP, code="CAN+USA.Q.RHP.IX",
                               priority=1, frequency="Q", params={"startPeriod": "2024-Q1"})
    raw = oecd.fetch(spec)
    rows = oecd.normalize(raw, spec)
    assert {r.country_id for r in rows} == {"CAN", "USA"}
    assert all(r.frequency == "Q" for r in rows)
    oecd.close()