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

针对于采购页面进行优化逻辑

Browse Source
This commit is contained in:
dxc
2026-01-28 11:22:08 +08:00
parent 87864a1c4f
commit 6f4917f57e
6 changed files with 1325 additions and 534 deletions
Download Patch File Download Diff File Expand all files Collapse all files

277
inventory-backend/app/services/inbound/buy_service.py
View File

@ -2,76 +2,88 @@ from app.extensions import db
from app.models.material import MaterialBase
from app.models.stock import StockBuy
from datetime import datetime
from sqlalchemy import or_, func # 引入 func 用于聚合计算
import traceback
class BuyInboundService:
@staticmethod
def handle_inbound(data):
"""新增入库：自动关联/创建基础信息 + 创建库存记录"""
def search_base_material(keyword):
try:
# 0. 基础校验
if not data.get('spec_model') or not data.get('material_name'):
raise ValueError("缺少必要的物料名称或规格型号")
# 1. 关联逻辑：通过规格型号(spec_model)查找基础库
material = MaterialBase.query.filter_by(spec_model=data['spec_model']).first()
# 如果不存在，则新建 MaterialBase
if not material:
material = MaterialBase(
name=data['material_name'],
spec_model=data['spec_model'],
category=data.get('category'),
material_type='采购件',
unit=data.get('unit'),
visibility_level=data.get('visibility_level', 0),
manual_link=data.get('manual_link'),
product_image=data.get('product_image'),
is_enabled=True
if not keyword:
return []
query = MaterialBase.query.filter(
MaterialBase.is_enabled == True,
or_(
MaterialBase.name.ilike(f'%{keyword}%'),
MaterialBase.spec_model.ilike(f'%{keyword}%')
)
db.session.add(material)
db.session.flush() # 立即执行，拿到 material.id
).limit(20)
# 2. 处理日期 (兼容性处理)
in_date_val = None
results = []
for item in query.all():
results.append({
'id': item.id,
'name': item.name,
'spec': item.spec_model,
'category': item.category,
'unit': item.unit,
'type': item.material_type,
'status': '启用'
})
return results
except Exception as e:
traceback.print_exc()
return []
@staticmethod
def handle_inbound(data):
try:
base_id = data.get('base_id')
if not base_id:
raise ValueError("必须选择基础物料进行入库 (缺少 base_id)")
material = MaterialBase.query.get(base_id)
if not material:
raise ValueError(f"ID为 {base_id} 的基础物料不存在")
in_date_val = datetime.utcnow().date()
if data.get('in_date'):
try:
in_date_val = datetime.strptime(data['in_date'], '%Y-%m-%d %H:%M:%S').date()
if len(str(data['in_date'])) > 10:
in_date_val = datetime.strptime(str(data['in_date'])[:10], '%Y-%m-%d').date()
else:
in_date_val = datetime.strptime(str(data['in_date']), '%Y-%m-%d').date()
except ValueError:
try:
in_date_val = datetime.strptime(data['in_date'], '%Y-%m-%d').date()
except ValueError:
in_date_val = datetime.utcnow().date()
pass
# --- 修改部分：增加字段兼容性，确保前端传参能被正确读取 ---
in_qty = float(data.get('in_quantity') or data.get('qty_inbound') or 0)
u_price = float(data.get('unit_price') or data.get('price_unit') or 0)
# ---------------------------------------------------
in_qty = float(data.get('in_quantity') or 0)
u_price = float(data.get('unit_price') or 0)
# 3. 创建 StockBuy
new_stock = StockBuy(
base_id=material.id,
sku=data.get('sku'),
in_date=in_date_val,
serial_number=data.get('serial_number'),
batch_number=data.get('batch_number'),
barcode=data.get('barcode'),
status='在库',
inspection_status=data.get('inspection_status'),
in_quantity=in_qty,
stock_quantity=in_qty,
available_quantity=in_qty,
warehouse_location=data.get('warehouse_location') or data.get('warehouse_loc'),
inspection_status=data.get('inspection_status', '未检'),
warehouse_location=data.get('warehouse_location'),
unit_price=u_price,
total_price=in_qty * u_price,
currency=data.get('currency', 'CNY'),
exchange_rate=data.get('exchange_rate', 1.0),
supplier_name=data.get('supplier_name'),
buyer_name=data.get('buyer_name'),
buyer_email=data.get('buyer_email'),
original_link=data.get('original_link'),
buyer_name=data.get('purchaser'),
buyer_email=data.get('purchaser_email'),
original_link=data.get('source_link'),
detail_link=data.get('detail_link'),
arrival_photo=data.get('arrival_photo')
arrival_photo=data.get('arrival_photo'),
remark=data.get('remark')
)
db.session.add(new_stock)
@ -84,101 +96,180 @@ class BuyInboundService:
@staticmethod
def update_inbound(stock_id, data):
"""更新入库：支持级联更新基础信息 + 自动重算总价"""
try:
print(f"----- UPDATE DEBUG: ID={stock_id} -----")
print(f"Payload: {data}")
stock = StockBuy.query.get(stock_id)
if not stock:
raise ValueError("记录不存在")
# 1. 更新普通字段 (增加对 warehouse_loc 的兼容)
if 'serial_number' in data: stock.serial_number = data['serial_number']
if 'batch_number' in data: stock.batch_number = data['batch_number']
if 'warehouse_location' in data: stock.warehouse_location = data['warehouse_location']
if 'warehouse_loc' in data: stock.warehouse_location = data['warehouse_loc']
if 'supplier_name' in data: stock.supplier_name = data['supplier_name']
if 'status' in data: stock.status = data['status']
if 'inspection_status' in data: stock.inspection_status = data['inspection_status']
if 'arrival_photo' in data: stock.arrival_photo = data['arrival_photo']
if 'remark' in data: stock.remark = data['remark']
field_mapping = {
'sku': 'sku',
'barcode': 'barcode',
'warehouse_location': 'warehouse_location',
'serial_number': 'serial_number',
'batch_number': 'batch_number',
'status': 'status',
'inspection_status': 'inspection_status',
'supplier_name': 'supplier_name',
'detail_link': 'detail_link',
'arrival_photo': 'arrival_photo',
'remark': 'remark',
'currency': 'currency',
'exchange_rate': 'exchange_rate',
'purchaser': 'buyer_name',
'purchaser_email': 'buyer_email',
'source_link': 'original_link'
}
# 2. 级联更新基础信息 (MaterialBase)
if stock.material:
if 'material_name' in data: stock.material.name = data['material_name']
if 'category' in data: stock.material.category = data['category']
if 'unit' in data: stock.material.unit = data['unit']
for frontend_key, db_attr in field_mapping.items():
if frontend_key in data:
setattr(stock, db_attr, data[frontend_key])
# 3. 核心逻辑：数量与价格联动 (增加对前端返回字段名 qty_inbound 和 price_unit 的识别)
qty_changed = False
price_changed = False
# (A) 数量变更 -> 更新库存和可用量
new_qty_input = data.get('in_quantity') or data.get('qty_inbound')
if new_qty_input is not None:
new_qty = float(new_qty_input)
if 'in_quantity' in data:
new_qty = float(data['in_quantity'])
old_qty = float(stock.in_quantity)
diff = new_qty - old_qty
if diff != 0:
if new_qty != old_qty:
diff = new_qty - old_qty
stock.in_quantity = new_qty
stock.stock_quantity = float(stock.stock_quantity) + diff
stock.available_quantity = float(stock.available_quantity) + diff
qty_changed = True
# (B) 单价变更
new_price_input = data.get('unit_price') or data.get('price_unit')
if new_price_input is not None:
new_price = float(new_price_input)
if new_price != float(stock.unit_price):
if 'unit_price' in data:
new_price = float(data['unit_price'])
old_price = float(stock.unit_price)
if new_price != old_price:
stock.unit_price = new_price
price_changed = True
# (C) 重算总价
if qty_changed or price_changed:
stock.total_price = float(stock.in_quantity) * float(stock.unit_price)
db.session.commit()
print("----- UPDATE SUCCESS -----")
return stock
except Exception as e:
db.session.rollback()
print(f"----- UPDATE FAILED: {str(e)} -----")
traceback.print_exc()
raise e
@staticmethod
def delete_inbound(stock_id):
"""删除逻辑：孤儿策略（如果MaterialBase无其他引用则一并删除）"""
try:
stock = StockBuy.query.get(stock_id)
if not stock:
raise ValueError("记录不存在")
# 1. 记下 base_id
material_id = stock.base_id
# 2. 删除库存记录
db.session.delete(stock)
db.session.flush()
# 3. 检查是否还有残留
remaining_count = StockBuy.query.filter_by(base_id=material_id).count()
if remaining_count == 0:
print(f"触发级联删除: MaterialBase ID {material_id} 已无关联，执行清理。")
material = MaterialBase.query.get(material_id)
if material:
db.session.delete(material)
db.session.commit()
return True
except Exception as e:
db.session.rollback()
print(f"删除失败: {e}")
raise e
@staticmethod
def get_list(page, limit):
def get_list(page, limit, keyword=None):
try:
pagination = StockBuy.query.order_by(StockBuy.id.desc()).paginate(page=page, per_page=limit)
items = [item.to_dict() for item in pagination.items]
# 1. 查询分页数据
query = db.session.query(StockBuy).outerjoin(MaterialBase, StockBuy.base_id == MaterialBase.id)
if keyword:
query = query.filter(
or_(
MaterialBase.name.ilike(f'%{keyword}%'),
MaterialBase.spec_model.ilike(f'%{keyword}%'),
StockBuy.batch_number.ilike(f'%{keyword}%'),
StockBuy.serial_number.ilike(f'%{keyword}%'),
StockBuy.sku.ilike(f'%{keyword}%')
)
)
pagination = query.order_by(StockBuy.id.desc()).paginate(page=page, per_page=limit, error_out=False)
# ---------------------------------------------------------------------
# 新增逻辑：计算总库存
# ---------------------------------------------------------------------
# 2. 提取当前页所有涉及的 base_id
current_items = pagination.items
base_ids = list(set([item.base_id for item in current_items if item.base_id]))
# 3. 聚合查询：一次性查出这些 base_id 对应的 stock_quantity 和 available_quantity 总和
stock_map = {}
if base_ids:
# SELECT base_id, SUM(stock_quantity), SUM(available_quantity) FROM stock_buy WHERE base_id IN (...) GROUP BY base_id
aggregates = db.session.query(
StockBuy.base_id,
func.sum(StockBuy.stock_quantity).label('total_stock'),
func.sum(StockBuy.available_quantity).label('total_avail')
).filter(StockBuy.base_id.in_(base_ids)).group_by(StockBuy.base_id).all()
for agg in aggregates:
stock_map[agg.base_id] = {
'total_stock': float(agg.total_stock or 0),
'total_avail': float(agg.total_avail or 0)
}
# ---------------------------------------------------------------------
items = []
for item in current_items:
mat_name = item.material.name if item.material else '未知物料'
mat_spec = item.material.spec_model if item.material else ''
mat_cat = item.material.category if item.material else ''
mat_unit = item.material.unit if item.material else ''
mat_type = item.material.material_type if item.material else ''
# 获取该物料的统计数据
stats = stock_map.get(item.base_id, {'total_stock': 0, 'total_avail': 0})
d = {
'id': item.id,
'base_id': item.base_id,
'material_name': mat_name,
'spec_model': mat_spec,
'category': mat_cat,
'unit': mat_unit,
'material_type': mat_type,
'sku': item.sku,
'inbound_date': str(item.in_date) if item.in_date else '',
'barcode': item.barcode,
'serial_number': item.serial_number,
'batch_number': item.batch_number,
'status': item.status,
'inspection_status': item.inspection_status,
# --- 原始批次数据 (用于编辑) ---
'qty_inbound': float(item.in_quantity or 0),
'qty_stock': float(item.stock_quantity or 0),
'qty_available': float(item.available_quantity or 0),
# --- [新增] 聚合统计数据 (用于列表显示) ---
'sum_stock': stats['total_stock'],
'sum_available': stats['total_avail'],
'warehouse_loc': item.warehouse_location,
'unit_price': float(item.unit_price or 0),
'total_price': float(item.total_price or 0),
'currency': item.currency,
'exchange_rate': float(item.exchange_rate or 1),
'supplier_name': item.supplier_name,
'purchaser': item.buyer_name,
'purchaser_email': item.buyer_email,
'source_link': item.original_link,
'detail_link': item.detail_link,
'arrival_photo': item.arrival_photo,
'remark': item.remark
}
items.append(d)
return {"total": pagination.total, "items": items}
except Exception as e:
print(f"查询列表失败: {e}")
print(f"List Error: {e}")
traceback.print_exc()
return {"total": 0, "items": []}