import hashlib import hmac 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 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, metrics_used, to_excel, validate_column, ) from .models import FormulaExport, Membership FORMULA_VERSION = "3" MAX_ROWS = 100_000 MAX_GROUPS = 200 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) 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},""))' ), ) 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)), "升单率和自营收入口径", "否"], ["产品毛利率", str(rates), "直接毛利缺失时的可编辑计算输入", "是"], [ "Excel 数值精度", "IEEE-754 双精度", "精确 Decimal 在证据页按文本保存；公式不保证 15 位以上有效数字", "否", ], ["重算范围", f"最多 {MAX_ROWS:,} 行、{MAX_GROUPS} 组", "超限需缩小导出范围", "否"], ["兼容性", "SUMIFS/COUNTIFS", "移除动态数组人数公式；仍需目标 Excel 桌面版本验收", "否"], ]: parameters.append(row) for definition in custom_columns: parameters.append( [ "自定义列", _text_cell(parameters, definition["name"]), _text_cell(parameters, definition["meaning"]), "公式由服务端生成", ] ) instructions.append(_styled_row(instructions, ["主题", "说明"], green, white)) for row in [ ["用途", "公式分析工作簿；本地修改不回写系统。"], ["行粒度", "计算明细每行对应一个 Fact，不等同患者就诊数。"], ["维度列", "列名为数据导入映射中的维度键；显示名称见本页分组维度。"], ["毛利", "直接毛利优先，否则收入×产品毛利率；无法计算则不可用。毛利贡献不代表净利润。"], ["ROI", "按选定口径；缺成本、成本非正或所需指标缺失时不可用。"], ["人数", "客户 token 跨行去重；缺客户编码时不可用。"], [ "人数去重", "人数去重页按分组保存随机 token 及本期/基期/升单标记；身份、分组和期间标记属于固定快照，不可修改。", ], ["变化率", "差额/ABS(基期)；基期为0或期间缺失时不可用。"], ["精度", "原始 Decimal 以文本留存；Excel 双精度存在细微误差可能。"], ["隐私", "客户标识替换为本文件随机 token，文件仍按敏感财务信息保护。"], ]: 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] 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})<0.01,"通过","待复核"),"待复核"))', ] ) reconcile_row += 1 for index, definition in enumerate(custom_columns): formula_row = metric_rows[(label, f"custom_{index}")] reconcile.append( [ _text_cell(reconcile, definition["name"]), _text_cell(reconcile, label), side_name, "自定义公式无服务器参考", f"='分析汇总'!{col}{formula_row}", "", "需业务验收", ] ) reconcile_row += 1 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 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 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: 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 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 )