import stat
from decimal import Decimal
from io import BytesIO
from pathlib import Path

import pytest

from finance.exporting import build_formula_workbook
from finance.models import Fact, FormulaExport


def custom_column():
    return {
        "name": "收入成本差",
        "meaning": "实际收入减成本",
        "scope": "按所选分组及本期、基期计算",
        "missing": "结果不可用（#N/A）；零分母不可用",
        "expression": {
            "operator": "subtract",
            "left": {"metric": "revenue"},
            "right": {"metric": "cost"},
        },
    }


pytestmark = pytest.mark.django_db


def test_formula_workbook_fails_closed_when_row_or_group_limit_is_exceeded(workspace, monkeypatch):
    import finance.exporting as exporting

    ds, member, *_ = workspace
    args = (
        ds,
        member,
        {"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
        BytesIO(),
    )
    monkeypatch.setattr(exporting, "MAX_ROWS", 0)
    with pytest.raises(ValueError, match="最多支持"):
        build_formula_workbook(*args)

    monkeypatch.setattr(exporting, "MAX_ROWS", 100_000)
    monkeypatch.setattr(exporting, "MAX_GROUPS", 0)
    with pytest.raises(ValueError, match="分组"):
        build_formula_workbook(*args)


def test_formula_workbook_uses_streaming_writer(workspace, monkeypatch):
    import finance.exporting as exporting

    ds, member, *_ = workspace
    real_workbook = exporting.Workbook
    options = {}

    def workbook(*args, **kwargs):
        options.update(kwargs)
        return real_workbook(*args, **kwargs)

    monkeypatch.setattr(exporting, "Workbook", workbook)
    build_formula_workbook(
        ds,
        member,
        {"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
        BytesIO(),
    )
    assert options == {"write_only": True}


def test_formula_export_worker_publishes_complete_file_and_removes_temp_file(workspace):
    from finance.exporting import process_formula_export

    ds, _, client, *_ = workspace
    context = {
        "start": "2026-08",
        "end": "2026-08",
        "comparison": "yoy",
        "group": "department",
        "custom_columns": [custom_column()],
    }
    response = client.post(
        f"/api/datasets/{ds.pk}/formula-exports", data=context, content_type="application/json"
    )
    job = FormulaExport.objects.get(pk=response.json()["id"])
    stats = process_formula_export(job.pk)
    job.refresh_from_db()
    assert stats["rows"] == 5
    assert stats["formula_version"] == "3"
    assert job.status == "succeeded"
    assert job.expires_at is not None
    assert not job.path.endswith(".tmp")
    assert Path(job.path).read_bytes()[:2] == b"PK"
    assert stat.S_IMODE(Path(job.path).stat().st_mode) == 0o600
    assert stat.S_IMODE(Path(job.path).parent.stat().st_mode) == 0o700
    assert stat.S_IMODE(Path(job.path).parent.parent.stat().st_mode) == 0o700
    assert not Path(job.path).with_suffix(".tmp").exists()
    assert not list(Path(job.path).parent.glob(f"{job.pk}.*.tmp"))


def test_formula_export_worker_records_failure_and_removes_partial_file(workspace, monkeypatch):
    from finance.exporting import process_formula_export

    ds, _, client, *_ = workspace
    response = client.post(
        f"/api/datasets/{ds.pk}/formula-exports",
        data={"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
        content_type="application/json",
    )
    job = FormulaExport.objects.get(pk=response.json()["id"])

    def fail(*args, **kwargs):
        args[-1].write(b"partial")
        raise ValueError("controlled export failure")

    monkeypatch.setattr("finance.exporting.build_formula_workbook", fail)
    with pytest.raises(ValueError, match="controlled export failure"):
        process_formula_export(job.pk)
    job.refresh_from_db()
    assert job.status == "failed"
    assert job.error == "controlled export failure"
    response = client.get(f"/api/datasets/{ds.pk}/formula-exports/{job.pk}")
    assert response.json()["error"] == "生成失败，请联系管理员并提供任务编号"
    assert "controlled export failure" not in response.content.decode()
    assert not job.path
    temporary = (
        Path(job.dataset.file.storage.location) / "formula-exports" / str(job.user_id) / f"{job.pk}.tmp"
    )
    assert not temporary.exists()


def test_formula_export_worker_removes_stale_published_file_after_failed_retry(workspace, monkeypatch):
    from finance.exporting import process_formula_export

    ds, _, client, *_ = workspace
    response = client.post(
        f"/api/datasets/{ds.pk}/formula-exports",
        data={"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
        content_type="application/json",
    )
    job = FormulaExport.objects.get(pk=response.json()["id"])
    target = Path(job.dataset.file.storage.location) / "formula-exports" / str(job.user_id) / f"{job.pk}.xlsx"
    target.parent.mkdir(parents=True)
    target.write_bytes(b"stale workbook")

    monkeypatch.setattr(
        "finance.exporting.build_formula_workbook",
        lambda *_args, **_kwargs: (_ for _ in ()).throw(ValueError("retry failed")),
    )
    with pytest.raises(ValueError, match="retry failed"):
        process_formula_export(job.pk)

    job.refresh_from_db()
    assert job.status == "failed"
    assert job.path == ""
    assert job.expires_at is None
    assert not target.exists()


def test_formula_export_worker_treats_department_order_as_the_same_scope(workspace):
    from finance.exporting import process_formula_export

    ds, member, *_ = workspace
    member.departments = ["外科", "儿科"]
    member.save(update_fields=["departments"])
    job = FormulaExport.objects.create(
        dataset=ds,
        user=ds.created_by,
        context={"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
        departments=["外科", "儿科"],
        status="processing",
    )
    member.departments = ["儿科", "外科"]
    member.save(update_fields=["departments"])

    process_formula_export(job.pk)
    job.refresh_from_db()
    assert job.status == "succeeded"


def test_formula_export_worker_records_path_preparation_failure(workspace, settings):
    from finance.exporting import process_formula_export

    ds, _, client, *_ = workspace
    response = client.post(
        f"/api/datasets/{ds.pk}/formula-exports",
        data={"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
        content_type="application/json",
    )
    job = FormulaExport.objects.get(pk=response.json()["id"])
    job.status = "processing"
    job.save(update_fields=["status"])
    export_root = Path(settings.MEDIA_ROOT) / "formula-exports"
    export_root.symlink_to(Path(settings.MEDIA_ROOT) / "elsewhere", target_is_directory=True)

    with pytest.raises(ValueError, match="符号链接"):
        process_formula_export(job.pk)
    job.refresh_from_db()
    assert job.status == "failed"


def test_formula_export_failure_does_not_follow_a_symlinked_user_directory(workspace, settings):
    from finance.exporting import process_formula_export

    ds, _, client, *_ = workspace
    response = client.post(
        f"/api/datasets/{ds.pk}/formula-exports",
        data={"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
        content_type="application/json",
    )
    job = FormulaExport.objects.get(pk=response.json()["id"])
    root = Path(settings.MEDIA_ROOT) / "formula-exports"
    protected = Path(settings.MEDIA_ROOT) / "protected"
    protected.mkdir()
    target = protected / f"{job.pk}.xlsx"
    target.write_bytes(b"keep")
    root.mkdir()
    (root / str(job.user_id)).symlink_to(protected, target_is_directory=True)

    with pytest.raises(ValueError, match="符号链接"):
        process_formula_export(job.pk)

    job.refresh_from_db()
    assert job.status == "failed"
    assert target.read_bytes() == b"keep"


def test_formula_export_failed_cleanup_is_observable_and_reclaimed(workspace, monkeypatch):
    from django.core.management import call_command

    from finance.exporting import process_formula_export

    ds, _, client, *_ = workspace
    response = client.post(
        f"/api/datasets/{ds.pk}/formula-exports",
        data={"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
        content_type="application/json",
    )
    job = FormulaExport.objects.get(pk=response.json()["id"])
    target = Path(job.dataset.file.storage.location) / "formula-exports" / str(job.user_id) / f"{job.pk}.xlsx"
    target.parent.mkdir(parents=True)
    target.write_bytes(b"orphan")
    monkeypatch.setattr(
        "finance.exporting.build_formula_workbook",
        lambda *_a, **_k: (_ for _ in ()).throw(ValueError("retry failed")),
    )
    real_unlink = Path.unlink
    monkeypatch.setattr(Path, "unlink", lambda *_a, **_k: (_ for _ in ()).throw(OSError("read-only")))

    with pytest.raises(ValueError, match="retry failed"):
        process_formula_export(job.pk)
    job.refresh_from_db()
    assert job.status == "failed"
    assert job.path == str(target)

    monkeypatch.setattr(Path, "unlink", real_unlink)
    call_command("cleanup_formula_exports")
    job.refresh_from_db()
    assert job.path == ""
    assert target.exists() is False


def test_formula_workbook_rejects_unclassified_label_collision(workspace):
    from decimal import Decimal

    ds, member, *_ = workspace
    Fact.objects.create(
        dataset=ds,
        period="2025-08",
        dimensions={"department": None, "product": "医疗服务"},
        revenue=Decimal("10"),
        cost=Decimal("4"),
        gross=Decimal("6"),
        customer="missing-department",
        source_row=20,
    )
    Fact.objects.create(
        dataset=ds,
        period="2025-08",
        dimensions={"department": "（未分类）", "product": "医疗服务"},
        revenue=Decimal("10"),
        cost=Decimal("4"),
        gross=Decimal("6"),
        customer="literal-unclassified",
        source_row=21,
    )
    with pytest.raises(ValueError, match="与系统缺失值标签冲突"):
        build_formula_workbook(
            ds,
            member,
            {
                "start": "2026-08",
                "end": "2026-08",
                "comparison": "yoy",
                "group": "department",
            },
            BytesIO(),
        )


def test_formula_criteria_escape_excel_wildcards_and_quotes():
    from finance.exporting import _checked_formula, _criteria

    criteria = _criteria(
        {"subject": 'A*?~"科'},
        "department",
        "儿科?*",
        ["2026-08", "2026-09"],
    )
    assert 'Data[subject],"=A~*~?~~""科"' in criteria
    assert 'Data[department],"=儿科~?~*"' in criteria
    assert 'Data[Period],">=2026-08"' in criteria
    assert _criteria({}, "department", ">高值", ["2026-08"]).startswith('Data[department],"=>高值"')
    with pytest.raises(ValueError, match="8,192"):
        _checked_formula("=" + "A" * 8192)
    with pytest.raises(ValueError, match="255"):
        _criteria({}, "department", "科" * 255, ["2026-08"])


def test_formula_workbook_rejects_case_insensitive_group_collision(workspace):
    ds, member, *_ = workspace
    for source_row, label in ((20, "Team"), (21, "team")):
        Fact.objects.create(
            dataset=ds,
            period="2026-08",
            dimensions={"department": label, "product": "医疗服务"},
            revenue=Decimal(10),
            cost=Decimal(4),
            gross=Decimal(6),
            customer=f"case-{source_row}",
            source_row=source_row,
        )

    with pytest.raises(ValueError, match="大小写"):
        build_formula_workbook(
            ds,
            member,
            {"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
            BytesIO(),
        )


def test_formula_workbook_omits_default_sheet_and_writes_untrusted_labels_as_text(workspace):
    from openpyxl import load_workbook

    ds, member, *_ = workspace
    dangerous = '=HYPERLINK("https://invalid")'
    mapping = {**ds.mapping}
    mapping["sheet"] = dangerous
    mapping["columns"] = [dict(column) for column in mapping["columns"]]
    mapping["columns"][1]["name"] = dangerous
    mapping["columns"][3]["name"] = dangerous
    mapping["self_operated_products"] = [dangerous]
    ds.mapping = mapping
    ds.save(update_fields=["mapping"])
    ds.facts.update(dimensions={"department": dangerous, "product": "医疗服务"})

    definition = custom_column()
    definition["name"] = dangerous
    output = BytesIO()
    build_formula_workbook(
        ds,
        member,
        {
            "start": "2026-08",
            "end": "2026-08",
            "comparison": "yoy",
            "group": "department",
            "custom_columns": [definition],
        },
        output,
    )

    workbook = load_workbook(BytesIO(output.getvalue()), data_only=False)
    assert "Sheet" not in workbook.sheetnames
    cells = [
        workbook["分析汇总"]["D3"],
        workbook["原始证据"]["A2"],
        workbook["原始证据"]["C1"],
        workbook["参数与口径"]["B2"],
        workbook["参数与口径"]["B4"],
        workbook["对账检查"]["B2"],
    ]
    custom_reconcile = next(
        cell for row in workbook["对账检查"].iter_rows() for cell in row if cell.value == dangerous
    )
    cells.append(custom_reconcile)
    assert all(cell.value == dangerous and cell.data_type == "s" for cell in cells)


def test_formula_reconcile_uses_literal_unclassified_group(workspace):
    from openpyxl import load_workbook

    ds, member, *_ = workspace
    Fact.objects.create(
        dataset=ds,
        period="2026-08",
        dimensions={"department": "（未分类）", "product": "医疗服务"},
        revenue=Decimal(25),
        cost=Decimal(10),
        gross=Decimal(15),
        customer="unclassified",
        list_amount=Decimal(30),
        source_row=20,
        raw=["2026-08", "", "医疗服务", 25, 10, 15, "unclassified", 30],
    )
    output = BytesIO()
    build_formula_workbook(
        ds,
        member,
        {"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
        output,
    )

    workbook = load_workbook(BytesIO(output.getvalue()), data_only=False)
    reference = next(
        row[3].value
        for row in workbook["对账检查"].iter_rows(min_row=2)
        if row[0].value == "实际收入" and row[1].value == "（未分类）" and row[2].value == "本期"
    )
    assert Decimal(reference) == Decimal(25)


def test_formula_workbook_rejects_dimension_keys_that_collide_with_reserved_headers(workspace):
    ds, member, *_ = workspace
    mapping = {**ds.mapping}
    mapping["columns"] = [dict(column) for column in mapping["columns"]]
    mapping["columns"][1]["key"] = "revenue"
    ds.mapping = mapping
    ds.save(update_fields=["mapping"])

    with pytest.raises(ValueError, match="保留列名冲突"):
        build_formula_workbook(
            ds,
            member,
            {"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "revenue"},
            BytesIO(),
        )


def test_custom_reconcile_reference_preserves_large_decimal_and_zero_division():
    from finance.exporting import _evaluate_expression

    expression = {
        "operator": "divide",
        "left": {"metric": "revenue"},
        "right": {"number": 0},
    }
    assert (
        _evaluate_expression(expression, {"revenue": Decimal("1.2345678901234567890123456789012345")}) is None
    )
    assert _evaluate_expression(
        {"operator": "add", "left": {"metric": "revenue"}, "right": {"number": 1}},
        {"revenue": Decimal("1234567890123456789012345678901234")},
    ) == Decimal("1234567890123456789012345678901235")