fix(workbook): 参数页维度对照与使用说明输入字典（F06 收尾）

- 参数与口径页新增维度键与显示名对照行（只读）
- 使用说明页新增输入字典：每个可编辑输入逐项列出字段、单位、来源、
  修改位置与缺失含义；补充固定结构边界说明
- 毛利率参数行改为文本快照并指向计算明细 MarginRate 列
- 对账容差说明改为按指标类型独立容差、调整需财务批准
- 撤权校验区分账号停用与角色变更的报错信息
- 引入 test_formula_workbook_docs 与扩展的 test_worker_paths（F08 撤权、
  崩溃恢复语义对齐 PostgreSQL 咨询锁立即重认领）

finance/exporting.py

@@ -468,11 +468,10 @@ def build_formula_workbook(ds, member, export_context, output):
         ],
         ["ROI 口径", mapping["roi_formula"], "按导入映射保存的公式分支", "否"],
         ["自营产品集合", _text_cell(parameters, ", ".join(products)), "升单率和自营收入口径", "否"],
-        ["产品毛利率快照", str(rates), "导入时保存的只读口径；修改需重新确认导入映射", "否"],
         [
-            "对账容差",
+            "产品毛利率（快照）",
-            "人数 0；金额 0.01；比率 0.0001；自定义 0.01",
+            _text_cell(parameters, str(rates)),
-            "服务器 Decimal 参考与 Excel 结果的绝对差；正式财务验收前需确认",
+            "只读快照；导出时已按行写入计算明细 MarginRate 列，调整毛利率请在计算明细页按行修改",
             "否",
         ],
         [
@@ -483,8 +482,23 @@ def build_formula_workbook(ds, member, export_context, output):
         ],
         ["重算范围", f"最多 {MAX_ROWS:,} 行、{MAX_GROUPS} 组", "超限需缩小导出范围", "否"],
         ["兼容性", "SUMIFS/COUNTIFS", "移除动态数组人数公式；仍需目标 Excel 桌面版本验收", "否"],
+        [
+            "对账容差",
+            "人数严格相等；金额 0.01；比率 0.0001",
+            "对账检查页按指标类型使用独立容差；自定义列按金额容差 0.01。容差调整需财务负责人批准",
+            "否",
+        ],
     ]:
         parameters.append(row)
+    for d in dims:
+        parameters.append(
+            [
+                _text_cell(parameters, f"维度 {d['key']}"),
+                _text_cell(parameters, d["name"]),
+                "维度键与显示名对照（只读）；计算明细与原始证据的列名即维度键",
+                "否",
+            ]
+        )
     for definition in custom_columns:
         parameters.append(
             [
@@ -494,11 +508,34 @@ def build_formula_workbook(ds, member, export_context, output):
                 "公式由服务端生成",
             ]
         )
+
+    def source_name(role):
+        return next((c["name"] for c in mapping["columns"] if c["role"] == role), "未命名")
+
+    editable_inputs = [
+        ("Revenue", "实际收入", "金额", source_name("revenue"), "空值行不参与求和，汇总按现有行计算"),
+        ("Cost", "成本", "金额", source_name("cost"), "任一行缺失时本组成本及 ROI 等依赖指标不可用"),
+        ("GrossInput", "直接毛利", "金额", source_name("gross"), "空值行改按产品毛利率（lookup）计算毛利"),
+        (
+            "MarginRate",
+            "产品毛利率",
+            "小数比率",
+            "导入确认的产品毛利率表",
+            "仅 GrossSource=lookup 的行参与毛利计算；0.2 表示 20%",
+        ),
+        (
+            "ListAmount",
+            "原价金额",
+            "金额",
+            source_name("list_amount"),
+            "用于扣率与原价客单价；缺失则相关指标不可用",
+        ),
+    ]
     instructions.append(_styled_row(instructions, ["主题", "说明"], green, white))
     for row in [
         ["用途", "公式分析工作簿；本地修改不回写系统。"],
         ["行粒度", "计算明细每行对应一个 Fact，不等同患者就诊数。"],
-        ["维度列", "列名为数据导入映射中的维度键；显示名称见本页分组维度。"],
+        ["维度列", "列名为数据导入映射中的维度键；键与显示名对照见“参数与口径”页的维度行。"],
         ["毛利", "直接毛利优先，否则收入×产品毛利率；无法计算则不可用。毛利贡献不代表净利润。"],
         ["ROI", "按选定口径；缺成本、成本非正或所需指标缺失时不可用。"],
         ["人数", "客户 token 跨行去重；缺客户编码时不可用。"],
@@ -510,8 +547,23 @@ def build_formula_workbook(ds, member, export_context, output):
         ["精度", "原始 Decimal 以文本留存；Excel 双精度存在细微误差可能。"],
         ["隐私", "客户标识替换为本文件随机 token，文件仍按敏感财务信息保护。"],
         [
-            "编辑保护",
+            "输入字典",
-            "工作表结构、人数去重、原始证据和公式列已保护；计算明细仅收入、成本、毛利输入、毛利率和原价金额可编辑。行结构变化请重新导出。",
+            "以下输入可在“计算明细”页直接修改，全部汇总公式与对账实时引用；字段、单位、来源与缺失含义逐项列出。",
+        ],
+        *[
+            [
+                _text_cell(instructions, f"输入 {header}（{label}）"),
+                _text_cell(
+                    instructions,
+                    f"单位：{unit}；来源：{source}；修改位置：计算明细页 {header} 列；缺失含义：{missing}。",
+                ),
+            ]
+            for header, label, unit, source, missing in editable_inputs
+        ],
+        [
+            "固定结构",
+            "Period、维度列、CustomerToken、GrossSource、SelfOperated、CustomerPresent、RowCount 与“人数去重”页均为固定快照；"
+            "修改或增删这些列/行会使汇总、去重与对账失真，需要改变结构时请重新导出。",
         ],
     ]:
         instructions.append(row)
@@ -686,8 +738,10 @@ def process_formula_export(export_id):
         target.parent.chmod(0o700)
         temporary = target.with_name(f"{job.pk}.{secrets.token_hex(8)}.tmp")
         member = Membership.objects.get(user=job.user, organization=job.dataset.organization)
-        if not job.user.is_active or member.role not in ("admin", "analyst"):
+        if not job.user.is_active:
-            raise PermissionError("账号已撤权，请重新提交导出")
+            raise PermissionError("账号已停用，请重新提交导出")
+        if member.role not in ("admin", "analyst"):
+            raise PermissionError("角色已变更，请重新提交导出")
         if set(member.departments) != set(job.departments):
             raise PermissionError("科室授权已变更，请重新提交导出")
         descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)

tests/test_formula_workbook_docs.py

@@ -0,0 +1,155 @@
+from io import BytesIO
+
+import pytest
+from openpyxl import load_workbook
+
+from finance.exporting import build_formula_workbook
+
+pytestmark = pytest.mark.django_db
+
+EDITABLE_INPUT_HEADERS = ["Revenue", "Cost", "GrossInput", "MarginRate", "ListAmount"]
+
+
+def _workbook(workspace, **extra):
+    ds, member, *_ = workspace
+    output = BytesIO()
+    build_formula_workbook(
+        ds,
+        member,
+        {"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department", **extra},
+        output,
+    )
+    return load_workbook(BytesIO(output.getvalue()), data_only=False)
+
+
+def test_parameter_sheet_is_read_only_snapshot(workspace):
+    workbook = _workbook(workspace)
+    rows = list(workbook["参数与口径"].iter_rows(values_only=True))
+    header = list(rows[0])
+    editable = header.index("可编辑")
+    assert all(row[editable] == "否" for row in rows[1:] if row[0])
+
+    margin = next(row for row in rows if row[0] and "产品毛利率" in str(row[0]))
+    assert "只读" in margin[2]
+    assert "MarginRate" in margin[2]
+    assert "计算明细" in margin[2]
+
+    # 维度键与显示名完整对照
+    texts = [str(row[0]) for row in rows if row[0]]
+    assert "维度 department" in texts and "维度 product" in texts
+    department = next(row for row in rows if row[0] == "维度 department")
+    assert department[1] == "科室"
+    product = next(row for row in rows if row[0] == "维度 product")
+    assert product[1] == "产品一级分类"
+
+
+def test_instructions_document_editable_inputs_with_real_dependencies(workspace):
+    workbook = _workbook(workspace)
+    instructions = {
+        str(row[0]): str(row[1]) for row in workbook["使用说明"].iter_rows(values_only=True) if row[0]
+    }
+
+    data_headers = [cell.value for cell in workbook["计算明细"][1]]
+    sources = {
+        "Revenue": "实际收入",
+        "Cost": "成本",
+        "GrossInput": "毛利贡献",
+        "MarginRate": "产品毛利率表",
+        "ListAmount": "应收金额",
+    }
+    for header in EDITABLE_INPUT_HEADERS:
+        key = next(k for k in instructions if k.startswith(f"输入 {header}（"))
+        text = instructions[key]
+        assert f"计算明细页 {header} 列" in text
+        assert sources[header] in text
+        assert header in data_headers
+
+    # 每个可修改输入都被汇总公式真实引用
+    summary_formula_cells = [
+        cell.value
+        for row in workbook["分析汇总"].iter_rows(min_row=6)
+        for cell in row
+        if isinstance(cell.value, str) and cell.value.startswith("=IF")
+    ]
+    joined = "\n".join(summary_formula_cells)
+    for header in ("Revenue", "Cost", "ListAmount"):
+        assert f"Data[{header}]" in joined
+    detail_formula = next(
+        cell.value
+        for row in workbook["计算明细"].iter_rows(min_row=2)
+        for cell in row
+        if isinstance(cell.value, str) and cell.value.startswith("=IF(")
+    )
+    assert "GrossInput" not in detail_formula  # 列名按相对引用参与 GrossCalc
+    assert detail_formula.startswith("=IF(")
+
+    boundary = instructions["固定结构"]
+    assert "人数去重" in boundary and "固定快照" in boundary
+    assert "重新导出" in boundary
+    assert "参数与口径" in instructions["维度列"]
+
+
+def test_margin_rate_lookup_edit_changes_gross_via_detail_recalculation(workspace, tmp_path):
+    import os
+    import shutil
+
+    from finance.models import Fact
+
+    if not (os.environ.get("SOFFICE_BIN") or shutil.which("soffice")):
+        pytest.skip("set SOFFICE_BIN to run the LibreOffice recalculation check")
+
+    ds, member, *_ = workspace
+    # 去掉直接毛利，使毛利全部走 lookup 毛利率
+    Fact.objects.filter(dataset=ds).update(gross=None)
+    mapping = {**ds.mapping, "margin_rates": {"医疗服务": "0.3"}}
+    ds.mapping = mapping
+    ds.save(update_fields=["mapping"])
+
+    source, output = tmp_path / "source", tmp_path / "out"
+    source.mkdir()
+    output.mkdir()
+    workbook_path = source / "lookup.xlsx"
+    with workbook_path.open("wb") as file:
+        build_formula_workbook(
+            ds,
+            member,
+            {"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
+            file,
+        )
+
+    editable = load_workbook(workbook_path)
+    # MarginRate 列在维度列之后：G 列（期间+2维度=3 列后第 4 个保留列）
+    header_row = [cell.value for cell in editable["计算明细"][1]]
+    rate_column = header_row.index("MarginRate") + 1
+    editable["计算明细"].cell(row=2, column=rate_column).value = 0.5
+    editable.save(workbook_path)
+
+    import subprocess
+
+    profile = tmp_path / "profile"
+    result = subprocess.run(
+        [
+            os.environ.get("SOFFICE_BIN") or shutil.which("soffice"),
+            "--headless",
+            f"-env:UserInstallation={profile.as_uri()}",
+            "--convert-to",
+            "xlsx",
+            "--outdir",
+            str(output),
+            str(workbook_path),
+        ],
+        capture_output=True,
+        text=True,
+        timeout=120,
+        check=False,
+    )
+    assert result.returncode == 0, result.stderr or result.stdout
+    values = load_workbook(output / workbook_path.name, data_only=True)
+    summary = values["分析汇总"]
+    gross_row = next(
+        row
+        for row in range(6, summary.max_row + 1)
+        if summary.cell(row, 1).value == "毛利贡献" and summary.cell(row, 2).value == "儿科"
+    )
+    # 2026-08 儿科收入 100，lookup 毛利率改为 0.5 后毛利应为 50
+    assert summary.cell(gross_row, 3).value == pytest.approx(50)

tests/test_operations.py

@@ -109,7 +109,7 @@ def test_formula_export_rechecks_role_when_worker_runs_and_file_is_downloaded(wo
     job = FormulaExport.objects.create(dataset=ds, user=ds.created_by, context={}, status="processing")
     member.role = "viewer"
     member.save(update_fields=["role"])
-    with pytest.raises(PermissionError, match="撤权"):
+    with pytest.raises(PermissionError, match="角色已变更"):
         process_formula_export(job.pk)
     job.refresh_from_db()
     assert job.status == "failed"

tests/test_worker_paths.py

@@ -1,7 +1,10 @@
+from datetime import timedelta
+from pathlib import Path
 from unittest.mock import patch
 
 import pytest
 from django.core.management import call_command
+from django.utils import timezone
 
 from finance.models import Dataset, FormulaExport
 
@@ -30,3 +33,109 @@ def test_worker_processes_queued_import_and_reports_export_failure(workspace):
         call_command("import_worker", once=True)
     export.refresh_from_db()
     assert export.status == "processing"
+
+
+def _export_job(ds, **kwargs):
+    return FormulaExport.objects.create(
+        dataset=ds,
+        user=ds.created_by,
+        context={"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
+        **kwargs,
+    )
+
+
+def test_worker_refuses_export_after_role_revocation(workspace):
+    from finance.exporting import process_formula_export
+
+    ds, member, *_ = workspace
+    job = _export_job(ds, status="processing")
+    member.role = "viewer"
+    member.save(update_fields=["role"])
+
+    with pytest.raises(PermissionError, match="角色"):
+        process_formula_export(job.pk)
+
+    job.refresh_from_db()
+    assert job.status == "failed"
+    assert not job.path
+    assert not list(Path(ds.file.storage.location).rglob(f"{job.pk}*"))
+
+
+def test_worker_refuses_export_after_account_deactivation(workspace):
+    from finance.exporting import process_formula_export
+
+    ds, _, _, _, _ = workspace
+    user = ds.created_by
+    user.is_active = False
+    user.save(update_fields=["is_active"])
+    job = _export_job(ds, status="processing")
+
+    with pytest.raises(PermissionError, match="停用"):
+        process_formula_export(job.pk)
+
+    job.refresh_from_db()
+    assert job.status == "failed"
+
+
+def test_status_and_download_deny_revoked_viewer(workspace, settings):
+    ds, member, client, *_ = workspace
+    job = _export_job(
+        ds,
+        status="succeeded",
+        finished_at=timezone.now(),
+        expires_at=timezone.now() + timedelta(hours=1),
+    )
+    target = Path(settings.MEDIA_ROOT) / "formula-exports" / str(job.user_id) / f"{job.pk}.xlsx"
+    target.parent.mkdir(parents=True)
+    target.write_bytes(b"PK workbook")
+    job.path = str(target)
+    job.save(update_fields=["path"])
+
+    assert client.get(f"/api/datasets/{ds.pk}/formula-exports/{job.pk}").status_code == 200
+    assert client.get(f"/api/datasets/{ds.pk}/formula-exports/{job.pk}/download").status_code == 200
+
+    member.role = "viewer"
+    member.save(update_fields=["role"])
+    assert client.get(f"/api/datasets/{ds.pk}/formula-exports/{job.pk}").status_code == 403
+    assert client.get(f"/api/datasets/{ds.pk}/formula-exports/{job.pk}/download").status_code == 403
+
+
+def test_half_finished_export_cannot_be_downloaded(workspace, settings):
+    ds, _, client, *_ = workspace
+    for status, code in (("processing", 409), ("failed", 409), ("queued", 409)):
+        job = _export_job(ds, status=status)
+        assert client.get(f"/api/datasets/{ds.pk}/formula-exports/{job.pk}/download").status_code == code, (
+            status
+        )
+    succeeded = _export_job(
+        ds,
+        status="succeeded",
+        path=str(Path(settings.MEDIA_ROOT) / "nowhere.xlsx"),
+        finished_at=timezone.now(),
+        expires_at=timezone.now() + timedelta(hours=1),
+    )
+    assert client.get(f"/api/datasets/{ds.pk}/formula-exports/{succeeded.pk}/download").status_code == 410
+
+
+def test_stale_processing_export_is_reclaimed_after_timeout(workspace):
+    ds, *_ = workspace
+    job = _export_job(ds, status="processing")
+    FormulaExport.objects.filter(pk=job.pk).update(created_at=timezone.now() - timedelta(minutes=20))
+
+    call_command("import_worker", once=True)
+
+    job.refresh_from_db()
+    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_recent_processing_export_is_reclaimed_immediately(workspace):
+    # PostgreSQL 咨询锁语义下 worker 立即重认领 processing 任务，不再等待超时。
+    ds, *_ = workspace
+    job = _export_job(ds, status="processing")
+
+    call_command("import_worker", once=True)
+
+    job.refresh_from_db()
+    assert job.status == "succeeded"