import pandas as pd
import pytest

from engine import pipeline
from helpers import csv_b64, xlsx_b64

TEMPLATE_A = {
    "name": "src_a",
    "display_name": "Source A",
    "source_role": "settlement_bank",
    "file_format": "csv",
    "fields": [
        {"canonical": "ref", "aliases": ["Ref"]},
        {"canonical": "amount", "aliases": ["Amt"], "dtype": "number",
         "normalize": [{"step": "to_number"}]},
    ],
}

TEMPLATE_B = {
    "name": "src_b",
    "display_name": "Source B",
    "source_role": "gateway",
    "file_format": "csv",
    "fields": [
        {"canonical": "ref", "aliases": ["ref"]},
        {"canonical": "amount", "aliases": ["amount"], "dtype": "number",
         "normalize": [{"step": "to_number"}]},
        {"canonical": "transaction_id", "aliases": ["txn"]},
    ],
}

TEMPLATE_C = {
    "name": "src_c",
    "display_name": "Source C (internal)",
    "source_role": "internal_system",
    "file_format": "csv",
    "fields": [
        {"canonical": "transaction_id", "aliases": ["gateway_txn"]},
        {"canonical": "amount", "aliases": ["amount"], "dtype": "number",
         "normalize": [{"step": "to_number"}]},
    ],
}


def two_stage_rule_set():
    return {
        "name": "chain_test",
        "stages": [
            {
                "stage": "a_vs_b",
                "side_a": {"source": "template:src_a"},
                "side_b": {"source": "template:src_b"},
                "tiers": [{"name": "exact", "keys": [["ref", "ref"], ["amount", "amount"]]}],
                "output": {"matched_sheet": "M1", "unmatched_sheet": "U1",
                           "carry_fields": ["transaction_id"]},
            },
            {
                "stage": "c_vs_matched",
                "side_a": {"source": "template:src_c"},
                "side_b": {"source": "stage:a_vs_b:matched"},
                "tiers": [{"name": "txn", "keys": [["transaction_id", "transaction_id"],
                                                     ["amount", "amount"]]}],
                "output": {"matched_sheet": "M2", "unmatched_sheet": "U2"},
            },
        ],
    }


def test_chained_stages_carry_fields_flow_through():
    files = {
        "src_a": csv_b64(pd.DataFrame({"Ref": ["R1", "R2"], "Amt": [100, 200]})),
        "src_b": csv_b64(pd.DataFrame({"ref": ["R1"], "amount": [100], "txn": ["TXN1"]})),
        "src_c": csv_b64(pd.DataFrame({"gateway_txn": ["TXN1", "TXNX"], "amount": [100, 999]})),
    }
    templates = {"src_a": TEMPLATE_A, "src_b": TEMPLATE_B, "src_c": TEMPLATE_C}

    result = pipeline.run(two_stage_rule_set(), templates, files, {})

    s1, s2 = result["stages"]
    assert s1["matched_count"] == 1 and s1["unmatched_a_count"] == 1
    # Stage 2 matches the internal row against stage 1's matched output via
    # the carried transaction_id — the chaining that makes 3-way recon work.
    assert s2["matched_count"] == 1
    assert s2["matched_rows"][0]["transaction_id"] == "TXN1"
    assert s2["unmatched_a_count"] == 1

    assert result["summary"]["stage_count"] == 2
    assert result["summary"]["matched_count"] == 2


def test_disabled_stage_skipped():
    rule_set = two_stage_rule_set()
    rule_set["stages"][1]["enabled"] = False
    files = {
        "src_a": csv_b64(pd.DataFrame({"Ref": ["R1"], "Amt": [100]})),
        "src_b": csv_b64(pd.DataFrame({"ref": ["R1"], "amount": [100], "txn": ["T"]})),
    }
    result = pipeline.run(rule_set, {"src_a": TEMPLATE_A, "src_b": TEMPLATE_B}, files, {})
    assert result["summary"]["stage_count"] == 1


def test_missing_file_reports_template_name():
    rule_set = two_stage_rule_set()
    files = {"src_a": csv_b64(pd.DataFrame({"Ref": ["R1"], "Amt": [100]}))}
    with pytest.raises(pipeline.PipelineError, match="src_b"):
        pipeline.run(rule_set, {"src_a": TEMPLATE_A, "src_b": TEMPLATE_B}, files, {})


def test_forward_stage_reference_rejected():
    rule_set = {
        "name": "bad",
        "stages": [{
            "stage": "s1",
            "side_a": {"source": "stage:s2:matched"},
            "side_b": {"source": "template:src_a"},
            "tiers": [{"name": "t", "keys": [["ref", "ref"]]}],
            "output": {"matched_sheet": "M", "unmatched_sheet": "U"},
        }],
    }
    files = {"src_a": csv_b64(pd.DataFrame({"Ref": ["R1"], "Amt": [1]}))}
    with pytest.raises(pipeline.PipelineError, match="before stage 's2' ran"):
        pipeline.run(rule_set, {"src_a": TEMPLATE_A}, files, {})


def test_loader_auto_header_and_skip_rows():
    from engine.loader import load_source

    # Bank-portal-style file: two junk rows above the real header, a footer row.
    raw_rows = [
        ["Bank Statement Report", None],
        [None, None],
        ["Ref", "Amt"],
        ["R1", 100],
        ["Grand Total", 100],
    ]
    df = pd.DataFrame(raw_rows)
    buf_template = {
        "name": "auto_hdr",
        "display_name": "x",
        "source_role": "settlement_bank",
        "file_format": "xlsx",
        "header": {"row": "auto", "skip_rows_containing": ["Grand Total"]},
        "fields": [
            {"canonical": "ref", "aliases": ["Ref"]},
            {"canonical": "amount", "aliases": ["Amt"], "normalize": [{"step": "to_number"}]},
        ],
    }
    import io, base64
    buf = io.BytesIO()
    df.to_excel(buf, index=False, header=False)
    b64 = base64.b64encode(buf.getvalue()).decode()

    out = load_source(b64, buf_template, {})
    assert out["ref"].tolist() == ["R1"]
    assert out["amount"].tolist() == [100]


def test_loader_missing_required_column_names_template():
    from engine.loader import load_source, LoaderError
    b64 = csv_b64(pd.DataFrame({"Wrong": [1]}))
    with pytest.raises(LoaderError, match="src_a.*ref"):
        load_source(b64, TEMPLATE_A, {})


def test_loader_row_limit_guard():
    from engine.loader import load_source, LoaderError
    template = dict(TEMPLATE_A, row_limit=2)
    b64 = csv_b64(pd.DataFrame({"Ref": ["a", "b", "c"], "Amt": [1, 2, 3]}))
    with pytest.raises(LoaderError, match="exceeding the limit"):
        load_source(b64, template, {})