From 2a83de21d97d4bff715ace56e26c9572e1b36a32 Mon Sep 17 00:00:00 2001 From: Jayway Liu <14022498+bluejayway@user.noreply.gitee.com> Date: Sun, 1 Nov 2026 03:52:08 +0800 Subject: [PATCH] =?UTF-8?q?fix(workbook):=20=E5=8F=82=E6=95=B0=E9=A1=B5?= =?UTF-8?q?=E7=BB=B4=E5=BA=A6=E5=AF=B9=E7=85=A7=E4=B8=8E=E4=BD=BF=E7=94=A8?= =?UTF-8?q?=E8=AF=B4=E6=98=8E=E8=BE=93=E5=85=A5=E5=AD=97=E5=85=B8=EF=BC=88?= =?UTF-8?q?F06=20=E6=94=B6=E5=B0=BE=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 参数与口径页新增维度键与显示名对照行（只读） - 使用说明页新增输入字典：每个可编辑输入逐项列出字段、单位、来源、 修改位置与缺失含义；补充固定结构边界说明 - 毛利率参数行改为文本快照并指向计算明细 MarginRate 列 - 对账容差说明改为按指标类型独立容差、调整需财务批准 - 撤权校验区分账号停用与角色变更的报错信息 - 引入 test_formula_workbook_docs 与扩展的 test_worker_paths（F08 撤权、 崩溃恢复语义对齐 PostgreSQL 咨询锁立即重认领） --- finance/exporting.py | 72 +++++++++++-- tests/test_formula_workbook_docs.py | 155 ++++++++++++++++++++++++++++ tests/test_operations.py | 2 +- tests/test_worker_paths.py | 109 +++++++++++++++++++ 4 files changed, 328 insertions(+), 10 deletions(-) create mode 100644 tests/test_formula_workbook_docs.py diff --git a/finance/exporting.py b/finance/exporting.py index bf9641f..ed7fc05 100644 --- a/finance/exporting.py +++ b/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", - "服务器 Decimal 参考与 Excel 结果的绝对差；正式财务验收前需确认", + "产品毛利率（快照）", + _text_cell(parameters, str(rates)), + "只读快照；导出时已按行写入计算明细 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"): - raise PermissionError("账号已撤权，请重新提交导出") + if not job.user.is_active: + 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) diff --git a/tests/test_formula_workbook_docs.py b/tests/test_formula_workbook_docs.py new file mode 100644 index 0000000..9542b12 --- /dev/null +++ b/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) diff --git a/tests/test_operations.py b/tests/test_operations.py index 0ec0158..35ef4f3 100644 --- a/tests/test_operations.py +++ b/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" diff --git a/tests/test_worker_paths.py b/tests/test_worker_paths.py index 1b1bc6d..f33af7f 100644 --- a/tests/test_worker_paths.py +++ b/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"