duxingchen/KCGL
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: use highest historical unit price for material bases in export

Browse Source
This commit is contained in:
dxc
2026-03-02 12:22:30 +08:00
committed by dxc (aider)
parent 2f140e112f
commit d7dff943fc
1 changed files with 73 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

145
inventory-backend/app/services/inbound/base_service.py
View File

@ -6,7 +6,7 @@ from app.models.inbound.buy import StockBuy
from app.models.inbound.semi import StockSemi from app.models.inbound.semi import StockSemi
from app.models.inbound.product import StockProduct from app.models.inbound.product import StockProduct
# from app.models.inbound.service import StockService # from app.models.inbound.service import StockService
from sqlalchemy import or_, and_ from sqlalchemy import or_, and_, func
import traceback import traceback
import json import json
import io import io
@ -114,7 +114,6 @@ class MaterialBaseService:
""" """
获取基础信息列表 (带分页和筛选) 获取基础信息列表 (带分页和筛选)
""" """
from sqlalchemy import func
try: try:
# 构建聚合子查询 # 构建聚合子查询
buy_sub = db.session.query( buy_sub = db.session.query(
@ -148,8 +147,8 @@ class MaterialBaseService:
MaterialBase, MaterialBase,
total_inv.label('total_inv'), total_inv.label('total_inv'),
total_avail.label('total_avail') total_avail.label('total_avail')
).outerjoin(buy_sub, MaterialBase.id == buy_sub.c.base_id)\ ).outerjoin(buy_sub, MaterialBase.id == buy_sub.c.base_id) \
.outerjoin(semi_sub, MaterialBase.id == semi_sub.c.base_id)\ .outerjoin(semi_sub, MaterialBase.id == semi_sub.c.base_id) \
.outerjoin(prod_sub, MaterialBase.id == prod_sub.c.base_id) .outerjoin(prod_sub, MaterialBase.id == prod_sub.c.base_id)
if filters: if filters:
@ -263,7 +262,6 @@ class MaterialBaseService:
raise ValueError(f"已存在相同名称和规格的数据 (ID: {exist.id})") raise ValueError(f"已存在相同名称和规格的数据 (ID: {exist.id})")
new_material = MaterialBase( new_material = MaterialBase(
# [修改] 移除了 'IRIS' 默认值
company_name=data.get('companyName'), company_name=data.get('companyName'),
name=data['name'], name=data['name'],
common_name=data.get('commonName'), common_name=data.get('commonName'),
@ -353,14 +351,13 @@ class MaterialBaseService:
raise e raise e
# ============================================================================== # ==============================================================================
# [核心修改] 统一资产统计导出（增加用户权限脱敏） # [核心修改] 统一资产统计导出（增加最高单价计算逻辑）
# ============================================================================== # ==============================================================================
@staticmethod @staticmethod
def export_excel(filters=None, user_permissions=None): def export_excel(filters=None, user_permissions=None):
""" """
全口径资产统计报表： 全口径资产统计报表：
逻辑：先查库存表 (Buy, Semi, Product)，关联基础信息，只导出实际存在的库存记录。 根据基础信息列表和库存表，计算出每个物料的最高历史单价并进行导出
并根据用户权限对字段进行脱敏。
""" """
try: try:
# 1. 构造基础信息的筛选条件 (用于过滤库存) # 1. 构造基础信息的筛选条件 (用于过滤库存)
@ -412,65 +409,61 @@ class MaterialBaseService:
query_product = query_product.filter(cond) query_product = query_product.filter(cond)
list_product = query_product.all() list_product = query_product.all()
# 2.4 计算每个物料基础的最高单价/成本 # ====================================================
buy_max = defaultdict(float) # [核心新增] 预先计算每个 base_id 的全局最高历史单价
# 优先级：采购件 > 半成品 > 成品
# ====================================================
buy_max_prices = {}
for stock, base in list_buy: for stock, base in list_buy:
p = float(stock.pre_tax_unit_price or 0) price = float(stock.pre_tax_unit_price or 0)
if p > buy_max[base.id]: if price > buy_max_prices.get(base.id, 0):
buy_max[base.id] = p buy_max_prices[base.id] = price
semi_max = defaultdict(float) semi_max_prices = {}
for stock, base in list_semi: for stock, base in list_semi:
p = float(stock.raw_material_cost or 0) + float(stock.manual_cost or 0) # 兼容如果有 unit_total_cost 字段则用，否则用 原料+人工
if p > semi_max[base.id]: if hasattr(stock, 'unit_total_cost') and stock.unit_total_cost is not None:
semi_max[base.id] = p price = float(stock.unit_total_cost)
product_max = defaultdict(float)
for stock, base in list_product:
p = float(stock.raw_material_cost or 0) + float(stock.manual_cost or 0)
if p > product_max[base.id]:
product_max[base.id] = p
highest_price = {}
all_base_ids = set()
for stock, base in list_buy:
all_base_ids.add(base.id)
for stock, base in list_semi:
all_base_ids.add(base.id)
for stock, base in list_product:
all_base_ids.add(base.id)
for base_id in all_base_ids:
price = None
buy_val = buy_max.get(base_id)
if buy_val is not None and buy_val > 0:
price = buy_val
else: else:
semi_val = semi_max.get(base_id) price = float(stock.raw_material_cost or 0) + float(stock.manual_cost or 0)
if semi_val is not None and semi_val > 0:
price = semi_val if price > semi_max_prices.get(base.id, 0):
semi_max_prices[base.id] = price
product_max_prices = {}
for stock, base in list_product:
if hasattr(stock, 'unit_total_cost') and stock.unit_total_cost is not None:
price = float(stock.unit_total_cost)
else: else:
prod_val = product_max.get(base_id) price = float(stock.raw_material_cost or 0) + float(stock.manual_cost or 0)
if prod_val is not None and prod_val > 0:
price = prod_val if price > product_max_prices.get(base.id, 0):
highest_price[base_id] = price or 0.0 product_max_prices[base.id] = price
# 构造获取某个物料最高价的闭包函数
def get_highest_price(base_id):
if base_id in buy_max_prices and buy_max_prices[base_id] > 0:
return buy_max_prices[base_id]
if base_id in semi_max_prices and semi_max_prices[base_id] > 0:
return semi_max_prices[base_id]
if base_id in product_max_prices and product_max_prices[base_id] > 0:
return product_max_prices[base_id]
return 0.0
# 3. 数据整合 # 3. 数据整合
all_rows = [] all_rows = []
# 处理采购件 # 处理采购件
for stock, base in list_buy: for stock, base in list_buy:
# 价格计算：使用当前物料基础的最高单价
unit_price = highest_price.get(base.id, 0.0)
tax_rate = float(stock.tax_rate or 0)
# 根据新单价重新计算含税单价
price_incl = unit_price * (1 + tax_rate / 100.0)
qty = float(stock.stock_quantity or 0) qty = float(stock.stock_quantity or 0)
# 使用该物料的全局最高单价作为不含税单价
highest_excl_price = get_highest_price(base.id)
tax_rate = float(stock.tax_rate or 0)
# 计算不含税总价 = 数量 * 不含税单价 # 计算含税单价和总额
total_val_excl = qty * unit_price highest_incl_price = highest_excl_price * (1 + tax_rate / 100.0)
# 计算含税总价 = 数量 * 含税单价 total_val_excl = qty * highest_excl_price
total_val_incl = qty * price_incl total_val_incl = qty * highest_incl_price
ident = stock.batch_number or stock.serial_number or stock.barcode or stock.sku ident = stock.batch_number or stock.serial_number or stock.barcode or stock.sku
@ -483,23 +476,21 @@ class MaterialBaseService:
"date": stock.in_date, "date": stock.in_date,
"qty": qty, "qty": qty,
"avail": float(stock.available_quantity or 0), "avail": float(stock.available_quantity or 0),
"price_excl": unit_price, "price_excl": highest_excl_price,
"total_val_excl": total_val_excl, "total_val_excl": total_val_excl,
"tax": tax_rate, "tax": tax_rate,
"price_incl": price_incl, "price_incl": highest_incl_price,
"total_val": total_val_incl "total_val": total_val_incl
}) })
# 处理半成品 # 处理半成品
for stock, base in list_semi: for stock, base in list_semi:
# 使用当前物料基础的最高单价
unit_price = highest_price.get(base.id, 0.0)
qty = float(stock.stock_quantity or 0) qty = float(stock.stock_quantity or 0)
# 使用该物料的全局最高单价作为成本
highest_cost = get_highest_price(base.id)
# 半成品不含税总价 = 数量 * 单价 total_val_excl = qty * highest_cost
total_val_excl = qty * unit_price total_val_incl = qty * highest_cost # 半成品无税
# 含税总价同上 (税率0)
total_val_incl = qty * unit_price
ident = stock.batch_number or stock.serial_number or stock.barcode or stock.sku ident = stock.batch_number or stock.serial_number or stock.barcode or stock.sku
@ -512,21 +503,21 @@ class MaterialBaseService:
"date": stock.production_date, "date": stock.production_date,
"qty": qty, "qty": qty,
"avail": float(stock.available_quantity or 0), "avail": float(stock.available_quantity or 0),
"price_excl": unit_price, "price_excl": highest_cost,
"total_val_excl": total_val_excl, "total_val_excl": total_val_excl,
"tax": 0.0, "tax": 0.0,
"price_incl": unit_price, "price_incl": highest_cost,
"total_val": total_val_incl "total_val": total_val_incl
}) })
# 处理成品 # 处理成品
for stock, base in list_product: for stock, base in list_product:
# 使用当前物料基础的最高单价
unit_price = highest_price.get(base.id, 0.0)
qty = float(stock.stock_quantity or 0) qty = float(stock.stock_quantity or 0)
# 使用该物料的全局最高单价作为成本
highest_cost = get_highest_price(base.id)
total_val_excl = qty * unit_price total_val_excl = qty * highest_cost
total_val_incl = qty * unit_price total_val_incl = qty * highest_cost
ident = stock.serial_number or stock.barcode or stock.sku ident = stock.serial_number or stock.barcode or stock.sku
@ -539,10 +530,10 @@ class MaterialBaseService:
"date": stock.production_date, "date": stock.production_date,
"qty": qty, "qty": qty,
"avail": float(stock.available_quantity or 0), "avail": float(stock.available_quantity or 0),
"price_excl": unit_price, "price_excl": highest_cost,
"total_val_excl": total_val_excl, "total_val_excl": total_val_excl,
"tax": 0.0, "tax": 0.0,
"price_incl": unit_price, "price_incl": highest_cost,
"total_val": total_val_incl "total_val": total_val_incl
}) })
@ -559,7 +550,7 @@ class MaterialBaseService:
ws = wb.active ws = wb.active
ws.title = "库存统计" ws.title = "库存统计"
# 表头 # 表头 (严格对应你的图 5)
headers = [ headers = [
"所属公司", "资产名称", "规格型号", "物料类型", "所属公司", "资产名称", "规格型号", "物料类型",
"类别一级", "类别二级", "类别三级", "类别四级", "类别五级", "类别一级", "类别二级", "类别三级", "类别四级", "类别五级",
@ -582,7 +573,7 @@ class MaterialBaseService:
col_idx['spec'] = idx col_idx['spec'] = idx
elif header == "物料类型": elif header == "物料类型":
col_idx['type'] = idx col_idx['type'] = idx
elif header in ("类别一级","类别二级","类别三级","类别四级","类别五级"): elif header in ("类别一级", "类别二级", "类别三级", "类别四级", "类别五级"):
col_idx.setdefault('category_cols', []).append(idx) col_idx.setdefault('category_cols', []).append(idx)
elif header == "计量单位": elif header == "计量单位":
col_idx['unit'] = idx col_idx['unit'] = idx
@ -679,13 +670,17 @@ class MaterialBaseService:
# 根据数据来源检查对应模块的权限 # 根据数据来源检查对应模块的权限
if row_type == '采购件': if row_type == '采购件':
# 校验采购模块的价格权限 # 校验采购模块的价格权限
has_price_perm = any(p in user_permissions for p in ['inbound_buy:postTaxUnitPrice', 'inbound_buy:preTaxUnitPrice', 'inbound_buy:totalAmount']) has_price_perm = any(p in user_permissions for p in
['inbound_buy:postTaxUnitPrice', 'inbound_buy:preTaxUnitPrice',
'inbound_buy:totalAmount'])
elif row_type == '半成品': elif row_type == '半成品':
# 校验半成品模块的成本权限 # 校验半成品模块的成本权限
has_price_perm = any(p in user_permissions for p in ['inbound_semi:rawMaterialCost', 'inbound_semi:manualCost']) has_price_perm = any(p in user_permissions for p in
['inbound_semi:rawMaterialCost', 'inbound_semi:manualCost'])
elif row_type == '成品': elif row_type == '成品':
# 校验成品模块的成本权限 # 校验成品模块的成本权限
has_price_perm = any(p in user_permissions for p in ['inbound_product:rawMaterialCost', 'inbound_product:manualCost']) has_price_perm = any(p in user_permissions for p in
['inbound_product:rawMaterialCost', 'inbound_product:manualCost'])
else: else:
# 未知类型，默认隐藏价格列 # 未知类型，默认隐藏价格列
has_price_perm = False has_price_perm = False