# 返利班次：领取戳、租约、查号落表、系统故障短码。

from contextlib import contextmanager
from datetime import datetime, timedelta

from job import (
    HEADER_FAILED,
    IMPORT_FAILED,
    PAGE_FAILED,
    SUBMIT_FAILED,
    recorder,
    run_batch,
)
from persist import (
    KIND_FAIL,
    KIND_NOT_FOUND,
    KIND_OK,
    USER_NAME,
    AgreementOutcome,
)
from rebate_flow_steps import RebateImportResult, RebateSubmitResult
from repo import MemoryRebateRepo


# 构造一条表 4 起草行。
def _t4(**overrides) -> dict:
    row = {
        "fid": 1,
        "oms_id": 2026100501,
        "action_code": 1,
        "contract_status": 1,
        "contract_fail_reason": None,
        "contract_no": "HT1",
        "agreement_status": None,
        "agreement_fail_reason": None,
        "agreement_id": None,
        "agreement_dtl_id": None,
        "org_code": "901",
        "push_status": 0,
        "process_status": 0,
        "create_user_name": USER_NAME,
        "update_user_name": USER_NAME,
    }
    row.update(overrides)
    return row


# 构造一条表 3 只录协议行。
def _t3(**overrides) -> dict:
    row = {
        "fid": 10,
        "oms_id": 2026100503,
        "contract_no": "HT-OLD",
        "contract_entry_status": 0,
        "process_status": 0,
        "process_result": "",
        "org_code": "901",
    }
    row.update(overrides)
    return row


# 总单替身，返回固定协议总单号。
def _ok_header(contract_no: str) -> str:
    return "RB1"


# 细单替身，全部成功。
def _ok_importer(contract_no: str) -> RebateImportResult:
    return RebateImportResult(1, 0, (), True)


# 送审替身，带总单/细单号。
def _ok_submitter(contract_no: str) -> RebateSubmitResult:
    return RebateSubmitResult(True, "AGR1", "AGRD1")


# 搜到并送审成功：起草行 process_status=1，每 oms_id 插一条 action_code=2。
def test_job_finds_contract_and_writes_agreement():
    repo = MemoryRebateRepo(
        table4=[
            _t4(),
            _t4(fid=2, oms_id=2026100502),
            _t4(fid=3, contract_status=2, contract_no="HT-DONE", process_status=1),
        ]
    )
    steps: list[str] = []

    # 记下合同号并回成功。
    def runner(group):
        steps.append(group.contract_no)
        return AgreementOutcome(KIND_OK, "AGR1", "AGRD1", None)

    summary = run_batch(repo, runner=runner)
    assert summary.claimed == 2
    assert summary.ok == 1
    assert steps == ["HT1"]
    drafts = [row for row in repo.table4.values() if row["fid"] in (1, 2)]
    assert {row["contract_status"] for row in drafts} == {2}
    assert {row["action_code"] for row in drafts} == {1}
    assert {row["process_status"] for row in drafts} == {1}
    assert {row["update_user_name"] for row in drafts} == {USER_NAME}
    assert {row["push_status"] for row in drafts} == {0}
    agreements = [row for row in repo.table4.values() if row["action_code"] == 2]
    assert sorted(row["oms_id"] for row in agreements) == [2026100501, 2026100502]
    assert {row["contract_no"] for row in agreements} == {"HT1"}
    assert {row["agreement_status"] for row in agreements} == {1}
    assert {row["agreement_id"] for row in agreements} == {"AGR1"}
    assert {row["agreement_dtl_id"] for row in agreements} == {"AGRD1"}
    assert {row["push_status"] for row in agreements} == {0}
    assert {row["process_status"] for row in agreements} == {1}
    assert {row["create_user_name"] for row in agreements} == {USER_NAME}
    assert {row["update_user_name"] for row in agreements} == {USER_NAME}
    assert repo.table4[3]["contract_status"] == 2
    assert repo.table4[3]["action_code"] == 1


# 送审失败：起草行回到可领，agreement_fail_reason 可查，不插协议行。
def test_job_failed_draft_stays_claimable_with_reason():
    repo = MemoryRebateRepo(
        table4=[_t4(), _t4(fid=2, oms_id=2026100502)]
    )

    # 模拟送审失败。
    def runner(group):
        return AgreementOutcome(KIND_FAIL, None, None, "送审失败")

    summary = run_batch(repo, runner=runner)
    assert summary.failed == 1
    assert summary.ok == 0
    for fid in (1, 2):
        row = repo.table4[fid]
        assert row["contract_status"] == 1
        assert row["process_status"] == 0
        assert row["agreement_fail_reason"] == "送审失败"
        assert row["action_code"] == 1
    assert [row["action_code"] for row in repo.table4.values()] == [1, 1]
    again = repo.claim_pending()
    assert [row["fid"] for row in again.drafts] == [1, 2]


# 搜不到：起草行 process_status 改回 0，不插协议行，留给下一班。
def test_job_leaves_not_found_for_next_shift():
    repo = MemoryRebateRepo(table4=[_t4()])
    later = {"n": 0}

    # 模拟查号未命中。
    def runner(group):
        later["n"] += 1
        return AgreementOutcome(KIND_NOT_FOUND, None, None, None)

    run_batch(repo, runner=runner)
    assert later["n"] == 1
    row = repo.table4[1]
    assert row["contract_status"] == 1
    assert row["action_code"] == 1
    assert row["process_status"] == 0
    assert row["update_user_name"] == USER_NAME
    assert [item["action_code"] for item in repo.table4.values()] == [1]


# 表 3 只录协议行领走后写表 4，已完成行不领。
def test_job_claims_protocol_only_and_writes_table4():
    repo = MemoryRebateRepo(table3=[_t3(), _t3(fid=11, process_status=1, contract_no="HT-SKIP")])
    steps: list[str] = []

    # 记下合同号并回成功。
    def runner(group):
        steps.append(group.contract_no)
        return AgreementOutcome(KIND_OK, "AGR-OLD", "AGRD-OLD", None)

    run_batch(repo, runner=runner)
    assert steps == ["HT-OLD"]
    assert repo.table3[10]["process_status"] == 1
    assert repo.table3[11]["process_status"] == 1
    success = [row for row in repo.table4.values() if row["action_code"] == 1]
    agreements = [row for row in repo.table4.values() if row["action_code"] == 2]
    assert len(success) == 1
    assert success[0]["oms_id"] == 2026100503
    assert success[0]["contract_status"] == 2
    assert success[0]["contract_no"] == "HT-OLD"
    assert success[0]["push_status"] == 0
    assert len(agreements) == 1
    assert agreements[0]["agreement_status"] == 1
    assert agreements[0]["agreement_id"] == "AGR-OLD"
    assert agreements[0]["agreement_dtl_id"] == "AGRD-OLD"


# 待查号行领走后 process_status=3，已回写行不领。
def test_claim_pending_stamps_drafts_and_skips_done():
    repo = MemoryRebateRepo(
        table4=[
            _t4(),
            _t4(fid=2, contract_status=2, process_status=1),
        ]
    )
    claimed = repo.claim_pending()
    assert [row["fid"] for row in claimed.drafts] == [1]
    assert claimed.drafts[0]["process_status"] == 3
    assert claimed.drafts[0]["update_user_name"] == USER_NAME
    assert repo.table4[1]["process_status"] == 3
    assert repo.table4[2]["process_status"] == 1


# 表 4 process_status=3 且 update_time 超过租约的行可以再领。
def test_claim_reclaims_expired_claiming_stamp():
    now = datetime(2026, 10, 22, 12, 0, 0)
    repo = MemoryRebateRepo(
        table4=[_t4(process_status=3, update_time=now - timedelta(minutes=31))]
    )
    claimed = repo.claim_pending(now=now)
    assert [row["fid"] for row in claimed.drafts] == [1]
    assert repo.table4[1]["process_status"] == 3


# 表 4 process_status=3 且未过租约的行不领。
def test_claim_skips_fresh_claiming_stamp():
    now = datetime(2026, 10, 22, 12, 0, 0)
    repo = MemoryRebateRepo(
        table4=[_t4(process_status=3, update_time=now - timedelta(minutes=1))]
    )
    claimed = repo.claim_pending(now=now)
    assert claimed.drafts == []
    assert repo.table4[1]["process_status"] == 3


# limit 不切开同一合同号；组内有未过租约的 3 则整组不领。
def test_claim_keeps_whole_contract_and_skips_fresh_sibling():
    now = datetime(2026, 10, 22, 12, 0, 0)
    repo = MemoryRebateRepo(
        table4=[
            _t4(fid=1, contract_no="HT1"),
            _t4(fid=2, oms_id=2026100502, contract_no="HT1"),
            _t4(fid=3, oms_id=2026100503, contract_no="HT2"),
            _t4(fid=4, oms_id=2026100504, contract_no="HT3"),
            _t4(
                fid=5,
                oms_id=2026100505,
                contract_no="HT3",
                process_status=3,
                update_time=now - timedelta(minutes=1),
            ),
        ]
    )
    claimed = repo.claim_pending(limit=1, now=now)
    assert [row["fid"] for row in claimed.drafts] == [1, 2]
    assert repo.table4[3]["process_status"] == 0
    assert repo.table4[4]["process_status"] == 0
    assert repo.table4[5]["process_status"] == 3


# 表 3 只录协议行停在 3 且 process_time 超过租约可以再领。
def test_claim_reclaims_expired_protocol_status_3():
    now = datetime(2026, 10, 22, 12, 0, 0)
    repo = MemoryRebateRepo(
        table3=[_t3(process_status=3, process_time=now - timedelta(minutes=31))]
    )
    claimed = repo.claim_pending(now=now)
    assert [row["fid"] for row in claimed.protocols] == [10]
    assert repo.table3[10]["process_status"] == 3


# 表 3 只录协议行停在 3 且未过租约的不领。
def test_claim_skips_fresh_protocol_status_3():
    now = datetime(2026, 10, 22, 12, 0, 0)
    repo = MemoryRebateRepo(
        table3=[_t3(process_status=3, process_time=now - timedelta(minutes=1))]
    )
    claimed = repo.claim_pending(now=now)
    assert claimed.protocols == []
    assert repo.table3[10]["process_status"] == 3


# recorder 按总单 → 细单 → 送审顺序把合同号传下去。
def test_recorder_passes_contract_no_to_header_import_submit():
    steps: list[str] = []

    # 记下总单调用。
    def header(contract_no: str) -> str:
        steps.append(f"header:{contract_no}")
        return "RB1"

    # 记下细单调用。
    def importer(contract_no: str) -> RebateImportResult:
        steps.append(f"import:{contract_no}")
        return RebateImportResult(1, 0, (), True)

    # 记下送审调用。
    def submitter(contract_no: str) -> RebateSubmitResult:
        steps.append(f"submit:{contract_no}")
        return RebateSubmitResult(True, "AGR1", "AGRD1")

    outcome = recorder("HT1", header, importer, submitter)
    assert steps == ["header:HT1", "import:HT1", "submit:HT1"]
    assert outcome.kind == KIND_OK
    assert outcome.agreement_id == "AGR1"
    assert outcome.agreement_dtl_id == "AGRD1"


# 总单抛错只写 HEADER_FAILED，不含异常原文。
def test_recorder_maps_header_error_to_short_code():
    # 模拟总单页面未捕获异常。
    def header(contract_no: str) -> str:
        raise RuntimeError("ORA-00001 unique constraint (RPA.UK)")

    outcome = recorder("HT1", header, _ok_importer, _ok_submitter)
    assert outcome.kind == KIND_FAIL
    assert outcome.reason == HEADER_FAILED
    assert "ORA-" not in (outcome.reason or "")
    assert outcome.error_text == "ORA-00001 unique constraint (RPA.UK)"


# 细单抛错只写 IMPORT_FAILED，不含异常原文。
def test_recorder_maps_import_error_to_short_code():
    # 模拟细单页面未捕获异常。
    def importer(contract_no: str) -> RebateImportResult:
        raise RuntimeError("traceback here")

    outcome = recorder("HT1", _ok_header, importer, _ok_submitter)
    assert outcome.kind == KIND_FAIL
    assert outcome.reason == IMPORT_FAILED
    assert "traceback" not in (outcome.reason or "")


# 送审抛错只写 SUBMIT_FAILED，不含异常原文。
def test_recorder_maps_submit_error_to_short_code():
    # 模拟送审页面未捕获异常。
    def submitter(contract_no: str) -> RebateSubmitResult:
        raise RuntimeError("ORA-12541 boom")

    outcome = recorder("HT1", _ok_header, _ok_importer, submitter)
    assert outcome.kind == KIND_FAIL
    assert outcome.reason == SUBMIT_FAILED
    assert "ORA-" not in (outcome.reason or "")


# runner 抛未捕获异常时表 3 原因列只写 PAGE_FAILED。
def test_job_maps_unexpected_error_to_short_code():
    repo = MemoryRebateRepo(table3=[_t3()])

    # 模拟查号后页面未捕获异常。
    def runner(group):
        raise RuntimeError("traceback here")

    run_batch(repo, runner=runner)
    assert repo.table3[10]["process_status"] == 2
    assert repo.table3[10]["process_result"] == PAGE_FAILED
    assert "traceback" not in repo.table3[10]["process_result"]
    assert len(repo.shift_errors) == 1
    assert repo.shift_errors[0].shift_name == "返利"
    assert repo.shift_errors[0].biz_key == "HT-OLD"
    assert repo.shift_errors[0].error_code == PAGE_FAILED
    assert repo.shift_errors[0].error_text == "traceback here"


def test_default_limit_caps_claim_at_10():
    """不传 --limit 时返利班最多领 10 行。"""
    from run_batch import DEFAULT_CLAIM_LIMIT, main as batch_main

    repo = MemoryRebateRepo(
        table4=[
            _t4(fid=i, oms_id=2026100500 + i, contract_no=f"HT{i}")
            for i in range(1, 16)
        ]
    )
    seen: dict[str, int | None] = {}
    original = repo.claim_pending

    def claim_pending(limit=None, **kwargs):
        """记下入口传给领取的上限和实际领走的行数。"""
        seen["limit"] = limit
        claimed = original(limit=limit, **kwargs)
        seen["count"] = len(claimed.drafts) + len(claimed.protocols)
        return claimed

    repo.claim_pending = claim_pending

    def runner(_group):
        """不打开页面，当作没搜到。"""
        return AgreementOutcome(KIND_NOT_FOUND, None, None, None)

    batch_main([], repo=repo, runner=runner)
    assert seen["limit"] == DEFAULT_CLAIM_LIMIT
    assert seen["count"] == 10
    assert seen["count"] <= DEFAULT_CLAIM_LIMIT


class HoldingRebateRepo(MemoryRebateRepo):
    """内存仓库加上可计数的 session，用来断言页面期间连接已归还。"""

    def __init__(self, table4=None, table3=None):
        """在内存行上加 held 计数。"""
        super().__init__(table4=table4, table3=table3)
        self.held = 0

    @contextmanager
    def session(self):
        """借连接时 held 加一，退出时减回。"""
        self.held += 1
        try:
            yield
        finally:
            self.held -= 1


def test_page_runs_after_connection_is_returned():
    """两组查号时连接已归还，落表时才再借。"""
    repo = HoldingRebateRepo(
        table4=[
            _t4(fid=1, contract_no="HT1"),
            _t4(fid=2, oms_id=2026100502, contract_no="HT2"),
        ]
    )
    during_page: list[int] = []
    during_save: list[int] = []
    original_claim = repo.claim_pending
    original_save = repo.save_group
    original_errors = repo.save_shift_errors

    def claim_pending(limit=None, **kwargs):
        """领取时必须占着本班的连接。"""
        assert repo.held == 1
        return original_claim(limit=limit, **kwargs)

    def save_group(payload):
        """落表时必须占着本班的连接。"""
        during_save.append(repo.held)
        return original_save(payload)

    def save_shift_errors(items):
        """审计走连接池，不能在 session 里写。"""
        assert repo.held == 0
        return original_errors(items)

    def runner(_group):
        """查号和录协议时连接必须已经归还。"""
        during_page.append(repo.held)
        return AgreementOutcome(KIND_OK, "AGR1", "AGRD1", None)

    repo.claim_pending = claim_pending
    repo.save_group = save_group
    repo.save_shift_errors = save_shift_errors
    summary = run_batch(repo, runner=runner)
    assert during_page == [0, 0]
    assert during_save == [1, 1]
    assert repo.held == 0
    assert summary.groups == 2
    assert summary.ok == 2


def test_existing_agreement_is_not_created_again():
    """对方已有协议、本地没写上时，重跑不产生第二份。"""
    repo = MemoryRebateRepo(
        table4=[_t4(), _t4(fid=2, oms_id=2026100502)]
    )
    created: list[str] = []

    def runner(group):
        """已有总单号则补写，不再走总单、细单和送审。"""

        def header(_number: str) -> str:
            """本用例不应新建总单。"""
            created.append("header")
            return "NEW"

        def importer(_number: str) -> RebateImportResult:
            """本用例不应新建细单。"""
            created.append("import")
            return RebateImportResult(1, 0, (), True)

        def submitter(_number: str) -> RebateSubmitResult:
            """本用例不应再送审。"""
            created.append("submit")
            return RebateSubmitResult(True, "AGR-NEW", "DTL-NEW")

        def existing(number: str) -> str:
            """对方这一合同已经有协议总单。"""
            assert number == "HT1"
            return "AGR-ALREADY"

        return recorder(
            group.contract_no,
            header,
            importer,
            submitter,
            existing=existing,
        )

    summary = run_batch(repo, runner=runner)
    assert created == []
    assert summary.ok == 1
    agreements = [row for row in repo.table4.values() if row["action_code"] == 2]
    assert sorted(row["oms_id"] for row in agreements) == [2026100501, 2026100502]
    assert {row["agreement_id"] for row in agreements} == {"AGR-ALREADY"}
    assert {row["agreement_status"] for row in agreements} == {1}


def test_unreadable_agreement_stays_in_process():
    """读号失败保持处理中。确认没有单据才放回待领。"""
    repo = MemoryRebateRepo(table4=[_t4()])

    def runner(group):
        """查已有协议时读不到号。"""

        def existing(_number: str) -> str:
            """页面没有把总单号读出来。"""
            raise RuntimeError("页面没有总单号")

        return recorder(
            group.contract_no,
            _ok_header,
            _ok_importer,
            _ok_submitter,
            existing=existing,
        )

    run_batch(repo, runner=runner)
    row = repo.table4[1]
    assert row["process_status"] == 3
    assert row["contract_status"] == 1
    assert row["action_code"] == 1
    assert row["agreement_fail_reason"] == "读号失败"
    assert [item for item in repo.table4.values() if item["action_code"] == 2] == []
    assert repo.claim_pending().drafts == []


def test_confirmed_absent_agreement_still_creates_once():
    """确认没有协议时才新建一份，空串不算已有总单。"""
    steps: list[str] = []

    def header(contract_no: str) -> str:
        """确认没有协议后才填总单。"""
        steps.append(f"header:{contract_no}")
        return "RB1"

    def importer(contract_no: str) -> RebateImportResult:
        """确认没有协议后才新增细单。"""
        steps.append(f"import:{contract_no}")
        return RebateImportResult(1, 0, (), True)

    def submitter(contract_no: str) -> RebateSubmitResult:
        """确认没有协议后才送审。"""
        steps.append(f"submit:{contract_no}")
        return RebateSubmitResult(True, "AGR1", "AGRD1")

    def existing(_number: str) -> str:
        """对方确认没有这份协议。"""
        return ""

    outcome = recorder("HT1", header, importer, submitter, existing=existing)
    assert steps == ["header:HT1", "import:HT1", "submit:HT1"]
    assert outcome.kind == KIND_OK
    assert outcome.agreement_id == "AGR1"