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-02-10 17:20:06 +08:00
parent b5b0677b01
commit 8ee2a9a45b
4 changed files with 254 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
inventory-backend/app/api/v1/inbound/buy.py
View File

@ -29,7 +29,7 @@ def search_base():
# ------------------------------------------------------------------
# 1. 获取列表 (修改：支持状态筛选)
# 1. 获取列表
# ------------------------------------------------------------------
@inbound_buy_bp.route('/list', methods=['GET'])
def get_list():
@ -38,7 +38,7 @@ def get_list():
limit = request.args.get('pageSize', 15, type=int)
keyword = request.args.get('keyword', '')
# 获取状态列表参数，前端传参格式: statuses=在库,借库
# 获取状态列表参数
statuses_str = request.args.get('statuses', '')
statuses = statuses_str.split(',') if statuses_str else []
@ -126,10 +126,22 @@ def get_supplier_suggestions():
# ------------------------------------------------------------------
# 7. 系统用户建议
# 7. 系统用户建议 (采购人)
# ------------------------------------------------------------------
@inbound_buy_bp.route('/suggestions/users', methods=['GET'])
def get_user_suggestions():
keyword = request.args.get('keyword', '')
data = BuyInboundService.search_system_users(keyword)
return jsonify({"code": 200, "msg": "success", "data": data})
# ------------------------------------------------------------------
# 8. [新增] 链接建议
# ------------------------------------------------------------------
@inbound_buy_bp.route('/suggestions/links', methods=['GET'])
def get_link_suggestions():
base_id = request.args.get('base_id', type=int)
link_type = request.args.get('type', 'original') # original or detail
if not base_id:
return jsonify({"code": 400, "msg": "base_id required"}), 400
data = BuyInboundService.get_history_links(base_id, link_type)
return jsonify({"code": 200, "msg": "success", "data": data})

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

@ -17,19 +17,14 @@ import json
class BuyInboundService:
# ============================================================
# 0. 辅助：唯一性校验 (核心修复)
# 0. 辅助：唯一性校验
# ============================================================
@staticmethod
def _check_unique(base_id, serial_number, batch_number, exclude_id=None):
"""
校验序列号和批号的唯一性逻辑
:param base_id: 当前物料的基础ID
:param serial_number: 序列号
:param batch_number: 批号
:param exclude_id: 排除的ID (用于编辑模式)
"""
# 1. 序列号 (SN) 全局唯一校验
# 解释: 不同规格的物料通常也不应该有相同的SN，防止扫码混淆
if serial_number:
query = StockBuy.query.filter(StockBuy.serial_number == serial_number)
if exclude_id:
@ -37,12 +32,10 @@ class BuyInboundService:
exists = query.first()
if exists:
# [修改] 获取占用该SN的物料名称 (material -> base)
occupied_name = exists.base.name if exists.base else "未知物料"
raise ValueError(f"序列号【{serial_number}】已存在！被物料 [{occupied_name}] 占用，请核查。")
# 2. 批号 (BN) 同物料唯一校验
# 解释: 不同规格的物料可以有相同的批号(如都有 001 批次)，但同一个物料不能重复建单
if batch_number and base_id:
query = StockBuy.query.filter(
StockBuy.base_id == base_id,
@ -60,7 +53,7 @@ class BuyInboundService:
@staticmethod
def search_base_material(keyword):
try:
# [核心修改] 只查询已启用的物料，防止选择已禁用的历史物料
# 只查询已启用的物料
query = MaterialBase.query.filter(MaterialBase.is_enabled == True)
if keyword:
@ -68,7 +61,7 @@ class BuyInboundService:
or_(
MaterialBase.name.ilike(f'%{keyword}%'),
MaterialBase.spec_model.ilike(f'%{keyword}%'),
MaterialBase.pinyin.ilike(f'%{keyword}%') # 假设有拼音搜索
MaterialBase.pinyin.ilike(f'%{keyword}%')
)
)
query = query.order_by(MaterialBase.id.desc()).limit(20)
@ -77,7 +70,7 @@ class BuyInboundService:
results.append({
'id': item.id,
'name': item.name,
'spec': item.spec_model, # 确保这里字段对应正确
'spec': item.spec_model,
'category': item.category,
'unit': item.unit,
'type': item.material_type,
@ -102,18 +95,16 @@ class BuyInboundService:
if not material:
raise ValueError("所选物料不存在")
# [核心修改] 后端二次校验：如果物料已停用，禁止入库
if not material.is_enabled:
raise ValueError(f"物料【{material.name}】已停用，无法办理新入库。")
# --- [修复点] 执行唯一性校验 ---
BuyInboundService._check_unique(
base_id=base_id,
serial_number=data.get('serial_number'),
batch_number=data.get('batch_number')
)
# 时间处理 (强制北京时间)
# 时间处理
beijing_tz = timezone(timedelta(hours=8))
current_time = datetime.now(beijing_tz).replace(tzinfo=None)
in_date_val = current_time
@ -162,7 +153,7 @@ class BuyInboundService:
batch_number=data.get('batch_number'),
status=data.get('status', '在库'),
in_quantity=in_qty,
stock_quantity=in_qty, # 初始库存等于入库数
stock_quantity=in_qty,
available_quantity=in_qty,
inspection_status=data.get('inspection_status', '未检'),
warehouse_location=data.get('warehouse_location'),
@ -195,8 +186,6 @@ class BuyInboundService:
if not stock:
raise ValueError("记录不存在")
# --- [修复点] 编辑时也要校验唯一性 (排除自身ID) ---
# 如果修改了物料(base_id)，或者修改了SN/BN，都需要校验
new_base_id = data.get('base_id', stock.base_id)
new_sn = data.get('serial_number', stock.serial_number)
new_bn = data.get('batch_number', stock.batch_number)
@ -315,7 +304,6 @@ class BuyInboundService:
d = {
'id': item.id,
'base_id': item.base_id,
# [核心修改] 确保这里从关联的 .base 获取信息
'material_name': item.base.name if item.base else '',
'spec_model': item.base.spec_model if item.base else '',
'category': item.base.category if item.base else '',
@ -354,7 +342,7 @@ class BuyInboundService:
return {"total": 0, "items": []}
# ============================================================
# 6. 供应商历史查询
# 6. 供应商历史查询 (根据 base_id)
# ============================================================
@staticmethod
def get_history_suppliers(base_id):
@ -362,7 +350,8 @@ class BuyInboundService:
try:
query = db.session.query(StockBuy.supplier_name).filter(
StockBuy.base_id == base_id,
StockBuy.supplier_name.isnot(None)
StockBuy.supplier_name.isnot(None),
StockBuy.supplier_name != ''
).distinct().order_by(StockBuy.supplier_name)
suppliers = [row[0] for row in query.all()]
return suppliers
@ -370,7 +359,7 @@ class BuyInboundService:
return []
# ============================================================
# 7. 系统用户搜索
# 7. 系统用户搜索 (全局)
# ============================================================
@staticmethod
def search_system_users(keyword):
@ -394,3 +383,22 @@ class BuyInboundService:
return users
except Exception:
return []
# ============================================================
# 8. [新增] 链接建议 (根据 base_id)
# ============================================================
@staticmethod
def get_history_links(base_id, link_type='original'):
"""查询该物料的历史链接，方便复用"""
try:
target_col = StockBuy.original_link if link_type == 'original' else StockBuy.detail_link
query = db.session.query(target_col).filter(
StockBuy.base_id == base_id,
target_col.isnot(None),
target_col != ''
).distinct().limit(10)
links = [row[0] for row in query.all()]
return links
except Exception:
return []