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 e31ef59df0
6 changed files with 1228 additions and 526 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,77 +1,111 @@
# app/services/inbound/buy_service.py
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_
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("缺少必要的物料名称或规格型号")
if not keyword:
return []
# 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
# 过滤条件：名称或规格包含关键词，且 is_enabled 为 True
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:
# 1. 核心校验
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} 的基础物料不存在")
# 2. 处理日期 (防止空字符串报错)
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()
date_str = str(data['in_date'])
# 截取前10位 YYYY-MM-DD
if len(date_str) > 10:
in_date_val = datetime.strptime(date_str[:10], '%Y-%m-%d').date()
else:
in_date_val = datetime.strptime(date_str, '%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)
# ---------------------------------------------------
# 3. 数据转换
in_qty = float(data.get('in_quantity') or 0)
u_price = float(data.get('unit_price') or 0)
# 3. 创建 StockBuy
# 4. 创建 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'),
stock_quantity=in_qty, # 初始库存 = 入库量
available_quantity=in_qty, # 初始可用 = 入库量
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'),
# [字段映射] 前端 -> DB
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)
@ -80,105 +114,182 @@ class BuyInboundService:
except Exception as e:
db.session.rollback()
print(f"Insert Error: {str(e)}")
raise e
@staticmethod
def update_inbound(stock_id, data):
"""更新入库：支持级联更新基础信息 + 自动重算总价"""
"""
更新入库逻辑
"""
try:
print(f"----- UPDATE DEBUG: ID={stock_id} -----")
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']
# 1. 字段映射字典：前端Key -> Model属性名
# 使用映射可以避免写大量 if...else，且逻辑更清晰
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',
# 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']
# 关键映射
'purchaser': 'buyer_name',
'purchaser_email': 'buyer_email',
'source_link': 'original_link'
}
# 3. 核心逻辑：数量与价格联动 (增加对前端返回字段名 qty_inbound 和 price_unit 的识别)
# 遍历更新 (排除日期、数量、单价，下面单独处理)
for frontend_key, db_attr in field_mapping.items():
if frontend_key in data:
setattr(stock, db_attr, data[frontend_key])
# 2. 核心数值逻辑 (数量 & 单价 & 总价)
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)
# (A) 处理入库数量变更
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:
print(f"Quantity Changed: {old_qty} -> {new_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):
# (B) 处理单价变更
if 'unit_price' in data:
new_price = float(data['unit_price'])
old_price = float(stock.unit_price)
if new_price != old_price:
print(f"Price Changed: {old_price} -> {new_price}")
stock.unit_price = new_price
price_changed = True
# (C) 重算总价
# (C) 强制重算总价
if qty_changed or price_changed:
stock.total_price = float(stock.in_quantity) * float(stock.unit_price)
print(f"New Total Price: {stock.total_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]
# 使用 Outer Join 确保即使 MaterialBase 物理删除，库存记录也不会报错或消失
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)
items = []
for item in pagination.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 ''
# 构建返回字典
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),
'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,
# [关键映射] DB -> Frontend
'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}")
# 打印错误到 Docker 日志
print(f"List Error: {e}")
traceback.print_exc()
return {"total": 0, "items": []}