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

feat(inbound): handle batch duplication gracefully, support spec search, and auto-fill historical locations

Browse Source
This commit is contained in:
DXC
2026-04-08 09:41:04 +08:00
parent c58a0a6d14
commit 9380508a89
9 changed files with 141 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,6 +1,7 @@
# inventory-backend/app/services/inbound/buy_service.py
from app.extensions import db
from app.models.inbound.buy import StockBuy
from app.models.inbound.product import StockProduct
from app.models.base import MaterialBase
from datetime import datetime, timedelta, timezone
from sqlalchemy import or_, func, text, and_
@ -16,6 +17,10 @@ class BuyInboundService:
# ============================================================
@staticmethod
def _check_unique(base_id, serial_number, batch_number, exclude_id=None):
"""
校验序列号/批号是否已存在
返回: None 表示校验通过，str 类型的错误信息表示校验失败
"""
if serial_number:
query = StockBuy.query.filter(StockBuy.serial_number == serial_number)
if exclude_id:
@ -23,7 +28,7 @@ class BuyInboundService:
exists = query.first()
if exists:
occupied_name = exists.base.name if exists.base else "未知物料"
raise ValueError(f"序列号【{serial_number}】已存在！被物料 [{occupied_name}] 占用，请核查。")
return f"序列号【{serial_number}】已存在！被物料 [{occupied_name}] 占用，请核查。"
if batch_number and base_id:
query = StockBuy.query.filter(
@ -33,7 +38,9 @@ class BuyInboundService:
if exclude_id:
query = query.filter(StockBuy.id != exclude_id)
if query.first():
raise ValueError(f"该物料已存在批号【{batch_number}】，请勿重复录入，可直接在该批次下追加库存。")
return f"该物料已存在批号【{batch_number}】，请勿重复录入，可直接在该批次下追加库存。"
return None
# ============================================================
# 1. 基础物料搜索
@ -59,9 +66,25 @@ class BuyInboundService:
items = []
for item in pagination.items:
# 查询最近一次入库的库位（采购入库 + 成品入库）
last_location = ''
# 查采购入库
last_buy = StockBuy.query.filter(
StockBuy.base_id == item.id
).order_by(StockBuy.in_date.desc()).first()
if last_buy and last_buy.warehouse_location:
last_location = last_buy.warehouse_location
else:
# 查成品入库
last_product = StockProduct.query.filter(
StockProduct.base_id == item.id
).order_by(StockProduct.in_date.desc()).first()
if last_product and last_product.warehouse_location:
last_location = last_product.warehouse_location
items.append({
'id': item.id,
'company_name': item.company_name, # [新增]
'company_name': item.company_name,
'name': item.name,
'spec': item.spec_model,
'category': item.category,
@ -70,7 +93,8 @@ class BuyInboundService:
'brand': getattr(item, 'brand', ''),
'manufacturer': getattr(item, 'manufacturer', ''),
'pinyin': getattr(item, 'pinyin', ''),
'status': '启用'
'status': '启用',
'history_location': last_location
})
return {
@ -113,11 +137,14 @@ class BuyInboundService:
if not has_report_file:
raise ValueError(f"物料【{material.name}】为强管控物料，必须提供检测报告文件或外部链接")
BuyInboundService._check_unique(
# 校验批号/序列号唯一性
unique_error = BuyInboundService._check_unique(
base_id=base_id,
serial_number=data.get('serial_number'),
batch_number=data.get('batch_number')
)
if unique_error:
return {'error': unique_error}
beijing_tz = timezone(timedelta(hours=8))
current_time = datetime.now(beijing_tz).replace(tzinfo=None)

37
inventory-backend/app/services/inbound/product_service.py
View File

@ -1,6 +1,8 @@
# app/services/inbound/product_service.py
from app.extensions import db
from app.models.base import MaterialBase
from app.models.inbound.buy import StockBuy
from app.models.inbound.semi import StockSemi
from app.models.outbound import TransOutbound
from datetime import datetime, timedelta, timezone
from sqlalchemy import or_, func, text, and_
@ -21,7 +23,9 @@ class ProductInboundService:
exists = query.first()
if exists:
occupied_name = exists.base.name if (hasattr(exists, 'base') and exists.base) else "未知物料"
raise ValueError(f"序列号【{serial_number}】已存在！被成品 [{occupied_name}] 占用，请核查。")
return f"序列号【{serial_number}】已存在！被成品 [{occupied_name}] 占用，请核查。"
return None
@staticmethod
def search_base_material(keyword, page=1, limit=50):
@ -40,6 +44,29 @@ class ProductInboundService:
pagination = query.paginate(page=page, per_page=limit, error_out=False)
results = []
for item in pagination.items:
# 查询最近一次入库的库位（成品入库 + 采购入库 + 半成品入库）
last_location = ''
# 查成品入库
last_product = StockProduct.query.filter(
StockProduct.base_id == item.id
).order_by(StockProduct.in_date.desc()).first()
if last_product and last_product.warehouse_location:
last_location = last_product.warehouse_location
else:
# 查采购入库
last_buy = StockBuy.query.filter(
StockBuy.base_id == item.id
).order_by(StockBuy.in_date.desc()).first()
if last_buy and last_buy.warehouse_location:
last_location = last_buy.warehouse_location
else:
# 查半成品入库
last_semi = StockSemi.query.filter(
StockSemi.base_id == item.id
).order_by(StockSemi.in_date.desc()).first()
if last_semi and last_semi.warehouse_location:
last_location = last_semi.warehouse_location
results.append({
'id': item.id,
'company_name': item.company_name,
@ -48,7 +75,8 @@ class ProductInboundService:
'category': item.category,
'unit': item.unit,
'type': item.material_type,
'status': '启用'
'status': '启用',
'history_location': last_location
})
return {
"items": results,
@ -107,9 +135,12 @@ class ProductInboundService:
if not material.is_enabled:
raise ValueError(f"物料【{material.name}】已停用，无法办理新入库。")
ProductInboundService._check_unique(
# 校验序列号唯一性
unique_error = ProductInboundService._check_unique(
serial_number=data.get('serial_number')
)
if unique_error:
return {'error': unique_error}
beijing_tz = timezone(timedelta(hours=8))
current_time = datetime.now(beijing_tz).replace(tzinfo=None)

39
inventory-backend/app/services/inbound/semi_service.py
View File

@ -1,6 +1,8 @@
# app/services/inbound/semi_service.py
from app.extensions import db
from app.models.base import MaterialBase
from app.models.inbound.buy import StockBuy
from app.models.inbound.product import StockProduct
from app.models.outbound import TransOutbound
from datetime import datetime, timedelta, timezone
from sqlalchemy import or_, func, text, and_
@ -21,7 +23,7 @@ class SemiInboundService:
exists = query.first()
if exists:
occupied_name = exists.base.name if (hasattr(exists, 'base') and exists.base) else "未知物料"
raise ValueError(f"序列号【{serial_number}】已存在！被半成品 [{occupied_name}] 占用，请核查。")
return f"序列号【{serial_number}】已存在！被半成品 [{occupied_name}] 占用，请核查。"
if batch_number and base_id:
query = StockSemi.query.filter(
@ -31,7 +33,9 @@ class SemiInboundService:
if exclude_id:
query = query.filter(StockSemi.id != exclude_id)
if query.first():
raise ValueError(f"该物料已存在批号【{batch_number}】，请勿重复建单，建议在原批次上追加库存。")
return f"该物料已存在批号【{batch_number}】，请勿重复建单，建议在原批次上追加库存。"
return None
@staticmethod
def search_base_material(keyword, page=1, limit=50):
@ -50,6 +54,29 @@ class SemiInboundService:
pagination = query.paginate(page=page, per_page=limit, error_out=False)
results = []
for item in pagination.items:
# 查询最近一次入库的库位（半成品入库 + 采购入库 + 成品入库）
last_location = ''
# 查半成品入库
last_semi = StockSemi.query.filter(
StockSemi.base_id == item.id
).order_by(StockSemi.in_date.desc()).first()
if last_semi and last_semi.warehouse_location:
last_location = last_semi.warehouse_location
else:
# 查采购入库
last_buy = StockBuy.query.filter(
StockBuy.base_id == item.id
).order_by(StockBuy.in_date.desc()).first()
if last_buy and last_buy.warehouse_location:
last_location = last_buy.warehouse_location
else:
# 查成品入库
last_product = StockProduct.query.filter(
StockProduct.base_id == item.id
).order_by(StockProduct.in_date.desc()).first()
if last_product and last_product.warehouse_location:
last_location = last_product.warehouse_location
results.append({
'id': item.id,
'company_name': item.company_name,
@ -58,7 +85,8 @@ class SemiInboundService:
'category': item.category,
'unit': item.unit,
'type': item.material_type,
'status': '启用'
'status': '启用',
'history_location': last_location
})
return {"items": results, "total": pagination.total, "page": page, "has_next": pagination.has_next}
except Exception as e:
@ -114,11 +142,14 @@ class SemiInboundService:
if not material.is_enabled:
raise ValueError(f"物料【{material.name}】已停用，无法办理新入库。")
SemiInboundService._check_unique(
# 校验批号/序列号唯一性
unique_error = SemiInboundService._check_unique(
base_id=base_id,
serial_number=data.get('serial_number'),
batch_number=data.get('batch_number')
)
if unique_error:
return {'error': unique_error}
beijing_tz = timezone(timedelta(hours=8))
current_time = datetime.now(beijing_tz).replace(tzinfo=None)