import os
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 _queued_job(ds):
    return FormulaExport.objects.create(
        dataset=ds,
        user=ds.created_by,
        context={"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
    )


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

    ds, *_ = workspace
    job = _queued_job(ds)
    root = Path(settings.MEDIA_ROOT) / "formula-exports"
    root.mkdir(parents=True)
    protected = Path(settings.MEDIA_ROOT) / "protected"
    protected.mkdir()
    victim = protected / f"{job.pk}.xlsx"
    victim.write_bytes(b"protected workbook")
    (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 not job.path
    assert victim.read_bytes() == b"protected workbook"
    assert list(protected.glob("*.tmp")) == []


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

    ds, *_ = workspace
    job = _queued_job(ds)
    user_dir = Path(settings.MEDIA_ROOT) / "formula-exports" / str(job.user_id)
    user_dir.mkdir(parents=True)
    protected_file = Path(settings.MEDIA_ROOT) / "protected.xlsx"
    protected_file.write_bytes(b"protected workbook")
    (user_dir / f"{job.pk}.xlsx").symlink_to(protected_file)

    with pytest.raises(ValueError, match="普通文件"):
        process_formula_export(job.pk)

    job.refresh_from_db()
    assert job.status == "failed"
    assert protected_file.read_bytes() == b"protected workbook"
    assert (user_dir / f"{job.pk}.xlsx").is_symlink()
    assert list(user_dir.glob("*.tmp")) == []


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

    ds, *_ = workspace
    job = _queued_job(ds)
    root = Path(settings.MEDIA_ROOT) / "formula-exports"
    user_dir = root / str(job.user_id)
    user_dir.mkdir(parents=True)
    protected = Path(settings.MEDIA_ROOT) / "protected"
    protected.mkdir()
    victim = protected / f"{job.pk}.xlsx"
    victim.write_bytes(b"protected workbook")

    def swap_and_fail(*args, **kwargs):
        args[-1].write(b"partial")
        user_dir.rename(user_dir.with_name("swapped"))
        user_dir.symlink_to(protected, target_is_directory=True)
        raise ValueError("controlled failure after path validation")

    monkeypatch.setattr("finance.exporting.build_formula_workbook", swap_and_fail)
    with pytest.raises(ValueError, match="controlled failure"):
        process_formula_export(job.pk)

    job.refresh_from_db()
    assert job.status == "failed"
    assert victim.read_bytes() == b"protected workbook"
    assert user_dir.is_symlink()
    # 本次创建的临时文件残留在被换名的原目录中，由清理命令按保留期回收。
    assert list((root / "swapped").glob(f"{job.pk}.*.tmp"))


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

    ds, *_ = workspace
    job = _queued_job(ds)

    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)
    monkeypatch.undo()

    stats = process_formula_export(job.pk)
    job.refresh_from_db()
    assert stats["rows"] == 5
    assert job.status == "succeeded"
    assert Path(job.path).read_bytes()[:2] == b"PK"
    assert not list(Path(job.path).parent.glob(f"{job.pk}.*.tmp"))


def test_cleanup_failure_is_logged_and_reclaimed_after_permissions_restored(
    workspace, settings, monkeypatch, caplog
):
    import logging as logging_module
    from datetime import timedelta

    from django.core.management import call_command
    from django.utils import timezone

    from finance.exporting import process_formula_export

    ds, *_ = workspace
    job = _queued_job(ds)
    user_dir = Path(settings.MEDIA_ROOT) / "formula-exports" / str(job.user_id)
    user_dir.mkdir(parents=True)
    target = user_dir / f"{job.pk}.xlsx"
    target.write_bytes(b"stale workbook")
    other = FormulaExport.objects.create(
        dataset=ds,
        user=ds.created_by,
        context={"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
        status="succeeded",
        expires_at=timezone.now() + timedelta(hours=1),
    )
    kept = user_dir / f"{other.pk}.xlsx"
    other.path = str(kept)
    other.save(update_fields=["path"])
    kept.write_bytes(b"active workbook")

    def fail(*args, **kwargs):
        args[-1].write(b"partial")
        os.chmod(user_dir, 0o500)
        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.path == str(target)
    warnings = [r for r in caplog.records if r.levelno == logging_module.WARNING]
    assert warnings and all(str(job.pk) in record.getMessage() for record in warnings)

    os.chmod(user_dir, 0o700)
    old = (timezone.now() - timedelta(hours=25)).timestamp()
    for leftover in [target, *user_dir.glob(f"{job.pk}.*.tmp"), kept]:
        os.utime(leftover, (old, old))
    call_command("cleanup_formula_exports")

    job.refresh_from_db()
    assert not target.exists()
    assert not list(user_dir.glob(f"{job.pk}.*.tmp"))
    assert job.path == ""
    assert kept.read_bytes() == b"active workbook"


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(),
        )