# 选品班次：领取戳、按条形码搜、系统故障短码。

from __future__ import annotations

from datetime import datetime, timedelta

from repo import MappingRow, MemorySelectRepo
from rules import Candidate, normalize_vendor
from job import run_batch


HANGZHOU = "杭州小药药自营旗舰店"
GUANGDONG = "广东壹号药业有限公司"


# 构造一条待选品表 1 行。
def _row(**overrides) -> MappingRow:
    data = {
        "fid": 1,
        "oms_id": 2026100201,
        "bar_code": "06901234567890",
        "sale_price": 100.0,
        "plan_purchase_qty": 10,
        "approval_no": "国药准字H20100001",
        "goods_name": "阿莫西林",
        "goods_spec": "0.25g",
        "manufacturer": "示例制药",
        "org_code": "907",
        "deal_flag": 0,
    }
    data.update(overrides)
    return MappingRow(**data)


# 构造一条能通过规则的候选。
def _ok_candidate(mall: str = "1号药城", **overrides) -> Candidate:
    vendor = overrides.pop("vendor", GUANGDONG if mall == "1号药城" else HANGZHOU)
    data = {
        "mall_name": mall,
        "sku_id": "S1",
        "name": "阿莫西林",
        "spec": "0.25g",
        "price": 90.0,
        "expiry_raw": "13",
        "expiry_months": 13.0,
        "vendor_raw": vendor,
        "vendor_norm": normalize_vendor(vendor),
        "stock": 20.0,
    }
    data.update(overrides)
    if "vendor_raw" in overrides and "vendor_norm" not in overrides:
        data["vendor_norm"] = normalize_vendor(overrides["vendor_raw"])
    return Candidate(**data)


# deal_flag=0 领走后变成 3，已完成的 1 不领。
def test_claim_pending_sets_processing_flag():
    repo = MemorySelectRepo([_row(), _row(fid=2, oms_id=2026100202, deal_flag=1)])
    claimed = repo.claim_pending()
    assert [row.oms_id for row in claimed] == [2026100201]
    assert repo.mapping[2026100201].deal_flag == 3
    assert repo.mapping[2026100202].deal_flag == 1


# deal_flag=3 且 deal_time 超过租约的行可以再领。
def test_claim_reclaims_expired_flag_3():
    now = datetime(2026, 10, 22, 12, 0, 0)
    expired = now - timedelta(minutes=31)
    repo = MemorySelectRepo(
        [_row(deal_flag=3, deal_time=expired)]
    )
    claimed = repo.claim_pending(now=now)
    assert [row.oms_id for row in claimed] == [2026100201]
    assert repo.mapping[2026100201].deal_flag == 3


# deal_flag=3 且未过租约的行不领。
def test_claim_skips_fresh_flag_3():
    now = datetime(2026, 10, 22, 12, 0, 0)
    fresh = now - timedelta(minutes=1)
    repo = MemorySelectRepo([_row(deal_flag=3, deal_time=fresh)])
    claimed = repo.claim_pending(now=now)
    assert claimed == []
    assert repo.mapping[2026100201].deal_flag == 3


# 班次按条形码搜，不拿批准文号去搜。
def test_job_searches_barcode_not_approval_no():
    repo = MemorySelectRepo([_row()])
    seen: list[str] = []

    # 记下查询键并返回一条合格候选。
    def search_yaoex(query_key: str):
        seen.append(query_key)
        return [_ok_candidate()]

    summary = run_batch(
        repo,
        platforms=["1号药城"],
        searchers={"1号药城": search_yaoex},
        login_errors={},
    )
    assert seen == ["06901234567890"]
    assert summary.added == 1
    assert repo.mapping[2026100201].deal_flag == 1
    assert len(repo.cart) == 1
    assert repo.cart[0]["bar_code"] == "06901234567890"
    assert repo.cart[0]["status"] == 1
    assert repo.cart[0]["purchase_qty"] == 10


def test_job_saves_each_claimed_row_before_the_next():
    """一行搜完就落这一行，不攒到班次结束。"""

    class _Repo(MemorySelectRepo):
        """记下每次 save_results 的行。"""

        def __init__(self, rows):
            super().__init__(rows)
            self.batches: list[list] = []

        def save_results(self, items):
            """记下这一次落库，再交给内存仓库。"""
            self.batches.append(list(items))
            return super().save_results(items)

    repo = _Repo(
        [
            _row(fid=1, oms_id=1, bar_code="111"),
            _row(fid=2, oms_id=2, bar_code="222"),
        ]
    )

    # 两行都给一条合格候选。
    def search(_query_key: str):
        return [_ok_candidate()]

    summary = run_batch(
        repo,
        platforms=["1号药城"],
        searchers={"1号药城": search},
        login_errors={},
    )
    assert summary.claimed == 2
    assert summary.added == 2
    assert len(repo.batches) == 2
    assert [item.fid for batch in repo.batches for item in batch] == [1, 2]
    assert repo.mapping[1].deal_flag == 1
    assert repo.mapping[2].deal_flag == 1


def test_expired_lease_second_process_does_not_insert_second_cart_set():
    """租约过期后第二个进程不再插入第二套表 2。"""
    from repo import FinishWrite

    now = datetime(2026, 10, 22, 12, 0, 0)
    repo = MemorySelectRepo([_row(deal_flag=3, deal_time=now - timedelta(minutes=31))])
    claimed = repo.claim_pending(now=now)
    assert [row.fid for row in claimed] == [1]
    repo.cart.append(
        {
            "oms_id": 2026100201,
            "platform_type": 66,
            "purchase_qty": 10,
            "status": 1,
        }
    )
    repo.save_results(
        [
            FinishWrite(
                oms_id=2026100201,
                fid=1,
                deal_flag=1,
                deal_result="选品成功：1号药城",
                cart_rows=[
                    {
                        "oms_id": 2026100201,
                        "platform_type": 66,
                        "purchase_qty": 10,
                        "status": 1,
                    }
                ],
            )
        ]
    )
    assert len(repo.cart) == 1
    assert repo.cart[0]["purchase_qty"] == 10
    assert repo.mapping[2026100201].deal_flag == 1


# 没有条形码时不搜平台，业务失败。
def test_job_skips_search_when_barcode_missing():
    repo = MemorySelectRepo([_row(bar_code="")])
    called = {"n": 0}

    # 本用例不应被调用。
    def search_yaoex(_query_key: str):
        called["n"] += 1
        return [_ok_candidate()]

    summary = run_batch(
        repo,
        platforms=["1号药城"],
        searchers={"1号药城": search_yaoex},
        login_errors={},
    )
    assert called["n"] == 0
    assert summary.skipped == 1
    assert repo.mapping[2026100201].deal_flag == 2
    assert repo.cart[0]["status"] == 2
    assert "条形码" in (repo.cart[0]["fail_reason"] or "")


# limit 只领 fid 最小的前 N 条。
def test_job_limit_claims_first_n():
    repo = MemorySelectRepo(
        [
            _row(fid=1, oms_id=1, bar_code="111"),
            _row(fid=2, oms_id=2, bar_code="222"),
            _row(fid=3, oms_id=3, bar_code="333"),
        ]
    )
    claimed = repo.claim_pending(limit=2)
    assert [row.oms_id for row in claimed] == [1, 2]
    assert repo.mapping[3].deal_flag == 0


def test_both_malls_login_failed_marks_system_fault():
    """两边登录失败且凑不齐时，deal_flag 改回 0，不写表 2。"""
    repo = MemorySelectRepo([_row()])
    summary = run_batch(
        repo,
        platforms=["1号药城", "药帮忙"],
        searchers={},
        login_errors={"1号药城": "chrome down", "药帮忙": "timeout"},
    )
    assert summary.added == 0
    assert repo.mapping[2026100201].deal_flag == 0
    assert repo.mapping[2026100201].deal_result.startswith("系统故障")
    assert repo.cart == []


# 杭州 3 + 广东 7 落两行成功，不插多余行。
def test_split_across_malls_writes_two_success_rows():
    repo = MemorySelectRepo([_row()])

    # 广东仓库存 20。
    def search_yaoex(_key: str):
        return [_ok_candidate("1号药城", stock=20.0)]

    # 杭州仓库存 3。
    def search_ybm(_key: str):
        return [_ok_candidate("药帮忙", vendor=HANGZHOU, sku_id="HZ", stock=3.0)]

    summary = run_batch(
        repo,
        platforms=["1号药城", "药帮忙"],
        searchers={"1号药城": search_yaoex, "药帮忙": search_ybm},
        login_errors={},
    )
    assert summary.added == 2
    assert repo.mapping[2026100201].deal_flag == 1
    ybm = next(row for row in repo.cart if row["platform_type"] == 48)
    yaoex = next(row for row in repo.cart if row["platform_type"] == 66)
    assert ybm["purchase_qty"] == 3
    assert yaoex["purchase_qty"] == 7
    assert ybm["multi_platform_flag"] == 1


# 一边登录失败、另一边能凑齐计划量时按成功落。
def test_peer_login_failed_but_other_fills_is_success():
    repo = MemorySelectRepo([_row()])

    # 广东仓库存 20，能单独凑齐。
    def search_yaoex(_key: str):
        return [_ok_candidate(stock=20.0)]

    summary = run_batch(
        repo,
        platforms=["1号药城", "药帮忙"],
        searchers={"1号药城": search_yaoex},
        login_errors={"药帮忙": "timeout"},
    )
    assert summary.added == 1
    assert repo.mapping[2026100201].deal_flag == 1
    assert len(repo.cart) == 1
    assert repo.cart[0]["platform_type"] == 66


def test_peer_login_failed_and_short_stock_is_system_fault():
    """一边故障且剩下的凑不齐时改回待领，不写成库存不足，不写表 2。"""
    repo = MemorySelectRepo([_row()])

    def search_yaoex(_key: str):
        """广东仓库存 4，凑不齐计划 10。"""
        return [_ok_candidate(stock=4.0)]

    summary = run_batch(
        repo,
        platforms=["1号药城", "药帮忙"],
        searchers={"1号药城": search_yaoex},
        login_errors={"药帮忙": "timeout"},
    )
    assert summary.added == 0
    assert repo.mapping[2026100201].deal_flag == 0
    assert repo.mapping[2026100201].deal_result.startswith("系统故障")
    assert "库存不足" not in repo.mapping[2026100201].deal_result
    assert repo.cart == []


# 两边都搜到但合计不够是业务失败库存不足。
def test_combined_short_stock_is_business_fail():
    repo = MemorySelectRepo([_row()])

    # 广东仓库存 2。
    def search_yaoex(_key: str):
        return [_ok_candidate(stock=2.0)]

    # 杭州仓库存 2。
    def search_ybm(_key: str):
        return [_ok_candidate("药帮忙", vendor=HANGZHOU, sku_id="HZ", stock=2.0)]

    summary = run_batch(
        repo,
        platforms=["1号药城", "药帮忙"],
        searchers={"1号药城": search_yaoex, "药帮忙": search_ybm},
        login_errors={},
    )
    assert summary.added == 0
    assert summary.skipped == 2
    assert repo.mapping[2026100201].deal_flag == 2
    assert "库存不足" in repo.mapping[2026100201].deal_result
    assert all(row["status"] == 2 for row in repo.cart)


# 查询抛错时原因列只写 QUERY_FAILED，不含异常原文。
def test_job_query_failed_writes_short_code_not_exception():
    repo = MemorySelectRepo([_row()])

    # 抛带 ORA 原文的错，确认原因列不收录。
    def boom(_key: str):
        raise RuntimeError("ORA-03114 secret-traceback")

    summary = run_batch(
        repo,
        platforms=["1号药城"],
        searchers={"1号药城": boom},
        login_errors={},
    )
    assert summary.added == 0
    assert repo.mapping[2026100201].deal_flag == 0
    assert repo.cart == []
    deal_result = repo.mapping[2026100201].deal_result
    assert deal_result.startswith("系统故障")
    assert "QUERY_FAILED" in deal_result
    assert "ORA-03114" not in deal_result
    assert "secret-traceback" not in deal_result
    assert len(repo.query_audits) == 1
    audit = repo.query_audits[0]
    assert audit.oms_id == 2026100201
    assert audit.mapping_fid == 1
    assert audit.mall_name == "1号药城"
    assert audit.bar_code == "06901234567890"
    assert audit.error_text == "ORA-03114 secret-traceback"


# 凑齐后按这次拆定的 sku 加车，同仓两条都不并成低价那条。
def test_success_adds_each_split_sku_without_rereading_table2():
    repo = MemorySelectRepo([_row()])
    seen: list[tuple[str, int, str]] = []

    # 同仓低价 6、高价 10，计划 10，应拆成 6 + 4。
    def search_ybm(_key: str):
        return [
            _ok_candidate(
                "药帮忙",
                vendor=HANGZHOU,
                sku_id="CHEAP",
                price=90.0,
                stock=6.0,
                extra={"raw": {"skuId": "CHEAP"}},
            ),
            _ok_candidate(
                "药帮忙",
                vendor=HANGZHOU,
                sku_id="DEAR",
                price=92.0,
                stock=10.0,
                extra={"raw": {"skuId": "DEAR"}},
            ),
        ]

    # 记下加车入参。行必须已经落成加购中，fid 来自这次插入。
    def finish(lines):
        assert [row["update_user_name"] for row in repo.cart] == ["加购中", "加购中"]
        for line in lines:
            seen.append((line.sku_id, line.qty, line.raw.get("skuId")))
            matched = next(row for row in repo.cart if row["fid"] == line.fid)
            matched["update_user_name"] = "加购班"
        return len(lines), 0

    summary = run_batch(
        repo,
        platforms=["药帮忙"],
        searchers={"药帮忙": search_ybm},
        login_errors={},
        cart_finish=finish,
    )
    assert summary.added == 2
    assert summary.cart_added == 2
    assert summary.cart_failed == 0
    assert seen == [("CHEAP", 6, "CHEAP"), ("DEAR", 4, "DEAR")]
    assert [row["update_user_name"] for row in repo.cart] == ["加购班", "加购班"]
    assert repo.mapping[2026100201].deal_flag == 1


# 凑不齐不加车，失败行戳保持系统管理员。
def test_business_fail_does_not_add_cart():
    repo = MemorySelectRepo([_row()])
    called = {"n": 0}

    def search_ybm(_key: str):
        return [_ok_candidate("药帮忙", vendor=HANGZHOU, sku_id="HZ", stock=2.0)]

    def finish(_lines):
        called["n"] += 1
        return 0, 0

    summary = run_batch(
        repo,
        platforms=["药帮忙"],
        searchers={"药帮忙": search_ybm},
        login_errors={},
        cart_finish=finish,
    )
    assert called["n"] == 0
    assert summary.cart_added == 0
    assert summary.added == 0
    assert repo.mapping[2026100201].deal_flag == 2
    assert repo.cart
    assert all(row["update_user_name"] == "系统管理员" for row in repo.cart)


# 表 2 已有这单时不再加一份。
def test_existing_cart_rows_are_not_added_again():
    repo = MemorySelectRepo(
        [_row(deal_flag=3, deal_time=datetime.now() - timedelta(minutes=31))]
    )
    repo.cart.append(
        {
            "oms_id": 2026100201,
            "platform_type": 66,
            "purchase_qty": 10,
            "status": 1,
            "update_user_name": "系统管理员",
        }
    )
    called = {"n": 0}

    def search_yaoex(_key: str):
        return [_ok_candidate()]

    def finish(_lines):
        called["n"] += 1
        return 1, 0

    summary = run_batch(
        repo,
        platforms=["1号药城"],
        searchers={"1号药城": search_yaoex},
        login_errors={},
        cart_finish=finish,
    )
    assert called["n"] == 0
    assert summary.cart_added == 0
    assert len(repo.cart) == 1
    assert repo.mapping[2026100201].deal_flag == 1