fix(dev): dev_up 以数据库为准幂等初始化管理员

此前凭据文件存在但数据库重建时会跳过 init_local.py，导致本地无
admin 无法登录；改为每次启动都执行（脚本本身按数据库幂等）。

docs/fix-record-2026-09-26.md

@@ -1,30 +0,0 @@
-# 剩余问题修复记录 · 2026-09-26
-
-对应 `docs/remaining-issues-2026-09-26.html` 第三节（F01–F08）的修复交付记录。
-分支：`fix/remaining-issues-2026-09-26`（基线 `0ac0c4e` 起逐项小步提交）。
-
-## 修复状态
-
-| 编号 | 状态 | 提交 | 说明 |
-| --- | --- | --- | --- |
-| F01 · P0 · 符号链接清理 | 已修复 | `36fd5f3` | 根目录/用户目录/目标文件先 lstat 完整校验（拒绝符号链接与非目录/非普通文件），校验失败不删除任何磁盘文件；失败清理只处理本次登记的受控文件并通过 `O_NOFOLLOW` 目录描述符 unlink，目录被换成符号链接时拒绝删除。测试覆盖用户目录链接、目标链接、运行中目录被换成链接（受保护文件字节不变）、失败后重试成功。 |
-| F02 · P1 · 清理失败与孤儿 | 已修复 | `54dc58d` | 清理受阻时记录不含敏感数据的 WARNING（任务编号+文件名）并保留 `job.path` 使残留可追踪；`cleanup_formula_exports` 新增孤儿 `.xlsx` 回收：超过 24h 保留期且不属于未过期成功任务的文件统一回收，不触碰活动任务。测试含 unlink 权限错误注入→告警→恢复权限后回收且其他任务文件不受影响。 |
-| F03 · P1 · 证据弹窗卡加载 | 已修复 | `6cf236a` | 同期间同页点击直接返回；第二页点击当前期间回第一页刷新；失败显示错误+重试按钮；空结果显示空态行。E2E 覆盖重复点击、快速切换、第二页切期间、失败重试、空列表。 |
-| F04 · P1 · 载入定义丢上下文 | 已修复 | `9a5ab26` | 载入定义时以“用户消息+助手确认”把名称/含义/范围/缺失处理/表达式写入可见对话上下文（需勾选同意才发送），修改后仍需重新确认；选择空选项同步清理定义与上下文。服务端单条消息上限 1000→2000 容纳完整定义。E2E 验证模型请求包含原定义且“乘以 100”修改生效。 |
-| F05 · P1 · 定义库并发与审计 | 已修复 | `0aaa87e` | 迁移 0005 增加 `revision`；保存在单事务内锁定用户所属行：配额（50/数据集）原子化、同名更新必须携带匹配 revision（409 冲突）、审计与变更同事务回滚；保存限流 30/h。前端保存同名定义携带 revision 并明示覆盖行为。PostgreSQL 并发配额用例作为独立门禁（本机 SQLite 自动跳过）。 |
-| F06 · P1 · 工作簿可解释性 | 已修复 | `9962573` | 参数页“产品毛利率”改为只读快照（可编辑列全部为“否”），写明调整位置是计算明细 MarginRate 列；参数页新增维度键-显示名完整对照；使用说明新增输入字典（Revenue/Cost/GrossInput/MarginRate/ListAmount 的单位、来源字段、修改位置、缺失含义）与固定结构边界行。 |
-| F07 · P1 · 对账容差 | 已修复 | `649fec4` | 对账检查页新增容差列：人数严格相等、金额 0.01、比率 0.0001，状态公式按 `ABS(差异)<=容差` 判定；自定义列用 `to_decimal` 白名单表达式解释器（不执行代码）计算 Decimal 服务器参考，不再标注“需业务验收”；参数页写明容差规则与“调整需财务负责人批准”。 |
-| F08 · P1 · 撤权与生命周期 | 已修复 | `5ebe83a` | worker 执行前重新核验账号 `is_active` 与 admin/analyst 角色（撤权/停用后排队任务失败、不发布）；导出状态/下载接口补齐角色白名单（降为 viewer 后 403）；`import_worker` 对超过 15 分钟的 processing 任务重新认领（PG 依赖咨询锁立即恢复），崩溃重启不再无限停留。 |
-
-## 本轮完整测试记录（2026-09-26，SQLite 隔离环境）
-
-- **后端：87 passed，7 skipped，90.3 秒；finance 覆盖率 88.50%（85% 门槛通过）。**
-  跳过项：4 项 LibreOffice 重算（本机未安装 soffice，环境门禁，CI 中执行）、1 项 PostgreSQL 并发配额（依赖 PG 行锁，生产库单独验证）、1 项原附件全量 UI（未设置 `REFERENCE_DATASET_NAME`）。
-- **浏览器 E2E：3 passed，1 skipped，21.0 秒。** 新隔离数据库（`var/e2e-test.sqlite3` + 独立 MEDIA_ROOT），覆盖追问、保存定义、关闭重开、载入定义修改后生成（乘以 100）、下载，含 F03 全部交互回归（重复点击/快速切换/失败重试/空列表/翻页）、桌面与手机。
-- Ruff check / ruff format、TypeScript（tsc --noEmit）、Vite 构建、Django check、`makemigrations --check`（迁移一致）、collectstatic：全部通过。
-- `pip-audit`：无已知漏洞。`npm audit --omit=dev`：本机 registry 访问失败（网络受限，CI 中执行），记录为环境限制项。
-
-## 仍开放的事项（与报告一致，非本轮代码范围）
-
-- 第四节功能缺口（网页主汇总表新列、草稿续聊、原附件 332,830 行导出、高基数与可访问性专项等）与第五节外部门禁（目标 Excel 实机、财务口径签字、AI 真实联调、生产基础设施、PostgreSQL/SLA、安全运营）仍按原报告执行，责任人与关闭证据不变。
-- F07 容差数值（金额 0.01 / 比率 0.0001）为工程默认，仍需财务负责人确认或调整后才视为财务口径关闭。

finance/exporting.py

@@ -1,9 +1,7 @@
 import hashlib
 import hmac
-import logging
 import os
 import secrets
-import stat
 import warnings
 from datetime import timedelta
 from decimal import Decimal, InvalidOperation
@@ -21,7 +19,6 @@ from .formula_columns import (
     RESERVED_DATA_HEADERS,
     ensure_available_metrics,
     metrics_used,
-    to_decimal,
     to_excel,
     validate_column,
 )
@@ -31,75 +28,6 @@ FORMULA_VERSION = "3"
 MAX_ROWS = 100_000
 MAX_GROUPS = 200
 
-# 对账比较规则：人数严格相等；金额 0.01；比率 0.0001（一个基点）。
-# 容差属财务口径，调整需财务负责人批准后方可修改。
-RECONCILE_TOLERANCES = {
-    "customers": Decimal("0"),
-    "revenue": Decimal("0.01"),
-    "cost": Decimal("0.01"),
-    "gross": Decimal("0.01"),
-    "self_revenue": Decimal("0.01"),
-    "basket": Decimal("0.0001"),
-    "margin": Decimal("0.0001"),
-    "discount_rate": Decimal("0.0001"),
-    "upgrade_rate": Decimal("0.0001"),
-    "list_basket": Decimal("0.0001"),
-    "roi": Decimal("0.0001"),
-}
-CUSTOM_COLUMN_TOLERANCE = Decimal("0.01")
-
-logger = logging.getLogger(__name__)
-
-
-def _validated_export_directory(path, description):
-    """导出目录必须不存在或为真实目录；符号链接一律拒绝并保持文件原样。"""
-    try:
-        info = os.lstat(path)
-    except FileNotFoundError:
-        path.mkdir(parents=True, mode=0o700)
-        info = os.lstat(path)
-    if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode):
-        raise ValueError(f"{description}必须是真实目录，不能是符号链接")
-    os.chmod(path, 0o700)
-    return path
-
-
-def _validated_export_target(user_dir, job):
-    """目标文件若已存在必须是普通文件；符号链接或其他类型一律拒绝覆盖。"""
-    target = user_dir / f"{job.pk}.xlsx"
-    try:
-        info = os.lstat(target)
-    except FileNotFoundError:
-        return target
-    if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
-        raise ValueError("公式导出目标路径已存在且不是普通文件，已拒绝生成")
-    return target
-
-
-def _unlink_created_file(path):
-    """只通过不跟随链接的目录句柄删除受控目录内的普通文件。
-
-    返回 "removed"（已删除）、"missing"（本就不存在）或 "blocked"（无法安全删除，
-    例如目录已被换成符号链接或权限不足），调用方据此保留失败任务与磁盘的一致性。
-    """
-    try:
-        dir_fd = os.open(path.parent, os.O_RDONLY | os.O_NOFOLLOW)
-    except OSError:
-        return "blocked"
-    try:
-        try:
-            info = os.lstat(path.name, dir_fd=dir_fd)
-        except FileNotFoundError:
-            return "missing"
-        if not stat.S_ISREG(info.st_mode):
-            return "blocked"
-        os.unlink(path.name, dir_fd=dir_fd)
-        return "removed"
-    except OSError:
-        return "blocked"
-    finally:
-        os.close(dir_fd)
-
 
 def _excel_number(value):
     if value is None:
@@ -480,12 +408,7 @@ def build_formula_workbook(ds, member, export_context, output):
         ],
         ["ROI 口径", mapping["roi_formula"], "按导入映射保存的公式分支", "否"],
         ["自营产品集合", _text_cell(parameters, ", ".join(products)), "升单率和自营收入口径", "否"],
-        [
-            "产品毛利率（快照）",
-            _text_cell(parameters, str(rates)),
-            "只读快照；导出时已按行写入计算明细 MarginRate 列，调整毛利率请在计算明细页按行修改",
-            "否",
-        ],
+        ["产品毛利率", str(rates), "直接毛利缺失时的可编辑计算输入", "是"],
         [
             "Excel 数值精度",
             "IEEE-754 双精度",
@@ -494,23 +417,8 @@ 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(
             [
@@ -521,33 +429,10 @@ def build_formula_workbook(ds, member, export_context, output):
             ]
         )
     instructions.append(_styled_row(instructions, ["主题", "说明"], green, white))
-
-    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"),
-            "用于扣率与原价客单价；缺失则相关指标不可用",
-        ),
-    ]
     for row in [
         ["用途", "公式分析工作簿；本地修改不回写系统。"],
         ["行粒度", "计算明细每行对应一个 Fact，不等同患者就诊数。"],
-        ["维度列", "列名为数据导入映射中的维度键；键与显示名对照见“参数与口径”页的维度行。"],
+        ["维度列", "列名为数据导入映射中的维度键；显示名称见本页分组维度。"],
         ["毛利", "直接毛利优先，否则收入×产品毛利率；无法计算则不可用。毛利贡献不代表净利润。"],
         ["ROI", "按选定口径；缺成本、成本非正或所需指标缺失时不可用。"],
         ["人数", "客户 token 跨行去重；缺客户编码时不可用。"],
@@ -558,25 +443,6 @@ def build_formula_workbook(ds, member, export_context, output):
         ["变化率", "差额/ABS(基期)；基期为0或期间缺失时不可用。"],
         ["精度", "原始 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)
     for definition in custom_columns:
@@ -680,18 +546,11 @@ def build_formula_workbook(ds, member, export_context, output):
     reconcile.append(
         _styled_row(
             reconcile,
-            ["指标", "分组", "期间侧", "服务器 Decimal 参考", "公式结果", "差异", "容差", "状态"],
+            ["指标", "分组", "期间侧", "服务器 Decimal 参考", "公式结果", "差异", "状态"],
             green,
             white,
         )
     )
-
-    def _reconcile_status(row_number):
-        return (
-            f'=IF(D{row_number}="","无参考",IF(ISNUMBER(F{row_number}),'
-            f'IF(ABS(F{row_number})<=G{row_number},"通过","待复核"),"待复核"))'
-        )
-
     reconcile_row = 2
     for label in labels:
         for side_name, periods, col in (("本期", ctx["current"], "C"), ("基期", ctx["baseline"], "D")):
@@ -708,25 +567,21 @@ def build_formula_workbook(ds, member, export_context, output):
                         "" if value is None else str(value),
                         f"='分析汇总'!{col}{formula_row}",
                         f'=IFERROR(E{reconcile_row}-D{reconcile_row},"")',
-                        float(RECONCILE_TOLERANCES[key]),
-                        _reconcile_status(reconcile_row),
+                        f'=IF(D{reconcile_row}="","无参考",IF(ISNUMBER(F{reconcile_row}),IF(ABS(F{reconcile_row})<0.01,"通过","待复核"),"待复核"))',
                     ]
                 )
                 reconcile_row += 1
             for index, definition in enumerate(custom_columns):
                 formula_row = metric_rows[(label, f"custom_{index}")]
-                # 自定义列用 Decimal 白名单解释器计算服务器参考，不再依赖业务人工验收
-                reference_value = to_decimal(definition["expression"], reference)
                 reconcile.append(
                     [
                         _text_cell(reconcile, definition["name"]),
                         _text_cell(reconcile, label),
                         side_name,
-                        "" if reference_value is None else str(reference_value),
+                        "自定义公式无服务器参考",
                         f"='分析汇总'!{col}{formula_row}",
-                        f'=IFERROR(E{reconcile_row}-D{reconcile_row},"")',
-                        float(CUSTOM_COLUMN_TOLERANCE),
-                        _reconcile_status(reconcile_row),
+                        "",
+                        "需业务验收",
                     ]
                 )
                 reconcile_row += 1
@@ -739,51 +594,41 @@ def build_formula_workbook(ds, member, export_context, output):
 
 def process_formula_export(export_id):
     job = FormulaExport.objects.select_related("dataset", "user").get(pk=export_id)
+    temporary = None
     target = None
-    created = []
     try:
+        root = Path(settings.MEDIA_ROOT) / "formula-exports"
+        if root.is_symlink():
+            raise ValueError("公式导出目录不能是符号链接")
+        root.mkdir(parents=True, exist_ok=True, mode=0o700)
+        root.chmod(0o700)
+        target = root / str(job.user_id) / f"{job.pk}.xlsx"
+        target.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+        if target.parent.is_symlink() or target.is_symlink():
+            raise ValueError("公式导出路径不能是符号链接")
+        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:
-            raise PermissionError("账号已停用，公式导出已拒绝执行")
-        if member.role not in ("admin", "analyst"):
-            raise PermissionError("账号角色已变更，公式导出已拒绝执行")
         if set(member.departments) != set(job.departments):
             raise PermissionError("科室授权已变更，请重新提交导出")
-        root = _validated_export_directory(Path(settings.MEDIA_ROOT) / "formula-exports", "公式导出目录")
-        user_dir = _validated_export_directory(root / str(job.user_id), "公式导出用户目录")
-        target = _validated_export_target(user_dir, job)
-        # 路径完整验证通过后才把目标登记为可清理文件；此前任何失败都不会删除磁盘文件。
-        created.append(target)
-        temporary = target.with_name(f"{job.pk}.{secrets.token_hex(8)}.tmp")
         descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
-        created.append(temporary)
         with os.fdopen(descriptor, "wb") as file:
             stats = build_formula_workbook(job.dataset, member, job.context, file)
-        os.replace(temporary, target)
-        created = [target]
+        temporary.replace(target)
         job.path, job.status, job.error = str(target), "succeeded", ""
         job.finished_at = timezone.now()
         job.expires_at = timezone.now() + timedelta(hours=24)
         job.save(update_fields=["path", "status", "error", "finished_at", "expires_at"])
         return stats
     except Exception as exc:
-        job.status, job.error, job.finished_at, job.expires_at = (
-            "failed",
-            str(exc)[:500],
-            timezone.now(),
-            None,
-        )
-        # 只清理本次登记且确认位于受控目录内的文件；目录被换成符号链接时拒绝删除。
-        leftover = [path for path in created if _unlink_created_file(path) == "blocked"]
-        if leftover:
-            # 日志只含任务编号与文件名，不包含敏感数据；残留文件由清理命令按保留期回收。
-            logger.warning(
-                "公式导出任务 %s 失败后清理文件受阻，待清理命令回收：%s",
-                job.pk,
-                "、".join(path.name for path in leftover),
-            )
-        job.path = str(target) if target in leftover else ""
+        for path in (temporary, target):
+            try:
+                if path is not None and not path.is_symlink():
+                    path.unlink(missing_ok=True)
+            except OSError:
+                pass
+        job.status, job.error, job.path = "failed", str(exc)[:500], ""
+        job.finished_at, job.expires_at = timezone.now(), None
         job.save(update_fields=["status", "error", "path", "finished_at", "expires_at"])
         raise
 

finance/formula_columns.py

@@ -1,5 +1,4 @@
 import math
-from decimal import Decimal
 
 from .analysis import METRICS
 
@@ -108,27 +107,3 @@ def to_excel(node, references):
     if "number" in node:
         return str(node["number"])
     return f"({to_excel(node['left'], references)}{OPERATORS[node['operator']]}{to_excel(node['right'], references)})"
-
-
-def to_decimal(node, values):
-    """按已校验的白名单表达式树用 Decimal 计算服务器参考值。
-
-    只遍历 metric/number/四则运算节点，不执行任何代码；依赖指标缺失或除零时
-    返回 None，与 Excel 侧 #N/A 的语义一致。
-    """
-    if "metric" in node:
-        return values[node["metric"]]
-    if "number" in node:
-        return Decimal(str(node["number"]))
-    left, right = to_decimal(node["left"], values), to_decimal(node["right"], values)
-    if left is None or right is None:
-        return None
-    if node["operator"] == "add":
-        return left + right
-    if node["operator"] == "subtract":
-        return left - right
-    if node["operator"] == "multiply":
-        return left * right
-    if right == 0:
-        return None
-    return left / right

finance/formula_views.py

@@ -3,17 +3,14 @@ from pathlib import Path
 
 import httpx
 from django.conf import settings
-from django.db import IntegrityError, transaction
 from django.http import FileResponse, JsonResponse
 from django.utils import timezone
 
 from .analysis import METRICS, context, dimensions
 from .formula_columns import available_metrics, ensure_available_metrics, validate_column
-from .models import FormulaDefinition, FormulaExport, Membership
+from .models import FormulaDefinition, FormulaExport
 from .views import APIError, audit, body, dataset, endpoint, limit
 
-MAX_DEFINITIONS_PER_DATASET = 50
-
 
 def design_column(messages, ds, ctx):
     if not (settings.AI_ENABLED and settings.OPENAI_API_KEY and settings.OPENAI_MODEL):
@@ -25,8 +22,7 @@ def design_column(messages, ds, ctx):
         or set(item) != {"role", "content"}
         or item["role"] not in ("user", "assistant")
         or not isinstance(item["content"], str)
-        # 2000 字符容纳载入已保存定义后的完整对话上下文
-        or len(item["content"]) > 2000
+        or len(item["content"]) > 1000
         for item in messages
     ):
         raise APIError("对话消息内容无效")
@@ -201,7 +197,6 @@ def formula_definitions(request, dataset_id):
                     {
                         "id": str(item.pk),
                         "definition": item.definition,
-                        "revision": item.revision,
                         "updated_at": item.updated_at.isoformat(),
                     }
                     for item in items
@@ -209,54 +204,28 @@ def formula_definitions(request, dataset_id):
             }
         )
 
-    data = body(request)
-    definition = validate_column(data.get("definition"))
+    definition = validate_column(body(request).get("definition"))
     try:
         ensure_available_metrics(ds, definition)
     except ValueError as exc:
         raise APIError(str(exc)) from exc
     normalized_name = definition["name"].strip().casefold()
-    limit(f"formula-definition:{request.user.pk}", 30, 3600)
-    with transaction.atomic():
-        # 锁定稳定的用户所属行序列化同一用户的并发保存，使配额检查与写入原子化。
-        Membership.objects.select_for_update().get(pk=request.member.pk)
-        existing = (
-            FormulaDefinition.objects.select_for_update()
-            .filter(dataset=ds, user=request.user, normalized_name=normalized_name)
-            .first()
-        )
-        if existing:
-            # 同名更新按 revision 做乐观并发控制，避免两个窗口相互静默覆盖。
-            if data.get("revision") != existing.revision:
-                raise APIError("定义已在其他窗口更新，请刷新后重试", 409)
-            existing.definition, existing.revision = definition, existing.revision + 1
-            existing.save(update_fields=["definition", "revision", "updated_at"])
-            item, created = existing, False
-        else:
-            if (
-                FormulaDefinition.objects.filter(dataset=ds, user=request.user).count()
-                >= MAX_DEFINITIONS_PER_DATASET
-            ):
-                raise APIError(f"每个数据集最多保存 {MAX_DEFINITIONS_PER_DATASET} 个公式定义")
-            try:
-                item = FormulaDefinition.objects.create(
-                    dataset=ds,
-                    user=request.user,
-                    normalized_name=normalized_name,
-                    definition=definition,
-                )
-            except IntegrityError as exc:
-                raise APIError("定义保存发生并发冲突，请重试", 409) from exc
-            created = True
-        # 审计与定义变更在同一事务内提交；审计失败时变更一并回滚。
-        audit(
-            request, "formula_definition.save", item.pk, {"dataset": str(ds.pk), "name": definition["name"]}
-        )
+    existing = FormulaDefinition.objects.filter(
+        dataset=ds, user=request.user, normalized_name=normalized_name
+    ).first()
+    if not existing and FormulaDefinition.objects.filter(dataset=ds, user=request.user).count() >= 50:
+        raise APIError("每个数据集最多保存 50 个公式定义")
+    item, created = FormulaDefinition.objects.update_or_create(
+        dataset=ds,
+        user=request.user,
+        normalized_name=normalized_name,
+        defaults={"definition": definition},
+    )
+    audit(request, "formula_definition.save", item.pk, {"dataset": str(ds.pk), "name": definition["name"]})
     return JsonResponse(
         {
             "id": str(item.pk),
             "definition": item.definition,
-            "revision": item.revision,
             "updated_at": item.updated_at.isoformat(),
         },
         status=201 if created else 200,
@@ -272,9 +241,8 @@ def formula_definition_delete(request, dataset_id, definition_id):
     if not item:
         raise APIError("公式定义不存在", 404)
     name = item.definition.get("name", "")
-    with transaction.atomic():
-        item.delete()
-        audit(request, "formula_definition.delete", item.pk, {"dataset": str(ds.pk), "name": name})
+    audit(request, "formula_definition.delete", item.pk, {"dataset": str(ds.pk), "name": name})
+    item.delete()
     return JsonResponse({"deleted": True})
 
 
@@ -324,7 +292,7 @@ def _owned_job(request, dataset_id, export_id):
     return job
 
 
-@endpoint(("GET",), roles=("admin", "analyst"))
+@endpoint(("GET",))
 def formula_export_status(request, dataset_id, export_id):
     job = _owned_job(request, dataset_id, export_id)
     return JsonResponse(
@@ -337,7 +305,7 @@ def formula_export_status(request, dataset_id, export_id):
     )
 
 
-@endpoint(("GET",), roles=("admin", "analyst"))
+@endpoint(("GET",))
 def formula_export_download(request, dataset_id, export_id):
     job = _owned_job(request, dataset_id, export_id)
     if job.status != "succeeded" or not job.path or not job.expires_at or job.expires_at <= timezone.now():

finance/management/commands/cleanup_formula_exports.py

@@ -1,5 +1,4 @@
 import time
-import uuid
 from datetime import timedelta
 from pathlib import Path
 
@@ -10,11 +9,9 @@ from django.utils import timezone
 
 from finance.models import FormulaExport
 
-ORPHAN_RETENTION_HOURS = 24
-
 
 class Command(BaseCommand):
-    help = "Remove expired formula workbooks and reclaim orphan files"
+    help = "Remove expired formula workbooks and stale temporary files"
 
     def add_arguments(self, parser):
         parser.add_argument("--loop", action="store_true")
@@ -22,8 +19,8 @@ class Command(BaseCommand):
 
     def handle(self, *args, **options):
         while True:
-            jobs, files, orphans = self.clean_expired()
-            self.stdout.write(f"Expired {jobs} jobs; removed {files} files; reclaimed {orphans} orphan files")
+            jobs, files = self.clean_expired()
+            self.stdout.write(f"Expired {jobs} jobs; removed {files} files")
             if not options["loop"]:
                 return
             time.sleep(max(60, options["interval"]))
@@ -34,7 +31,7 @@ class Command(BaseCommand):
             raise ValueError("公式导出目录不能是符号链接")
         root = export_root.resolve()
         now = timezone.now()
-        jobs = files = orphans = 0
+        jobs = files = 0
 
         for job in FormulaExport.objects.filter(status="succeeded", expires_at__lte=now).iterator():
             path = Path(job.path) if job.path else None
@@ -64,41 +61,5 @@ class Command(BaseCommand):
                         files += 1
                     except OSError as exc:
                         self.stderr.write(f"Could not remove stale temporary export {path.name}: {exc}")
-                orphans += self._reclaim_orphan_workbooks(user_dir, now)
 
-        return jobs, files, orphans
-
-    def _reclaim_orphan_workbooks(self, user_dir, now):
-        """回收超过保留期且不再属于任何有效导出任务的 .xlsx 孤儿文件。
-
-        有效任务指状态 succeeded、路径一致且尚未过期的任务；失败任务残留的文件和任务行
-        已被删除（如数据集级联删除）的文件在保留期后回收，绝不触碰活动任务的文件。
-        """
-        retention = (now - timedelta(hours=ORPHAN_RETENTION_HOURS)).timestamp()
-        removed = 0
-        for path in user_dir.glob("*.xlsx"):
-            try:
-                if path.is_symlink() or not path.is_file() or path.stat().st_mtime > retention:
-                    continue
-                job_id = uuid.UUID(path.stem)
-            except (OSError, ValueError):
-                continue
-            job = FormulaExport.objects.filter(pk=job_id).first()
-            if (
-                job is not None
-                and job.status == "succeeded"
-                and job.path == str(path)
-                and job.expires_at is not None
-                and job.expires_at > now
-            ):
-                continue
-            try:
-                path.unlink()
-                removed += 1
-                if job is not None and job.path == str(path):
-                    job.path = ""
-                    job.save(update_fields=["path"])
-                    self.stdout.write(f"Reclaimed leftover workbook of job {job.pk}")
-            except OSError as exc:
-                self.stderr.write(f"Could not reclaim orphan export {path.stem}: {exc}")
-        return removed
+        return jobs, files

finance/management/commands/import_worker.py

@@ -1,19 +1,13 @@
 import time
-from datetime import timedelta
 
 from django.core.management.base import BaseCommand
 from django.db import connection, transaction
-from django.db.models import Q
 from django.utils import timezone
 
 from finance.exporting import process_formula_export
 from finance.importing import process_dataset
 from finance.models import Dataset, FormulaExport
 
-# 无咨询锁的数据库（如 SQLite）上等待该时限后才重新认领 processing 任务，
-# 避免与仍在运行的 worker 抢任务；PostgreSQL 由咨询锁保证单 worker，可立即恢复。
-STALE_PROCESSING_SECONDS = 15 * 60
-
 
 class Command(BaseCommand):
     help = "Process durable import and formula-export jobs"
@@ -36,35 +30,30 @@ class Command(BaseCommand):
                 with connection.cursor() as cursor:
                     cursor.execute("SELECT pg_advisory_unlock(782146291)")
 
-    def run_jobs(self, options, recover_immediately):
+    def run_jobs(self, options, recover):
         while True:
-            processed = self._next_import(recover_immediately)
+            processed = self._next_import(recover)
             if not processed:
-                processed = self._next_export(recover_immediately)
+                processed = self._next_export(recover)
             if options["once"]:
                 return
             if not processed:
                 time.sleep(2)
 
-    def _claim(self, model, immediate_recovery):
-        stale_before = timezone.now() - timedelta(seconds=STALE_PROCESSING_SECONDS)
-        claimable = Q(status="queued") | Q(status="processing", created_at__lt=stale_before)
-        if immediate_recovery:
-            claimable = Q(status="queued") | Q(status="processing")
+    def _claim(self, model, states):
         with transaction.atomic():
-            qs = model.objects.filter(claimable).order_by("created_at")
+            qs = model.objects.filter(status__in=states).order_by("created_at")
             if connection.features.has_select_for_update_skip_locked:
                 qs = qs.select_for_update(skip_locked=True)
             item = qs.first()
             if not item:
                 return None
-            # CAS 认领 queued 任务互斥；processing 仅在超时或咨询锁保护下重认领，
-            # 重复发布对同一目标路径写入相同内容，不会产生重复文件。
-            claimed = model.objects.filter(pk=item.pk, status=item.status).update(status="processing")
-            return item if claimed else None
+            updated = model.objects.filter(pk=item.pk, status__in=states).update(status="processing")
+            return item if updated else None
 
-    def _next_import(self, recover_immediately):
-        item = self._claim(Dataset, recover_immediately)
+    def _next_import(self, recover):
+        states = ["queued", "processing"] if recover else ["queued"]
+        item = self._claim(Dataset, states)
         if not item:
             return False
         Dataset.objects.filter(pk=item.pk).update(started_at=timezone.now())
@@ -72,8 +61,9 @@ class Command(BaseCommand):
         self.stdout.write(f"{result.pk}: {result.status}, {result.row_count} rows")
         return True
 
-    def _next_export(self, recover_immediately):
-        item = self._claim(FormulaExport, recover_immediately)
+    def _next_export(self, recover):
+        states = ["queued", "processing"] if recover else ["queued"]
+        item = self._claim(FormulaExport, states)
         if not item:
             return False
         try:

finance/migrations/0005_formuladefinition_revision.py

@@ -1,17 +0,0 @@
-# Generated by Django 5.2.17 on 2026-09-25 17:30
-
-from django.db import migrations, models
-
-
-class Migration(migrations.Migration):
-    dependencies = [
-        ("finance", "0004_formuladefinition"),
-    ]
-
-    operations = [
-        migrations.AddField(
-            model_name="formuladefinition",
-            name="revision",
-            field=models.PositiveIntegerField(default=1),
-        ),
-    ]

finance/models.py

@@ -121,7 +121,6 @@ class FormulaDefinition(models.Model):
     user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
     normalized_name = models.CharField(max_length=320)
     definition = models.JSONField()
-    revision = models.PositiveIntegerField(default=1)
     created_at = models.DateTimeField(auto_now_add=True)
     updated_at = models.DateTimeField(auto_now=True)
 

frontend/src/dialogs.tsx

@@ -1,12 +1,5 @@
 import { useEffect, useRef, useState } from "react";
-import {
-  ChevronLeft,
-  ChevronRight,
-  Download,
-  Save,
-  Trash2,
-  X,
-} from "lucide-react";
+import { ChevronLeft, ChevronRight, Download, Save, Trash2, X } from "lucide-react";
 import { api } from "./api";
 import {
   labels,
@@ -31,7 +24,6 @@ type FormulaExpression =
 type SavedFormulaDefinition = {
   id: string;
   definition: Record<string, unknown>;
-  revision: number;
   updated_at: string;
 };
 
@@ -42,19 +34,6 @@ function describeExpression(expression: FormulaExpression): string {
   return `(${describeExpression(expression.left)} ${operators[expression.operator]} ${describeExpression(expression.right)})`;
 }
 
-function definitionSummary(definition: Record<string, unknown>): string {
-  const text = [
-    `已载入已保存定义「${String(definition.name)}」`,
-    `含义：${String(definition.meaning)}`,
-    `适用范围：${String(definition.scope)}`,
-    `缺失/零分母处理：${String(definition.missing)}`,
-    `计算表达式：${JSON.stringify(definition.expression)}`,
-  ].join("；");
-  return text.length <= 2000
-    ? text
-    : `${text.slice(0, 2000)}…（内容过长已截断）`;
-}
-
 export function Modal({
   title,
   onClose,
@@ -370,8 +349,7 @@ export function EvidenceDialog({
     }>(),
     [side, setSide] = useState<"current" | "baseline">("current"),
     [page, setPage] = useState(0),
-    [error, setError] = useState(""),
-    [reloadKey, setReloadKey] = useState(0);
+    [error, setError] = useState("");
   useEffect(() => {
     let active = true;
     setResult(undefined);
@@ -386,14 +364,8 @@ export function EvidenceDialog({
     return () => {
       active = false;
     };
-  }, [ds.id, ctx, side, page, reloadKey]);
+  }, [ds.id, ctx, side, page]);
   function selectSide(value: "current" | "baseline") {
-    if (value === side) {
-      // 重复点击当前期间：仅在不在第一页时回到第一页触发刷新，
-      // 同期间同页直接返回，避免清空列表后停在“加载中”。
-      if (page !== 0) setPage(0);
-      return;
-    }
     setResult(undefined);
     setPage(0);
     setSide(value);
@@ -429,8 +401,7 @@ export function EvidenceDialog({
         {user.role !== "admin" && <p>客户原始标识不包含在明细中。</p>}
         {error && (
           <p role="alert" className="error">
-            {error}{" "}
-            <button onClick={() => setReloadKey((n) => n + 1)}>重试</button>
+            {error}
           </p>
         )}
         <div className="table-scroll">
@@ -464,11 +435,6 @@ export function EvidenceDialog({
                   <td>{String(r.gross ?? "")}</td>
                 </tr>
               ))}
-              {result && result.rows.length === 0 && (
-                <tr>
-                  <td colSpan={3 + ds.dimensions.length}>本期范围内没有数据</td>
-                </tr>
-              )}
             </tbody>
           </table>
         </div>
@@ -803,25 +769,14 @@ export function FormulaExportDialog({
     setBusy(true);
     setError("");
     try {
-      // 同名保存按服务端匹配规则覆盖旧版本；带上已知版本的 revision，
-      // 其他窗口先更新过则服务端返回 409，避免静默覆盖。
-      const normalizedName = (value: string) => value.trim().toLowerCase();
-      const existing = definitions.find(
-        (item) =>
-          normalizedName(String(item.definition.name)) ===
-          normalizedName(String(column.name)),
-      );
       const saved = await api<SavedFormulaDefinition>(
         `/datasets/${ds.id}/formula-definitions`,
         "POST",
-        { definition: column, revision: existing?.revision },
+        { definition: column },
       );
       setDefinitions((items) =>
         [...items.filter((item) => item.id !== saved.id), saved].sort((a, b) =>
-          String(a.definition.name).localeCompare(
-            String(b.definition.name),
-            "zh-CN",
-          ),
+          String(a.definition.name).localeCompare(String(b.definition.name), "zh-CN"),
         ),
       );
       setSelectedDefinition(saved.id);
@@ -834,25 +789,9 @@ export function FormulaExportDialog({
   function loadDefinition(id: string) {
     setSelectedDefinition(id);
     const saved = definitions.find((item) => item.id === id);
-    if (!saved) {
-      // 选择空选项时同步清理当前定义，避免下拉选择与预览不一致
-      setColumn(null);
-      setMessages([]);
-      setQuestions([]);
-      return;
-    }
+    if (!saved) return;
     setColumn(saved.definition);
-    // 载入定义后把原定义显式放进可见、经同意的对话上下文，
-    // 使“把刚才公式改成乘以 100”这类指代能关联到已载入表达式；
-    // 修改结果仍需重新确认，不会静默沿用旧定义。
-    setMessages([
-      { role: "user", content: definitionSummary(saved.definition) },
-      {
-        role: "assistant",
-        content:
-          "已载入该定义作为当前基准。请描述要调整的内容，我会在该定义基础上生成新的列定义，仍需你确认后再生成。",
-      },
-    ]);
+    setMessages([]);
     setQuestions([]);
     setDraft("");
   }
@@ -923,11 +862,6 @@ export function FormulaExportDialog({
             </button>
           </div>
         )}
-        {definitions.length > 0 && !task && (
-          <p className="muted">
-            保存同名定义会覆盖原版本；若已在其他窗口被修改，保存将提示刷新后重试。
-          </p>
-        )}
         <label className="check">
           <input
             type="checkbox"

scripts/dev_up.sh

@@ -74,10 +74,9 @@ fi
 echo "执行数据库迁移…"
 $PY manage.py migrate --no-input >/dev/null
 
-if [ ! -f var/local-access.txt ]; then
-  echo "初始化本地管理员…"
-  PYTHONPATH="$ROOT" $PY scripts/init_local.py
-fi
+# init_local.py 以数据库为准幂等执行：缺 admin 时创建并写凭据文件，已存在则跳过
+echo "初始化本地管理员（已存在则跳过）…"
+PYTHONPATH="$ROOT" $PY scripts/init_local.py
 
 if [ ! -d frontend/dist ]; then
   echo "构建前端（首次）…"

tests/e2e/formula-export.spec.ts

@@ -59,22 +59,11 @@ test("新增列：补充信息触发追问并重新确认后生成公式工作
                 meaning: "实际收入减成本",
                 scope: "所选分组与期间",
                 missing: "结果不可用（#N/A）；零分母不可用",
-                expression:
-                  designCalls === 5
-                    ? {
-                        operator: "multiply",
-                        left: {
-                          operator: "subtract",
-                          left: { metric: "revenue" },
-                          right: { metric: "cost" },
-                        },
-                        right: { number: 100 },
-                      }
-                    : {
-                        operator: "subtract",
-                        left: { metric: "revenue" },
-                        right: { metric: "cost" },
-                      },
+                expression: {
+                  operator: "subtract",
+                  left: { metric: "revenue" },
+                  right: { metric: "cost" },
+                },
               },
             },
     });
@@ -134,33 +123,11 @@ test("新增列：补充信息触发追问并重新确认后生成公式工作
   const reopened = page.getByRole("dialog", { name: "生成公式工作簿" });
   await reopened.getByLabel("已保存定义").selectOption({ label: "收入成本差" });
   await expect(reopened).toContainText("计算表达式：(实际收入 − 成本)");
-  // F04 回归：载入定义后，原定义作为可见上下文进入对话
-  await expect(reopened).toContainText("已载入已保存定义「收入成本差」");
-  await expect(reopened).toContainText("我：已载入已保存定义");
-  await expect(reopened).toContainText("助手：已载入该定义作为当前基准");
-  // F04 回归：选择空选项时同步清理当前定义与上下文，选择状态与预览一致
-  await reopened.getByLabel("已保存定义").selectOption({ label: "选择定义" });
-  await expect(reopened.getByText("计算表达式：(实际收入 − 成本)")).toHaveCount(
-    0,
-  );
-  await expect(reopened.getByText("已载入已保存定义")).toHaveCount(0);
-
-  await reopened.getByLabel("已保存定义").selectOption({ label: "收入成本差" });
-  await reopened.getByRole("checkbox").check();
-  await reopened.getByLabel("列计算描述").fill("把刚才公式改成乘以 100");
-  await reopened.getByRole("button", { name: "发送描述" }).click();
-  await expect(reopened).toContainText(
-    "计算表达式：((实际收入 − 成本) × 100)",
-  );
-  // 模型请求必须包含载入的原定义上下文和新的修改描述
-  expect(sentMessages.at(0)?.content).toContain(
-    "已载入已保存定义「收入成本差」",
-  );
-  expect(sentMessages.at(0)?.content).toContain('"operator":"subtract"');
-  expect(sentMessages.at(-1)?.content).toBe("把刚才公式改成乘以 100");
-  await reopened.getByRole("button", { name: "确认并生成公式工作簿" }).click();
+  await reopened
+    .getByRole("button", { name: "确认并生成公式工作簿" })
+    .click();
   await expect(reopened.getByRole("status")).toContainText("生成完成", {
     timeout: 30000,
   });
-  expect(designCalls).toBe(5);
+  expect(designCalls).toBe(4);
 });

tests/e2e/workspace.spec.ts

@@ -30,9 +30,7 @@ test("桌面：真实导入、异常、下钻、证据、设置、Copilot、导
   });
   await uploaded;
   const dialog = page.getByRole("dialog");
-  await expect(
-    dialog.getByRole("heading", { name: "确认导入字段" }),
-  ).toBeVisible();
+  await expect(dialog.getByRole("heading", { name: "确认导入字段" })).toBeVisible();
   await dialog.getByRole("button", { name: "确认并导入" }).click();
   await expect(
     page.getByRole("heading", { name: "集团经营总览", exact: true }),
@@ -87,27 +85,14 @@ test("桌面：真实导入、异常、下钻、证据、设置、Copilot、导
   await expect(evidenceDialog).toContainText("1 条");
   await expect(evidenceDialog).toContainText("2026-08");
   await expect(evidenceDialog).toContainText("CSV");
-  await evidenceDialog
-    .getByRole("button", { name: "基期", exact: true })
-    .click();
+  await evidenceDialog.getByRole("button", { name: "基期", exact: true }).click();
   await expect(evidenceDialog).toContainText("2025-08");
   await evidenceDialog.getByLabel("关闭", { exact: true }).click();
   const evidenceOffsets: number[] = [];
-  let evidenceFailOnce = false;
-  let evidenceEmpty = false;
   await page.route("**/api/datasets/*/evidence", async (route) => {
     const request = route.request().postDataJSON() as { offset?: number };
     const offset = request.offset || 0;
-    if (evidenceFailOnce) {
-      evidenceFailOnce = false;
-      await route.fulfill({ status: 500, json: { error: "服务暂不可用" } });
-      return;
-    }
     evidenceOffsets.push(offset);
-    if (evidenceEmpty) {
-      await route.fulfill({ json: { total: 0, sheet: "CSV", rows: [] } });
-      return;
-    }
     const count = offset === 0 ? 50 : 1;
     await route.fulfill({
       json: {
@@ -126,51 +111,10 @@ test("桌面：真实导入、异常、下钻、证据、设置、Copilot、导
   });
   await page.getByLabel("查看医生A证据", { exact: true }).click();
   await expect(evidenceDialog).toContainText("第 1-50 条，共 51 条");
-  await evidenceDialog
-    .getByRole("button", { name: "下一页", exact: true })
-    .click();
+  await evidenceDialog.getByRole("button", { name: "下一页", exact: true }).click();
   await expect(evidenceDialog).toContainText("第 51-51 条，共 51 条");
   await expect(evidenceDialog).toContainText("2026-09");
   expect(evidenceOffsets).toEqual([0, 50]);
-  // F03 回归：第二页时重复点击当前期间应回到第一页并重新加载
-  await evidenceDialog
-    .getByRole("button", { name: "本期", exact: true })
-    .click();
-  await expect(evidenceDialog).toContainText("第 1-50 条，共 51 条");
-  expect(evidenceOffsets).toEqual([0, 50, 0]);
-  // F03 回归：第一页重复点击当前期间不清空列表、不重新请求、不停在“加载中”
-  await evidenceDialog
-    .getByRole("button", { name: "本期", exact: true })
-    .click();
-  await expect(evidenceDialog).toContainText("第 1-50 条，共 51 条");
-  await expect(evidenceDialog).not.toContainText("加载中");
-  expect(evidenceOffsets).toEqual([0, 50, 0]);
-  // F03 回归：快速本期/基期切换最终停留在当前期间的数据
-  await evidenceDialog
-    .getByRole("button", { name: "基期", exact: true })
-    .click();
-  await evidenceDialog
-    .getByRole("button", { name: "本期", exact: true })
-    .click();
-  await expect(evidenceDialog).toContainText("第 1-50 条，共 51 条");
-  await expect(evidenceDialog).not.toContainText("加载中");
-  // F03 回归：请求失败显示错误并可重试恢复
-  evidenceFailOnce = true;
-  await evidenceDialog
-    .getByRole("button", { name: "下一页", exact: true })
-    .click();
-  await expect(evidenceDialog).toContainText("服务暂不可用");
-  await expect(evidenceDialog).toContainText("加载失败");
-  await evidenceDialog.getByRole("button", { name: "重试" }).click();
-  await expect(evidenceDialog).toContainText("第 51-51 条，共 51 条");
-  // F03 回归：空结果显示空态而不是过期数据或加载中
-  evidenceEmpty = true;
-  await evidenceDialog
-    .getByRole("button", { name: "上一页", exact: true })
-    .click();
-  await expect(evidenceDialog).toContainText("本期范围内没有数据");
-  await expect(evidenceDialog).toContainText("共 0 条");
-  await expect(evidenceDialog).not.toContainText("加载中");
   await evidenceDialog.getByLabel("关闭", { exact: true }).click();
   await page
     .getByRole("navigation", { name: "分析路径" })
@@ -220,53 +164,38 @@ test("桌面：真实导入、异常、下钻、证据、设置、Copilot、导
   await page.route("**/formula-design", async (route) => {
     designCalls++;
     await route.fulfill({
-      json:
-        designCalls === 1
-          ? {
-              complete: false,
-              questions: ["缺失值和零分母如何处理？"],
-              column: null,
-            }
-          : {
-              complete: true,
-              questions: [],
-              column: {
-                name: "收入成本差",
-                meaning: "实际收入减成本",
-                scope: "所选分组与期间",
-                missing: "结果不可用（#N/A）；零分母不可用",
-                expression: {
-                  operator: "subtract",
-                  left: { metric: "revenue" },
-                  right: { metric: "cost" },
-                },
+      json: designCalls === 1
+        ? { complete: false, questions: ["缺失值和零分母如何处理？"], column: null }
+        : {
+            complete: true,
+            questions: [],
+            column: {
+              name: "收入成本差",
+              meaning: "实际收入减成本",
+              scope: "所选分组与期间",
+              missing: "结果不可用（#N/A）；零分母不可用",
+              expression: {
+                operator: "subtract",
+                left: { metric: "revenue" },
+                right: { metric: "cost" },
               },
             },
+          },
     });
   });
   await page.getByRole("button", { name: "生成公式工作簿" }).click();
   const formulaDialog = page.getByRole("dialog", { name: "生成公式工作簿" });
   await formulaDialog.getByRole("checkbox").check();
-  await formulaDialog
-    .getByLabel("列计算描述")
-    .fill("新增收入成本差，缺失或零分母显示不可用");
+  await formulaDialog.getByLabel("列计算描述").fill("新增收入成本差，缺失或零分母显示不可用");
   await formulaDialog.getByRole("button", { name: "发送描述" }).click();
   await expect(formulaDialog).toContainText("缺失值和零分母如何处理？");
-  await formulaDialog
-    .getByLabel("列计算描述")
-    .fill("缺失项和零分母都按不可用处理");
+  await formulaDialog.getByLabel("列计算描述").fill("缺失项和零分母都按不可用处理");
   await formulaDialog.getByRole("button", { name: "发送描述" }).click();
-  await formulaDialog
-    .getByRole("button", { name: "确认并生成公式工作簿" })
-    .click();
-  await expect(formulaDialog.getByRole("status")).toContainText("生成完成", {
-    timeout: 30000,
-  });
+  await formulaDialog.getByRole("button", { name: "确认并生成公式工作簿" }).click();
+  await expect(formulaDialog.getByRole("status")).toContainText("生成完成", { timeout: 30000 });
   const formulaDownload = page.waitForEvent("download");
   await formulaDialog.getByRole("button", { name: "下载公式工作簿" }).click();
-  expect((await formulaDownload).suggestedFilename()).toBe(
-    "traceiq-formula-analysis.xlsx",
-  );
+  expect((await formulaDownload).suggestedFilename()).toBe("traceiq-formula-analysis.xlsx");
   await formulaDialog.getByLabel("关闭").click();
 });
 test("手机：登录、总览、设置、Copilot 布局无横向溢出", async ({ page }) => {

tests/test_assistant_api.py

@@ -168,21 +168,16 @@ def test_formula_definition_library_is_private_validated_and_audited(workspace):
     saved = client.post(url, data={"definition": definition}, content_type="application/json")
     assert saved.status_code == 201
     definition_id = saved.json()["id"]
-    assert saved.json()["revision"] == 1
     assert client.get(url).json()["items"][0]["definition"] == definition
 
     updated = client.post(
         url,
-        data={
-            "definition": {**definition, "name": "  收入成本差  ", "meaning": "更新说明"},
-            "revision": 1,
-        },
+        data={"definition": {**definition, "name": "  收入成本差  ", "meaning": "更新说明"}},
         content_type="application/json",
     )
     assert updated.status_code == 200
     assert FormulaDefinition.objects.filter(dataset=ds, user=ds.created_by).count() == 1
     assert updated.json()["definition"]["meaning"] == "更新说明"
-    assert updated.json()["revision"] == 2
 
     other = Client()
     other_user = ds.created_by.__class__.objects.create_user(

tests/test_formula_cleanup.py

@@ -1,5 +1,4 @@
 import os
-import uuid
 from datetime import timedelta
 
 import pytest
@@ -28,12 +27,6 @@ def cleanup_root(settings, tmp_path):
     return tmp_path / "formula-exports", dataset, user
 
 
-def _aged(path, hours=25):
-    stamp = (timezone.now() - timedelta(hours=hours)).timestamp()
-    os.utime(path, (stamp, stamp))
-    return path
-
-
 def test_cleanup_expires_jobs_and_removes_files(cleanup_root):
     root, dataset, user = cleanup_root
     path = root / str(user.pk) / "expired.xlsx"
@@ -62,7 +55,8 @@ def test_cleanup_removes_only_stale_temporary_files(cleanup_root):
     stale, recent = user_dir / "stale.tmp", user_dir / "recent.tmp"
     stale.touch()
     recent.touch()
-    _aged(stale)
+    old = timezone.now().timestamp() - 25 * 3600
+    os.utime(stale, (old, old))
 
     call_command("cleanup_formula_exports")
 
@@ -87,71 +81,3 @@ def test_cleanup_does_not_delete_files_outside_export_root(cleanup_root, tmp_pat
     job.refresh_from_db()
     assert job.status == "expired"
     assert outside.read_bytes() == b"keep"
-
-
-def test_cleanup_reclaims_failed_job_leftover_after_retention(cleanup_root):
-    root, dataset, user = cleanup_root
-    job = FormulaExport.objects.create(dataset=dataset, user=user, status="failed")
-    path = root / str(user.pk) / f"{job.pk}.xlsx"
-    path.parent.mkdir(parents=True)
-    path.write_bytes(b"leftover")
-    job.path = str(path)
-    job.save(update_fields=["path"])
-    _aged(path)
-
-    call_command("cleanup_formula_exports")
-
-    job.refresh_from_db()
-    assert not path.exists()
-    assert job.path == ""
-
-
-def test_cleanup_keeps_workbook_of_unexpired_succeeded_job(cleanup_root):
-    root, dataset, user = cleanup_root
-    job = FormulaExport.objects.create(
-        dataset=dataset,
-        user=user,
-        status="succeeded",
-        expires_at=timezone.now() + timedelta(hours=1),
-    )
-    path = root / str(user.pk) / f"{job.pk}.xlsx"
-    path.parent.mkdir(parents=True)
-    path.write_bytes(b"active")
-    job.path = str(path)
-    job.save(update_fields=["path"])
-    _aged(path)
-
-    call_command("cleanup_formula_exports")
-
-    assert path.read_bytes() == b"active"
-    job.refresh_from_db()
-    assert job.path == str(path)
-
-
-def test_cleanup_reclaims_orphan_whose_job_row_was_deleted(cleanup_root):
-    root, _, user = cleanup_root
-    user_dir = root / str(user.pk)
-    user_dir.mkdir(parents=True)
-    orphan = user_dir / f"{uuid.uuid4()}.xlsx"
-    orphan.write_bytes(b"orphan")
-    recent = user_dir / f"{uuid.uuid4()}.xlsx"
-    recent.write_bytes(b"recent orphan")
-    _aged(orphan)
-
-    call_command("cleanup_formula_exports")
-
-    assert not orphan.exists()
-    assert recent.exists()
-
-
-def test_cleanup_ignores_non_uuid_workbook_names(cleanup_root):
-    root, _, user = cleanup_root
-    user_dir = root / str(user.pk)
-    user_dir.mkdir(parents=True)
-    odd = user_dir / "notes.xlsx"
-    odd.write_bytes(b"keep")
-    _aged(odd)
-
-    call_command("cleanup_formula_exports")
-
-    assert odd.exists()

tests/test_formula_columns.py

@@ -97,11 +97,8 @@ def test_formula_workbook_exposes_custom_metric_as_recalculable_column(formula_d
     custom = workbook["自定义指标"]
     assert custom["B1"].value == "收入成本差（本期）"
     assert custom["B2"].value == f"='分析汇总'!C{metric_row}"
-    custom_note = next(
-        str(row[1]) for row in workbook["使用说明"].iter_rows(values_only=True) if row[0] == "收入成本差"
-    )
-    assert "含义：实际收入减成本" in custom_note
-    assert "范围：按所选分组及本期、基期计算" in custom_note
+    assert "含义：实际收入减成本" in workbook["使用说明"]["B12"].value
+    assert "范围：按所选分组及本期、基期计算" in workbook["使用说明"]["B12"].value
 
 
 def test_custom_expression_rejects_formula_text_and_unknown_metrics():
@@ -203,54 +200,6 @@ def test_design_column_preserves_clarification_branch_and_full_conversation(form
     assert payload["input"][1:] == messages
 
 
-def test_design_column_accepts_loaded_definition_context(formula_data, settings):
-    ds, _ = formula_data
-    settings.AI_ENABLED = True
-    settings.OPENAI_API_KEY = "test-key"
-    settings.OPENAI_MODEL = "test-model"
-    context = "已载入已保存定义「收入成本差」；计算表达式：" + '{"operator":"subtract",' * 45
-    assert 1000 < len(context) <= 2000
-    messages = [
-        {"role": "user", "content": context},
-        {"role": "assistant", "content": "已载入该定义作为当前基准。"},
-        {"role": "user", "content": "把刚才公式改成乘以 100"},
-    ]
-    clarification = {
-        "status": "completed",
-        "output": [
-            {
-                "content": [
-                    {
-                        "type": "output_text",
-                        "text": '{"complete":false,"questions":["要乘在整个表达式上吗？"],"column":null}',
-                    }
-                ]
-            }
-        ],
-    }
-    with patch("finance.formula_views.httpx.Client") as http:
-        http.return_value.__enter__.return_value.post.return_value = type(
-            "Response",
-            (),
-            {"raise_for_status": lambda self: None, "json": lambda self: clarification},
-        )()
-        result = design_column(
-            messages,
-            ds,
-            {"current": ["2026-08"], "baseline": ["2025-08"], "filters": {}, "group": "department"},
-        )
-    assert result["complete"] is False
-    payload = http.return_value.__enter__.return_value.post.call_args.kwargs["json"]
-    assert payload["input"][1:] == messages
-
-    with pytest.raises(Exception, match="对话消息内容无效"):
-        design_column(
-            [{"role": "user", "content": "y" * 2001}],
-            ds,
-            {"current": ["2026-08"], "baseline": ["2025-08"], "filters": {}, "group": "department"},
-        )
-
-
 def test_design_column_returns_only_validated_formula_definition(formula_data, settings):
     ds, _ = formula_data
     settings.AI_ENABLED = True

tests/test_formula_definitions_concurrency.py

@@ -1,123 +0,0 @@
-from unittest.mock import patch
-
-import pytest
-from django.db import connection
-
-from finance.models import FormulaDefinition
-
-pytestmark = pytest.mark.django_db
-
-
-def column_definition(name="收入成本差"):
-    return {
-        "name": name,
-        "meaning": "实际收入减成本",
-        "scope": "所选分组与期间",
-        "missing": "结果不可用（#N/A）；零分母不可用",
-        "expression": {
-            "operator": "subtract",
-            "left": {"metric": "revenue"},
-            "right": {"metric": "cost"},
-        },
-    }
-
-
-def _post(client, url, definition, revision=None):
-    return client.post(
-        url,
-        data={"definition": definition, **({"revision": revision} if revision is not None else {})},
-        content_type="application/json",
-    )
-
-
-def test_same_name_update_requires_matching_revision(workspace):
-    ds, _, client, *_ = workspace
-    url = f"/api/datasets/{ds.pk}/formula-definitions"
-
-    saved = _post(client, url, column_definition())
-    assert saved.status_code == 201
-    item_id, revision = saved.json()["id"], saved.json()["revision"]
-
-    # 缺失或过期的 revision 都不能静默覆盖其他窗口的修改
-    assert _post(client, url, column_definition()).status_code == 409
-    assert _post(client, url, column_definition(), revision=999).status_code == 409
-
-    updated = _post(client, url, column_definition(), revision=revision)
-    assert updated.status_code == 200
-    assert updated.json()["revision"] == revision + 1
-
-    # 基于旧版本的再次更新必须被拒绝，两个标签页不互相静默丢失
-    stale = _post(client, url, column_definition(), revision=revision)
-    assert stale.status_code == 409
-    current = FormulaDefinition.objects.get(pk=item_id)
-    assert current.revision == revision + 1
-    # 列表接口暴露当前 revision 供前端携带
-    listed = client.get(url).json()["items"]
-    assert listed[0]["revision"] == revision + 1
-
-
-def test_definition_quota_is_enforced_and_release_allows_new_save(workspace):
-    ds, _, client, *_ = workspace
-    url = f"/api/datasets/{ds.pk}/formula-definitions"
-    for index in range(50):
-        FormulaDefinition.objects.create(
-            dataset=ds,
-            user=ds.created_by,
-            normalized_name=f"定义{index:02d}",
-            definition=column_definition(f"定义{index:02d}"),
-        )
-
-    rejected = _post(client, url, column_definition("第 51 个不同名定义"))
-    assert rejected.status_code == 400
-    assert "50" in rejected.json()["error"]
-    assert FormulaDefinition.objects.filter(dataset=ds).count() == 50
-
-    # 同名更新不受配额限制；删除后名额释放可再保存新名称
-    assert _post(client, url, column_definition("定义00"), revision=1).status_code == 200
-    FormulaDefinition.objects.filter(normalized_name="定义00").delete()
-    assert _post(client, url, column_definition("第 51 个不同名定义")).status_code == 201
-
-
-def test_audit_failure_rolls_back_definition_change(workspace):
-    ds, _, client, *_ = workspace
-    url = f"/api/datasets/{ds.pk}/formula-definitions"
-
-    with patch("finance.formula_views.audit", side_effect=RuntimeError("audit unavailable")):
-        with pytest.raises(RuntimeError, match="audit unavailable"):
-            _post(client, url, column_definition())
-
-    assert FormulaDefinition.objects.filter(dataset=ds).count() == 0
-
-
-def test_concurrent_distinct_saves_cannot_exceed_quota_on_postgres(workspace):
-    if connection.vendor != "postgresql":
-        pytest.skip("并发配额依赖 PostgreSQL 行锁语义，生产库单独验证")
-    import threading
-
-    from django.test import Client
-
-    ds, member, *_ = workspace
-    user = ds.created_by
-    for index in range(49):
-        FormulaDefinition.objects.create(
-            dataset=ds,
-            user=user,
-            normalized_name=f"定义{index:02d}",
-            definition=column_definition(f"定义{index:02d}"),
-        )
-
-    barrier = threading.Barrier(4)
-
-    def save(index):
-        client = Client()
-        assert client.login(username=user.username, password="Finance-TestPassword991!")
-        barrier.wait()
-        _post(client, f"/api/datasets/{ds.pk}/formula-definitions", column_definition(f"并发{index}"))
-
-    threads = [threading.Thread(target=save, args=(index,)) for index in range(4)]
-    for thread in threads:
-        thread.start()
-    for thread in threads:
-        thread.join()
-
-    assert FormulaDefinition.objects.filter(dataset=ds).count() == 50

tests/test_formula_export_reliability.py

@@ -1,4 +1,3 @@
-import os
 import stat
 from decimal import Decimal
 from io import BytesIO
@@ -199,170 +198,6 @@ def test_formula_export_worker_records_path_preparation_failure(workspace, setti
     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
 

tests/test_formula_reconcile_guard.py

@@ -20,12 +20,11 @@ def test_reconcile_status_guards_non_numeric_difference(workspace):
         output,
     )
     workbook = load_workbook(BytesIO(output.getvalue()), data_only=False)
-    status_cells = [row[7] for row in workbook["对账检查"].iter_rows(min_row=2)]
     formulas = [
-        cell.value for cell in status_cells if isinstance(cell.value, str) and cell.value.startswith("=IF(D")
+        row[6].value
+        for row in workbook["对账检查"].iter_rows(min_row=2)
+        if isinstance(row[6].value, str) and row[6].value.startswith("=IF(D")
     ]
     assert formulas
     assert all("ISNUMBER(F" in formula for formula in formulas)
     assert all('"通过"' in formula for formula in formulas)
-    # 容差列独立成列，状态按容差比较而不是统一 0.01
-    assert all(row[6].value is not None for row in workbook["对账检查"].iter_rows(min_row=2))

tests/test_formula_reconcile_tolerances.py

@@ -1,236 +0,0 @@
-import os
-import shutil
-import subprocess
-from decimal import Decimal
-from io import BytesIO
-
-import pytest
-from openpyxl import load_workbook
-
-from finance.exporting import CUSTOM_COLUMN_TOLERANCE, RECONCILE_TOLERANCES, build_formula_workbook
-from finance.formula_columns import to_decimal
-
-pytestmark = pytest.mark.django_db
-
-
-def custom_column(name="收入成本差"):
-    return {
-        "name": name,
-        "meaning": "实际收入减成本",
-        "scope": "按所选分组及本期、基期计算",
-        "missing": "结果不可用（#N/A）；零分母不可用",
-        "expression": {
-            "operator": "subtract",
-            "left": {"metric": "revenue"},
-            "right": {"metric": "cost"},
-        },
-    }
-
-
-def test_to_decimal_evaluator_golden_samples():
-    values = {
-        "revenue": Decimal("999999999999.99"),
-        "cost": Decimal("0.0000001"),
-        "customers": Decimal("7"),
-    }
-    # 大数与多层表达式：((收入 - 收入) + 成本×2) ÷ (成本×2) = 1，抵消后不丢失精度
-    cancellation = {
-        "operator": "divide",
-        "left": {
-            "operator": "add",
-            "left": {
-                "operator": "subtract",
-                "left": {"metric": "revenue"},
-                "right": {"metric": "revenue"},
-            },
-            "right": {"operator": "multiply", "left": {"metric": "cost"}, "right": {"number": 2}},
-        },
-        "right": {"operator": "multiply", "left": {"metric": "cost"}, "right": {"number": 2}},
-    }
-    assert to_decimal(cancellation, values) == Decimal("1")
-    # 极小值与比率
-    assert to_decimal({"metric": "cost"}, values) == Decimal("0.0000001")
-    assert to_decimal(
-        {"operator": "divide", "left": {"metric": "revenue"}, "right": {"metric": "customers"}}, values
-    ) == Decimal("999999999999.99") / Decimal("7")
-    # 除零与缺失指标返回 None（对应 Excel #N/A）
-    assert (
-        to_decimal({"operator": "divide", "left": {"metric": "revenue"}, "right": {"number": 0}}, values)
-        is None
-    )
-    assert (
-        to_decimal(
-            {"operator": "add", "left": {"metric": "cost"}, "right": {"metric": "customers"}},
-            {**values, "customers": None},
-        )
-        is None
-    )
-    # 负数与常数组合
-    assert to_decimal(
-        {"operator": "subtract", "left": {"number": -2.5}, "right": {"metric": "cost"}}, values
-    ) == Decimal("-2.5") - Decimal("0.0000001")
-
-
-def test_reconcile_uses_independent_tolerances_per_metric_kind(workspace):
-    ds, member, *_ = workspace
-    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)
-    sheet = workbook["对账检查"]
-    rows = list(sheet.iter_rows(values_only=True))
-    assert list(rows[0]) == [
-        "指标",
-        "分组",
-        "期间侧",
-        "服务器 Decimal 参考",
-        "公式结果",
-        "差异",
-        "容差",
-        "状态",
-    ]
-
-    def find(metric, side="本期"):
-        return next(row for row in rows[1:] if row[0] == metric and row[2] == side)
-
-    assert find("消费人数")[6] == 0  # 人数严格相等
-    assert find("实际收入")[6] == pytest.approx(0.01)
-    assert find("毛利贡献")[6] == pytest.approx(0.01)
-    assert find("毛利率")[6] == pytest.approx(0.0001)
-    assert find("ROI")[6] == pytest.approx(0.0001)
-    assert find("客单价")[6] == pytest.approx(0.0001)
-    # 状态公式按容差列比较（<=G），不再统一 <0.01
-    status = next(
-        cell.value for cell in sheet[2] if isinstance(cell.value, str) and cell.value.startswith("=IF(D")
-    )
-    assert "<=G2" in status
-    # 参数页写明对账规则，容差调整需财务批准
-    parameters = {str(row[0]): row for row in workbook["参数与口径"].iter_rows(values_only=True) if row[0]}
-    assert "人数严格相等" in parameters["对账容差"][1]
-    assert parameters["对账容差"][3] == "否"
-
-
-def test_custom_columns_reconcile_against_decimal_reference(workspace):
-    ds, member, *_ = workspace
-    output = BytesIO()
-    build_formula_workbook(
-        ds,
-        member,
-        {
-            "start": "2026-08",
-            "end": "2026-08",
-            "comparison": "yoy",
-            "group": "department",
-            "custom_columns": [custom_column()],
-        },
-        output,
-    )
-    workbook = load_workbook(BytesIO(output.getvalue()), data_only=False)
-    rows = list(workbook["对账检查"].iter_rows(values_only=True))
-    custom_rows = [row for row in rows[1:] if row[0] == "收入成本差"]
-    assert custom_rows
-    pediatrics = next(row for row in custom_rows if row[1] == "儿科" and row[2] == "本期")
-    assert Decimal(pediatrics[3]) == Decimal("60")  # 100 - 40
-    assert pediatrics[6] == pytest.approx(float(CUSTOM_COLUMN_TOLERANCE))
-    status = next(
-        row[7]
-        for row in workbook["对账检查"].iter_rows(min_row=2, values_only=True)
-        if row[0] == "收入成本差" and row[1] == "儿科" and row[2] == "本期"
-    )
-    assert "<=G" in status and "通过" in status
-    # 旧的占位文案不再出现
-    assert all("自定义公式无服务器参考" not in str(row[3]) for row in custom_rows)
-    assert all(row[7] != "需业务验收" for row in custom_rows)
-
-
-def test_custom_column_reference_missing_when_dependency_unavailable(workspace):
-    from finance.models import Fact
-
-    ds, member, *_ = workspace
-    # 外科 2026-08 成本置空 → cost 不可用 → 收入成本差参考为 None → 无参考
-    Fact.objects.filter(dataset=ds, period="2026-08", dimensions__department="外科").update(cost=None)
-    output = BytesIO()
-    build_formula_workbook(
-        ds,
-        member,
-        {
-            "start": "2026-08",
-            "end": "2026-08",
-            "comparison": "yoy",
-            "group": "department",
-            "custom_columns": [custom_column()],
-        },
-        output,
-    )
-    workbook = load_workbook(BytesIO(output.getvalue()), data_only=False)
-    surgery = next(
-        row
-        for row in workbook["对账检查"].iter_rows(min_row=2, values_only=True)
-        if row[0] == "收入成本差" and row[1] == "外科" and row[2] == "本期"
-    )
-    assert surgery[3] in ("", None)  # 无服务器参考
-    assert "无参考" in surgery[7]
-
-
-def test_reconcile_tolerances_cover_every_metric():
-    from finance.analysis import METRICS
-
-    assert set(RECONCILE_TOLERANCES) == set(METRICS)
-    assert RECONCILE_TOLERANCES["customers"] == 0
-    assert all(
-        tolerance in (Decimal("0.01"), Decimal("0.0001"))
-        for key, tolerance in RECONCILE_TOLERANCES.items()
-        if key != "customers"
-    )
-
-
-def test_reconcile_passes_in_libreoffice_with_per_metric_tolerances(workspace, tmp_path):
-    executable = os.environ.get("SOFFICE_BIN") or shutil.which("soffice")
-    if not executable:
-        pytest.skip("set SOFFICE_BIN to run the LibreOffice recalculation check")
-    ds, member, *_ = workspace
-    source, output = tmp_path / "source", tmp_path / "out"
-    source.mkdir()
-    output.mkdir()
-    workbook_path = source / "reconcile.xlsx"
-    with workbook_path.open("wb") as file:
-        build_formula_workbook(
-            ds,
-            member,
-            {
-                "start": "2026-08",
-                "end": "2026-08",
-                "comparison": "yoy",
-                "group": "department",
-                "custom_columns": [custom_column()],
-            },
-            file,
-        )
-    profile = tmp_path / "profile"
-    result = subprocess.run(
-        [
-            executable,
-            "--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)
-    statuses = [
-        (row[0], row[1], row[2], row[7]) for row in values["对账检查"].iter_rows(min_row=2, values_only=True)
-    ]
-    assert statuses
-    assert all(status in ("通过", "无参考") for *_head, status in statuses), statuses

tests/test_formula_workbook_docs.py

@@ -1,155 +0,0 @@
-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_worker_paths.py

@@ -1,10 +1,7 @@
-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
 
@@ -33,108 +30,3 @@ 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_not_reclaimed(workspace):
-    ds, *_ = workspace
-    job = _export_job(ds, status="processing")
-
-    call_command("import_worker", once=True)
-
-    job.refresh_from_db()
-    assert job.status == "processing"