import hashlib import hmac import logging import os import secrets import warnings from datetime import timedelta from decimal import Decimal, InvalidOperation from pathlib import Path 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, 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): if value is None: return None number = float(value) if not Decimal(str(number)).is_finite(): raise ValueError("金额超出 Excel 数值范围，公式工作簿已拒绝生成") return number def _text_cell(sheet, value): cell = WriteOnlyCell(sheet, value="" if value is None else str(value)) cell.data_type = "s" return cell def _styled_row(sheet, values, fill, font): cells = [] for value in values: cell = value if hasattr(value, "data_type") else WriteOnlyCell(sheet, value=value) cell.fill, cell.font = fill, font cells.append(cell) return cells def _table(name, ref, headers, style): table = Table(displayName=name, ref=ref) table.tableColumns = [TableColumn(id=index, name=header) for index, header in enumerate(headers, 1)] table.tableStyleInfo = TableStyleInfo(name=style, showRowStripes=True, showColumnStripes=False) return table def _add_table(sheet, table): with warnings.catch_warnings(): warnings.filterwarnings("ignore", message="In write-only mode you must add table columns manually") sheet.add_table(table) def _quote(value): escaped = str(value).replace("~", "~~").replace("*", "~*").replace("?", "~?").replace('"', '""') return '"' + escaped + '"' def _text_criterion(value): criterion = "=" + str(value) escaped = criterion.replace("~", "~~").replace("*", "~*").replace("?", "~?") if len(escaped) > 255: raise ValueError("分组或筛选值超过 Excel criteria 的 255 字符上限，请缩短标签") return _quote(criterion) def _checked_formula(value): if len(value) > 8192: raise ValueError("生成公式超过 Excel 8,192 字符上限，请减少筛选条件或缩短标签") return value def _criteria(filters, group_key, label, periods): terms = [] for key, value in sorted(filters.items()): terms.extend((f"Data[{key}]", _text_criterion(value))) terms.extend((f"Data[{group_key}]", _text_criterion(label))) terms.extend(("Data[Period]", _quote(">=" + periods[0]), "Data[Period]", _quote("<=" + periods[-1]))) return ",".join(terms) def _sum(column, criteria): return f"SUMIFS(Data[{column}],{criteria})" def _column_name(index): name = "" while index: index, remainder = divmod(index - 1, 26) name = chr(65 + remainder) + name return name def _metric_formula(key, label, periods, side_col, metric_rows, filters, group_key, products, roi_formula): criteria = _criteria(filters, group_key, label, periods) rows = f"COUNTIFS({criteria})" revenue, cost = _sum("Revenue", criteria), _sum("Cost", criteria) gross, list_sum = _sum("GrossCalc", criteria), _sum("ListAmount", criteria) cost_n = f'COUNTIFS({criteria},Data[Cost],"<>")' gross_n = f'COUNTIFS({criteria},Data[GrossCalc],"<>")' list_n = f'COUNTIFS({criteria},Data[ListAmount],"<>")' customer_period = "Current" if side_col == "C" else "Baseline" customer_upgrade = f"{customer_period}Upgraded" customers = ( f"COUNTIFS(CustomerRoster[Group],{_text_criterion(label)},CustomerRoster[{customer_period}],1)" ) if key == "revenue": return f"=IF({rows}=0,0,{revenue})" if key == "cost": return f"=IF({rows}=0,0,IF({cost_n}={rows},{cost},NA()))" if key == "gross": return f"=IF({rows}=0,0,IF({gross_n}={rows},{gross},NA()))" if key == "customers": return f"=IF({rows}=0,NA(),IF(COUNTIFS({criteria},Data[CustomerPresent],0)>0,NA(),{customers}))" if key == "basket": ref = f"{side_col}{metric_rows[(label, 'customers')]}" return f"=IF(OR({rows}=0,ISNA({ref}),{ref}=0),NA(),{revenue}/{ref})" if key == "margin": return f"=IF(OR({rows}=0,{revenue}<=0,{gross_n}<>{rows}),NA(),{gross}/{revenue})" if key == "discount_rate": return f"=IF(OR({rows}=0,{list_n}<>{rows},{list_sum}<=0),NA(),{revenue}/{list_sum})" if key == "self_revenue": return ( "=NA()" if not products else f"=IF({rows}=0,0,SUMIFS(Data[Revenue],{criteria},Data[SelfOperated],1))" ) if key == "upgrade_rate": if not products: return "=NA()" ref = f"{side_col}{metric_rows[(label, 'customers')]}" upgraded = ( f"COUNTIFS(CustomerRoster[Group],{_text_criterion(label)},CustomerRoster[{customer_upgrade}],1)" ) return f"=IF(OR(ISNA({ref}),{ref}<=0),NA(),{upgraded}/{ref})" if key == "list_basket": ref = f"{side_col}{metric_rows[(label, 'customers')]}" return f"=IF(OR({rows}=0,{list_n}<>{rows},ISNA({ref}),{ref}<=0),NA(),{list_sum}/{ref})" numerator = revenue if roi_formula == "profit_over_cost" else gross complete = "FALSE" if roi_formula == "profit_over_cost" else f"{gross_n}<>{rows}" return f"=IF(OR({rows}=0,{cost_n}<>{rows},{cost}<=0,{complete}),NA(),({numerator}-{cost})/{cost})" def build_formula_workbook(ds, member, export_context, output): custom_columns = [validate_column(item) for item in export_context.get("custom_columns", [])] for item in custom_columns: ensure_available_metrics(ds, item) if len(custom_columns) > 10: raise ValueError("每次最多添加 10 个自定义指标列") ctx = context(ds, export_context) qs = ( scoped_facts(ds, member, ctx["filters"]) .filter(period__in=ctx["current"] + ctx["baseline"]) .order_by("source_row") ) count = qs.count() if count > MAX_ROWS: raise ValueError(f"公式工作簿最多支持 {MAX_ROWS:,} 行，请缩小范围") group_key = ctx["group"] groups = list( qs.order_by().values_list(f"dimensions__{group_key}", flat=True).distinct()[: MAX_GROUPS + 1] ) group_values = {} excel_labels = {} for value in groups: label = str(value or "（未分类）") if label in group_values and group_values[label] != value: raise ValueError("分组存在与系统缺失值标签冲突的真实值，公式工作簿已拒绝生成") folded = label.casefold() if folded in excel_labels and excel_labels[folded] != label: raise ValueError("分组名称仅大小写不同，Excel 无法可靠区分，公式工作簿已拒绝生成") group_values[label] = value excel_labels[folded] = label if len(groups) > MAX_GROUPS: raise ValueError(f"所选维度超过 {MAX_GROUPS} 个分组，请缩小范围") labels = sorted(group_values) mapping = ds.mapping products, rates = mapping.get("self_operated_products", []), mapping.get("margin_rates", {}) if len(products) > 100 or len(rates) > 2000: raise ValueError("产品口径超过公式工作簿的支持上限") dims = dimensions(ds) dim_headers = [d["key"] for d in dims] dim_names = {d["key"]: d["name"] for d in dims} if RESERVED_DATA_HEADERS.intersection(key.casefold() for key in dim_headers): raise ValueError("维度标识与公式工作簿保留列名冲突") wb = Workbook(write_only=True) summary = wb.create_sheet("分析汇总") instructions = wb.create_sheet("使用说明") parameters = wb.create_sheet("参数与口径") data = wb.create_sheet("计算明细") customers_sheet = wb.create_sheet("人数去重") evidence = wb.create_sheet("原始证据") reconcile = wb.create_sheet("对账检查") custom_sheet = wb.create_sheet("自定义指标") if custom_columns else None green, white = PatternFill("solid", fgColor="17664F"), Font(color="FFFFFF", bold=True) summary.freeze_panes = "A2" summary.sheet_view.showGridLines = False for sheet in (instructions, parameters, reconcile): sheet.freeze_panes = "A2" sheet.sheet_view.showGridLines = False data.freeze_panes = "A2" customers_sheet.freeze_panes = "A2" customers_sheet.sheet_view.showGridLines = False if custom_sheet: custom_sheet.freeze_panes = "B2" custom_sheet.sheet_view.showGridLines = False custom_sheet.auto_filter.ref = f"A1:{_column_name(1 + 4 * len(custom_columns))}{len(labels) + 1}" summary.append(_styled_row(summary, ["公式分析工作簿", "本地可重算副本", "修改不回写系统"], green, white)) summary.append(["数据版本", ds.sha256, "公式版本", FORMULA_VERSION]) summary.append(["筛选范围", str(ctx["filters"]), "分组维度", _text_cell(summary, dim_names[group_key])]) summary.append( [ "本期", f"{ctx['current'][0]} 至 {ctx['current'][-1]}", "基期", f"{ctx['baseline'][0]} 至 {ctx['baseline'][-1]}", ] ) summary.append( _styled_row(summary, ["指标", "分组", "本期", "基期", "差额", "变化率", "状态"], green, white) ) all_metrics = {**METRICS, **{f"custom_{i}": item["name"] for i, item in enumerate(custom_columns)}} metric_rows = {} row = 6 for label in labels: for key in all_metrics: metric_rows[(label, key)] = row row += 1 headers = [ "Period", *dim_headers, "CustomerToken", "Revenue", "Cost", "GrossInput", "MarginRate", "GrossSource", "ListAmount", "SelfOperated", "CustomerPresent", "RowCount", "GrossCalc", ] data.freeze_panes = "A2" if count: _add_table( data, _table( "Data", f"A1:{_column_name(len(headers))}{count + 1}", headers, "TableStyleMedium2", ), ) 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) for row_number, fact in enumerate(qs.iterator(chunk_size=2000), start=2): product = fact.dimensions.get("product", "") direct = None if ( gross_index is not None and gross_index < len(fact.raw) and fact.raw[gross_index] not in (None, "") ): try: direct = _excel_number(Decimal(str(fact.raw[gross_index]).replace(",", ""))) except (InvalidOperation, ValueError): pass rate = _excel_number(Decimal(str(rates[product]))) if product in rates else None customer = fact.customer or "" token = hmac.new(token_secret, customer.encode(), hashlib.sha256).hexdigest()[:24] if customer else "" if customer: group_label = str(fact.dimensions.get(group_key) or "（未分类）") record = customer_roster.setdefault( (group_label, token), {"current": False, "baseline": False, "current_upgraded": False, "baseline_upgraded": False}, ) if fact.period in ctx["current"]: record["current"] = True record["current_upgraded"] |= product in products if fact.period in ctx["baseline"]: record["baseline"] = True record["baseline_upgraded"] |= product in products dim_values = [fact.dimensions.get(d["key"], "（未分类）") or "（未分类）" for d in dims] values = [ fact.period, *dim_values, token, _excel_number(fact.revenue), _excel_number(fact.cost), direct, rate, "direct" if direct is not None else ("lookup" if rate is not None else "missing"), _excel_number(fact.list_amount), int(product in products), int(bool(customer)), 1, None, ] row = [ _text_cell(data, value) if i <= len(dims) + 1 or i == len(dims) + 6 else value for i, value in enumerate(values) ] row[-1] = WriteOnlyCell( data, value=( f'=IF({_column_name(len(dims) + 7)}{row_number}="direct",' f"{_column_name(len(dims) + 5)}{row_number}," f'IF({_column_name(len(dims) + 7)}{row_number}="lookup",' f"{_column_name(len(dims) + 3)}{row_number}*" 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", "CustomerToken", "Current", "Baseline", "CurrentUpgraded", "BaselineUpgraded", ] _add_table( customers_sheet, _table( "CustomerRoster", f"A1:F{len(customer_roster) + 1 if customer_roster else 2}", customer_headers, "TableStyleMedium4", ), ) customers_sheet.append(customer_headers) for (group_label, token), flags in customer_roster.items(): customers_sheet.append( [ _text_cell(customers_sheet, group_label), _text_cell(customers_sheet, token), int(flags["current"]), int(flags["baseline"]), int(flags["current_upgraded"]), int(flags["baseline_upgraded"]), ] ) if not customer_roster: customers_sheet.append(["", "", 0, 0, 0, 0]) evidence.append( [ "来源工作表", "来源行号", *[_text_cell(evidence, d["name"]) for d in dims], "期间", "客户 Token", "收入原值", "成本原值", "毛利原值", "原价原值", ] ) for source_row, period, values, customer, revenue, cost, gross, list_amount in qs.values_list( "source_row", "period", "dimensions", "customer", "revenue", "cost", "gross", "list_amount" ).iterator(chunk_size=2000): token = ( hmac.new(token_secret, (customer or "").encode(), hashlib.sha256).hexdigest()[:24] if customer else "" ) values = [ mapping["sheet"], source_row, *[values.get(d["key"], "（未分类）") for d in dims], period, token, str(revenue), "" if cost is None else str(cost), "" if gross is None else str(gross), "" if list_amount is None else str(list_amount), ] evidence.append([_text_cell(evidence, value) for value in values]) parameters.append(_styled_row(parameters, ["参数", "值", "说明", "可编辑"], green, white)) for row in [ [ "收入字段", _text_cell( parameters, next((c["name"] for c in mapping["columns"] if c["role"] == "revenue"), "未命名"), ), "使用导入时确认的收入字段", "否", ], ["ROI 口径", mapping["roi_formula"], "按导入映射保存的公式分支", "否"], ["自营产品集合", _text_cell(parameters, ", ".join(products)), "升单率和自营收入口径", "否"], [ "产品毛利率（快照）", _text_cell(parameters, str(rates)), "只读快照；导出时已按行写入计算明细 MarginRate 列，调整毛利率请在计算明细页按行修改", "否", ], [ "Excel 数值精度", "IEEE-754 双精度", "精确 Decimal 在证据页按文本保存；公式不保证 15 位以上有效数字", "否", ], ["重算范围", 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( [ "自定义列", _text_cell(parameters, definition["name"]), _text_cell(parameters, definition["meaning"]), "公式由服务端生成", ] ) def source_name(role): return next((c["name"] for c in mapping["columns"] if c["role"] == role), "未命名") editable_inputs = [ ("Revenue", "实际收入", "金额", source_name("revenue"), "空值行不参与求和，汇总按现有行计算"), ("Cost", "成本", "金额", source_name("cost"), "任一行缺失时本组成本及 ROI 等依赖指标不可用"), ("GrossInput", "直接毛利", "金额", source_name("gross"), "空值行改按产品毛利率（lookup）计算毛利"), ( "MarginRate", "产品毛利率", "小数比率", "导入确认的产品毛利率表", "仅 GrossSource=lookup 的行参与毛利计算；0.2 表示 20%", ), ( "ListAmount", "原价金额", "金额", source_name("list_amount"), "用于扣率与原价客单价；缺失则相关指标不可用", ), ] instructions.append(_styled_row(instructions, ["主题", "说明"], green, white)) for row in [ ["用途", "公式分析工作簿；本地修改不回写系统。"], ["行粒度", "计算明细每行对应一个 Fact，不等同患者就诊数。"], ["维度列", "列名为数据导入映射中的维度键；键与显示名对照见“参数与口径”页的维度行。"], ["毛利", "直接毛利优先，否则收入×产品毛利率；无法计算则不可用。毛利贡献不代表净利润。"], ["ROI", "按选定口径；缺成本、成本非正或所需指标缺失时不可用。"], ["人数", "客户 token 跨行去重；缺客户编码时不可用。"], [ "人数去重", "人数去重页按分组保存随机 token 及本期/基期/升单标记；身份、分组和期间标记属于固定快照，不可修改。", ], ["变化率", "差额/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: instructions.append( [ _text_cell(instructions, definition["name"]), _text_cell( instructions, f"含义：{definition['meaning']}；范围：{definition['scope']}；缺失/零分母：{definition['missing']}。", ), ] ) if custom_sheet: custom_sheet.append( [ "分组", *[ _text_cell(custom_sheet, f"{item['name']}（{suffix}）") for item in custom_columns for suffix in ("本期", "基期", "差额", "变化率") ], ] ) for label in labels: values = [_text_cell(custom_sheet, label)] for index, _definition in enumerate(custom_columns): formula_row = metric_rows[(label, f"custom_{index}")] values.extend( [ f"='分析汇总'!C{formula_row}", f"='分析汇总'!D{formula_row}", f"='分析汇总'!E{formula_row}", f"='分析汇总'!F{formula_row}", ] ) custom_sheet.append(values) for label in labels: for key in METRICS: row = metric_rows[(label, key)] formulas = [] for side, periods in enumerate((ctx["current"], ctx["baseline"])): if set(periods) & set(ctx["missing"]): formulas.append("=NA()") continue col = "C" if side == 0 else "D" formulas.append( _checked_formula( _metric_formula( key, label, periods, col, metric_rows, ctx["filters"], group_key, products, mapping["roi_formula"], ) ) ) summary.append( [ METRICS[key], _text_cell(summary, label), formulas[0], formulas[1], f"=IFERROR(C{row}-D{row},NA())", f"=IFERROR(IF(D{row}=0,NA(),E{row}/ABS(D{row})),NA())", f'=IF(ISNA(C{row}),"本期不可用",IF(ISNA(D{row}),"基期不可用","可用"))', ] ) for index, definition in enumerate(custom_columns): key, row = f"custom_{index}", metric_rows[(label, f"custom_{index}")] formulas = [] for side in range(2): periods = ctx["current"] if side == 0 else ctx["baseline"] if set(periods) & set(ctx["missing"]): formulas.append("=NA()") continue col = "C" if side == 0 else "D" refs = {metric: f"{col}{metric_rows[(label, metric)]}" for metric in METRICS} expr = to_excel(definition["expression"], refs) used = metrics_used(definition["expression"]) invalid = ",".join(f"ISNA({refs[metric]})" for metric in sorted(used)) formula = ( f"IF(OR({invalid}),NA(),IFERROR({expr},NA()))" if invalid else f"IFERROR({expr},NA())" ) formulas.append(_checked_formula(f"={formula}")) summary.append( [ _text_cell(summary, definition["name"]), _text_cell(summary, label), formulas[0], formulas[1], f"=IFERROR(C{row}-D{row},NA())", f"=IFERROR(IF(D{row}=0,NA(),E{row}/ABS(D{row})),NA())", f'=IF(ISNA(C{row}),"本期不可用",IF(ISNA(D{row}),"基期不可用","可用"))', ] ) reconcile.append( _styled_row( reconcile, ["指标", "分组", "期间侧", "服务器 Decimal 参考", "公式结果", "差异", "状态"], green, white, ) ) reconcile_row = 2 for label in labels: for side_name, periods, col in (("本期", ctx["current"], "C"), ("基期", ctx["baseline"], "D")): side_qs = qs.filter(period__in=periods, **{f"dimensions__{group_key}": group_values[label]}) reference = metrics(side_qs.aggregate(**aggregations(products)), mapping["roi_formula"], products) 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, _text_cell(reconcile, label), side_name, "" 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})<={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" wb.save(output) return {"rows": count, "groups": len(labels), "formula_version": FORMULA_VERSION} 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: 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("科室授权已变更，请重新提交导出") descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(descriptor, "wb") as file: stats = build_formula_workbook(job.dataset, member, job.context, file) 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: remaining = "" for path in (temporary, target): 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 def csv_safe(value): text = str(value) if value is not None else "" if not isinstance(value, str): return text return ( "'" + text if text.startswith(("\t", "\r", "\n")) or text.lstrip().startswith(("=", "+", "-", "@")) else text )