import pandas as pd

from engine.matcher import run_stage, apply_filters


def stage(tiers, filters_a=None, filters_b=None, carry=None):
    return {
        "stage": "test",
        "side_a": {"source": "template:a", "filters": filters_a},
        "side_b": {"source": "template:b", "filters": filters_b},
        "tiers": tiers,
        "output": {"matched_sheet": "M", "unmatched_sheet": "U",
                   "carry_fields": carry or []},
    }


def test_one_to_one_consumption_no_cartesian_duplication():
    # Two identical A rows, ONE matching B row: exactly one match —
    # the reference implementation's LEFT JOIN would have matched both.
    df_a = pd.DataFrame({"key": ["K1", "K1"], "amount": [100.0, 100.0]})
    df_b = pd.DataFrame({"key": ["K1"], "amount": [100.0]})

    result = run_stage(df_a, df_b, stage([{"name": "t1", "keys": [["key", "key"]]}]), {})

    assert len(result["matched"]) == 1
    assert len(result["unmatched_a"]) == 1
    assert len(result["unmatched_b"]) == 0


def test_duplicate_keys_pair_up_positionally():
    # 2 A rows and 2 B rows with the same key: both pair (occurrence matching).
    df_a = pd.DataFrame({"key": ["K1", "K1"]})
    df_b = pd.DataFrame({"key": ["K1", "K1"]})
    result = run_stage(df_a, df_b, stage([{"name": "t1", "keys": [["key", "key"]]}]), {})
    assert len(result["matched"]) == 2


def test_tier_cascade_consumes_and_labels():
    df_a = pd.DataFrame({
        "card": ["1111", "2222", "3333"],
        "approval": ["A1", "XX", "A3"],
    })
    df_b = pd.DataFrame({
        "card": ["1111", "2222"],
        "approval": ["A1", "YY"],
    })
    tiers = [
        {"name": "exact-2key", "keys": [["card", "card"], ["approval", "approval"]]},
        {"name": "1key", "keys": [["card", "card"]]},
    ]
    result = run_stage(df_a, df_b, stage(tiers), {})

    matched = result["matched"]
    assert len(matched) == 2
    by_card = dict(zip(matched["card"], matched["match_tier"]))
    assert by_card == {"1111": "exact-2key", "2222": "1key"}
    assert result["tier_counts"] == {"exact-2key": 1, "1key": 1}
    assert result["unmatched_a"]["card"].tolist() == ["3333"]


def test_disabled_tier_is_skipped():
    df_a = pd.DataFrame({"card": ["1111"]})
    df_b = pd.DataFrame({"card": ["1111"]})
    tiers = [{"name": "off", "enabled": False, "keys": [["card", "card"]]}]
    result = run_stage(df_a, df_b, stage(tiers), {})
    assert len(result["matched"]) == 0


def test_null_keys_never_match():
    df_a = pd.DataFrame({"key": [None, "K1"]})
    df_b = pd.DataFrame({"key": [None, "K1"]})
    result = run_stage(df_a, df_b, stage([{"name": "t", "keys": [["key", "key"]]}]), {})
    assert len(result["matched"]) == 1  # only K1 = K1; null != null


def test_numeric_keys_match_across_representations():
    df_a = pd.DataFrame({"amount": pd.to_numeric(pd.Series(["500"]))})
    df_b = pd.DataFrame({"amount": pd.to_numeric(pd.Series(["500.00"]))})
    result = run_stage(df_a, df_b, stage([{"name": "t", "keys": [["amount", "amount"]]}]), {})
    assert len(result["matched"]) == 1


def test_amount_tolerance_tier():
    df_a = pd.DataFrame({"tid": ["T1", "T2"], "amount": [100.0, 200.0]})
    df_b = pd.DataFrame({"tid": ["T1", "T2"], "amount": [100.6, 250.0]})
    tiers = [{
        "name": "tolerant",
        "keys": [["tid", "tid"]],
        "amount_tolerance": {"type": "abs", "value": 1.0, "fields": ["amount", "amount"]},
    }]
    result = run_stage(df_a, df_b, stage(tiers), {})
    assert result["matched"]["tid"].tolist() == ["T1"]
    assert result["unmatched_a"]["tid"].tolist() == ["T2"]


def test_date_window_tier():
    df_a = pd.DataFrame({
        "tid": ["T1", "T2"],
        "date": pd.to_datetime(["2026-06-10", "2026-06-10"]),
    })
    df_b = pd.DataFrame({
        "tid": ["T1", "T2"],
        "date": pd.to_datetime(["2026-06-12", "2026-06-14"]),
    })
    tiers = [{
        "name": "window",
        "keys": [["tid", "tid"]],
        "date_window": {"days": 2, "fields": ["date", "date"]},
    }]
    result = run_stage(df_a, df_b, stage(tiers), {})
    assert result["matched"]["tid"].tolist() == ["T1"]


def test_carry_fields_copied_from_side_b():
    df_a = pd.DataFrame({"key": ["K1"]})
    df_b = pd.DataFrame({"key": ["K1"], "transaction_id": ["TXN9"]})
    result = run_stage(
        df_a, df_b,
        stage([{"name": "t", "keys": [["key", "key"]]}], carry=["transaction_id"]),
        {},
    )
    assert result["matched"]["transaction_id"].tolist() == ["TXN9"]


def test_matched_b_aligned_with_matched():
    df_a = pd.DataFrame({"key": ["K1", "K2"]})
    df_b = pd.DataFrame({"key": ["K2", "K1"], "ref": ["R2", "R1"]})
    result = run_stage(df_a, df_b, stage([{"name": "t", "keys": [["key", "key"]]}]), {})
    pairs = list(zip(result["matched"]["key"], result["matched_b"]["ref"]))
    assert sorted(pairs) == [("K1", "R1"), ("K2", "R2")]


def test_filters():
    df = pd.DataFrame({"mode": ["CARD", "UPI", None], "x": [1, 2, 3]})
    assert apply_filters(df, [{"field": "mode", "op": "eq", "value": "UPI"}])["x"].tolist() == [2]
    assert apply_filters(df, [{"field": "mode", "op": "in", "values": ["CARD", "UPI"]}])["x"].tolist() == [1, 2]
    assert apply_filters(df, [{"field": "mode", "op": "not_null"}])["x"].tolist() == [1, 2]
    assert apply_filters(df, [{"field": "mode", "op": "is_null"}])["x"].tolist() == [3]