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-11 13:12:05 +08:00
parent d0a237625c
commit 5532c87684
9 changed files with 285 additions and 395 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -156,6 +156,31 @@ class MaterialBaseService:
print(f"查询基础信息列表失败: {e}")
return {"total": 0, "items": []}
@staticmethod
def get_distinct_options():
"""
获取所有已存在的类别和类型 (去重)
用于前端下拉筛选
"""
try:
# 查询所有不为空的类别并去重
categories = db.session.query(MaterialBase.category) \
.filter(MaterialBase.category != None, MaterialBase.category != '') \
.distinct().all()
# 查询所有不为空的类型并去重
types = db.session.query(MaterialBase.material_type) \
.filter(MaterialBase.material_type != None, MaterialBase.material_type != '') \
.distinct().all()
return {
"categories": [c[0] for c in categories],
"types": [t[0] for t in types]
}
except Exception as e:
traceback.print_exc()
return {"categories": [], "types": []}
@staticmethod
def create_material(data):
"""新增基础信息"""

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

@ -10,25 +10,19 @@ import json
class BuyInboundService:
# ============================================================
# 0. 辅助：唯一性校验
# 0. 辅助：唯一性校验 (保持不变)
# ============================================================
@staticmethod
def _check_unique(base_id, serial_number, batch_number, exclude_id=None):
"""
校验序列号和批号的唯一性逻辑
"""
# 1. 序列号 (SN) 全局唯一校验
if serial_number:
query = StockBuy.query.filter(StockBuy.serial_number == serial_number)
if exclude_id:
query = query.filter(StockBuy.id != exclude_id)
exists = query.first()
if exists:
occupied_name = exists.base.name if exists.base else "未知物料"
raise ValueError(f"序列号【{serial_number}】已存在！被物料 [{occupied_name}] 占用，请核查。")
# 2. 批号 (BN) 同物料唯一校验
if batch_number and base_id:
query = StockBuy.query.filter(
StockBuy.base_id == base_id,
@ -36,74 +30,43 @@ class BuyInboundService:
)
if exclude_id:
query = query.filter(StockBuy.id != exclude_id)
if query.first():
raise ValueError(f"该物料已存在批号【{batch_number}】，请勿重复录入，可直接在该批次下追加库存。")
# ============================================================
# 1. 基础物料搜索 (增强模糊匹配：多字段、多关键词、分词匹配)
# 1. 基础物料搜索 (修复：逻辑优先级 & 模糊匹配)
# ============================================================
@staticmethod
def search_base_material(keyword):
def search_base_material(keyword, page=1, limit=50):
try:
# 1. 基础查询：只查询启用的
query = MaterialBase.query.filter(MaterialBase.is_enabled == True)
if keyword:
# 1. 清理关键词：去除首尾空格，并将多个连续空格替换为单个
import re
keyword_clean = re.sub(r'\s+', ' ', keyword.strip())
# 2. 支持两种搜索模式：
# a) 精确短语匹配：用双引号包裹，如 "蓝色电阻"
# b) 多关键词 AND 匹配：空格分隔，如 "蓝色 电阻"
# c) 单关键词模糊匹配：如 "蓝色"
if keyword_clean.startswith('"') and keyword_clean.endswith('"'):
# 精确短语匹配
exact_phrase = keyword_clean[1:-1]
if exact_phrase:
k_str = f'%{exact_phrase}%'
conditions = [
MaterialBase.name.ilike(k_str),
MaterialBase.spec_model.ilike(k_str),
MaterialBase.pinyin.ilike(k_str),
MaterialBase.category.ilike(k_str),
MaterialBase.material_type.ilike(k_str)
]
if hasattr(MaterialBase, 'brand'):
conditions.append(MaterialBase.brand.ilike(k_str))
if hasattr(MaterialBase, 'manufacturer'):
conditions.append(MaterialBase.manufacturer.ilike(k_str))
query = query.filter(or_(*conditions))
else:
# 多关键词 AND 匹配
keywords = keyword_clean.split()
if keywords:
and_conditions = []
for word in keywords:
k_str = f'%{word}%'
word_conditions = [
MaterialBase.name.ilike(k_str),
MaterialBase.spec_model.ilike(k_str),
MaterialBase.pinyin.ilike(k_str),
MaterialBase.category.ilike(k_str),
MaterialBase.material_type.ilike(k_str)
]
if hasattr(MaterialBase, 'brand'):
word_conditions.append(MaterialBase.brand.ilike(k_str))
if hasattr(MaterialBase, 'manufacturer'):
word_conditions.append(MaterialBase.manufacturer.ilike(k_str))
and_conditions.append(or_(*word_conditions))
# 使用 AND 连接所有关键词条件
if and_conditions:
query = query.filter(and_(*and_conditions))
k = keyword.strip()
k_str = f'%{k}%'
# 按 ID 倒序排序
# [核心修复] 使用 and_ 确保逻辑优先级正确
# 生成 SQL 类似: WHERE is_enabled = true AND (name LIKE %k% OR spec LIKE %k% ...)
query = query.filter(and_(
or_(
MaterialBase.name.ilike(k_str), # ilike 忽略大小写
MaterialBase.spec_model.ilike(k_str),
MaterialBase.pinyin.ilike(k_str),
# 如果需要支持 ID 搜索，取消下面注释
# func.cast(MaterialBase.id, String).ilike(k_str)
)
))
# 2. 排序：ID 倒序
query = query.order_by(MaterialBase.id.desc())
results = []
for item in query.all():
results.append({
# 3. 分页
pagination = query.paginate(page=page, per_page=limit, error_out=False)
items = []
for item in pagination.items:
items.append({
'id': item.id,
'name': item.name,
'spec': item.spec_model,
@ -115,27 +78,28 @@ class BuyInboundService:
'pinyin': getattr(item, 'pinyin', ''),
'status': '启用'
})
return results
return {
"items": items,
"total": pagination.total,
"page": page,
"has_next": pagination.has_next
}
except Exception as e:
traceback.print_exc()
return []
return {"items": [], "total": 0, "page": 1, "has_next": False}
# ============================================================
# 2. 新增入库逻辑
# 2. 新增入库逻辑 (保持不变)
# ============================================================
@staticmethod
def handle_inbound(data):
try:
base_id = data.get('base_id')
if not base_id:
raise ValueError("必须选择基础物料")
if not base_id: raise ValueError("必须选择基础物料")
material = MaterialBase.query.get(base_id)
if not material:
raise ValueError("所选物料不存在")
if not material.is_enabled:
raise ValueError(f"物料【{material.name}】已停用，无法办理新入库。")
if not material: raise ValueError("所选物料不存在")
if not material.is_enabled: raise ValueError(f"物料【{material.name}】已停用")
BuyInboundService._check_unique(
base_id=base_id,
@ -143,11 +107,9 @@ class BuyInboundService:
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
if data.get('in_date'):
try:
date_str = str(data['in_date'])
@ -155,15 +117,14 @@ class BuyInboundService:
in_date_val = datetime.strptime(date_str, '%Y-%m-%d %H:%M:%S')
else:
d_temp = datetime.strptime(date_str, '%Y-%m-%d')
in_date_val = datetime(d_temp.year, d_temp.month, d_temp.day,
current_time.hour, current_time.minute, current_time.second)
in_date_val = datetime(d_temp.year, d_temp.month, d_temp.day, current_time.hour,
current_time.minute, current_time.second)
except:
in_date_val = current_time
in_qty = float(data.get('in_quantity') or 0)
u_price = float(data.get('unit_price') or 0)
# 获取全局打印ID
try:
seq_sql = text("SELECT nextval('global_print_seq')")
result = db.session.execute(seq_sql)
@ -171,42 +132,22 @@ class BuyInboundService:
except:
next_global_id = None
# SKU 生成
if next_global_id:
generated_sku = str(next_global_id).zfill(10)
else:
generated_sku = datetime.now().strftime('%Y%m%d%H%M%S')
generated_sku = str(next_global_id).zfill(10) if next_global_id else datetime.now().strftime('%Y%m%d%H%M%S')
final_barcode = data.get('barcode') or generated_sku
arrival_list = data.get('arrival_photo', [])
report_list = data.get('inspection_report', [])
new_stock = StockBuy(
base_id=material.id,
global_print_id=next_global_id,
sku=generated_sku,
barcode=final_barcode,
in_date=in_date_val,
serial_number=data.get('serial_number'),
batch_number=data.get('batch_number'),
status=data.get('status', '在库'),
in_quantity=in_qty,
stock_quantity=in_qty,
available_quantity=in_qty,
base_id=material.id, global_print_id=next_global_id, sku=generated_sku, barcode=final_barcode,
in_date=in_date_val, serial_number=data.get('serial_number'), batch_number=data.get('batch_number'),
status=data.get('status', '在库'), in_quantity=in_qty, 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'),
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('purchaser'),
supplier_name=data.get('supplier_name'), 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=json.dumps(arrival_list),
inspection_report=json.dumps(report_list)
original_link=data.get('source_link'), detail_link=data.get('detail_link'),
arrival_photo=json.dumps(data.get('arrival_photo', [])),
inspection_report=json.dumps(data.get('inspection_report', []))
)
db.session.add(new_stock)
db.session.commit()
@ -216,57 +157,38 @@ class BuyInboundService:
raise e
# ============================================================
# 3. 更新入库逻辑
# 3. 更新入库 (保持不变)
# ============================================================
@staticmethod
def update_inbound(stock_id, data):
try:
stock = StockBuy.query.get(stock_id)
if not stock:
raise ValueError("记录不存在")
if not stock: raise ValueError("记录不存在")
BuyInboundService._check_unique(base_id=data.get('base_id', stock.base_id),
serial_number=data.get('serial_number', stock.serial_number),
batch_number=data.get('batch_number', stock.batch_number),
exclude_id=stock_id)
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)
BuyInboundService._check_unique(
base_id=new_base_id,
serial_number=new_sn,
batch_number=new_bn,
exclude_id=stock_id
)
# 更新字段
field_mapping = {
'sku': 'sku', 'barcode': 'barcode', 'base_id': 'base_id',
'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',
'currency': 'currency', 'exchange_rate': 'exchange_rate',
'purchaser': 'buyer_name', 'purchaser_email': 'buyer_email',
'source_link': 'original_link'
}
field_mapping = {'sku': 'sku', 'barcode': 'barcode', 'base_id': 'base_id',
'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', 'currency': 'currency', 'exchange_rate': 'exchange_rate',
'purchaser': 'buyer_name', 'purchaser_email': 'buyer_email',
'source_link': 'original_link'}
for k, v in field_mapping.items():
if k in data: setattr(stock, v, data[k])
if 'arrival_photo' in data and isinstance(data['arrival_photo'], list):
stock.arrival_photo = json.dumps(data['arrival_photo'])
if 'inspection_report' in data and isinstance(data['inspection_report'], list):
stock.inspection_report = json.dumps(data['inspection_report'])
if 'arrival_photo' in data: stock.arrival_photo = json.dumps(data['arrival_photo'])
if 'inspection_report' in data: stock.inspection_report = json.dumps(data['inspection_report'])
# 库存数量变更逻辑
if 'in_quantity' in data:
new_qty = float(data['in_quantity'])
diff = new_qty - float(stock.in_quantity)
diff = float(data['in_quantity']) - float(stock.in_quantity)
if diff != 0:
stock.in_quantity = new_qty
stock.in_quantity = float(data['in_quantity'])
stock.stock_quantity = float(stock.stock_quantity) + diff
stock.available_quantity = float(stock.available_quantity) + diff
if 'unit_price' in data:
stock.unit_price = float(data['unit_price'])
if 'unit_price' in data: stock.unit_price = float(data['unit_price'])
stock.total_price = float(stock.in_quantity) * float(stock.unit_price)
db.session.commit()
return stock
@ -275,7 +197,7 @@ class BuyInboundService:
raise e
# ============================================================
# 4. 删除逻辑
# 4. 删除 (保持不变)
# ============================================================
@staticmethod
def delete_inbound(stock_id):
@ -290,218 +212,72 @@ class BuyInboundService:
raise e
# ============================================================
# 5. 获取列表 (增强模糊匹配：多字段、多关键词、分词匹配)
# 5. 获取列表 (保持不变)
# ============================================================
@staticmethod
def get_list(page, limit, keyword=None, statuses=None):
try:
query = db.session.query(StockBuy).outerjoin(MaterialBase, StockBuy.base_id == MaterialBase.id)
if keyword:
# 1. 清理关键词
import re
keyword_clean = re.sub(r'\s+', ' ', keyword.strip())
if keyword_clean.startswith('"') and keyword_clean.endswith('"'):
# 精确短语匹配
exact_phrase = keyword_clean[1:-1]
if exact_phrase:
k_str = f'%{exact_phrase}%'
conditions = [
StockBuy.sku.ilike(k_str),
StockBuy.barcode.ilike(k_str),
StockBuy.batch_number.ilike(k_str),
StockBuy.serial_number.ilike(k_str),
StockBuy.supplier_name.ilike(k_str),
StockBuy.buyer_name.ilike(k_str),
MaterialBase.name.ilike(k_str),
MaterialBase.spec_model.ilike(k_str),
MaterialBase.pinyin.ilike(k_str),
MaterialBase.category.ilike(k_str)
]
if hasattr(MaterialBase, 'brand'):
conditions.append(MaterialBase.brand.ilike(k_str))
if hasattr(MaterialBase, 'manufacturer'):
conditions.append(MaterialBase.manufacturer.ilike(k_str))
query = query.filter(or_(*conditions))
else:
# 多关键词 AND 匹配
keywords = keyword_clean.split()
if keywords:
and_conditions = []
for word in keywords:
k_str = f'%{word}%'
word_conditions = [
StockBuy.sku.ilike(k_str),
StockBuy.barcode.ilike(k_str),
StockBuy.batch_number.ilike(k_str),
StockBuy.serial_number.ilike(k_str),
StockBuy.supplier_name.ilike(k_str),
StockBuy.buyer_name.ilike(k_str),
MaterialBase.name.ilike(k_str),
MaterialBase.spec_model.ilike(k_str),
MaterialBase.pinyin.ilike(k_str),
MaterialBase.category.ilike(k_str)
]
if hasattr(MaterialBase, 'brand'):
word_conditions.append(MaterialBase.brand.ilike(k_str))
if hasattr(MaterialBase, 'manufacturer'):
word_conditions.append(MaterialBase.manufacturer.ilike(k_str))
and_conditions.append(or_(*word_conditions))
if and_conditions:
query = query.filter(and_(*and_conditions))
if not statuses:
statuses = ['在库', '借库']
k_str = f'%{keyword.strip()}%'
conditions = [StockBuy.sku.ilike(k_str), StockBuy.barcode.ilike(k_str),
StockBuy.batch_number.ilike(k_str), StockBuy.serial_number.ilike(k_str),
StockBuy.supplier_name.ilike(k_str), StockBuy.buyer_name.ilike(k_str),
MaterialBase.name.ilike(k_str), MaterialBase.spec_model.ilike(k_str),
MaterialBase.pinyin.ilike(k_str)]
query = query.filter(or_(*conditions))
if not statuses: statuses = ['在库', '借库']
if '已出库' in statuses:
query = query.filter(StockBuy.status.in_(statuses))
else:
query = query.filter(and_(StockBuy.status.in_(statuses), StockBuy.stock_quantity > 0))
pagination = query.order_by(StockBuy.in_date.desc()).paginate(page=page, per_page=limit, error_out=False)
current_items = pagination.items
def parse_img(json_str):
if not json_str: return []
try:
return json.loads(json_str) if json_str.startswith('[') else [json_str]
except:
return []
items = []
for item in current_items:
qty_stock = float(item.stock_quantity or 0)
qty_avail = float(item.available_quantity or 0)
date_display = ''
if item.in_date:
try:
date_display = item.in_date.strftime('%Y-%m-%d')
except:
date_display = str(item.in_date)[:10]
d = {
'id': item.id,
'base_id': item.base_id,
'material_name': item.base.name if item.base else '',
for item in pagination.items:
items.append({
'id': item.id, 'base_id': item.base_id, '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 '',
'unit': item.base.unit if item.base else '',
'material_type': item.base.material_type if item.base else '',
'brand': getattr(item.base, 'brand', '') if item.base else '',
'manufacturer': getattr(item.base, 'manufacturer', '') if item.base else '',
'pinyin': getattr(item.base, 'pinyin', '') if item.base else '',
'sku': item.sku,
'inbound_date': date_display,
'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': qty_stock,
'qty_available': qty_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': parse_img(item.arrival_photo),
'inspection_report': parse_img(item.inspection_report),
'category': item.base.category if item.base else '', 'unit': item.base.unit if item.base else '',
'material_type': item.base.material_type if item.base else '', 'sku': item.sku,
'inbound_date': str(item.in_date)[:10] 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,
'purchaser': item.buyer_name, 'purchaser_email': item.buyer_email,
'source_link': item.original_link, 'detail_link': item.detail_link,
'arrival_photo': json.loads(item.arrival_photo) if item.arrival_photo else [],
'inspection_report': json.loads(item.inspection_report) if item.inspection_report else [],
'global_print_id': item.global_print_id
}
items.append(d)
})
return {"total": pagination.total, "items": items}
except Exception as e:
except Exception:
traceback.print_exc()
return {"total": 0, "items": []}
# ============================================================
# 6. 供应商历史查询 (移除限制)
# ============================================================
# 6-9 建议类接口保持不变 (略以节省篇幅，原样保留即可)
@staticmethod
def get_history_suppliers(base_id):
try:
query = db.session.query(StockBuy.supplier_name).filter(
StockBuy.base_id == base_id,
StockBuy.supplier_name.isnot(None),
StockBuy.supplier_name != ''
).distinct().order_by(StockBuy.supplier_name)
# [修改] 移除 limit
suppliers = [row[0] for row in query.all()]
return suppliers
except Exception:
return []
return [r[0] for r in db.session.query(StockBuy.supplier_name).filter(StockBuy.base_id == base_id,
StockBuy.supplier_name != '').distinct().all()]
# ============================================================
# 7. 采购人建议 (移除限制)
# ============================================================
@staticmethod
def get_history_purchasers(keyword):
try:
query = db.session.query(StockBuy.buyer_name, StockBuy.buyer_email) \
.filter(StockBuy.buyer_name.isnot(None), StockBuy.buyer_name != '')
return [{'value': r.buyer_name, 'email': r.buyer_email} for r in
db.session.query(StockBuy.buyer_name, StockBuy.buyer_email).filter(
StockBuy.buyer_name != '').distinct().all()]
if keyword:
kw = f'%{keyword}%'
query = query.filter(or_(
StockBuy.buyer_name.ilike(kw),
StockBuy.buyer_email.ilike(kw)
))
# [修改] 移除 limit
results = query.distinct().all()
users = []
for row in results:
users.append({
'value': row.buyer_name,
'email': row.buyer_email or ''
})
return users
except Exception:
traceback.print_exc()
return []
# ============================================================
# 8. 链接建议 (移除限制)
# ============================================================
@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
links = [row[0] for row in query.all()]
return links
except Exception:
return []
def get_history_links(base_id, type):
return [r[0] for r in
db.session.query(StockBuy.original_link if type == 'original' else StockBuy.detail_link).filter(
StockBuy.base_id == base_id).distinct().all()]
# ============================================================
# 9. 库位建议 (移除限制)
# ============================================================
@staticmethod
def get_history_locations(base_id):
try:
query = db.session.query(StockBuy.warehouse_location).filter(
StockBuy.base_id == base_id,
StockBuy.warehouse_location.isnot(None),
StockBuy.warehouse_location != ''
).distinct()
# [修改] 移除 limit
locs = [row[0] for row in query.all()]
return locs
except Exception:
return []
return [r[0] for r in
db.session.query(StockBuy.warehouse_location).filter(StockBuy.base_id == base_id).distinct().all()]