人财事物信息化 - purchase_invoice.py

以下是ERPNext采购发票模块的Python代码详细分析：

  1. 核心类结构
  • 继承自BuyingController，处理采购业务流程
  • 包含采购发票的完整生命周期管理：创建、验证、提交、取消
  1. 初始化与状态管理
  • __init__初始化状态更新器，跟踪采购订单项状态
  • status_updater配置：
    • 更新采购订单项计费金额和百分比
    • 处理库存接收数量更新
    • 退货数量跟踪
  1. 验证体系
  • 基础验证：
    • 必填项检查（供应商、日期等）
    • 单据状态验证（是否冻结/关闭）
    • 货币汇率验证
  • 业务规则验证：
    • 采购订单/收货单强制要求（根据系统配置）
    • 库存更新时的仓库验证
    • 供应商发票唯一性校验
    • 预付款与核销金额匹配
  • 财务校验：
    • 费用科目有效性
    • 税务账户配置
    • 暂记账户处理
  1. 会计处理
  • 总账分录：
    • 供应商科目处理（应付账款）
    • 库存科目调整（永续库存）
    • 税款分离处理（进项税、代扣税）
    • 汇兑损益计算
  • 特殊处理：
    • 内部转移交易处理
    • 资产类目(CWIP)特殊科目
    • 暂估入库处理
  • 付款相关：
    • 现金/银行科目处理
    • 预付款分配
    • 坏账核销处理
  1. 库存管理
  • 库存更新逻辑：
    • 库存价值调整
    • 批次/序列号跟踪
    • 供应商仓库特殊处理
  • 关联文档更新：
    • 采购订单计费状态
    • 收货单接收数量
    • 物料请求状态
  1. 税务处理
  • 代扣所得税(TDS)：
    • 供应商级税务配置
    • 预付款税款分配
    • 多凭证税款跟踪
  • 增值税处理：
    • 进项税额分离
    • 不同税率处理
    • 南非特殊处理
  1. 高级功能
  • 项目成本跟踪：
    • 自动更新项目采购成本
    • 多项目分配处理
  • 内部交易：
    • 公司间交易处理
    • 未实现损益计算
  • 暂记处理：
    • 暂估费用科目
    • 库存暂估调整
  • 资产处理：
    • 固定资产采购
    • 在建工程核算
  1. 工作流控制
  • 单据冻结/解冻：
    • 冻结原因记录
    • 解冻日期控制
  • 审批流程：
    • 授权金额校验
    • 修改跟踪
  • 关联文档状态同步：
    • 采购订单
    • 收货单
    • 付款凭证
  1. 取消与冲销
  • 库存反向处理
  • 会计凭证冲销
  • 项目成本回滚
  • 关联状态恢复
  • 税款分配撤销
  1. 特殊业务场景
  • 退货处理：
    • 反向库存移动
    • 红字发票处理
    • 关联文档逆向更新
  • 暂估入库：
    • 临时科目处理
    • 正式发票匹配
  • 代采购处理：
    • 供应商仓库特殊流程
    • 代购费用分配
  1. 性能优化
  • 批量处理：
    • 库存条目批量更新
    • 项目成本批量计算
  • 智能合并：
    • 相似总账分录合并
    • 重复校验优化
  • 异步处理：
    • 未来库存重算
    • 延迟过账处理
  1. 国际化支持
  • 多币种处理：
    • 本位币转换
    • 外币支付处理
  • 区域化定制：
    • 南非增值税处理
    • 不同国家税务模板
  1. 扩展性设计
  • 钩子函数：
    • 验证扩展点
    • 状态更新扩展
  • 模块化设计：
    • 税务组件分离
    • 库存组件解耦
  • 配置驱动：
    • 公司级默认科目
    • 系统参数控制流程

该模块完整实现了采购发票的全生命周期管理，深度集成了财务、库存、采购、项目等多个模块，体现了ERPNext高度模块化和集成化的设计理念。通过严谨的状态管理和完善的验证体系，确保业务数据的完整性和准确性。

根据上传的发票明细直接创建对应采购发票（同时创建缺失的供应商和商品服务）

以下是针对ERPNext采购发票模块的本地化增强方案，实现通过发票明细文件自动创建采购发票及关联主数据的功能：

一、功能架构设计

  1. 文件处理层
   class InvoiceFileParser:
       SUPPORTED_FORMATS = {'csv', 'xlsx'}

       def parse(self, file_path):
           ext = file_path.split('.')-1
           if ext == 'csv':
               return self._parse_csv(file_path)
           elif ext == 'xlsx':
               return self._parse_excel(file_path)

       def _parse_csv(self, file_path):
            实现CSV解析逻辑
           pass

       def _parse_excel(self, file_path):
            实现Excel解析逻辑
           pass
  1. 数据映射层
   class FieldMapper:
       DEFAULT_MAPPING = {
           'supplier_name': '供应商名称',
           'tax_id': '税号',
           'item_code': '商品编码',
           'qty': '数量'
       }

       def __init__(self, custom_mapping=None):
           self.mapping = {self.DEFAULT_MAPPING, (custom_mapping or {})}

       def transform(self, raw_data):
            执行字段转换和清洗
           pass
  1. 业务处理层
   class InvoiceCreator:
       def create_invoices(self, processed_data):
           invoices = 
           for record in processed_data:
               self._create_missing_master(record)
               pi = self._build_invoice(record)
               pi.insert()
               invoices.append(pi)
           return invoices

       def _create_missing_master(self, record):
           if not frappe.db.exists('Supplier', record'supplier_name'):
               self._create_supplier(record)

           if not frappe.db.exists('Item', record'item_code'):
               self._create_item(record)

       def _create_supplier(self, data):
           supplier = frappe.new_doc('Supplier')
           supplier.update({
               'supplier_name': data'supplier_name',
               'tax_id': data.get('tax_id'),
               'supplier_type': '本地供应商'
           })
           supplier.insert()

       def _create_item(self, data):
           item = frappe.new_doc('Item')
           item.update({
               'item_code': data'item_code',
               'item_name': data.get('item_name'),
               'item_group': '服务类',
               'is_purchase_item': 1
           })
           item.insert()

二、 核心功能实现

  1. 智能供应商匹配
   def _match_supplier(self, record):
        多维度匹配逻辑
       match_conditions = 
           ('tax_id', '=', record.get('tax_id')),
           ('mobile_no', '=', record.get('contact_phone')),
           ('supplier_name', 'like', f"%{record'supplier_name'}%")


       for field, op, val in match_conditions:
           if val:
               suppliers = frappe.get_all('Supplier', 
                   filters={field: op, val},
                   fields='name'
               )
               if suppliers:
                   return suppliers0.name
       return None
  1. 自动化商品创建
   def _create_item(self, data):
       item = frappe.new_doc('Item')
       defaults = {
           'description': data.get('item_desc'),
           'stock_uom': data.get('uom', '个'),
           'is_stock_item': 0,
           'taxes': {
               'item_tax_template': self._get_tax_template(data.get('tax_rate'))
           }
       }
       item.update({data, defaults})
       item.insert()
  1. 发票生成引擎
   def _build_invoice(self, record):
       pi = frappe.new_doc('Purchase Invoice')
       pi.update({
           'supplier': record'supplier',
           'posting_date': record'invoice_date',
           'due_date': self._calculate_due_date(record),
           'items': {
               'item_code': record'item_code',
               'qty': record'qty',
               'rate': record'unit_price',
               'cost_center': '主成本中心'
           },
           'taxes_and_charges': self._get_tax_rule(record)
       })
       pi.run_method('set_missing_values')
       return pi

三、异常处理机制

  1. 数据校验流水线
   class DataValidator:
       REQUIRED_FIELDS = 'supplier_name', 'item_code', 'total_amount'

       def validate(self, records):
           errors = 
           for idx, rec in enumerate(records):
               if not all(rec.get(f) for f in self.REQUIRED_FIELDS):
                   errors.append(f"行{idx+1}: 缺少必填字段")
               if rec.get('tax_id') and not self._validate_tax_id(rec'tax_id'):
                   errors.append(f"行{idx+1}: 税号格式错误")
           return errors
  1. 事务回滚机制
   def safe_import(file_path):
       try:
           data = parse_file(file_path)
           errors = validate(data)
           if errors:
               raise ValidationError("\n".join(errors))

           frappe.enqueue(
               '_execute_import',
               data=data,
               timeout=3600
           )
       except Exception as e:
           frappe.log_error("发票导入失败")
           rollback_transaction()

四、用户界面增强

  1. 文件上传组件
   frappe.ui.form.on('Bulk Invoice Import', {
       upload_complete: function(frm, attachment) {
           frm.call({
               method: 'process_uploaded_file',
               args: {
                   file_url: attachment.file_url
               },
               callback: function(r) {
                   frm.refresh_field('import_results')
               }
           })
       }
   })
  1. 映射配置界面
   <div class="form-group">
       <label>供应商名称映射:</label>
       <select class="form-control" data-fieldname="supplier_mapping">
           <option value="vendor_name">供应商名称</option>
           <option value="company_name">公司名称</option>
       </select>
   </div>

五、部署配置项

  1. 参数配置文件
   class LocalConfig:
       AUTO_CREATE_MASTER = True
       DEFAULT_TAX_RATE = 0.13
       ALLOWED_FILE_SIZE = 50   MB
       REQUIRED_TAX_FIELDS = 'tax_id', 'bank_account'
  1. 定时任务配置
   python
   scheduler.add_job(
       process_pending_invoices,
       'cron',
       hour=2,
       kwargs={'batch_size': 100}
   )

六、安全增强措施

  1. 文件安全扫描
   def check_file_safety(file_path):
       if not antivirus_scan(file_path):
           raise SecurityException("文件包含恶意代码")

       if detect_script_tags(file_path):
           raise ValidationError("文件包含非法脚本")
  1. 访问控制
   def check_permission():
       if not frappe.has_permission('Purchase Invoice', 'create'):
           frappe.throw("无权限执行此操作", frappe.PermissionError)

七、技术验证点

  1. 性能基准测试
   class PerformanceTest:
       @benchmark
       def test_1000_records(self):
           data = generate_test_data(1000)
           creator = InvoiceCreator()
           return creator.create_invoices(data)
  1. 数据一致性检查
   def verify_integrity(original_data, created_invoices):
       total_amount = sum(d'total' for d in original_data)
       system_total = sum(pi.grand_total for pi in created_invoices)
       assert abs(total_amount - system_total) < 0.01

该方案实现了从发票文件解析到系统创建的完整自动化流程，通过模块化设计保证扩展性，结合异常处理和事务机制确保数据安全，适用于处理大批量发票数据录入场景。实施时需注意：

  1. 供应商匹配算法的优化（可引入模糊匹配）
  2. 商品税率逻辑与当地税务政策保持同步
  3. 定期清理临时文件
  4. 建立操作日志审计追踪
  5. 提供数据修正界面处理导入异常
Discard
Save
Draft

Previous Page

Left

Next Page

Right
Review Changes ← Back to Content
Message Status Space Raised By Last update on