a8a3c82331
available_quantity 一旦为负，预占/释放/盘点的全部算术都会失真。此前有两处 缺口，本次一并堵上，使 available_quantity >= 0 成为不变量。 1) 执行阶段逐行扣减无下限校验（inventory_reservation.restore_then_deduct） 物料级校验只保证「Σ实扫 ≤ Σ可用」这一总量关系，拦不住「总量守恒但单行 穿仓」：同物料下 A 批可用 2、B 批可用 8，工人把 5 件全压在 A 批上，总量 5 ≤ 10 通过，A 批却被扣成 -3。借库走 deduct_stock=False，连实物数校验都 跳过，是裸扣。 已实测复现：借库与出库路径均可把单行扣成 -3。 校验放在 release_reserved() 之后，故不会误拒合法的换批次（物理覆盖）： 释放后每行 available 已含本单预占，扫自己预占过的批次时 raw >= 0 保证 必然放行；改扫其它批次时，该批次实时可用量就是它自己的上限。 2) 下调入库数量可把可用数压成负（buy/product/semi 三处 update_inbound） 按 diff 同步增减 stock/available，但无任何下限检查。该批次若已有部分被 预占/出库/借出，向下调整即产生负可用数。现在下调前校验可用数是否够扣。 正常数据下 stock >= available 恒成立，故守住 available 同时守住 stock。 配套：三个 update_* 端点此前只捕获 Exception → 500，没有 ValueError 分支 （同文件的 delete_* 早就有）。补上 400 分支，使业务校验失败不再被记成 服务端故障。 验证（事务内执行并回滚，未落库）：跨批次穿仓被拦、合法换批次放行、全额执行 本单预占批次放行、下调击穿被拦（API 返回 400，三个端点一致）、上调不受影响、 真实借库单 52 行全额执行正常、全库无负可用数。
728 lines
34 KiB
Python
728 lines
34 KiB
Python
# app/services/inbound/product_service.py
|
||
from app.extensions import db
|
||
from app.models.base import MaterialBase
|
||
from app.utils.decorators import get_current_company_filter
|
||
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_
|
||
from sqlalchemy.exc import IntegrityError
|
||
import traceback
|
||
import json
|
||
import numpy as np
|
||
from app.utils.ai_vision import extract_and_embed
|
||
from app.services.image_embedding_service import ImageEmbeddingService
|
||
from app.services.track_webhook_service import notify_track, get_current_operator
|
||
|
||
|
||
class ProductInboundService:
|
||
|
||
@staticmethod
|
||
def _check_unique(serial_number, exclude_id=None):
|
||
from app.models.inbound.product import StockProduct
|
||
if serial_number:
|
||
query = StockProduct.query.filter(StockProduct.serial_number == serial_number)
|
||
if exclude_id:
|
||
query = query.filter(StockProduct.id != exclude_id)
|
||
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}] 占用，请核查。")
|
||
|
||
@staticmethod
|
||
def search_base_material(keyword, page=1, limit=50):
|
||
try:
|
||
query = MaterialBase.query.filter(MaterialBase.is_enabled == True)
|
||
if keyword:
|
||
kw = f'%{keyword}%'
|
||
query = query.filter(
|
||
or_(
|
||
MaterialBase.name.ilike(kw),
|
||
MaterialBase.spec_model.ilike(kw),
|
||
MaterialBase.company_name.ilike(kw)
|
||
)
|
||
)
|
||
# ★ 行级公司隔离：只能搜索/选择本公司的物料（超管/跨域不受限）
|
||
from app.utils.decorators import get_current_company_filter
|
||
company_limit = get_current_company_filter()
|
||
if company_limit is not None:
|
||
query = query.filter(MaterialBase.company_name == company_limit)
|
||
query = query.order_by(MaterialBase.id.desc())
|
||
pagination = query.paginate(page=page, per_page=limit, error_out=False)
|
||
results = []
|
||
for item in pagination.items:
|
||
results.append({
|
||
'id': item.id,
|
||
'company_name': item.company_name,
|
||
'name': item.name,
|
||
'spec': item.spec_model,
|
||
'category': item.category,
|
||
'unit': item.unit,
|
||
'type': item.material_type,
|
||
'status': '启用'
|
||
})
|
||
return {
|
||
"items": results,
|
||
"total": pagination.total,
|
||
"page": page,
|
||
"has_next": pagination.has_next
|
||
}
|
||
except Exception:
|
||
traceback.print_exc()
|
||
return {"items": [], "total": 0, "page": 1, "has_next": False}
|
||
|
||
@staticmethod
|
||
def search_bom_options(keyword, parent_spec=None):
|
||
from app.models.bom import BomTable
|
||
try:
|
||
query = db.session.query(
|
||
BomTable.bom_no,
|
||
BomTable.version,
|
||
MaterialBase.name.label('parent_name'),
|
||
MaterialBase.spec_model.label('parent_spec')
|
||
).join(MaterialBase, BomTable.parent_id == MaterialBase.id)
|
||
|
||
if hasattr(BomTable, 'is_enabled'):
|
||
query = query.filter(BomTable.is_enabled == True)
|
||
|
||
if parent_spec:
|
||
query = query.filter(MaterialBase.spec_model == parent_spec)
|
||
|
||
if keyword:
|
||
kw = f'%{keyword}%'
|
||
query = query.filter(
|
||
or_(
|
||
BomTable.bom_no.ilike(kw),
|
||
MaterialBase.name.ilike(kw),
|
||
MaterialBase.spec_model.ilike(kw)
|
||
)
|
||
)
|
||
|
||
results = query.distinct().limit(20).all()
|
||
|
||
return [{
|
||
'bom_no': r.bom_no,
|
||
'version': r.version,
|
||
'parent_name': r.parent_name,
|
||
'parent_spec': r.parent_spec or ''
|
||
} for r in results]
|
||
except Exception:
|
||
traceback.print_exc()
|
||
return []
|
||
|
||
@staticmethod
|
||
def handle_inbound(data):
|
||
from app.models.inbound.product import StockProduct
|
||
try:
|
||
base_id = data.get('base_id')
|
||
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.category or '/成品' not in material.category:
|
||
raise ValueError(f"物料【{material.name}】属于【{material.category or '未分类'}】，只有【成品】才允许进行成品入库！")
|
||
|
||
ProductInboundService._check_unique(
|
||
serial_number=data.get('serial_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'])
|
||
if len(date_str) > 10:
|
||
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)
|
||
except:
|
||
in_date_val = current_time
|
||
|
||
in_qty = float(data.get('in_quantity') or 0)
|
||
raw_cost = float(data.get('raw_material_cost') or 0)
|
||
manual_cost = 0.0 # 字段已弃用，保持向后兼容
|
||
unit_total_cost = float(data.get('unit_total_cost') or raw_cost or 0)
|
||
total_price = unit_total_cost * in_qty
|
||
|
||
p_start = data.get('production_start_time', '')
|
||
p_end = data.get('production_end_time', '')
|
||
time_range = f"{p_start} ~ {p_end}" if p_start or p_end else None
|
||
|
||
try:
|
||
seq_sql = text("SELECT nextval('global_print_seq')")
|
||
result = db.session.execute(seq_sql)
|
||
next_global_id = result.scalar()
|
||
except:
|
||
next_global_id = None
|
||
|
||
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
|
||
|
||
photo_list = data.get('product_photo', [])
|
||
quality_list = data.get('quality_report_link', [])
|
||
inspection_list = data.get('inspection_report_link', [])
|
||
if not isinstance(photo_list, list): photo_list = []
|
||
if not isinstance(quality_list, list): quality_list = []
|
||
if not isinstance(inspection_list, list): inspection_list = []
|
||
|
||
new_stock = StockProduct(
|
||
base_id=material.id,
|
||
global_print_id=next_global_id,
|
||
sku=generated_sku,
|
||
production_date=in_date_val,
|
||
barcode=final_barcode,
|
||
serial_number=data.get('serial_number'),
|
||
status=data.get('status', '在库'),
|
||
warehouse_location=data.get('warehouse_location'),
|
||
in_quantity=in_qty,
|
||
stock_quantity=in_qty,
|
||
available_quantity=in_qty,
|
||
bom_code=data.get('bom_code'),
|
||
bom_version=data.get('bom_version'),
|
||
work_order_code=data.get('work_order_code'),
|
||
production_manager=data.get('production_manager'),
|
||
production_time_range=time_range,
|
||
raw_material_cost=raw_cost,
|
||
manual_cost=unit_total_cost,
|
||
quality_status=data.get('quality_status', '合格'),
|
||
product_photo=json.dumps(photo_list),
|
||
quality_report_link=json.dumps(quality_list),
|
||
inspection_report_link=json.dumps(inspection_list),
|
||
detail_link=data.get('detail_link'),
|
||
remark=data.get('remark'),
|
||
sale_price=float(data.get('sale_price') or 0),
|
||
order_id=data.get('order_id')
|
||
)
|
||
db.session.add(new_stock)
|
||
db.session.flush() # 获取 new_stock.id
|
||
|
||
# 先提交主事务，图片向量异步后台提取
|
||
db.session.commit()
|
||
|
||
if isinstance(photo_list, list) and photo_list:
|
||
from flask import current_app
|
||
from app.utils.executor import run_embedding_task
|
||
run_embedding_task(
|
||
ImageEmbeddingService.save_embeddings_background,
|
||
current_app._get_current_object(),
|
||
ImageEmbeddingService.MODULE_STOCK_PRODUCT,
|
||
new_stock.id,
|
||
photo_list
|
||
)
|
||
|
||
# 真实扫码入库成功，异步通知 Track 系统（Webhook）
|
||
notify_track({
|
||
'event': 'inbound.created',
|
||
'source_table': 'stock_product',
|
||
'sku': material.spec_model or material.name,
|
||
'serial_number': new_stock.serial_number,
|
||
'quantity': float(new_stock.in_quantity or 0),
|
||
'operator': get_current_operator(),
|
||
})
|
||
return new_stock
|
||
except Exception as e:
|
||
db.session.rollback()
|
||
raise e
|
||
|
||
@staticmethod
|
||
def update_inbound(stock_id, data):
|
||
from app.models.inbound.product import StockProduct
|
||
try:
|
||
stock = StockProduct.query.get(stock_id)
|
||
if not stock: raise ValueError("记录不存在")
|
||
|
||
if 'serial_number' in data:
|
||
ProductInboundService._check_unique(
|
||
serial_number=data['serial_number'],
|
||
exclude_id=stock_id
|
||
)
|
||
|
||
fields = [
|
||
'barcode', 'serial_number', 'warehouse_location',
|
||
'status', 'quality_status', 'bom_code', 'bom_version',
|
||
'work_order_code', 'production_manager',
|
||
'detail_link', 'order_id', 'remark'
|
||
]
|
||
for f in fields:
|
||
if f in data: setattr(stock, f, data[f])
|
||
|
||
# ★ 仅当 payload 明确携带 product_photo 时才更新（含 [] 表示清空）；改其它字段不动图片
|
||
if 'product_photo' in data:
|
||
new_photo_list = data['product_photo']
|
||
stock.product_photo = json.dumps(new_photo_list)
|
||
# 立即触发异步向量提取，不阻塞主事务提交
|
||
if isinstance(new_photo_list, list) and new_photo_list:
|
||
from flask import current_app
|
||
from app.utils.executor import run_embedding_task
|
||
run_embedding_task(
|
||
ImageEmbeddingService.save_embeddings_background,
|
||
current_app._get_current_object(),
|
||
ImageEmbeddingService.MODULE_STOCK_PRODUCT,
|
||
stock.id,
|
||
new_photo_list
|
||
)
|
||
else:
|
||
# 图片被清空([])时，同步清理该记录的图片向量
|
||
ImageEmbeddingService.delete_embeddings(
|
||
ImageEmbeddingService.MODULE_STOCK_PRODUCT, stock.id
|
||
)
|
||
if 'quality_report_link' in data:
|
||
imgs = data['quality_report_link']
|
||
if isinstance(imgs, list): stock.quality_report_link = json.dumps(imgs)
|
||
if 'inspection_report_link' in data:
|
||
imgs = data['inspection_report_link']
|
||
if isinstance(imgs, list): stock.inspection_report_link = json.dumps(imgs)
|
||
|
||
if 'sale_price' in data: stock.sale_price = float(data['sale_price'])
|
||
if 'raw_material_cost' in data: stock.raw_material_cost = float(data['raw_material_cost'])
|
||
if 'unit_total_cost' in data: stock.manual_cost = float(data['unit_total_cost']) # 映射到 manual_cost 物理字段
|
||
|
||
if 'in_quantity' in data:
|
||
new_qty = float(data['in_quantity'])
|
||
diff = new_qty - float(stock.in_quantity)
|
||
# ★ 下调数量前校验可用数下限（同 buy_service）：
|
||
# 该批次可能已有部分被预占/出库/借出，硬扣会让 available
|
||
# 变成负数，破坏预占释放与盘点的算术。
|
||
if diff < 0:
|
||
avail_now = float(stock.available_quantity or 0)
|
||
if -diff > avail_now:
|
||
raise ValueError(
|
||
f"无法下调入库数量：该批次已有部分被预占/出库/借出，"
|
||
f"当前可用数（{avail_now}）不足以支持向下调整 {-diff}。"
|
||
f"请先处理相关单据或改为调整盘点差异。"
|
||
)
|
||
stock.in_quantity = new_qty
|
||
stock.stock_quantity = float(stock.stock_quantity) + diff
|
||
stock.available_quantity = float(stock.available_quantity) + diff
|
||
|
||
|
||
if 'production_start_time' in data or 'production_end_time' in data:
|
||
old_range = stock.production_time_range or " ~ "
|
||
parts = old_range.split(' ~ ')
|
||
old_start = parts[0] if len(parts) > 0 else ''
|
||
old_end = parts[1] if len(parts) > 1 else ''
|
||
start = data.get('production_start_time', old_start)
|
||
end = data.get('production_end_time', old_end)
|
||
stock.production_time_range = f"{start} ~ {end}"
|
||
|
||
db.session.commit()
|
||
return stock
|
||
except Exception as e:
|
||
db.session.rollback()
|
||
raise e
|
||
|
||
@staticmethod
|
||
def delete_inbound(stock_id):
|
||
from app.models.inbound.product import StockProduct
|
||
try:
|
||
stock = StockProduct.query.get(stock_id)
|
||
if stock:
|
||
material_name = stock.base.name if stock.base else '未知物料'
|
||
# 删除时同步清理向量记录
|
||
ImageEmbeddingService.delete_embeddings(
|
||
ImageEmbeddingService.MODULE_STOCK_PRODUCT, stock.id
|
||
)
|
||
db.session.delete(stock)
|
||
db.session.commit()
|
||
return material_name
|
||
return None
|
||
except IntegrityError:
|
||
db.session.rollback()
|
||
raise ValueError("该入库单已被出库、盘点或借还等业务关联，为保证账目完整，禁止删除！")
|
||
except Exception as e:
|
||
db.session.rollback()
|
||
raise e
|
||
|
||
@staticmethod
|
||
def get_outbound_history(stock_id):
|
||
try:
|
||
records = TransOutbound.query.filter_by(
|
||
source_table='stock_product', stock_id=stock_id
|
||
).order_by(TransOutbound.outbound_time.desc()).all()
|
||
return [r.to_dict() for r in records]
|
||
except:
|
||
return []
|
||
|
||
@staticmethod
|
||
def get_list(page, limit, keyword=None, sku=None, search_field='all', statuses=None, category=None, material_type=None, company=None,
|
||
order_by_column=None, is_asc=None, advanced_filters=None):
|
||
from app.models.inbound.product import StockProduct
|
||
try:
|
||
query = db.session.query(StockProduct).outerjoin(MaterialBase, StockProduct.base_id == MaterialBase.id)
|
||
# 1. 通用关键词搜索（支持指定字段精准搜索）
|
||
if keyword:
|
||
kw = f'%{keyword}%'
|
||
if search_field == 'name':
|
||
query = query.filter(MaterialBase.name.ilike(kw))
|
||
elif search_field == 'spec':
|
||
query = query.filter(MaterialBase.spec_model.ilike(kw))
|
||
elif search_field == 'common_name':
|
||
query = query.filter(MaterialBase.common_name.ilike(kw))
|
||
elif search_field == 'serial_number':
|
||
query = query.filter(StockProduct.serial_number.ilike(kw))
|
||
elif search_field == 'work_order_code':
|
||
query = query.filter(StockProduct.work_order_code.ilike(kw))
|
||
elif search_field == 'order_id':
|
||
query = query.filter(StockProduct.order_id.ilike(kw))
|
||
else: # 'all' 默认全局模糊匹配
|
||
query = query.filter(or_(
|
||
MaterialBase.name.ilike(kw),
|
||
MaterialBase.spec_model.ilike(kw),
|
||
MaterialBase.company_name.ilike(kw),
|
||
StockProduct.serial_number.ilike(kw),
|
||
StockProduct.work_order_code.ilike(kw),
|
||
StockProduct.order_id.ilike(kw)
|
||
))
|
||
# 1.1 SKU 独立搜索
|
||
if sku and sku.strip():
|
||
sku_str = f'%{sku.strip()}%'
|
||
query = query.filter(StockProduct.sku.ilike(sku_str))
|
||
if category and category.strip():
|
||
# 级联选择器：中间节点用前缀匹配，与 base_service.get_list 行为一致
|
||
query = query.filter(MaterialBase.category.ilike(f"{category.strip()}%"))
|
||
if material_type and material_type.strip():
|
||
query = query.filter(MaterialBase.material_type == material_type.strip())
|
||
|
||
# ============================================================
|
||
# 【行级数据隔离】基于 JWT 多租户公司过滤
|
||
# ============================================================
|
||
company_limit = get_current_company_filter()
|
||
if company_limit is not None:
|
||
query = query.filter(MaterialBase.company_name == company_limit)
|
||
|
||
# ============================================================
|
||
# 【全局特权】基于 JWT 与 global:cross_company_op 的跨组织隔离
|
||
# ============================================================
|
||
from flask_jwt_extended import get_jwt
|
||
|
||
claims = get_jwt()
|
||
user_role = claims.get('role', '').upper() if claims.get('role') else ''
|
||
user_company = claims.get('company_name', '')
|
||
|
||
# 获取用户权限列表（用于检查 global:cross_company_op 特权）
|
||
from app.api.v1.inbound.base import get_current_user_permissions
|
||
user_perms = get_current_user_permissions() or []
|
||
normalized_perms = set(p.lower().replace('_', '').replace(':', '') for p in user_perms)
|
||
|
||
# 检查是否拥有全局特权或超管角色（crossDomain 与 global:cross_company_op 都视为跨域）
|
||
has_cross_company = 'globalcrosscompanyop' in normalized_perms or 'crossdomain' in normalized_perms
|
||
|
||
if user_role != 'SUPER_ADMIN' and not has_cross_company:
|
||
# 无特权：严禁查其他公司，强制绑定本公司
|
||
if company and company.strip() and company.strip() != user_company:
|
||
from flask import abort
|
||
abort(403, description=f'越权访问：您无权查询 {company} 的数据')
|
||
if user_company:
|
||
query = query.filter(MaterialBase.company_name == user_company)
|
||
elif user_role == 'SUPER_ADMIN' or has_cross_company:
|
||
# 有特权：允许下拉框传过来的 company 参数生效
|
||
if company and company.strip():
|
||
query = query.filter(MaterialBase.company_name == company.strip())
|
||
|
||
if not statuses:
|
||
statuses = ['在库', '借库']
|
||
if '已出库' in statuses:
|
||
query = query.filter(StockProduct.status.in_(statuses))
|
||
else:
|
||
query = query.filter(
|
||
and_(
|
||
StockProduct.status.in_(statuses),
|
||
StockProduct.stock_quantity > 0
|
||
)
|
||
)
|
||
|
||
# 高级动态筛选
|
||
if advanced_filters:
|
||
if isinstance(advanced_filters, str):
|
||
try:
|
||
import json
|
||
advanced_filters = json.loads(advanced_filters)
|
||
except:
|
||
advanced_filters = []
|
||
if isinstance(advanced_filters, list):
|
||
field_mapping = {
|
||
'id': StockProduct.id,
|
||
'base_id': StockProduct.base_id,
|
||
'company_name': MaterialBase.company_name,
|
||
'material_name': MaterialBase.name,
|
||
'spec_model': MaterialBase.spec_model,
|
||
'category': MaterialBase.category,
|
||
'material_type': MaterialBase.material_type,
|
||
'unit': MaterialBase.unit,
|
||
'sku': StockProduct.sku,
|
||
'inbound_date': StockProduct.production_date,
|
||
'barcode': StockProduct.barcode,
|
||
'serial_number': StockProduct.serial_number,
|
||
'batch_number': StockProduct.serial_number,
|
||
'status': StockProduct.status,
|
||
'quality_status': StockProduct.quality_status,
|
||
'in_quantity': StockProduct.in_quantity,
|
||
'stock_quantity': StockProduct.stock_quantity,
|
||
'available_quantity': StockProduct.available_quantity,
|
||
'warehouse_location': StockProduct.warehouse_location,
|
||
'bom_code': StockProduct.bom_code,
|
||
'bom_version': StockProduct.bom_version,
|
||
'work_order_code': StockProduct.work_order_code,
|
||
'raw_material_cost': StockProduct.raw_material_cost,
|
||
'unit_total_cost': StockProduct.manual_cost,
|
||
'order_id': StockProduct.order_id,
|
||
'sale_price': StockProduct.sale_price,
|
||
'production_manager': StockProduct.production_manager,
|
||
'total_price': (StockProduct.manual_cost * StockProduct.in_quantity),
|
||
}
|
||
for cond in advanced_filters:
|
||
field = cond.get('field')
|
||
operator = cond.get('operator')
|
||
value = cond.get('value')
|
||
if not field or not operator:
|
||
continue
|
||
model_field = field_mapping.get(field)
|
||
if model_field is None:
|
||
continue
|
||
# 防止 SQL 注入，只允许映射的字段
|
||
if operator == '=':
|
||
query = query.filter(model_field == value)
|
||
elif operator == '!=':
|
||
query = query.filter(model_field != value)
|
||
elif operator == 'like':
|
||
query = query.filter(model_field.ilike(f'%{value}%'))
|
||
elif operator == 'not_like':
|
||
query = query.filter(~model_field.ilike(f'%{value}%'))
|
||
elif operator == '>':
|
||
if value.replace('.', '', 1).isdigit():
|
||
query = query.filter(model_field > float(value))
|
||
elif operator == '<':
|
||
if value.replace('.', '', 1).isdigit():
|
||
query = query.filter(model_field < float(value))
|
||
elif operator == '>=':
|
||
if value.replace('.', '', 1).isdigit():
|
||
query = query.filter(model_field >= float(value))
|
||
elif operator == '<=':
|
||
if value.replace('.', '', 1).isdigit():
|
||
query = query.filter(model_field <= float(value))
|
||
|
||
# 动态排序
|
||
order_field = None
|
||
if order_by_column and is_asc is not None:
|
||
order_mapping = {
|
||
'id': StockProduct.id,
|
||
'base_id': StockProduct.base_id,
|
||
'company_name': MaterialBase.company_name,
|
||
'material_name': MaterialBase.name,
|
||
'category': MaterialBase.category,
|
||
'material_type': MaterialBase.material_type,
|
||
'spec_model': MaterialBase.spec_model,
|
||
'unit': MaterialBase.unit,
|
||
'sku': StockProduct.sku,
|
||
'inbound_date': StockProduct.production_date,
|
||
'barcode': StockProduct.barcode,
|
||
'serial_number': StockProduct.serial_number,
|
||
'batch_number': StockProduct.serial_number,
|
||
'status': StockProduct.status,
|
||
'quality_status': StockProduct.quality_status,
|
||
'in_quantity': StockProduct.in_quantity,
|
||
'stock_quantity': StockProduct.stock_quantity,
|
||
'available_quantity': StockProduct.available_quantity,
|
||
'warehouse_location': StockProduct.warehouse_location,
|
||
'bom_code': StockProduct.bom_code,
|
||
'bom_version': StockProduct.bom_version,
|
||
'work_order_code': StockProduct.work_order_code,
|
||
'raw_material_cost': StockProduct.raw_material_cost,
|
||
'unit_total_cost': StockProduct.manual_cost,
|
||
'order_id': StockProduct.order_id,
|
||
'sale_price': StockProduct.sale_price,
|
||
'production_manager': StockProduct.production_manager,
|
||
'total_price': (StockProduct.manual_cost * StockProduct.in_quantity),
|
||
}
|
||
order_field = order_mapping.get(order_by_column)
|
||
if order_field is not None:
|
||
if is_asc == 'true' or is_asc == True:
|
||
query = query.order_by(order_field.asc())
|
||
else:
|
||
query = query.order_by(order_field.desc())
|
||
if order_field is None:
|
||
query = query.order_by(StockProduct.production_date.desc())
|
||
|
||
pagination = query.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:
|
||
item_dict = item.to_dict()
|
||
item_dict['unit_total_cost'] = float(item.manual_cost or 0)
|
||
items.append(item_dict)
|
||
return {"total": pagination.total, "items": items}
|
||
except Exception as e:
|
||
traceback.print_exc()
|
||
return {"total": 0, "items": []}
|
||
|
||
@staticmethod
|
||
def search_system_users(keyword):
|
||
from app.models.system import SysUser
|
||
try:
|
||
query = SysUser.query.filter(SysUser.status == 'active')
|
||
if keyword:
|
||
kw = f'%{keyword}%'
|
||
query = query.filter(db.or_(
|
||
SysUser.username.ilike(kw),
|
||
SysUser.email.ilike(kw)
|
||
))
|
||
query = query.order_by(SysUser.username)
|
||
users = []
|
||
for u in query.limit(20).all():
|
||
users.append({
|
||
'value': u.username,
|
||
'email': u.email
|
||
})
|
||
return users
|
||
except Exception:
|
||
return []
|
||
|
||
@staticmethod
|
||
def get_filter_options():
|
||
try:
|
||
from app.models.base import MaterialBase
|
||
categories = db.session.query(MaterialBase.category).filter(MaterialBase.category != None,
|
||
MaterialBase.category != '').distinct().all()
|
||
sorted_categories = sorted([r[0] for r in categories])
|
||
|
||
types = db.session.query(MaterialBase.material_type).filter(MaterialBase.material_type != None,
|
||
MaterialBase.material_type != '').distinct().all()
|
||
sorted_types = sorted([r[0] for r in types])
|
||
|
||
companies = db.session.query(MaterialBase.company_name).filter(MaterialBase.company_name != None,
|
||
MaterialBase.company_name != '').distinct().all()
|
||
sorted_companies = sorted([r[0] for r in companies])
|
||
|
||
return {
|
||
"categories": sorted_categories,
|
||
"types": sorted_types,
|
||
"companies": sorted_companies
|
||
}
|
||
except Exception:
|
||
import traceback
|
||
traceback.print_exc()
|
||
return {"categories": [], "types": [], "companies": []}
|
||
|
||
@staticmethod
|
||
def get_history_managers(keyword=None):
|
||
from app.models.inbound.product import StockProduct
|
||
try:
|
||
query = db.session.query(StockProduct.production_manager).filter(
|
||
StockProduct.production_manager.isnot(None),
|
||
StockProduct.production_manager != ''
|
||
)
|
||
if keyword:
|
||
query = query.filter(StockProduct.production_manager.ilike(f'%{keyword}%'))
|
||
records = query.distinct().all()
|
||
return [r[0] for r in records if r[0]]
|
||
except Exception:
|
||
traceback.print_exc()
|
||
return []
|
||
|
||
# ============================================================
|
||
# 9. BOM 原材料成本自动核算 (新增)
|
||
# ============================================================
|
||
@staticmethod
|
||
def calculate_bom_cost(bom_no, bom_version):
|
||
"""
|
||
根据 BOM 编号和版本计算原材料总成本
|
||
遍历 BOM 子件，使用原生 SQL 查物理表 bom_table，取每个子件在采购、半成品、成品三个表中的最高单价，乘以用量后累加
|
||
"""
|
||
from app.models.inbound.buy import StockBuy
|
||
from app.models.inbound.semi import StockSemi
|
||
from app.models.inbound.product import StockProduct
|
||
from sqlalchemy import func, text
|
||
try:
|
||
# 使用原生 SQL 精准查询 bom_table，避免模型映射错误
|
||
sql = text("""
|
||
SELECT child_id, dosage
|
||
FROM bom_table
|
||
WHERE bom_no = :bom_no AND version = :version
|
||
""")
|
||
bom_lines = db.session.execute(sql, {'bom_no': bom_no, 'version': bom_version}).fetchall()
|
||
|
||
total_cost = 0.0
|
||
for line in bom_lines:
|
||
component_base_id = line[0] # child_id
|
||
usage_qty = float(line[1] or 1.0) # dosage
|
||
|
||
# 1. 查采购表最高价 (不含税)
|
||
buy_price = db.session.query(func.max(StockBuy.pre_tax_unit_price)).filter(
|
||
StockBuy.base_id == component_base_id
|
||
).scalar() or 0.0
|
||
|
||
# 2. 查半成品表最高价 (单件成本映射存在 manual_cost 里了)
|
||
semi_price = db.session.query(func.max(StockSemi.manual_cost)).filter(
|
||
StockSemi.base_id == component_base_id
|
||
).scalar() or 0.0
|
||
|
||
# 3. 查成品表最高价 (同样存储在 manual_cost 字段里)
|
||
product_price = db.session.query(func.max(StockProduct.manual_cost)).filter(
|
||
StockProduct.base_id == component_base_id
|
||
).scalar() or 0.0
|
||
|
||
# 4. 取三个表中的最大值，乘以用量 (dosage)
|
||
max_price = max(float(buy_price), float(semi_price), float(product_price))
|
||
total_cost += max_price * usage_qty
|
||
|
||
return round(total_cost, 2)
|
||
except Exception as e:
|
||
traceback.print_exc()
|
||
raise e
|
||
|
||
@staticmethod
|
||
def get_last_location_by_base_id(base_id):
|
||
"""
|
||
获取指定物料最近一次入库的库位（跨表查询）
|
||
查询顺序：成品入库 -> 采购入库 -> 半成品入库，返回最新入库的库位
|
||
"""
|
||
from app.models.inbound.product import StockProduct
|
||
|
||
# 1. 查询成品入库最新记录
|
||
last_product = StockProduct.query.filter(
|
||
StockProduct.base_id == base_id
|
||
).order_by(StockProduct.production_date.desc()).first()
|
||
|
||
# 2. 查询采购入库最新记录
|
||
last_buy = StockBuy.query.filter(
|
||
StockBuy.base_id == base_id
|
||
).order_by(StockBuy.in_date.desc()).first()
|
||
|
||
# 3. 查询半成品入库最新记录
|
||
last_semi = StockSemi.query.filter(
|
||
StockSemi.base_id == base_id
|
||
).order_by(StockSemi.production_date.desc()).first()
|
||
|
||
# 比较三个表中的最新入库时间，返回最新的库位
|
||
candidates = []
|
||
if last_product and last_product.warehouse_location:
|
||
candidates.append((last_product.production_date, last_product.warehouse_location))
|
||
if last_buy and last_buy.warehouse_location:
|
||
candidates.append((last_buy.in_date, last_buy.warehouse_location))
|
||
if last_semi and last_semi.warehouse_location:
|
||
candidates.append((last_semi.production_date, last_semi.warehouse_location))
|
||
|
||
if not candidates:
|
||
return ""
|
||
|
||
# 按时间倒序排序，返回最新的库位
|
||
candidates.sort(key=lambda x: x[0] if x[0] else datetime.min, reverse=True)
|
||
return candidates[0][1] if candidates[0][1] else ""
|