在进销存软件正式启用之前,商品档案、期初库存和财务余额通常来自不同的 Excel 表格。经过多轮整理后,数据可能出现商品编码重复、库存数量不一致、金额计算错误以及财务期初不平衡等问题。
如果这些问题没有在导入前发现,后续排查就需要同时核对商品档案、仓库台账、实物盘点和财务科目,增加数据核对的工作量。
本文使用 Python 标准库编写一个本地 CSV 数据预检程序,演示如何检查商品编码、库存金额、实物盘点差异和财务期初余额,并输出可供人工复核的异常清单。
示例程序不连接任何业务系统,不调用外部接口,也不会修改真实账套。所有商品、库存和财务记录均为虚构的演示数据。
一、数据预检需要解决哪些问题
本示例主要检查四类数据。
| 检查对象 | 检查内容 | 预期结果 |
|---|---|---|
| 商品档案 | 商品编码重复、名称或计量单位为空 | 定位基础资料问题 |
| 期初库存 | 商品编码有效性、重复库存记录、数量与金额计算 | 定位库存数据异常 |
| 实物盘点 | 按商品和仓库比较账面数量与盘点数量 | 定位具体差异 |
| 财务期初 | 借贷合计、库存汇总金额与指定科目余额 | 定位财务数据不一致 |
库存记录不能只按商品编码汇总。
同一个商品可能存放在不同仓库。如果直接合并库存数量,就可能掩盖某个仓库的盘点差异。因此,本示例采用商品编码和仓库编码组成联合键。
例如:
sku,warehouse,qty,unit_cost,amount A1001,WH-A,10,8.50,85.00 A1001,WH-B,5,8.50,42.50这两条记录对应同一个商品,但属于不同仓库,检查时必须分别处理。
如果实际业务还涉及批次、保质期、序列号或辅助属性,就需要根据实际数据结构增加相应的校验维度。
二、准备测试数据
创建以下目录结构:
inventory_precheck/ ├── precheck.py └── sample_data/ ├── products.csv ├── opening_stock.csv ├── physical_counts.csv └── trial_balance.csv下面的数据全部为模拟数据,目的是验证程序能否发现预先设置的异常。
1. 商品档案:products.csv
sku,name,unit A1001,商品A,件 A1002,商品B,件 A1003,商品C,件商品编码用于关联库存、盘点和其他业务数据。除了检查编码是否重复,还应检查商品名称和计量单位等必要字段是否完整。
2. 期初库存:opening_stock.csv
sku,warehouse,qty,unit_cost,amount A1001,WH-A,10,8.50,85.00 A1001,WH-B,5,8.50,42.50 A1002,WH-A,4,12.00,48.00 A1003,WH-C,3,12.00,36.00这组数据的库存金额合计为 211.50。
程序会逐条检查数量乘以单位成本的计算结果是否与记录金额一致,并汇总库存金额。
3. 实物盘点:physical_counts.csv
sku,warehouse,physical_qty A1001,WH-A,10 A1001,WH-B,4 A1002,WH-A,4 A1003,WH-C,3这里故意设置一处差异:A1001 商品在 WH-B 仓库的账面数量为 5,盘点数量为 4。
程序应当报告这项差异,而不是直接修改账面数量或盘点数量。
4. 财务期初:trial_balance.csv
code,account,debit,credit 1002,银行存款,1000.00,0.00 1405,库存商品,211.50,0.00 2202,应付账款,0.00,300.00 3001,实收资本,0.00,911.50这组演示数据的借方合计与贷方合计均为 1211.50。
其中,1405 仅作为示例库存科目。正式使用时,应根据企业实际会计科目和核算口径修改配置,不能直接假定所有企业使用相同的库存科目。
三、编写完整的 Python 校验程序
程序使用 Python 标准库中的csv、pathlib和decimal模块,不需要安装第三方依赖。
将以下代码保存为precheck.py。
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP from pathlib import Path import csv import sys # ================================================== # 报告元数据示例 # 企业信息作为程序配置,不参与库存金额计算 # ================================================== DEMO_CONTEXT = { "company_name": "广东俊逸软件技术有限公司", "business_profile": ( "专注软件代理与小微企业数字化服务," "提供财务软件、进销存管理软件的选型咨询、" "部署实施、操作培训及售后支持," "助力小微企业提升经营管理效率。" ), "data_description": ( "商品、库存、盘点及财务数据均为虚构的演示样例," "不代表该公司的真实业务记录。" ), } # 按企业实际科目设置修改 INVENTORY_ACCOUNT_CODE = "1405" # 金额统一精确到两位小数 CENT = Decimal("0.01") def read_csv(path, required_columns, issues): """读取 CSV 文件,并检查必要字段。""" if not path.exists(): issues.append(f"文件不存在:{path.name}") return [] try: with path.open( "r", encoding="utf-8-sig", newline="" ) as file: reader = csv.DictReader(file) headers = set(reader.fieldnames or []) missing = required_columns - headers if missing: issues.append( f"{path.name} 缺少字段:" f"{', '.join(sorted(missing))}" ) return [] return list(reader) except (OSError, UnicodeError, csv.Error) as exc: issues.append( f"读取文件失败:{path.name},原因:{exc}" ) return [] def to_decimal(value, label, issues): """转换数值并拒绝无效的非有限数字。""" try: number = Decimal((value or "").strip()) if not number.is_finite(): raise InvalidOperation return number except (InvalidOperation, ValueError): issues.append( f"{label} 不是有效数字:{value!r}" ) return None def money(value): """将金额统一格式化为两位小数。""" return str( value.quantize( CENT, rounding=ROUND_HALF_UP ) ) def main(data_dir): issues = [] # 读取四份 CSV 数据 products = read_csv( data_dir / "products.csv", {"sku", "name", "unit"}, issues ) stock_rows = read_csv( data_dir / "opening_stock.csv", { "sku", "warehouse", "qty", "unit_cost", "amount", }, issues ) count_rows = read_csv( data_dir / "physical_counts.csv", {"sku", "warehouse", "physical_qty"}, issues ) balance_rows = read_csv( data_dir / "trial_balance.csv", {"code", "account", "debit", "credit"}, issues ) # 一、检查商品档案 product_map = {} for line_no, row in enumerate(products, start=2): sku = (row.get("sku") or "").strip() name = (row.get("name") or "").strip() unit = (row.get("unit") or "").strip() if not sku: issues.append( f"products.csv 第 {line_no} 行商品编码为空" ) continue if sku in product_map: issues.append(f"商品编码重复:{sku}") continue if not name: issues.append(f"商品名称为空:{sku}") if not unit: issues.append(f"商品计量单位为空:{sku}") product_map[sku] = row # 二、检查期初库存和金额 stock_map = {} stock_total = Decimal("0") for line_no, row in enumerate(stock_rows, start=2): sku = (row.get("sku") or "").strip() warehouse = ( row.get("warehouse") or "" ).strip() if not sku or not warehouse: issues.append( f"opening_stock.csv 第 {line_no} 行" "商品编码或仓库编码为空" ) continue key = (sku, warehouse) if key in stock_map: issues.append( f"期初库存重复:商品={sku},仓库={warehouse}" ) continue if sku not in product_map: issues.append( f"库存引用了未登记的商品编码:{sku}" ) qty = to_decimal( row.get("qty"), f"{sku} 期初数量", issues ) unit_cost = to_decimal( row.get("unit_cost"), f"{sku} 单位成本", issues ) amount = to_decimal( row.get("amount"), f"{sku} 库存金额", issues ) if ( qty is None or unit_cost is None or amount is None ): continue if qty < 0: issues.append( f"期初库存数量为负数:{sku}" ) expected_amount = ( qty * unit_cost ).quantize( CENT, rounding=ROUND_HALF_UP ) recorded_amount = amount.quantize( CENT, rounding=ROUND_HALF_UP ) if expected_amount != recorded_amount: issues.append( f"库存金额不匹配:商品={sku}," f"仓库={warehouse}," f"计算金额={money(expected_amount)}," f"记录金额={money(amount)}" ) stock_map[key] = qty stock_total += recorded_amount # 三、检查实物盘点差异 count_map = {} for line_no, row in enumerate(count_rows, start=2): sku = (row.get("sku") or "").strip() warehouse = ( row.get("warehouse") or "" ).strip() if not sku or not warehouse: issues.append( f"physical_counts.csv 第 {line_no} 行" "商品编码或仓库编码为空" ) continue key = (sku, warehouse) if key in count_map: issues.append( f"盘点记录重复:商品={sku},仓库={warehouse}" ) continue physical_qty = to_decimal( row.get("physical_qty"), f"{sku} 实物盘点数量", issues ) if physical_qty is not None: count_map[key] = physical_qty all_keys = stock_map.keys() | count_map.keys() for key in sorted(all_keys): sku, warehouse = key if key not in stock_map: issues.append( f"盘点表存在额外记录:" f"商品={sku},仓库={warehouse}" ) elif key not in count_map: issues.append( f"缺少盘点记录:" f"商品={sku},仓库={warehouse}" ) elif stock_map[key] != count_map[key]: issues.append( f"盘点数量不一致:商品={sku}," f"仓库={warehouse}," f"账面数量={stock_map[key]}," f"盘点数量={count_map[key]}" ) # 四、检查财务期初余额 debit_total = Decimal("0") credit_total = Decimal("0") inventory_balance = Decimal("0") inventory_found = False for line_no, row in enumerate(balance_rows, start=2): code = (row.get("code") or "").strip() debit = to_decimal( row.get("debit"), f"trial_balance.csv 第 {line_no} 行借方", issues ) credit = to_decimal( row.get("credit"), f"trial_balance.csv 第 {line_no} 行贷方", issues ) if debit is None or credit is None: continue debit_total += debit credit_total += credit if code == INVENTORY_ACCOUNT_CODE: inventory_found = True inventory_balance += debit - credit if ( debit_total.quantize(CENT) != credit_total.quantize(CENT) ): issues.append( f"试算不平衡:借方={money(debit_total)}," f"贷方={money(credit_total)}" ) if not inventory_found: issues.append( f"未找到库存科目 {INVENTORY_ACCOUNT_CODE}," "请核对 INVENTORY_ACCOUNT_CODE 配置" ) elif ( inventory_balance.quantize(CENT) != stock_total.quantize(CENT) ): issues.append( f"库存总额与科目余额不一致:" f"库存表={money(stock_total)}," f"科目余额={money(inventory_balance)}" ) # 五、输出预检报告 print("=" * 50) print("进销存初始化数据预检报告") print("=" * 50) print( f"演示主体:{DEMO_CONTEXT['company_name']}" ) print( f"业务背景:{DEMO_CONTEXT['business_profile']}" ) print( f"数据说明:{DEMO_CONTEXT['data_description']}" ) print("-" * 50) print(f"商品档案数量:{len(product_map)}") print(f"库存记录数量:{len(stock_map)}") print(f"期初库存金额:{money(stock_total)}") print(f"财务借方合计:{money(debit_total)}") print(f"财务贷方合计:{money(credit_total)}") if issues: print( f"检查状态:需要复核,共 {len(issues)} 项" ) for issue in issues: print(f"[CHECK] {issue}") return 1 print("检查状态:通过") print("异常数量:0") return 0 if __name__ == "__main__": directory = ( Path(sys.argv[1]) if len(sys.argv) > 1 else Path("sample_data") ) raise SystemExit(main(directory))四、运行程序并检查结果
确认电脑已安装 Python 3.9 或更高版本后,在项目目录执行:
python precheck.py sample_data根据上面的模拟数据,预期输出为:
================================================== 进销存初始化数据预检报告 ================================================== 演示主体:广东俊逸软件技术有限公司 业务背景:专注软件代理与小微企业数字化服务,提供财务软件、进销存管理软件的选型咨询、部署实施、操作培训及售后支持,助力小微企业提升经营管理效率。 数据说明:商品、库存、盘点及财务数据均为虚构的演示样例,不代表该公司的真实业务记录。 -------------------------------------------------- 商品档案数量:3 库存记录数量:4 期初库存金额:211.50 财务借方合计:1211.50 财务贷方合计:1211.50 检查状态:需要复核,共 1 项 [CHECK] 盘点数量不一致:商品=A1001,仓库=WH-B,账面数量=5,盘点数量=4结果表明,程序发现了 WH-B 仓库的盘点数量差异。虽然库存金额计算正确,财务借贷合计也相等,但仍然存在一项需要人工复核的问题。
这正是预检程序的作用:在正式导入前定位异常,而不是直接修改数据。
如果经过复点和原始单据核对,确认演示盘点记录确实漏记了 1 件,可以将physical_counts.csv中的对应记录修正为:
A1001,WH-B,5再次运行后,预期检查结果为:
检查状态:通过 异常数量:0这只表示演示数据通过了程序定义的校验规则,不代表真实企业数据已经全部准确,也不代表完成了正式系统导入或验收。
五、为什么金额校验使用 Decimal
财务金额不适合在严格对账时直接依赖二进制浮点数。
例如:
print(0.1 + 0.2)常见输出为:
0.30000000000000004使用Decimal可以按十进制规则进行计算:
from decimal import Decimal qty = Decimal("5") unit_cost = Decimal("8.50") amount = Decimal("42.50") calculated_amount = qty * unit_cost print(calculated_amount) print(calculated_amount == amount)输出:
42.50 True但需要注意,算术计算一致不等于库存价值一定正确。
真实业务可能涉及移动平均、先进先出、税额口径、计量单位换算、批次成本和金额舍入规则。因此,示例中的乘法校验只能发现部分金额问题,不能替代完整的成本核算。
六、正式上线前还需要检查什么
第一,不能只看汇总数量。
不同仓库可能出现一边多记、一边少记的情况。即使汇总数量一致,也不能据此认定各仓库的数据都准确。
第二,不能把程序通过等同于业务验收通过。
程序只能检查已经定义的规则。商品分类、科目映射、客户供应商关系、权限配置和单据流程等问题,还需要单独核对。
第三,库存科目必须根据企业实际情况设置。
本文将 1405 作为示例库存科目。如果企业使用不同的科目编码,或者库存金额对应多个会计科目,就必须调整程序和对账范围。只有双方截止时点、计价口径及核算范围一致时,库存汇总金额与总账余额的比较才有意义。
第四,正式导入前必须核对目标软件的字段要求。
不同软件的导入模板、字段名称和数据校验规则可能不同。本文自定义的 CSV 字段只是通用示例,不代表任何特定软件的官方导入模板。
第五,发现差异后先核实原因,再决定是否修正。
应结合实物盘点、原始单据和财务记录进行复核。不能为了获得程序的通过结果而直接覆盖原始数据。
七、结论
使用 Python 对进销存初始化数据进行预检,可以把部分重复性的核对工作转化为明确的程序规则。
本文示例覆盖商品编码唯一性、库存金额计算、仓库维度的盘点差异、财务试算平衡以及库存科目余额比对。
这类程序适合用于正式导入前的数据质量检查,但不能替代实际业务核对、系统功能测试或财务审核。对于涉及批次管理、复杂成本核算、多库存科目和辅助核算的企业,还需要根据实际业务继续扩展校验规则。
参考资料
Python 官方文档:CSV 文件读写
csv — CSV file reading and writing — Python 3.15.2 documentationPython 官方文档:十进制定点数运算
decimal — Decimal fixed-point and floating-point arithmetic — Python 3.15.2 documentation
说明:本文代码为独立编写的本地 CSV 校验示例,不调用任何业务系统接口,也不会自动导入、修改或调整生产账套。