117 lines
4.5 KiB
Python
117 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
测试对账单单价刷新功能
|
|
"""
|
|
|
|
import sqlite3
|
|
import sys
|
|
import os
|
|
|
|
# 添加项目路径
|
|
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
def test_price_refresh():
|
|
"""测试价格刷新功能"""
|
|
|
|
# 连接数据库
|
|
conn = sqlite3.connect('server/data.db')
|
|
c = conn.cursor()
|
|
|
|
print("=== 测试对账单单价刷新功能 ===\n")
|
|
|
|
# 1. 检查数据库结构
|
|
print("1. 检查数据库表结构...")
|
|
c.execute("PRAGMA table_info(customer_orders)")
|
|
customer_orders_columns = c.fetchall()
|
|
print(f" customer_orders 表字段: {[col[1] for col in customer_orders_columns]}")
|
|
|
|
c.execute("PRAGMA table_info(reconciliations)")
|
|
reconciliations_columns = c.fetchall()
|
|
print(f" reconciliations 表字段: {[col[1] for col in reconciliations_columns]}")
|
|
|
|
# 2. 查看现有数据
|
|
print("\n2. 查看现有数据...")
|
|
c.execute("SELECT COUNT(*) FROM customer_orders")
|
|
customer_orders_count = c.fetchone()[0]
|
|
print(f" 客户订单记录数: {customer_orders_count}")
|
|
|
|
c.execute("SELECT COUNT(*) FROM reconciliations")
|
|
reconciliations_count = c.fetchone()[0]
|
|
print(f" 对账单记录数: {reconciliations_count}")
|
|
|
|
# 3. 显示部分示例数据
|
|
print("\n3. 客户订单示例数据:")
|
|
c.execute("SELECT order_no, material, unit_price FROM customer_orders LIMIT 5")
|
|
for row in c.fetchall():
|
|
print(f" 订单号: {row[0]}, 物料: {row[1][:30]}..., 单价: {row[2]}")
|
|
|
|
print("\n4. 对账单示例数据:")
|
|
c.execute("SELECT contract_no, material_name, unit_price FROM reconciliations LIMIT 5")
|
|
for row in c.fetchall():
|
|
print(f" 合同号: {row[0]}, 物料: {row[1][:30]}..., 单价: {row[2]}")
|
|
|
|
# 4. 测试价格匹配逻辑
|
|
print("\n5. 测试价格匹配逻辑...")
|
|
|
|
# 获取所有客户订单
|
|
c.execute('SELECT order_no, material, unit_price FROM customer_orders')
|
|
customer_orders = c.fetchall()
|
|
|
|
# 构建订单单价字典
|
|
order_prices = {}
|
|
for order in customer_orders:
|
|
order_no = order[0]
|
|
if order_no not in order_prices:
|
|
order_prices[order_no] = {}
|
|
# 处理物料名称（支持换行符分割的多个物料）
|
|
materials = str(order[1]).split('\n')
|
|
for material in materials:
|
|
material = material.strip()
|
|
if material:
|
|
order_prices[order_no][material] = order[2]
|
|
|
|
# 测试对账单匹配
|
|
c.execute('SELECT id, contract_no, material_name, unit_price FROM reconciliations LIMIT 10')
|
|
test_reconciliations = c.fetchall()
|
|
|
|
matched_count = 0
|
|
for recon in test_reconciliations:
|
|
recon_id, contract_no, material_name, current_price = recon
|
|
new_price = None
|
|
|
|
# 精确匹配合同号
|
|
if contract_no in order_prices:
|
|
# 尝试精确匹配物料名
|
|
if material_name in order_prices[contract_no]:
|
|
new_price = order_prices[contract_no][material_name]
|
|
else:
|
|
# 尝试部分匹配
|
|
for order_material, price in order_prices[contract_no].items():
|
|
if material_name in order_material or order_material in material_name:
|
|
new_price = price
|
|
break
|
|
|
|
if new_price is not None:
|
|
print(f" ✓ 匹配成功: 对账单ID={recon_id}, 合同号={contract_no}, 当前单价={current_price}, 匹配单价={new_price}")
|
|
matched_count += 1
|
|
else:
|
|
print(f" ✗ 未匹配: 对账单ID={recon_id}, 合同号={contract_no}, 物料={material_name[:30]}...")
|
|
|
|
print(f"\n 测试结果: {matched_count}/{len(test_reconciliations)} 条记录成功匹配")
|
|
|
|
conn.close()
|
|
|
|
print("\n=== 测试完成 ===")
|
|
print("\n功能说明:")
|
|
print("1. 已添加 /api/reconciliations/refresh-prices 接口，用于批量刷新所有对账单单价")
|
|
print("2. 已添加 /api/reconciliations/fetch-price 接口，用于根据合同号和物料名获取单个单价")
|
|
print("3. 前端已添加'刷新单价'按钮，支持手动批量刷新")
|
|
print("4. 新增/编辑对账单时，可通过'获取单价'按钮从客户订单获取最新单价")
|
|
print("\n使用方法:")
|
|
print("- 批量刷新: 在对账单管理页面点击'刷新单价'按钮")
|
|
print("- 单个获取: 在新增/编辑对账单时，填写合同号和物料名称后点击'获取单价'")
|
|
|
|
if __name__ == '__main__':
|
|
test_price_refresh()
|