finance/exporting.py

@@ -1,5 +1,6 @@
 import hashlib
 import hmac
+import logging
 import os
 import secrets
 import warnings
@@ -11,22 +12,62 @@ from django.conf import settings
 from django.utils import timezone
 from openpyxl import Workbook
 from openpyxl.cell import WriteOnlyCell
-from openpyxl.styles import Font, PatternFill
+from openpyxl.styles import Font, PatternFill, Protection
+from openpyxl.worksheet.datavalidation import DataValidation
 from openpyxl.worksheet.table import Table, TableColumn, TableStyleInfo
 
 from .analysis import METRICS, aggregations, context, dimensions, metrics, scoped_facts
 from .formula_columns import (
     RESERVED_DATA_HEADERS,
     ensure_available_metrics,
+    evaluate,
     metrics_used,
     to_excel,
     validate_column,
 )
 from .models import FormulaExport, Membership
 
+# Keep the private test seam while sharing the single evaluator implementation.
+_evaluate_expression = evaluate
+
 FORMULA_VERSION = "3"
 MAX_ROWS = 100_000
 MAX_GROUPS = 200
+RECONCILE_TOLERANCES = {
+    "customers": Decimal("0"),
+    "margin": Decimal("0.0001"),
+    "discount_rate": Decimal("0.0001"),
+    "upgrade_rate": Decimal("0.0001"),
+}
+DEFAULT_RECONCILE_TOLERANCE = Decimal("0.01")
+logger = logging.getLogger(__name__)
+
+
+def _safe_unlink_export(path, root):
+    """Delete only a file below the owned export root without following links."""
+    path, root = Path(path), Path(root).absolute()
+    if not path.is_absolute() or root.is_symlink() or path.is_symlink():
+        logger.warning("formula export cleanup refused unsafe path: %s", path)
+        return False
+    parent = path.parent
+    try:
+        if not parent.is_relative_to(root):
+            logger.warning("formula export cleanup refused out-of-root path: %s", path)
+            return False
+    except ValueError:
+        return False
+    current = parent
+    while current != root:
+        if current.is_symlink() or current.parent == current:
+            logger.warning("formula export cleanup refused linked parent: %s", path)
+            return False
+        current = current.parent
+    try:
+        path.unlink(missing_ok=True)
+        return True
+    except OSError:
+        logger.warning("formula export cleanup failed for %s", path, exc_info=True)
+        return False
 
 
 def _excel_number(value):
@@ -269,6 +310,19 @@ def build_formula_workbook(ds, member, export_context, output):
             ),
         )
     data.append(headers)
+    editable_indices = {len(dims) + 2, len(dims) + 3, len(dims) + 4, len(dims) + 5, len(dims) + 7}
+    if count:
+        numeric_input = DataValidation(
+            type="decimal", operator="between", formula1="-1e15", formula2="1e15", allow_blank=True
+        )
+        numeric_input.showErrorMessage = True
+        numeric_input.errorStyle = "stop"
+        numeric_input.errorTitle = "请输入数值"
+        numeric_input.error = "计算输入只接受范围内的数值或空白；请勿粘贴公式或文本。"
+        for index in editable_indices:
+            column_name = _column_name(index + 1)
+            numeric_input.add(f"{column_name}2:{column_name}{count + 1}")
+        data.data_validations.append(numeric_input)
     token_secret = secrets.token_bytes(32)
     customer_roster = {}
     gross_index = next((col["index"] for col in mapping["columns"] if col["role"] == "gross"), None)
@@ -329,6 +383,12 @@ def build_formula_workbook(ds, member, export_context, output):
                 f'{_column_name(len(dims) + 6)}{row_number},""))'
             ),
         )
+        for index in editable_indices:
+            cell = row[index]
+            if not hasattr(cell, "protection"):
+                cell = WriteOnlyCell(data, value=cell)
+            cell.protection = Protection(locked=False)
+            row[index] = cell
         data.append(row)
     customer_headers = [
         "Group",
@@ -408,7 +468,13 @@ def build_formula_workbook(ds, member, export_context, output):
         ],
         ["ROI 口径", mapping["roi_formula"], "按导入映射保存的公式分支", "否"],
         ["自营产品集合", _text_cell(parameters, ", ".join(products)), "升单率和自营收入口径", "否"],
-        ["产品毛利率", str(rates), "直接毛利缺失时的可编辑计算输入", "是"],
+        ["产品毛利率快照", str(rates), "导入时保存的只读口径；修改需重新确认导入映射", "否"],
+        [
+            "对账容差",
+            "人数 0；金额 0.01；比率 0.0001；自定义 0.01",
+            "服务器 Decimal 参考与 Excel 结果的绝对差；正式财务验收前需确认",
+            "否",
+        ],
         [
             "Excel 数值精度",
             "IEEE-754 双精度",
@@ -443,6 +509,10 @@ def build_formula_workbook(ds, member, export_context, output):
         ["变化率", "差额/ABS(基期)；基期为0或期间缺失时不可用。"],
         ["精度", "原始 Decimal 以文本留存；Excel 双精度存在细微误差可能。"],
         ["隐私", "客户标识替换为本文件随机 token，文件仍按敏感财务信息保护。"],
+        [
+            "编辑保护",
+            "工作表结构、人数去重、原始证据和公式列已保护；计算明细仅收入、成本、毛利输入、毛利率和原价金额可编辑。行结构变化请重新导出。",
+        ],
     ]:
         instructions.append(row)
     for definition in custom_columns:
@@ -559,6 +629,7 @@ def build_formula_workbook(ds, member, export_context, output):
             for key, name in METRICS.items():
                 formula_row = metric_rows[(label, key)]
                 value = reference[key]
+                tolerance = RECONCILE_TOLERANCES.get(key, DEFAULT_RECONCILE_TOLERANCE)
                 reconcile.append(
                     [
                         name,
@@ -567,24 +638,30 @@ 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},"")',
-                        f'=IF(D{reconcile_row}="","无参考",IF(ISNUMBER(F{reconcile_row}),IF(ABS(F{reconcile_row})<0.01,"通过","待复核"),"待复核"))',
+                        f'=IF(D{reconcile_row}="","无参考",IF(ISNUMBER(F{reconcile_row}),IF(ABS(F{reconcile_row})<={tolerance},"通过","待复核"),"待复核"))',
                     ]
                 )
                 reconcile_row += 1
             for index, definition in enumerate(custom_columns):
                 formula_row = metric_rows[(label, f"custom_{index}")]
+                custom_reference = evaluate(definition["expression"], reference)
                 reconcile.append(
                     [
                         _text_cell(reconcile, definition["name"]),
                         _text_cell(reconcile, label),
                         side_name,
-                        "自定义公式无服务器参考",
+                        "" if custom_reference is None else str(custom_reference),
                         f"='分析汇总'!{col}{formula_row}",
-                        "",
-                        "需业务验收",
+                        f'=IFERROR(E{reconcile_row}-D{reconcile_row},"")',
+                        f'=IF(D{reconcile_row}="","无参考",IF(ISNUMBER(F{reconcile_row}),IF(ABS(F{reconcile_row})<={DEFAULT_RECONCILE_TOLERANCE},"通过","待复核"),"待复核"))',
                     ]
                 )
                 reconcile_row += 1
+    for sheet in wb.worksheets:
+        sheet.protection.sheet = True
+        sheet.protection.selectLockedCells = False
+        sheet.protection.selectUnlockedCells = True
+    wb.security.lockStructure = True
     wb.calculation.fullCalcOnLoad = True
     wb.calculation.forceFullCalc = True
     wb.calculation.calcMode = "auto"
@@ -596,8 +673,8 @@ def process_formula_export(export_id):
     job = FormulaExport.objects.select_related("dataset", "user").get(pk=export_id)
     temporary = None
     target = None
+    root = Path(settings.MEDIA_ROOT) / "formula-exports"
     try:
-        root = Path(settings.MEDIA_ROOT) / "formula-exports"
         if root.is_symlink():
             raise ValueError("公式导出目录不能是符号链接")
         root.mkdir(parents=True, exist_ok=True, mode=0o700)
@@ -609,6 +686,8 @@ 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 set(member.departments) != set(job.departments):
             raise PermissionError("科室授权已变更，请重新提交导出")
         descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
@@ -621,13 +700,12 @@ def process_formula_export(export_id):
         job.save(update_fields=["path", "status", "error", "finished_at", "expires_at"])
         return stats
     except Exception as exc:
+        remaining = ""
         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], ""
+            if path is not None and not _safe_unlink_export(path, root):
+                if path.exists() or path.is_symlink():
+                    remaining = str(path)
+        job.status, job.error, job.path = "failed", str(exc)[:500], remaining
         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,4 +1,5 @@
 import math
+from decimal import Decimal, localcontext
 
 from .analysis import METRICS
 
@@ -107,3 +108,27 @@ 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 evaluate(node, references):
+    """Evaluate a validated expression with Decimal values and #N/A semantics."""
+    with localcontext() as decimal_context:
+        decimal_context.prec = 64
+        return _evaluate(node, references)
+
+
+def _evaluate(node, references):
+    if "metric" in node:
+        return references.get(node["metric"])
+    if "number" in node:
+        return Decimal(str(node["number"]))
+    left, right = _evaluate(node["left"], references), _evaluate(node["right"], references)
+    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
+    return None if right == 0 else left / right

finance/formula_views.py

@@ -3,12 +3,14 @@ from pathlib import Path
 
 import httpx
 from django.conf import settings
+from django.db import transaction
+from django.db.models import Q
 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
+from .models import Dataset, FormulaDefinition, FormulaExport
 from .views import APIError, audit, body, dataset, endpoint, limit
 
 
@@ -190,13 +192,23 @@ def formula_definitions(request, dataset_id):
     if request.member.departments:
         raise APIError("科室授权账号不可管理跨范围公式定义", 403)
     if request.method == "GET":
-        items = FormulaDefinition.objects.filter(dataset=ds, user=request.user).order_by("normalized_name")
+        items = (
+            FormulaDefinition.objects.filter(dataset=ds)
+            .filter(Q(user=request.user) | Q(shared=True))
+            .select_related("user")
+            .order_by("normalized_name")
+        )
         return JsonResponse(
             {
                 "items": [
                     {
                         "id": str(item.pk),
                         "definition": item.definition,
+                        "revision": item.revision,
+                        "shared": item.shared,
+                        "owned": item.user_id == request.user.pk,
+                        "owner": item.user.username,
+                        "history": item.history,
                         "updated_at": item.updated_at.isoformat(),
                     }
                     for item in items
@@ -204,28 +216,74 @@ def formula_definitions(request, dataset_id):
             }
         )
 
-    definition = validate_column(body(request).get("definition"))
+    data = body(request)
+    definition = validate_column(data.get("definition"))
+    shared = data.get("shared", False)
+    if type(shared) is not bool:
+        raise APIError("共享选项无效")
     try:
         ensure_available_metrics(ds, definition)
     except ValueError as exc:
         raise APIError(str(exc)) from exc
     normalized_name = definition["name"].strip().casefold()
-    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"]})
+    requested_revision = data.get("revision")
+    with transaction.atomic():
+        Dataset.objects.select_for_update().get(pk=ds.pk)
+        existing = (
+            FormulaDefinition.objects.select_for_update()
+            .filter(dataset=ds, user=request.user, normalized_name=normalized_name)
+            .first()
+        )
+        if existing:
+            if requested_revision is not None and (
+                type(requested_revision) is not int or requested_revision != existing.revision
+            ):
+                raise APIError("公式定义已在其他窗口更新，请重新载入", 409)
+            # ponytail: retain 20 revisions; use a separate history table if long-term retention is required.
+            existing.history = [
+                *existing.history,
+                {
+                    "revision": existing.revision,
+                    "definition": existing.definition,
+                    "updated_at": existing.updated_at.isoformat(),
+                },
+            ][-20:]
+            existing.definition = definition
+            existing.shared = shared if "shared" in data else existing.shared
+            existing.revision += 1
+            existing.save(update_fields=["definition", "revision", "shared", "history", "updated_at"])
+            item, created = existing, False
+        else:
+            if FormulaDefinition.objects.filter(dataset=ds, user=request.user).count() >= 50:
+                raise APIError("每个数据集最多保存 50 个公式定义")
+            item = FormulaDefinition.objects.create(
+                dataset=ds,
+                user=request.user,
+                normalized_name=normalized_name,
+                definition=definition,
+                shared=shared,
+            )
+            created = True
+        audit(
+            request,
+            "formula_definition.save",
+            item.pk,
+            {
+                "dataset": str(ds.pk),
+                "name": definition["name"],
+                "shared": item.shared,
+                "revision": item.revision,
+            },
+        )
     return JsonResponse(
         {
             "id": str(item.pk),
             "definition": item.definition,
+            "revision": item.revision,
+            "shared": item.shared,
+            "owned": True,
+            "owner": request.user.username,
+            "history": item.history,
             "updated_at": item.updated_at.isoformat(),
         },
         status=201 if created else 200,
@@ -241,8 +299,10 @@ def formula_definition_delete(request, dataset_id, definition_id):
     if not item:
         raise APIError("公式定义不存在", 404)
     name = item.definition.get("name", "")
-    audit(request, "formula_definition.delete", item.pk, {"dataset": str(ds.pk), "name": name})
-    item.delete()
+    with transaction.atomic():
+        item = FormulaDefinition.objects.select_for_update().get(pk=item.pk)
+        audit(request, "formula_definition.delete", item.pk, {"dataset": str(ds.pk), "name": name})
+        item.delete()
     return JsonResponse({"deleted": True})
 
 
@@ -279,6 +339,8 @@ def formula_export_create(request, dataset_id):
 
 
 def _owned_job(request, dataset_id, export_id):
+    if not request.user.is_active or request.member.role not in ("admin", "analyst"):
+        raise APIError("没有此导出任务的访问权限", 403)
     job = FormulaExport.objects.filter(
         pk=export_id,
         dataset_id=dataset_id,

finance/management/commands/cleanup_formula_exports.py

@@ -5,8 +5,10 @@ from pathlib import Path
 from django.conf import settings
 from django.core.management.base import BaseCommand
 from django.db import transaction
+from django.db.models import Q
 from django.utils import timezone
 
+from finance.exporting import _safe_unlink_export
 from finance.models import FormulaExport
 
 
@@ -29,22 +31,30 @@ class Command(BaseCommand):
         export_root = Path(settings.MEDIA_ROOT) / "formula-exports"
         if export_root.is_symlink():
             raise ValueError("公式导出目录不能是符号链接")
-        root = export_root.resolve()
+        root = export_root.absolute()
         now = timezone.now()
         jobs = files = 0
 
-        for job in FormulaExport.objects.filter(status="succeeded", expires_at__lte=now).iterator():
+        jobs_qs = FormulaExport.objects.filter(
+            Q(status="succeeded", expires_at__lte=now) | Q(status="failed", path__gt="")
+        ).iterator()
+        for job in jobs_qs:
             path = Path(job.path) if job.path else None
-            resolved = path.resolve() if path and path.is_absolute() and not path.is_symlink() else None
-            in_root = bool(resolved and resolved.is_relative_to(root))
             try:
                 with transaction.atomic():
-                    if in_root and resolved.is_file():
-                        resolved.unlink()
+                    deleted = bool(path and _safe_unlink_export(path, root))
+                    if job.status == "succeeded":
+                        if deleted or not path or not path.is_absolute() or not path.is_relative_to(root):
+                            if deleted:
+                                files += 1
+                            job.status, job.path = "expired", ""
+                            job.save(update_fields=["status", "path"])
+                            jobs += 1
+                    elif deleted:
                         files += 1
-                    job.status, job.path = "expired", ""
-                    job.save(update_fields=["status", "path"])
-                    jobs += 1
+                        job.path = ""
+                        job.save(update_fields=["path"])
+                        jobs += 1
             except OSError as exc:
                 self.stderr.write(f"Could not remove expired export {job.pk}: {exc}")
 

finance/management/commands/import_worker.py

@@ -24,7 +24,7 @@ class Command(BaseCommand):
                     self.stdout.write("Another worker owns the database lock")
                     return
         try:
-            self.run_jobs(options, postgres)
+            self.run_jobs(options, True)
         finally:
             if postgres:
                 with connection.cursor() as cursor:

finance/migrations/0006_formuladefinition_revision.py

@@ -0,0 +1,15 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("finance", "0005_aiconfig"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="formuladefinition",
+            name="revision",
+            field=models.PositiveIntegerField(default=1),
+        ),
+    ]

finance/migrations/0007_alter_dataset_file.py

@@ -0,0 +1,15 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("finance", "0006_formuladefinition_revision"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="dataset",
+            name="file",
+            field=models.FileField(max_length=500, upload_to="sources/%Y/%m"),
+        ),
+    ]

finance/migrations/0008_formuladefinition_history_formuladefinition_shared.py

@@ -0,0 +1,22 @@
+# Generated by Django 5.2.17 on 2026-10-07 23:50
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("finance", "0007_alter_dataset_file"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="formuladefinition",
+            name="history",
+            field=models.JSONField(default=list),
+        ),
+        migrations.AddField(
+            model_name="formuladefinition",
+            name="shared",
+            field=models.BooleanField(default=False),
+        ),
+    ]

finance/models.py

@@ -22,7 +22,7 @@ class Dataset(models.Model):
     organization = models.ForeignKey(Organization, on_delete=models.PROTECT)
     created_by = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.PROTECT)
     name = models.CharField(max_length=200)
-    file = models.FileField(upload_to="sources/%Y/%m")
+    file = models.FileField(upload_to="sources/%Y/%m", max_length=500)
     sha256 = models.CharField(max_length=64)
     status = models.CharField(max_length=16, default="mapping")
     profile = models.JSONField(default=dict)
@@ -133,6 +133,9 @@ 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)
+    shared = models.BooleanField(default=False)
+    history = models.JSONField(default=list)
     created_at = models.DateTimeField(auto_now_add=True)
     updated_at = models.DateTimeField(auto_now=True)
 

frontend/src/dialogs.tsx

@@ -6,6 +6,9 @@ import {
   roleLabels,
   type AIConfig,
   type Context,
+  type FormulaColumn,
+  type FormulaExpression,
+  type SavedFormulaDefinition,
   type Dataset,
   type Metric,
   type Preferences,
@@ -14,21 +17,6 @@ import {
 
 type User = NonNullable<Session["user"]>;
 
-type FormulaExpression =
-  | { metric: Metric }
-  | { number: number }
-  | {
-      operator: "add" | "subtract" | "multiply" | "divide";
-      left: FormulaExpression;
-      right: FormulaExpression;
-    };
-
-type SavedFormulaDefinition = {
-  id: string;
-  definition: Record<string, unknown>;
-  updated_at: string;
-};
-
 function describeExpression(expression: FormulaExpression): string {
   if ("metric" in expression) return labels[expression.metric];
   if ("number" in expression) return String(expression.number);
@@ -827,7 +815,7 @@ export function FormulaExportDialog({
       { role: "user" | "assistant"; content: string }[]
     >([]),
     [draft, setDraft] = useState(""),
-    [column, setColumn] = useState<Record<string, unknown> | null>(null),
+    [column, setColumn] = useState<FormulaColumn | null>(null),
     [questions, setQuestions] = useState<string[]>([]),
     [task, setTask] = useState<{
       id: string;
@@ -836,9 +824,30 @@ export function FormulaExportDialog({
     }>(),
     [definitions, setDefinitions] = useState<SavedFormulaDefinition[]>([]),
     [selectedDefinition, setSelectedDefinition] = useState(""),
+    [shared, setShared] = useState(false),
+    [loadedDefinition, setLoadedDefinition] = useState<FormulaColumn | null>(null),
     [consent, setConsent] = useState(false),
     [busy, setBusy] = useState(false),
     [error, setError] = useState("");
+  const draftKey = `traceiq.formula-draft.${ds.id}`;
+  const [draftRestored, setDraftRestored] = useState(false);
+  useEffect(() => {
+    try {
+      const stored = sessionStorage.getItem(draftKey);
+      if (stored) {
+        const saved = JSON.parse(stored);
+        setDraft(saved.draft || "");
+        setMessages(saved.messages || []);
+        setColumn(saved.column || null);
+      }
+    } catch {
+      sessionStorage.removeItem(draftKey);
+    }
+    setDraftRestored(true);
+  }, [draftKey]);
+  useEffect(() => {
+    if (draftRestored) sessionStorage.setItem(draftKey, JSON.stringify({ draft, messages, column }));
+  }, [draftKey, draftRestored, draft, messages, column]);
   useEffect(() => {
     api<{ items: SavedFormulaDefinition[] }>(
       `/datasets/${ds.id}/formula-definitions`,
@@ -865,7 +874,10 @@ export function FormulaExportDialog({
   async function send() {
     const text = draft.trim();
     if (!text || !consent || userTurns >= 6) return;
-    const next = [...messages, { role: "user" as const, content: text }];
+    const content = loadedDefinition
+      ? `已载入的原始定义：${JSON.stringify(loadedDefinition)}\n本次补充：${text}`
+      : text;
+    const next = [...messages, { role: "user" as const, content }];
     setBusy(true);
     setError("");
     setQuestions([]);
@@ -874,7 +886,7 @@ export function FormulaExportDialog({
       const result = await api<{
         complete: boolean;
         questions: string[];
-        column: Record<string, unknown> | null;
+        column: FormulaColumn | null;
       }>(`/datasets/${ds.id}/formula-design`, "POST", {
         context: ctx,
         messages: next,
@@ -891,6 +903,7 @@ export function FormulaExportDialog({
       ]);
       setQuestions(result.questions);
       setColumn(result.column);
+      setLoadedDefinition(null);
       setDraft("");
     } catch (e) {
       setError((e as Error).message);
@@ -899,7 +912,7 @@ export function FormulaExportDialog({
     }
   }
   async function create() {
-    if (!column) return;
+    if (!column && !ctx.custom_columns?.length) return;
     setBusy(true);
     setError("");
     try {
@@ -907,7 +920,12 @@ export function FormulaExportDialog({
         await api<{ id: string; status: string }>(
           `/datasets/${ds.id}/formula-exports`,
           "POST",
-          { ...ctx, custom_columns: [column] },
+          {
+            ...ctx,
+            custom_columns: column
+              ? [...(ctx.custom_columns || []).filter((item) => item.name !== column.name), column]
+              : ctx.custom_columns,
+          },
         ),
       );
     } catch (e) {
@@ -924,11 +942,24 @@ export function FormulaExportDialog({
       const saved = await api<SavedFormulaDefinition>(
         `/datasets/${ds.id}/formula-definitions`,
         "POST",
-        { definition: column },
+        {
+          definition: column,
+          shared,
+          revision: definitions.find(
+            (item) =>
+              (item.id === selectedDefinition && item.owned) ||
+              (item.owned &&
+              String(item.definition.name).trim().toLowerCase() ===
+                String(column.name).trim().toLowerCase()),
+          )?.revision,
+        },
       );
       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);
@@ -941,8 +972,14 @@ export function FormulaExportDialog({
   function loadDefinition(id: string) {
     setSelectedDefinition(id);
     const saved = definitions.find((item) => item.id === id);
-    if (!saved) return;
+    if (!saved) {
+      setLoadedDefinition(null);
+      setColumn(null);
+      return;
+    }
     setColumn(saved.definition);
+    setShared(saved.owned && saved.shared);
+    setLoadedDefinition(saved.definition);
     setMessages([]);
     setQuestions([]);
     setDraft("");
@@ -960,6 +997,7 @@ export function FormulaExportDialog({
         items.filter((item) => item.id !== selectedDefinition),
       );
       setSelectedDefinition("");
+      setLoadedDefinition(null);
       setColumn(null);
     } catch (e) {
       setError((e as Error).message);
@@ -969,15 +1007,10 @@ export function FormulaExportDialog({
   }
   async function getFile() {
     try {
-      const blob = await api<Blob>(
-          `/datasets/${ds.id}/formula-exports/${task!.id}/download`,
-        ),
-        url = URL.createObjectURL(blob),
-        link = document.createElement("a");
-      link.href = url;
+      const link = document.createElement("a");
+      link.href = `/api/datasets/${ds.id}/formula-exports/${task!.id}/download`;
       link.download = "traceiq-formula-analysis.xlsx";
       link.click();
-      window.setTimeout(() => URL.revokeObjectURL(url), 1000);
     } catch (e) {
       setError((e as Error).message);
     }
@@ -986,6 +1019,11 @@ export function FormulaExportDialog({
     <Modal title="生成公式工作簿" onClose={onClose} wide>
       <div className="dialog-body">
         <p>描述新列含义、计算方式、适用范围，以及缺失值或零分母处理方式。</p>
+        {!column && !task && !!ctx.custom_columns?.length && (
+          <button disabled={busy} onClick={create}>
+            按网页已选 {ctx.custom_columns.length} 列生成工作簿
+          </button>
+        )}
         {definitions.length > 0 && !task && (
           <div className="form-row">
             <label>
@@ -998,7 +1036,7 @@ export function FormulaExportDialog({
                 <option value="">选择定义</option>
                 {definitions.map((item) => (
                   <option key={item.id} value={item.id}>
-                    {String(item.definition.name)}
+                    {item.definition.name}{item.owned ? "" : `（${item.owner}共享）`}
                   </option>
                 ))}
               </select>
@@ -1007,13 +1045,27 @@ export function FormulaExportDialog({
               className="icon"
               aria-label="删除所选定义"
               title="删除所选定义"
-              disabled={busy || !selectedDefinition}
+              disabled={busy || !definitions.find((item) => item.id === selectedDefinition)?.owned}
               onClick={deleteDefinition}
             >
               <Trash2 />
             </button>
           </div>
         )}
+        {definitions.find((item) => item.id === selectedDefinition)?.history.length ? (
+          <details>
+            <summary>版本历史（最近 20 次，可载入后另存）</summary>
+            {definitions.find((item) => item.id === selectedDefinition)!.history.map((version) => (
+              <button key={version.revision} disabled={busy} onClick={() => {
+                setColumn(version.definition);
+                setLoadedDefinition(version.definition);
+                setMessages([]);
+              }}>
+                载入版本 {version.revision} · {version.definition.meaning}
+              </button>
+            ))}
+          </details>
+        ) : null}
         <label className="check">
           <input
             type="checkbox"
@@ -1056,6 +1108,10 @@ export function FormulaExportDialog({
                 </p>
                 <p>适用范围：{String(column.scope)}</p>
                 <p>缺失/零分母：{String(column.missing)}</p>
+                <label className="check">
+                  <input type="checkbox" checked={shared} onChange={(event) => setShared(event.target.checked)} />
+                  保存时共享给当前数据集的其他全范围分析账号
+                </label>
                 <button
                   disabled={busy || Boolean(draft.trim())}
                   onClick={saveDefinition}

frontend/src/types.ts

@@ -98,6 +98,32 @@ export interface Context {
   filters: Record<string, string>;
   group: string;
   metric: Metric;
+  custom_columns?: FormulaColumn[];
+}
+export type FormulaExpression =
+  | { metric: Metric }
+  | { number: number }
+  | {
+      operator: "add" | "subtract" | "multiply" | "divide";
+      left: FormulaExpression;
+      right: FormulaExpression;
+    };
+export interface FormulaColumn {
+  name: string;
+  meaning: string;
+  scope: string;
+  missing: string;
+  expression: FormulaExpression;
+}
+export interface SavedFormulaDefinition {
+  id: string;
+  definition: FormulaColumn;
+  revision: number;
+  updated_at: string;
+  shared: boolean;
+  owned: boolean;
+  owner: string;
+  history: { revision: number; definition: FormulaColumn; updated_at: string }[];
 }
 export interface Value {
   current: Numeric;

tests/e2e/formula-export.spec.ts

@@ -119,15 +119,36 @@ test("新增列：补充信息触发追问并重新确认后生成公式工作
   );
   await dialog.getByLabel("关闭", { exact: true }).click();
 
+  await page.getByRole("button", { name: "选择网页自定义列" }).click();
+  const columns = page.getByRole("dialog", { name: "选择网页自定义列" });
+  await columns.getByRole("checkbox", { name: /收入成本差/ }).check();
+  const customQuery = page.waitForResponse(
+    (response) =>
+      response.url().includes("/query") && response.request().method() === "POST",
+  );
+  await columns.getByRole("button", { name: "应用", exact: true }).click();
+  const customResult = await (await customQuery).json();
+  expect(customResult.summary.custom_0.current).toBe("1000");
+  expect(customResult.rows[0].values.custom_0.current).toBe("1000");
+
   await page.getByRole("button", { name: "生成公式工作簿" }).click();
   const reopened = page.getByRole("dialog", { name: "生成公式工作簿" });
   await reopened.getByLabel("已保存定义").selectOption({ label: "收入成本差" });
   await expect(reopened).toContainText("计算表达式：(实际收入 − 成本)");
+  await reopened.getByLabel("列计算描述").fill("把刚才公式乘以 100");
+  await expect(
+    reopened.getByRole("button", { name: "先发送补充内容" }),
+  ).toBeDisabled();
   await reopened
-    .getByRole("button", { name: "确认并生成公式工作簿" })
-    .click();
+    .getByLabel("同意将描述、指标定义和当前筛选范围发送至已配置的外部 AI 服务")
+    .check();
+  await reopened.getByRole("button", { name: "发送描述" }).click();
+  expect(sentMessages[0].content).toContain('"name":"收入成本差"');
+  expect(sentMessages[0].content).toContain('"operator":"subtract"');
+  expect(sentMessages[0].content).toContain("把刚才公式乘以 100");
+  await reopened.getByRole("button", { name: "确认并生成公式工作簿" }).click();
   await expect(reopened.getByRole("status")).toContainText("生成完成", {
     timeout: 30000,
   });
-  expect(designCalls).toBe(4);
+  expect(designCalls).toBe(5);
 });

tests/test_assistant_api.py

@@ -331,19 +331,34 @@ def test_formula_definition_library_is_private_validated_and_audited(workspace):
             "right": {"metric": "cost"},
         },
     }
-    saved = client.post(url, data={"definition": definition}, content_type="application/json")
+    saved = client.post(url, data={"definition": definition, "shared": True}, content_type="application/json")
     assert saved.status_code == 201
+    assert saved.json()["shared"] is True
     definition_id = saved.json()["id"]
     assert client.get(url).json()["items"][0]["definition"] == definition
 
     updated = client.post(
         url,
-        data={"definition": {**definition, "name": "  收入成本差  ", "meaning": "更新说明"}},
+        data={
+            "definition": {**definition, "name": "  收入成本差  ", "meaning": "更新说明"},
+            "shared": True,
+            "revision": saved.json()["revision"],
+        },
         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
+    assert len(updated.json()["history"]) == 1
+
+    stale = client.post(
+        url,
+        data={"definition": {**definition, "meaning": "旧标签页覆盖"}, "revision": 1},
+        content_type="application/json",
+    )
+    assert stale.status_code == 409
+    assert FormulaDefinition.objects.get(pk=definition_id).definition["meaning"] == "更新说明"
 
     other = Client()
     other_user = ds.created_by.__class__.objects.create_user(
@@ -351,7 +366,8 @@ def test_formula_definition_library_is_private_validated_and_audited(workspace):
     )
     Membership.objects.create(user=other_user, organization=ds.organization, role="analyst")
     assert other.login(username=other_user.username, password="Formula-Library-TestPassword991!")
-    assert other.get(url).json()["items"] == []
+    shared = other.get(url).json()["items"]
+    assert len(shared) == 1 and shared[0]["owned"] is False and shared[0]["shared"] is True
     assert other.delete(f"{url}/{definition_id}").status_code == 404
 
     deleted = client.delete(f"{url}/{definition_id}")
@@ -385,3 +401,26 @@ def test_formula_definition_library_rejects_unavailable_metric(workspace):
     )
     assert response.status_code == 400
     assert "成本" in response.json()["error"]
+
+
+def test_formula_definition_audit_failure_rolls_back_definition(workspace, monkeypatch):
+    ds, _, client, *_ = workspace
+    definition = {
+        "name": "审计回滚",
+        "meaning": "实际收入",
+        "scope": "所选分组与期间",
+        "missing": "结果不可用（#N/A）；零分母不可用",
+        "expression": {"metric": "revenue"},
+    }
+
+    def fail(*_args, **_kwargs):
+        raise RuntimeError("audit unavailable")
+
+    monkeypatch.setattr("finance.formula_views.audit", fail)
+    with pytest.raises(RuntimeError, match="audit unavailable"):
+        client.post(
+            f"/api/datasets/{ds.pk}/formula-definitions",
+            data={"definition": definition},
+            content_type="application/json",
+        )
+    assert not FormulaDefinition.objects.filter(dataset=ds, user=ds.created_by).exists()

tests/test_formula_columns.py

@@ -97,8 +97,30 @@ 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}"
-    assert "含义：实际收入减成本" in workbook["使用说明"]["B12"].value
-    assert "范围：按所选分组及本期、基期计算" in workbook["使用说明"]["B12"].value
+    instruction_text = " ".join(str(cell.value) for row in workbook["使用说明"].iter_rows() for cell in row)
+    assert "含义：实际收入减成本" in instruction_text
+    assert "范围：按所选分组及本期、基期计算" in instruction_text
+
+
+def test_formula_workbook_protects_structure_and_unlocks_calculation_inputs(formula_data):
+    ds, member = formula_data
+    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)
+    data = workbook["计算明细"]
+    assert data.protection.sheet
+    assert data["D2"].protection.locked is False
+    assert data["F2"].protection.locked is False
+    assert data["G2"].protection.locked is False
+    assert data["I2"].protection.locked is False
+    assert data["C2"].protection.locked is True
+    assert data["M2"].protection.locked is True
+    assert workbook["原始证据"].protection.sheet
 
 
 def test_custom_expression_rejects_formula_text_and_unknown_metrics():

tests/test_formula_export_reliability.py

@@ -198,6 +198,67 @@ def test_formula_export_worker_records_path_preparation_failure(workspace, setti
     assert job.status == "failed"
 
 
+def test_formula_export_failure_does_not_follow_a_symlinked_user_directory(workspace, settings):
+    from finance.exporting import process_formula_export
+
+    ds, _, client, *_ = workspace
+    response = client.post(
+        f"/api/datasets/{ds.pk}/formula-exports",
+        data={"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
+        content_type="application/json",
+    )
+    job = FormulaExport.objects.get(pk=response.json()["id"])
+    root = Path(settings.MEDIA_ROOT) / "formula-exports"
+    protected = Path(settings.MEDIA_ROOT) / "protected"
+    protected.mkdir()
+    target = protected / f"{job.pk}.xlsx"
+    target.write_bytes(b"keep")
+    root.mkdir()
+    (root / str(job.user_id)).symlink_to(protected, target_is_directory=True)
+
+    with pytest.raises(ValueError, match="符号链接"):
+        process_formula_export(job.pk)
+
+    job.refresh_from_db()
+    assert job.status == "failed"
+    assert target.read_bytes() == b"keep"
+
+
+def test_formula_export_failed_cleanup_is_observable_and_reclaimed(workspace, monkeypatch):
+    from django.core.management import call_command
+
+    from finance.exporting import process_formula_export
+
+    ds, _, client, *_ = workspace
+    response = client.post(
+        f"/api/datasets/{ds.pk}/formula-exports",
+        data={"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "department"},
+        content_type="application/json",
+    )
+    job = FormulaExport.objects.get(pk=response.json()["id"])
+    target = Path(job.dataset.file.storage.location) / "formula-exports" / str(job.user_id) / f"{job.pk}.xlsx"
+    target.parent.mkdir(parents=True)
+    target.write_bytes(b"orphan")
+    monkeypatch.setattr(
+        "finance.exporting.build_formula_workbook",
+        lambda *_a, **_k: (_ for _ in ()).throw(ValueError("retry failed")),
+    )
+    real_unlink = Path.unlink
+    monkeypatch.setattr(Path, "unlink", lambda *_a, **_k: (_ for _ in ()).throw(OSError("read-only")))
+
+    with pytest.raises(ValueError, match="retry failed"):
+        process_formula_export(job.pk)
+    job.refresh_from_db()
+    assert job.status == "failed"
+    assert job.path == str(target)
+
+    monkeypatch.setattr(Path, "unlink", real_unlink)
+    call_command("cleanup_formula_exports")
+    job.refresh_from_db()
+    assert job.path == ""
+    assert target.exists() is False
+
+
 def test_formula_workbook_rejects_unclassified_label_collision(workspace):
     from decimal import Decimal
 
@@ -374,3 +435,20 @@ def test_formula_workbook_rejects_dimension_keys_that_collide_with_reserved_head
             {"start": "2026-08", "end": "2026-08", "comparison": "yoy", "group": "revenue"},
             BytesIO(),
         )
+
+
+def test_custom_reconcile_reference_preserves_large_decimal_and_zero_division():
+    from finance.exporting import _evaluate_expression
+
+    expression = {
+        "operator": "divide",
+        "left": {"metric": "revenue"},
+        "right": {"number": 0},
+    }
+    assert (
+        _evaluate_expression(expression, {"revenue": Decimal("1.2345678901234567890123456789012345")}) is None
+    )
+    assert _evaluate_expression(
+        {"operator": "add", "left": {"metric": "revenue"}, "right": {"number": 1}},
+        {"revenue": Decimal("1234567890123456789012345678901234")},
+    ) == Decimal("1234567890123456789012345678901235")

tests/test_operations.py

@@ -46,6 +46,23 @@ def test_worker_processes_queued_import_and_formula_export(workspace):
     assert job.status == "processing"
 
 
+def test_worker_reclaims_processing_formula_export_after_restart(workspace):
+    ds, _, _, *_ = workspace
+    job = FormulaExport.objects.create(
+        dataset=ds,
+        user=ds.created_by,
+        context={"custom_columns": []},
+        status="processing",
+    )
+    from unittest.mock import patch
+
+    with patch(
+        "finance.management.commands.import_worker.process_formula_export", return_value={"rows": 5}
+    ) as process:
+        call_command("import_worker", once=True)
+    process.assert_called_once_with(job.pk)
+
+
 def test_formula_export_status_download_and_access_controls(workspace, settings, tmp_path):
     ds, _, client, _, other_client = workspace
     url = f"/api/datasets/{ds.pk}"
@@ -58,7 +75,7 @@ def test_formula_export_status_download_and_access_controls(workspace, settings,
     job.status = "succeeded"
     job.expires_at = timezone.now() + timedelta(hours=1)
     target = tmp_path / "formula-exports" / str(ds.created_by_id) / f"{job.pk}.xlsx"
-    target.parent.mkdir(parents=True)
+    target.parent.mkdir(parents=True, exist_ok=True)
     target.write_bytes(b"xlsx")
     job.path = str(target)
     job.save()
@@ -85,6 +102,29 @@ def test_formula_export_status_download_and_access_controls(workspace, settings,
     assert client.get(status_url).status_code == 403
 
 
+def test_formula_export_rechecks_role_when_worker_runs_and_file_is_downloaded(workspace, settings, tmp_path):
+    from finance.exporting import process_formula_export
+
+    ds, member, client, _, _ = workspace
+    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="撤权"):
+        process_formula_export(job.pk)
+    job.refresh_from_db()
+    assert job.status == "failed"
+
+    job.status = "succeeded"
+    job.expires_at = timezone.now() + timedelta(hours=1)
+    target = tmp_path / "formula-exports" / str(ds.created_by_id) / f"{job.pk}.xlsx"
+    target.parent.mkdir(parents=True, exist_ok=True)
+    target.write_bytes(b"xlsx")
+    job.path = str(target)
+    job.save(update_fields=["status", "expires_at", "path"])
+    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_formula_export_create_validates_definition_and_records_job(workspace):
     ds, _, client, *_ = workspace
     url = f"/api/datasets/{ds.pk}/formula-exports"