人财事物信息化 - report-utils.py

一、核心功能模块深度解析

财务模块（accounts）架构

（1）多维度货币管理体系 - 动态汇率缓存机制（utils.py）


python
__exchange_rates = {}  全局汇率缓存字典

def get_rate_as_at(date, from_currency, to_currency):
    rate_key = f"{from_currency}-{to_currency}@{date}"
    if not __exchange_rates.get(rate_key):
        rate = get_exchange_rate(from_currency, to_currency, date) or 1
        __exchange_ratesrate_key = rate  缓存计算结果
    return __exchange_ratesrate_key

设计特点： - 采用{from}-{to}@date的三段式键名设计，支持双向汇率查询 - 缓存生命周期与会话绑定，避免长期缓存导致汇率过期 - 通过or 1兜底处理，确保零值情况下的计算稳定性

（2）智能凭证转换引擎


python
def convert_to_presentation_currency(gl_entries, currency_info):
    converted_gl_list = 
    presentation_currency = currency_info"presentation_currency"

     智能判断币种复杂度
    account_currencies = list(set(e"account_currency" for e in gl_entries))

    for entry in gl_entries:
        if len(account_currencies) == 1 and account_currency == presentation_currency:
             单一币种直接映射
            entry"debit" = entry"debit_in_account_currency"
        else:
             复杂场景动态转换
            date = currency_info"report_date"
            converted_debit = convert(entry"debit", presentation_currency, 
                                    currency_info"company_currency", date)
            entry"debit" = converted_debit
        converted_gl_list.append(entry)
    return converted_gl_list

关键算法： - 使用集合(set)快速识别账套中存在的所有币种 - 当检测到len(account_currencies)==1时，跳过复杂计算直接映射 - 转换过程保留原始数据字段，避免数据丢失

（3）多维税务处理系统


python
def get_taxes_query(invoice_list, doctype, parenttype):
    taxes = frappe.qb.DocType(doctype)
    query = (
        frappe.qb.from_(taxes)
        .select(taxes.account_head)
        .distinct()
        .where(
            (taxes.parenttype == parenttype)
            & (taxes.docstatus == 1)
            & (taxes.parent.isin(inv.name for inv in invoice_list))
        )
    )
     分业务类型处理税务规则
    if doctype == "Purchase Taxes and Charges":
        return query.where(taxes.category.isin("Total", "Valuation and Total"))
    elif doctype == "Sales Taxes and Charges":
        return query.where(taxes.charge_type.isin("On Paid Amount", "Actual"))

业务逻辑： - 采购类税项：限定为「总额」和「计价总额」类别 - 销售类税项：限制为「实付金额」和「实际」计费类型 - 使用distinct()确保税务科目唯一性

业务模块协同机制

（1）采购-库存-应付联动 - 采购订单 → 库存接收 → 采购发票全链路凭证自动生成 - 通过get_opening_row计算供应商期初余额：


python
def get_opening_row(party_type, party, from_date, company):
    party_account = get_party_account(party_type, party, company, include_advance=True)
    return (
        frappe.qb.from_("GL Entry")
        .select(
            Sum("debit").as_("debit"),
            Sum("credit").as_("credit"),
            (Sum("debit") - Sum("credit")).as_("balance")
        )
        .where(
            (account.isin(party_account))
            & (posting_date < from_date)
            & (is_cancelled == 0)
        )
    )

算法特点： - include_advance=True包含预付款科目 - 使用Sum(debit) - Sum(credit)动态计算余额 - 排除已取消凭证(is_cancelled=0)

（2）销售-应收-现金流协同 - 销售订单 → 出库单 → 销售发票 → 收款单四联单闭环 - 通过get_payment_entries获取收款记录：


python
def get_payment_entries(filters, args):
    pe = frappe.qb.DocType("Payment Entry")
    query = (
        frappe.qb.from_(pe)
        .select(
            pe.name, pe.posting_date,
            pe.paid_amount.as_("base_net_total"),
            pe.paid_amount_after_tax.as_("base_grand_total")
        )
        .where(
            (pe.party == filters.get("party")) 
            & (peargs.account_fieldname.isin(args.party_account))
        )
        .orderby(pe.posting_date, order=Order.desc)
    )
    return apply_common_conditions(filters, query, doctype="Payment Entry")

设计亮点： - 动态字段映射：args.account_fieldname支持客户/供应商科目切换 - 统一条件过滤：复用apply_common_conditions处理日期/公司等通用条件

二、关键代码深度分析

多维核算引擎（utils.py）


python
def filter_invoices_based_on_dimensions(filters, query, parent_doc):
    dimensions = get_accounting_dimensions(as_list=False)
    for dim in dimensions:
        if filters.get(dim.fieldname):
            if frappe.get_cached_value("DocType", dim.document_type, "is_tree"):
                 处理树形结构维度
                filtersdim.fieldname = get_dimension_with_children(
                    dim.document_type, 
                    filtersdim.fieldname
                )
            query = query.where(
                parent_docdim.fieldname.isin(filtersdim.fieldname)
            )
    return query

核心技术： 1. 动态维度发现：get_accounting_dimensions获取用户自定义维度 2. 树形结构处理：对类似成本中心的层级维度，通过get_dimension_with_children获取所有子节点 3. 查询注入：使用isin()实现多值过滤，避免多次查询

复合查询构建器


python
def apply_common_conditions(filters, query, doctype, child_doctype=None):
    parent = frappe.qb.DocType(doctype)

     基础过滤
    if filters.get("company"):
        query = query.where(parent.company == filters"company")

     动态关联子表
    if filters.get("cost_center") or filters.get("warehouse"):
        child = frappe.qb.DocType(child_doctype)
        query = (
            query.inner_join(child)
            .on(parent.name == child.parent)
            .where(child.parenttype == doctype)
        )
        if filters.get("cost_center"):
            query = query.where(child.cost_center == filters"cost_center")

    return query.distinct()

优化策略： - 延迟关联：仅当需要子表字段时才进行JOIN操作 - 去重机制：使用distinct()避免主表记录重复 - 条件预判：先处理公司等高频过滤条件，提升查询效率

安全审计关键点


python
@frappe.whitelist()
def get_invoiced_item_gross_margin(sales_invoice=None, item_code=None, company=None):
     参数清洗
    sales_invoice = sales_invoice or frappe.form_dict.get("sales_invoice")
    company = company or frappe.get_cached_value("Sales Invoice", sales_invoice, "company")

     查询构造
    filters = {
        "sales_invoice": sales_invoice,
        "item_code": item_code,
        "company": company,
        "group_by": "Invoice"
    }

     业务逻辑执行
    gross_profit_data = GrossProfitGenerator(filters)
    return sum(d.gross_profit for d in gross_profit_data.grouped_data)

安全设计： 1. 权限控制：@whitelist装饰器限制入口访问 2. 参数清洗：优先使用传入参数，次取form_dict防止CSRF 3. 上下文隔离：强制传入company参数，避免越权访问

三、架构设计亮点

混合持久层设计

对高频操作（如汇率获取）使用内存缓存
复杂查询使用Frappe ORM构建，兼顾可读性与安全性
原始SQL仅用于特殊聚合场景（如期初余额计算）

币种隔离策略


python
 在convert函数中实现双向转换
def convert(value, from_, to, date):
    rate = get_rate_as_at(date, from_, to)
    return flt(value) / (rate or 1)   防止除零错误

采用from_→to的显式转换方向声明
统一使用flt()处理浮点精度问题
所有金额字段存储本位币值+交易币值双字段

维度扩展机制

通过get_accounting_dimensions实现动态维度发现
每个维度关联到具体的DocType，支持自定义字段
查询时自动注入维度过滤条件，业务代码无感知

四、性能优化实践

查询计划优化


python
 在get_opening_row中
.where(
    (gle.posting_date < from_date)
    & (gle.is_cancelled == 0) 
)

将posting_date作为首过滤条件，利用索引加速
is_cancelled=0使用位图索引快速过滤

批量处理模式


python
 在convert_to_presentation_currency中
account_currencies = list(set(e"account_currency" for e in gl_entries))

使用集合(set)预判币种复杂度，减少逐条判断
对大批量GL条目采用批处理转换

内存缓存策略

_exchangerates使用进程级缓存，适合汇率相对稳定的场景
getpartydetails中对合作伙伴信息进行缓存，减少数据库访问

该代码库体现了成熟ERP系统的典型特征：在保持财务严谨性的同时，通过灵活的架构设计支持业务扩展。其多维核算体系和币种处理机制尤其值得借鉴，展示了如何将复杂的会计准则转化为可维护的代码实现。