"""报表统计集成测试（M4 ⑤）。

守的口径（破坏了就是报表数字与台账对不上）：
- 个人明细与台账逐笔一致；summary 为**全筛选**合计而非当前页合计
- 部门汇总团队基数 = 成员绩效合计；12.5%/5% 分成按参数；无成员经理出 0 行
- 回款统计 8 个区间桶恒在（空桶为 0）；**二次分摊拆行时回款金额只计一次**
- 业务统计排除经理聚合行（calc_type=2），绩效不被整团队聚合额污染
- 导出 xlsx 可解析且含签字栏；导出行为入操作日志
- 权限：普通用户 403 且文案为「无权访问报表统计」
"""

import uuid
from decimal import Decimal

import pytest
from fastapi.testclient import TestClient

pytestmark = pytest.mark.integration

_REPORTS = "/api/v1/reports"
_PERSONNEL = "/api/v1/base/personnel"
_BUSINESSES = "/api/v1/business/businesses"
_SHIPMENTS = "/api/v1/business/shipments"
_RECEIPTS = "/api/v1/business/receipts"
_CALC = "/api/v1/calc"
_SETTLEMENTS = "/api/v1/settlements"


def _role_id(code: str) -> int:
    """取内置业务岗位 id（salesperson / sales_manager）。"""
    from sqlalchemy import select

    from app.core.db import get_session_factory
    from app.models import BusinessRole

    factory = get_session_factory()
    with factory() as session:
        return int(
            session.execute(
                select(BusinessRole.id).where(
                    BusinessRole.role_code == code, BusinessRole.status == 1
                )
            ).scalar_one()
        )


def _person(client: TestClient, headers: dict, *, role: str = "salesperson", manager_id=None) -> dict:
    body = {
        "personnel_name": f"t_报表_{uuid.uuid4().hex[:8]}",
        "business_role_id": _role_id(role),
        "salary_standard": "3000.00",
    }
    if manager_id is not None:
        body["manager_id"] = manager_id
    resp = client.post(_PERSONNEL, json=body, headers=headers)
    assert resp.status_code == 200, resp.text
    return resp.json()["data"]


def _full_chain(
    client: TestClient,
    headers: dict,
    *,
    owner_id: int,
    customer_id: int | None = None,
    allocations: list[dict] | None = None,
) -> tuple[dict, dict]:
    """业务（单行 100×100，成本 3000）→ 发货（60 天前）→ 回款 → 核算。"""
    from tests.integration.factories import business_item, business_payload, customer

    cust = customer(client, headers) if customer_id is None else {"id": customer_id}
    payload = business_payload(
        cust["id"],
        owner_personnel_id=owner_id,
        items=[business_item(cost_amount="3000.00")],
    )
    if allocations:
        payload["secondary_allocations"] = allocations
    resp = client.post(_BUSINESSES, json=payload, headers=headers)
    assert resp.status_code == 200, resp.text
    business = resp.json()["data"]

    from datetime import date, timedelta

    ship_date = (date.today() - timedelta(days=60)).isoformat()
    resp = client.post(
        _SHIPMENTS,
        json={
            "business_id": business["id"],
            "ship_date": ship_date,
            "items": [{"product_name": "t_发货物", "quantity": "1", "sale_price": "1.00"}],
        },
        headers=headers,
    )
    assert resp.status_code == 200, resp.text
    shipment = resp.json()["data"]

    resp = client.post(
        _RECEIPTS,
        json={
            "business_id": business["id"],
            "receipt_date": date.today().isoformat(),
            "amount": "10000.00",
            "shipment_id": shipment["id"],
        },
        headers=headers,
    )
    assert resp.status_code == 200, resp.text
    receipt = resp.json()["data"]

    resp = client.post(f"{_CALC}/receipts/{receipt['id']}/calc", headers=headers)
    assert resp.status_code == 200, resp.text
    return business, receipt


def _month() -> str:
    from datetime import date

    return date.today().strftime("%Y-%m")


def _ledger_sum(headers: dict, month: str) -> Decimal:
    """台账（单笔生效行）同月绩效合计——报表 summary 的对照基准。"""
    from sqlalchemy import func, select

    from app.core.db import get_session_factory
    from app.models import PerformanceLedger

    factory = get_session_factory()
    with factory() as session:
        return Decimal(
            str(
                session.execute(
                    select(func.coalesce(func.sum(PerformanceLedger.performance_amount), 0)).where(
                        PerformanceLedger.settlement_month == month,
                        PerformanceLedger.calc_type == 1,
                        PerformanceLedger.is_active == 1,
                    )
                ).scalar_one()
            )
        )


def test_个人明细_与台账一致且summary为全量合计(client: TestClient, admin: dict):
    """明细行金额必须等于台账；pageSize=1 时 summary 仍是全部筛选结果之和。"""
    person = _person(client, admin["headers"])
    _full_chain(client, admin["headers"], owner_id=person["id"])
    month = _month()

    resp = client.get(
        f"{_REPORTS}/personal-detail",
        params={"settlement_month": month, "pageSize": 1},
        headers=admin["headers"],
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    assert data["total"] == 1
    expected = _ledger_sum(admin["headers"], month)
    assert Decimal(str(data["list"][0]["performance_amount"])) == expected
    assert Decimal(str(data["summary"]["performance_total"])) == expected


def test_部门汇总_团队基数等于成员绩效且分成按参数(admin, client: TestClient):
    """团队基数=成员绩效合计；12.5% 团队分成与 5% 全公司分成分别可核。"""
    manager = _person(client, admin["headers"], role="sales_manager")
    salesperson = _person(client, admin["headers"], manager_id=manager["id"])
    _full_chain(client, admin["headers"], owner_id=salesperson["id"])
    month = _month()

    resp = client.get(
        f"{_REPORTS}/department-summary",
        params={"settlement_month": month},
        headers=admin["headers"],
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]

    team = next(t for t in data["teams"] if t["manager_id"] == manager["id"])
    ledger_total = _ledger_sum(admin["headers"], month)
    assert Decimal(str(team["team_base"])) == ledger_total
    expected_share = (ledger_total * Decimal("0.125")).quantize(Decimal("0.01"))
    assert Decimal(str(team["team_share_amount"])) == expected_share
    assert team["member_count"] == 1
    assert data["company"]["company_base"] == team["team_base"]


def test_部门汇总_无成员经理出零值行(admin, client: TestClient):
    """无团队成员 → 0 行展示（FR-030 异常处理）。"""
    manager = _person(client, admin["headers"], role="sales_manager")
    month = _month()
    resp = client.get(
        f"{_REPORTS}/department-summary",
        params={"settlement_month": month},
        headers=admin["headers"],
    )
    assert resp.status_code == 200, resp.text
    team = next(
        (t for t in resp.json()["data"]["teams"] if t["manager_id"] == manager["id"]), None
    )
    assert team is not None
    assert Decimal(str(team["team_base"])) == Decimal("0")
    assert team["member_count"] == 0


def test_回款统计_空桶恒在且分摊拆行金额只计一次(admin, client: TestClient):
    """8 个区间桶恒在；二次分配拆行使台账多行但回款金额不得翻倍。"""
    owner = _person(client, admin["headers"])
    participant = _person(client, admin["headers"])
    _full_chain(
        client,
        admin["headers"],
        owner_id=owner["id"],
        allocations=[{"personnel_id": participant["id"], "ratio": "30.00"}],
    )
    month = _month()

    resp = client.get(
        f"{_REPORTS}/payment-stats",
        params={"settlement_month": month},
        headers=admin["headers"],
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]

    assert len(data["buckets"]) == 8
    assert [b["code"] for b in data["buckets"]] == [
        "le0", "1-30", "31-60", "61-90", "91-180", "181-360", "361+", "none",
    ]
    # 60 天前发货 → 落在 61-90 桶（发货日首日计第 1 天）
    bucket = next(b for b in data["buckets"] if b["code"] == "61-90")
    assert Decimal(str(bucket["receipt_amount"])) == Decimal("10000.00")
    assert bucket["receipt_count"] == 1
    # 拆行后台账两行，但绩效合计仍等于台账总和
    assert bucket["ledger_count"] == 2
    empty = next(b for b in data["buckets"] if b["code"] == "361+")
    assert Decimal(str(empty["performance_amount"])) == Decimal("0")


def test_业务统计_绩效排除经理聚合行(admin, client: TestClient):
    """生成月结后出现 calc_type=2 聚合行，业务统计不得把聚合额算进单业务。"""
    manager = _person(client, admin["headers"], role="sales_manager")
    salesperson = _person(client, admin["headers"], manager_id=manager["id"])
    business, _ = _full_chain(client, admin["headers"], owner_id=salesperson["id"])

    resp = client.post(
        f"{_SETTLEMENTS}/generate", json={"month": _month()}, headers=admin["headers"]
    )
    assert resp.status_code == 200, resp.text

    resp = client.get(
        f"{_REPORTS}/business-stats",
        params={"keyword": business["business_no"]},
        headers=admin["headers"],
    )
    assert resp.status_code == 200, resp.text
    row = resp.json()["data"]["list"][0]
    assert Decimal(str(row["performance_amount"])) == _ledger_sum(admin["headers"], _month())


def test_导出_excel含签字栏且入操作日志(admin, client: TestClient):
    """导出是文件流（非统一信封）；模板含复核/签字栏；行为入操作日志。"""
    from io import BytesIO

    from openpyxl import load_workbook

    person = _person(client, admin["headers"])
    _full_chain(client, admin["headers"], owner_id=person["id"])

    resp = client.get(
        f"{_REPORTS}/personal-detail/export",
        params={"settlement_month": _month()},
        headers=admin["headers"],
    )
    assert resp.status_code == 200, resp.text
    assert resp.headers["content-type"].startswith(
        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
    )
    workbook = load_workbook(BytesIO(resp.content))
    text = "\n".join(
        str(cell.value)
        for row in workbook.active.iter_rows()
        for cell in row
        if cell.value is not None
    )
    assert "个人绩效工资明细表" in text
    assert "财务复核" in text
    assert "领导签字" in text

    from sqlalchemy import select

    from app.core.db import get_session_factory
    from app.models import OperationLog

    factory = get_session_factory()
    with factory() as session:
        log = session.execute(
            select(OperationLog)
            .where(OperationLog.module == "报表统计", OperationLog.action == "导出")
            .order_by(OperationLog.id.desc())
        ).scalars().first()
    assert log is not None


def test_权限_普通用户403文案(admin, normal, client: TestClient):
    """报表限管理员/财务/领导；普通用户被后端拦截且文案正确。"""
    resp = client.get(f"{_REPORTS}/personal-detail", headers=normal["headers"])
    assert resp.status_code == 403
    assert resp.json()["message"] == "无权访问报表统计"


# ==== 审查回归：归属、调整单、经理个人户、导出审计 ====


def _set_param(client: TestClient, headers: dict, code: str, value: str) -> None:
    """立即生效的参数变更（用例不依赖种子默认值，结束显式恢复）。"""
    resp = client.post(
        "/api/v1/system/params/changes",
        json={
            "change_type": 1,
            "remark": f"用例设置 {code}={value}",
            "items": [{"param_code": code, "dimension_value": "", "value": value}],
        },
        headers=headers,
    )
    assert resp.status_code == 200, resp.text


def _insert_adjustment(person: dict, month: str, amount: str, direction: int = 1) -> None:
    """直插生效调整单（status=2）：绕开审批链，聚焦报表归属口径。"""
    from app.core.db import get_session_factory
    from app.models import AdjustmentOrder

    factory = get_session_factory()
    with factory() as session:
        session.add(
            AdjustmentOrder(
                order_no=f"TZ{uuid.uuid4().hex[:12]}",
                settlement_month=month,
                personnel_id=person["id"],
                personnel_name=person["personnel_name"],
                direction=direction,
                amount=Decimal(amount),
                reason="t_报表用例",
                status=2,
                submitter_name="t_admin",
            )
        )
        session.commit()


def _insert_manager_personal_ledger(manager: dict, month: str, amount: str) -> None:
    """直插一条经理个人户台账行（模拟单笔侧已计提的结果）。"""
    from datetime import date

    from app.core.db import get_session_factory
    from app.models import PerformanceLedger

    factory = get_session_factory()
    with factory() as session:
        session.add(
            PerformanceLedger(
                receipt_id=None,
                calc_version=1,
                calc_type=1,
                business_id=0,
                business_no="t_个人户",
                personnel_id=manager["id"],
                personnel_name=manager["personnel_name"],
                role_code="sales_manager",
                customer_type="manager_personal",
                product_category="mixed",
                receipt_date=date.today(),
                rate_text="0",
                performance_amount=Decimal(amount),
                barter_applied=0,
                payable_amount=Decimal(amount),
                settlement_month=month,
                is_probation=0,
            )
        )
        session.commit()


def _latest_export_log():
    from sqlalchemy import select

    from app.core.db import get_session_factory
    from app.models import OperationLog

    factory = get_session_factory()
    with factory() as session:
        return session.execute(
            select(OperationLog)
            .where(OperationLog.module == "报表统计", OperationLog.action == "导出")
            .order_by(OperationLog.id.desc())
        ).scalars().first()


def test_部门汇总_无台账经理挂上级不报500(admin, client: TestClient):
    """经理 A 无台账且上级是经理 B：不能因 member_map 直接索引而 KeyError。"""
    manager_b = _person(client, admin["headers"], role="sales_manager")
    manager_a = _person(
        client, admin["headers"], role="sales_manager", manager_id=manager_b["id"]
    )

    resp = client.get(
        f"{_REPORTS}/department-summary",
        params={"settlement_month": _month()},
        headers=admin["headers"],
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    team_a = next(t for t in data["teams"] if t["manager_id"] == manager_a["id"])
    assert team_a["member_count"] == 0
    team_b = next(t for t in data["teams"] if t["manager_id"] == manager_b["id"])
    assert all(m["personnel_id"] != manager_a["id"] for m in team_b["members"])


def test_部门汇总_仅有调整单人员并入对应团队(admin, client: TestClient):
    """只有调整单、当月无台账行的人也要出现在团队明细中。"""
    manager = _person(client, admin["headers"], role="sales_manager")
    salesperson = _person(client, admin["headers"], manager_id=manager["id"])
    _insert_adjustment(salesperson, _month(), "500.00")

    resp = client.get(
        f"{_REPORTS}/department-summary",
        params={"settlement_month": _month()},
        headers=admin["headers"],
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    team = next(t for t in data["teams"] if t["manager_id"] == manager["id"])
    member = next(m for m in team["members"] if m["personnel_id"] == salesperson["id"])
    assert member["member_kind"] == "adjustment"
    assert Decimal(str(member["performance_sum"])) == Decimal("0")
    assert Decimal(str(member["adjust_net"])) == Decimal("500.00")
    assert Decimal(str(team["adjust_net"])) == Decimal("500.00")
    assert Decimal(str(data["reconciliation"]["adjust_net_total"])) == Decimal("500.00")


def test_部门汇总_仅调整单无经理人员进入未分配(admin, client: TestClient):
    """无经理人员的调整单不能被静默丢弃。"""
    salesperson = _person(client, admin["headers"])
    _insert_adjustment(salesperson, _month(), "300.00")

    resp = client.get(
        f"{_REPORTS}/department-summary",
        params={"settlement_month": _month()},
        headers=admin["headers"],
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    row = next(u for u in data["unassigned"] if u["personnel_id"] == salesperson["id"])
    assert row["member_kind"] == "adjustment"
    assert Decimal(str(row["adjust_net"])) == Decimal("300.00")


def test_部门汇总_经理个人户计入时基数等于成员合计(admin, client: TestClient):
    """开关 1：经理成员行只取个人户金额，团队基数 == 成员绩效合计。"""
    manager = _person(client, admin["headers"], role="sales_manager")
    _insert_manager_personal_ledger(manager, _month(), "100.00")
    _set_param(client, admin["headers"], "manager.personal.included", "1")
    try:
        resp = client.get(
            f"{_REPORTS}/department-summary",
            params={"settlement_month": _month()},
            headers=admin["headers"],
        )
        assert resp.status_code == 200, resp.text
        team = next(
            t for t in resp.json()["data"]["teams"] if t["manager_id"] == manager["id"]
        )
        assert Decimal(str(team["team_base"])) == Decimal("100.00")
        assert sum(
            (Decimal(str(m["performance_sum"])) for m in team["members"]), Decimal("0")
        ) == Decimal("100.00")
        assert all(m["member_kind"] == "manager_personal" for m in team["members"])
    finally:
        _set_param(client, admin["headers"], "manager.personal.included", "0")


def test_部门汇总_经理个人户关闭时不展示个人户金额(admin, client: TestClient):
    """开关 0：个人户金额既不计入团队基数，也不进未分配（与全公司口径一致）。"""
    manager = _person(client, admin["headers"], role="sales_manager")
    _insert_manager_personal_ledger(manager, _month(), "100.00")
    _set_param(client, admin["headers"], "manager.personal.included", "0")

    resp = client.get(
        f"{_REPORTS}/department-summary",
        params={"settlement_month": _month()},
        headers=admin["headers"],
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    team = next(t for t in data["teams"] if t["manager_id"] == manager["id"])
    assert team["member_count"] == 0
    assert all(u["personnel_id"] != manager["id"] for u in data["unassigned"])


def test_部门汇总_按经理筛选不误判其他团队下属(admin, client: TestClient):
    """筛选只影响展示的团队；其他经理的下属不能进入未分配。"""
    manager_1 = _person(client, admin["headers"], role="sales_manager")
    manager_2 = _person(client, admin["headers"], role="sales_manager")
    salesperson_1 = _person(client, admin["headers"], manager_id=manager_1["id"])
    salesperson_2 = _person(client, admin["headers"], manager_id=manager_2["id"])
    _full_chain(client, admin["headers"], owner_id=salesperson_1["id"])
    _full_chain(client, admin["headers"], owner_id=salesperson_2["id"])

    resp = client.get(
        f"{_REPORTS}/department-summary",
        params={"settlement_month": _month(), "manager_id": manager_1["id"]},
        headers=admin["headers"],
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    assert [t["manager_id"] for t in data["teams"]] == [manager_1["id"]]
    assert data["unassigned"] == []
    assert all(
        m["personnel_id"] != salesperson_2["id"]
        for team in data["teams"]
        for m in team["members"]
    )
    assert Decimal(str(data["reconciliation"]["other_team_base"])) > 0


def test_部门汇总_对账恒等式(admin, client: TestClient):
    """未筛选：团队基数 + 未分配绩效 = 全公司基数；应发公式成立。"""
    manager = _person(client, admin["headers"], role="sales_manager")
    salesperson = _person(client, admin["headers"], manager_id=manager["id"])
    unassigned = _person(client, admin["headers"])
    _full_chain(client, admin["headers"], owner_id=salesperson["id"])
    _full_chain(client, admin["headers"], owner_id=unassigned["id"])
    _insert_adjustment(salesperson, _month(), "200.00")

    resp = client.get(
        f"{_REPORTS}/department-summary",
        params={"settlement_month": _month()},
        headers=admin["headers"],
    )
    assert resp.status_code == 200, resp.text
    reconciliation = resp.json()["data"]["reconciliation"]
    assert Decimal(str(reconciliation["team_base_total"])) + Decimal(
        str(reconciliation["unassigned_performance_total"])
    ) == Decimal(str(reconciliation["company_base"]))
    assert Decimal(str(reconciliation["other_team_base"])) == Decimal("0")
    assert Decimal(str(reconciliation["payable_total"])) == (
        Decimal(str(reconciliation["team_member_payable_total"]))
        + Decimal(str(reconciliation["unassigned_payable_total"]))
        + Decimal(str(reconciliation["aggregate_payable_total"]))
        + Decimal(str(reconciliation["adjust_net_total"]))
    )
    # 分列合计 + 舍入差 = 聚合额（月结一次舍入口径）
    assert Decimal(str(reconciliation["aggregate_rounding_diff"])) == (
        Decimal(str(reconciliation["aggregate_payable_total"]))
        - Decimal(str(reconciliation["team_share_total"]))
        - Decimal(str(reconciliation["company_share_total"]))
    )


def _two_manager_scene(
    client: TestClient, headers: dict
) -> tuple[dict, dict, dict, dict, dict]:
    """两位经理各有下属与业务 + 一位零成员经理（无聚合行）。"""
    manager_1 = _person(client, headers, role="sales_manager")
    manager_2 = _person(client, headers, role="sales_manager")
    empty_manager = _person(client, headers, role="sales_manager")
    salesperson_1 = _person(client, headers, manager_id=manager_1["id"])
    salesperson_2 = _person(client, headers, manager_id=manager_2["id"])
    _full_chain(client, headers, owner_id=salesperson_1["id"])
    _full_chain(client, headers, owner_id=salesperson_2["id"])
    return manager_1, manager_2, empty_manager, salesperson_1, salesperson_2


def test_部门汇总_应发合计与月结对拍_两位经理(admin, client: TestClient):
    """月结每位有聚合行的经理各计一次 5%；签字稿应发必须等于 total_payable。"""
    _two_manager_scene(client, admin["headers"])

    resp = client.get(
        f"{_REPORTS}/department-summary",
        params={"settlement_month": _month()},
        headers=admin["headers"],
    )
    assert resp.status_code == 200, resp.text
    before = resp.json()["data"]
    assert before["company"]["company_share_count"] == 2
    assert Decimal(str(before["company"]["company_share_total"])) == (
        Decimal(str(before["company"]["company_share_amount"])) * 2
    )

    resp = client.post(
        f"{_SETTLEMENTS}/generate", json={"month": _month()}, headers=admin["headers"]
    )
    assert resp.status_code == 200, resp.text
    settlement = resp.json()["data"]

    resp = client.get(
        f"{_REPORTS}/department-summary",
        params={"settlement_month": _month()},
        headers=admin["headers"],
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    teams = {team["manager_id"]: team for team in data["teams"]}
    frozen_total = sum(
        (
            Decimal(str(team["frozen_amount"]))
            for team in teams.values()
            if team["frozen_amount"] is not None
        ),
        Decimal("0"),
    )
    reconciliation = data["reconciliation"]
    assert Decimal(str(reconciliation["aggregate_payable_total"])) == frozen_total
    assert Decimal(str(reconciliation["payable_total"])) == Decimal(
        str(settlement["total_payable"])
    )


def test_部门汇总_有调整单时应发等于月结加调整(admin, client: TestClient):
    """调整单不进月结 items；报表应发 = 月结总额 + 调整净额。"""
    _, _, _, salesperson, _ = _two_manager_scene(client, admin["headers"])
    _insert_adjustment(salesperson, _month(), "500.00")

    resp = client.post(
        f"{_SETTLEMENTS}/generate", json={"month": _month()}, headers=admin["headers"]
    )
    assert resp.status_code == 200, resp.text
    settlement = resp.json()["data"]

    resp = client.get(
        f"{_REPORTS}/department-summary",
        params={"settlement_month": _month()},
        headers=admin["headers"],
    )
    assert resp.status_code == 200, resp.text
    reconciliation = resp.json()["data"]["reconciliation"]
    assert Decimal(str(reconciliation["adjust_net_total"])) == Decimal("500.00")
    assert Decimal(str(reconciliation["payable_total"])) == (
        Decimal(str(settlement["total_payable"])) + Decimal("500.00")
    )


def test_部门汇总_按经理筛选时5百分比只计该团队(admin, client: TestClient):
    """筛选态只汇总展示团队；隐藏团队的金额进 other_*，不参与应发。"""
    manager_1, _, _, _, _ = _two_manager_scene(client, admin["headers"])

    resp = client.get(
        f"{_REPORTS}/department-summary",
        params={"settlement_month": _month(), "manager_id": manager_1["id"]},
        headers=admin["headers"],
    )
    assert resp.status_code == 200, resp.text
    data = resp.json()["data"]
    assert data["company"]["company_share_count"] == 1
    reconciliation = data["reconciliation"]
    assert Decimal(str(reconciliation["other_team_aggregate_total"])) > 0
    assert Decimal(str(reconciliation["other_team_company_share_total"])) > 0


def test_导出_筛选页眉ID转名称与中文标签(admin, client: TestClient):
    """页眉是签字稿追溯筛选范围的唯一凭据，ID 必须转名称。"""
    from io import BytesIO

    from openpyxl import load_workbook

    from app.core.db import get_session_factory
    from app.models import BusinessRole

    person = _person(client, admin["headers"])
    _full_chain(client, admin["headers"], owner_id=person["id"])
    factory = get_session_factory()
    with factory() as session:
        role_name = (
            session.query(BusinessRole.role_name)
            .filter(BusinessRole.role_code == "salesperson")
            .scalar()
        )

    resp = client.get(
        f"{_REPORTS}/personal-detail/export",
        params={
            "settlement_month": _month(),
            "personnel_id": person["id"],
            "role_code": "salesperson",
            "customer_type": "personal",
            "product_category": "municipal_water",
        },
        headers=admin["headers"],
    )
    assert resp.status_code == 200, resp.text
    workbook = load_workbook(BytesIO(resp.content))
    text = "\n".join(
        str(cell.value)
        for row in workbook.active.iter_rows()
        for cell in row
        if cell.value is not None
    )
    assert person["personnel_name"] in text
    assert f"人员 {person['personnel_name']}(#{person['id']})" in text
    assert "个人户" in text
    assert "市政水利" in text
    assert role_name in text


def test_导出_回款维度sheet透传筛选条件(admin, client: TestClient):
    """两个维度 sheet 的页眉不能固定写「全部」。"""
    from io import BytesIO

    from openpyxl import load_workbook

    resp = client.get(
        f"{_REPORTS}/payment-stats/export",
        params={"customer_type": "personal"},
        headers=admin["headers"],
    )
    assert resp.status_code == 200, resp.text
    workbook = load_workbook(BytesIO(resp.content))
    for sheet_name in ("客户类型对比", "产品类别对比"):
        text = "\n".join(
            str(cell.value)
            for row in workbook[sheet_name].iter_rows()
            for cell in row
            if cell.value is not None
        )
        assert "客户类型 个人户" in text
        assert "筛选条件：全部" not in text


def test_导出超限_失败审计留痕(admin, client: TestClient, monkeypatch):
    """超限 400 发生在审计上下文内，必须留下失败记录。"""
    from app.services import report_service

    person = _person(client, admin["headers"])
    _full_chain(client, admin["headers"], owner_id=person["id"])
    monkeypatch.setattr(report_service, "REPORT_EXPORT_MAX_ROWS", 0)

    resp = client.get(
        f"{_REPORTS}/personal-detail/export",
        params={"settlement_month": _month()},
        headers=admin["headers"],
    )
    assert resp.status_code == 400, resp.text
    log = _latest_export_log()
    assert log is not None
    assert log.success == 0
    assert "超过上限" in (log.error_message or "")


def test_导出取数异常_失败审计留痕(admin, client: TestClient, monkeypatch):
    """取数阶段的异常同样要留下失败记录。

    TestClient 默认重抛服务端异常，这里直接调 service 层，聚焦审计落库。
    """
    from app.core.db import get_session_factory
    from app.models import User
    from app.schemas.report import PersonalDetailQuery
    from app.services import report_service

    def _boom(*args, **kwargs):
        raise RuntimeError("t_取数异常")

    monkeypatch.setattr(report_service, "_export_payload", _boom)
    factory = get_session_factory()
    with factory() as session:
        operator = session.get(User, admin["id"])
        with pytest.raises(RuntimeError, match="t_取数异常"):
            report_service.export_report(
                session,
                "personal-detail",
                PersonalDetailQuery(settlement_month=_month()),
                operator=operator,
            )
    log = _latest_export_log()
    assert log is not None
    assert log.success == 0
    # 未预期异常按安全口径收敛为通用文案，摘要仍能定位到报表
    assert "导出个人绩效工资明细表" in (log.summary or "")


@pytest.mark.parametrize(
    "path", ["/personal-detail", "/department-summary", "/payment-stats"]
)
def test_非法月份_查询返回400(admin, client: TestClient, path: str):
    """非法月份必须是参数错误 400，不能落 500。"""
    resp = client.get(
        f"{_REPORTS}{path}",
        params={"settlement_month": "2026-13"},
        headers=admin["headers"],
    )
    assert resp.status_code == 400, resp.text


@pytest.mark.parametrize(
    "code", ["personal-detail", "department-summary", "payment-stats"]
)
def test_非法月份_导出返回400(admin, client: TestClient, code: str):
    """导出手工构造查询模型同样要转 400。"""
    resp = client.get(
        f"{_REPORTS}/{code}/export",
        params={"settlement_month": "2026-13"},
        headers=admin["headers"],
    )
    assert resp.status_code == 400, resp.text