Files
KCGL/inventory-backend/app/api/v1/inbound/stock.py
yueli dda6e4c787 feat(return): 退回、回库、报废与在管台账接口
打通逆向物流的全部后端入口。

新增接口(app/api/v1/inbound/stock.py):
- POST /stock/<id>/change-status      库存状态变更(在库/冻结/不良品)
- POST /stock/return-from-outbound    通用原单退回
- POST /stock/defective/<id>/restock  不良品修好回库(支持部分回库)
- POST /stock/defective/<id>/scrap    不良品报废销毁
- GET  /stock/defective               在管台账分页查询

设计要点:
- 良品退回加回原库存行;不良品退回则库存表分毫不动,只写独立在管台账。
  这样坏件从根上不会混进可分配池
- 全链路 Fail-Closed 守卫:良品退回到非「在库」行会被拒(status 是行级
  属性,加回已冻结/不良品的行会让良品被连带隔离);原库存行已不存在会被
  拒(入库模块会物理删除库存行,实测 1077 条出库记录中已有 7 条悬空)
- 三个写接口均加 with_for_update 行锁 + prevent_double_submit 幂等锁。
  装饰器顺序为 permission_required → prevent_double_submit,顺序颠倒会因
  JWT 未验证而抛错、被自身 except 捕获后 fail-open 降级
- 报废同时写 trans_scrap 台账(source_table 用 trans_defective_goods 并
  存台账自身主键,与 trans_borrow/trans_repair 作来源时的约定一致),
  成本按原库存行 best-effort 取价,取不到记 0 而不中断报废

报废报表集成(app/api/v1/scrap.py):
- _resolve_materials 补 trans_defective_goods 分支。物料名已在台账冗余存储,
  不联表——坏件的原库存行可能已被删除,联表取名称会得到空值
- 公司隔离补 defective_subq 分支。原先按 source_table 逐个构造子查询,
  未知来源会被整体过滤,导致这类记录对普通用户静默消失
- 无审批单号的分组前缀按来源分流:不良品直报不再套用 LEGACY-(它是新业务
  记录,不是历史脏数据)。分组键同步带上来源标记,且 _order_key_pred 的
  SQL 谓词改为同口径,否则页面分组与「按单筛选」结果会对不上
2026-09-16 15:45:31 +08:00

3149 lines
138 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

from flask import Blueprint, jsonify, request, send_file, current_app
from app.extensions import db, beijing_time
from datetime import datetime, timedelta
from flask_jwt_extended import jwt_required, get_jwt, get_jwt_identity
from app.utils.decorators import permission_required, get_current_company_filter, prevent_double_submit
from sqlalchemy.orm import joinedload
import uuid as uuid_module
import io
import traceback
from openpyxl import Workbook
from openpyxl.styles import Font, Alignment, Border, Side, PatternFill
# 导入模型
from app.models.inbound.buy import StockBuy
from app.models.inbound.stocktake import (
StocktakeDraft,
StocktakeSession,
STOCKTAKE_MODE_OPEN,
STOCKTAKE_MODE_BLIND,
STOCKTAKE_SCOPE_FULL,
STOCKTAKE_SCOPE_ACTIVE,
STOCKTAKE_STATUS_ACTIVE,
STOCKTAKE_STATUS_FINISHED,
)
from app.models.transaction import (
TransBorrow,
TransReturn,
TransScrap,
TransDefectiveGoods,
RETURN_TYPE_GOOD,
RETURN_TYPE_DEFECTIVE,
DEFECTIVE_STATUS_PENDING,
DEFECTIVE_STATUS_IN_PROGRESS,
RESTOCKABLE_DEFECTIVE_STATUSES,
SCRAPPABLE_DEFECTIVE_STATUSES,
VALID_DEFECTIVE_STATUSES,
defective_close_status,
)
from app.models.outbound import TransOutbound
from app.models.base import MaterialBase
# 库存状态语义的单一事实来源(与分配器共用同一套常量,避免两处定义漂移)
from app.services.inventory_reservation import (
VALID_STOCK_STATUSES,
STOCK_STATUS_IN_STOCK,
)
# 尝试导入用户模型
try:
from app.models.system import SysUser
except ImportError:
SysUser = None
# 尝试导入半成品和成品
import logging
try:
from app.models.inbound.semi import StockSemi
except Exception as e:
logging.error(f"❌ 致命错误StockSemi 模型导入失败: {e}")
StockSemi = None
try:
from app.models.inbound.product import StockProduct
except Exception as e:
logging.error(f"❌ 致命错误StockProduct 模型导入失败: {e}")
StockProduct = None
def _normalize_user_id(user_id=None):
"""规范化 user_id确保是有效字符串"""
# 优先使用传入的 user_id否则从 JWT 获取
if user_id and isinstance(user_id, str) and len(user_id) <= 100:
return user_id.strip()
# 从 JWT 获取当前用户
try:
return get_jwt().get('display_name') or get_jwt_identity()
except:
return 'unknown'
def _resolve_user_name(user_id):
"""把 user_id 解析为真实姓名,解析不出来就原样返回。
需要兼容三种来源:
1. "孙霞(sunxia)" —— 草稿/流水里实际存的格式,来自 JWT 的 display_name
2. "张三/zhangsan01" —— SysUser.username 的存储格式
3. 纯数字用户ID —— 直接按主键查
"""
if not user_id:
return '-'
raw = str(user_id).strip()
# 1. "姓名(账号)" / "姓名/账号" —— 分隔符前就是姓名,无需查库
if '(' in raw and raw.endswith(')'):
name = raw.split('(', 1)[0].strip()
if name:
return name
if '/' in raw:
name = raw.split('/', 1)[0].strip()
if name:
return name
if not SysUser:
return raw
try:
user = None
# 尝试通过ID或用户名查找
if raw.isdigit():
user = SysUser.query.get(int(raw))
if not user:
user = SysUser.query.filter(SysUser.username.like(f"%/{raw}")).first()
# 注意:此处不做 filter_by(username=...) 兜底,
# 避免 PostgreSQL 把 user_id 数字与 username 字符串列做类型比较导致报错
if not user:
return raw
# 2. 解析 username 格式: "张三/zhangsan01" -> 取前面的真实姓名
raw_username = getattr(user, 'username', None) or raw
if '/' in raw_username:
return raw_username.split('/')[0]
return raw_username
except Exception:
return raw
def get_stock_model(source_table):
"""根据source_table获取对应的库存模型"""
if source_table == 'stock_buy':
return StockBuy
elif source_table == 'stock_semi':
return StockSemi
elif source_table == 'stock_product':
return StockProduct
return None
from app.services.print.network_print_service import NetworkPrintService
bp = Blueprint('stock_ops', __name__)
# ============================================================
# 辅助函数:获取库存记录
# ============================================================
def get_stock_record(source_table, stock_id, for_update=False):
"""根据库存类型和ID获取库存记录
Args:
source_table: 库存类型 ('stock_buy' / 'stock_semi' / 'stock_product')
stock_id: 库存记录主键ID
for_update: 是否使用 SELECT ... FOR UPDATE 悲观行锁(默认 False
设为 True 时可防止并发调整/报废导致的 TOCTOU 竞态
"""
if source_table == 'stock_buy' and StockBuy:
q = StockBuy.query
return (q.with_for_update() if for_update else q).get(stock_id)
elif source_table == 'stock_semi' and StockSemi:
q = StockSemi.query
return (q.with_for_update() if for_update else q).get(stock_id)
elif source_table == 'stock_product' and StockProduct:
q = StockProduct.query
return (q.with_for_update() if for_update else q).get(stock_id)
return None
def _filter_by_company(query, model, company_name):
"""
对库存查询施加公司隔离。
三张库存表都没有 company_id公司维度挂在 material_base.company_name 上,
故经由 base 关系做子查询过滤。
company_name 为 None 表示不过滤(超管/跨域用户未指定公司);
'__NO_COMPANY__' 是 get_current_company_filter() 的哨兵值,会匹配不到任何行。
"""
if company_name is None:
return query
return query.filter(model.base.has(MaterialBase.company_name == company_name))
def get_active_locations(company_name, days=30, top_n=50):
"""
统计最近 N 天内异动最频繁的库位 —— 「动盘/抽盘」的范围来源。
数据来源(三张库存表都没有 company_id公司维度一律经 base_id 关联 material_base
- 出库 trans_outbound其 warehouse_location 历史上几乎没填900 条里 829 条为空,
且仅 2026-09 起才部分有值),故改为经 (source_table, stock_id) 回查库存表的
warehouse_location —— 实测近 30 天覆盖率 298/299。
- 入库:入库没有独立流水表,直接以 stock_buy.in_date /
stock_semi.production_date / stock_product.production_date 作为异动时间。
- 借出 trans_borrow用其 location 字段填充率同样很低80 条里仅 8 条有值)。
已知局限:
- 报废(trans_scrap)与维修(trans_repair)表没有库位字段,未纳入统计;
- 回查得到的是库存行**当前**的库位,不是异动发生当时的库位,
库存行事后迁库会让历史归因偏移;
- 库存行被删除后 JOIN 不到,该笔异动不计入。
:return: [{'location': str, 'moves': int, 'sku_count': int, 'last_move': str|None}, ...]
按异动次数降序,最多 top_n 条。
"""
since = beijing_time() - timedelta(days=days)
sql = db.text("""
WITH moves AS (
SELECT COALESCE(b.warehouse_location, s.warehouse_location, p.warehouse_location) AS loc,
o.sku AS sku, o.outbound_time AS ts,
COALESCE(m1.company_name, m2.company_name, m3.company_name) AS company
FROM trans_outbound o
LEFT JOIN stock_buy b ON o.source_table = 'stock_buy' AND b.id = o.stock_id
LEFT JOIN stock_semi s ON o.source_table = 'stock_semi' AND s.id = o.stock_id
LEFT JOIN stock_product p ON o.source_table = 'stock_product' AND p.id = o.stock_id
LEFT JOIN material_base m1 ON m1.id = b.base_id
LEFT JOIN material_base m2 ON m2.id = s.base_id
LEFT JOIN material_base m3 ON m3.id = p.base_id
WHERE o.outbound_time >= :since
UNION ALL
SELECT b2.warehouse_location, b2.sku, b2.in_date, m4.company_name
FROM stock_buy b2 JOIN material_base m4 ON m4.id = b2.base_id
WHERE b2.in_date >= :since
UNION ALL
SELECT s2.warehouse_location, s2.sku, s2.production_date, m5.company_name
FROM stock_semi s2 JOIN material_base m5 ON m5.id = s2.base_id
WHERE s2.production_date >= :since
UNION ALL
SELECT p2.warehouse_location, p2.sku, p2.production_date, m6.company_name
FROM stock_product p2 JOIN material_base m6 ON m6.id = p2.base_id
WHERE p2.production_date >= :since
UNION ALL
SELECT br.location, br.sku, br.borrow_time, m7.company_name
FROM trans_borrow br
LEFT JOIN stock_buy bs ON br.source_table = 'stock_buy' AND bs.id = br.stock_id
LEFT JOIN material_base m7 ON m7.id = bs.base_id
WHERE br.borrow_time >= :since
)
SELECT loc AS location,
count(*) AS moves,
count(DISTINCT sku) AS sku_count,
max(ts) AS last_move
FROM moves
WHERE loc IS NOT NULL AND loc <> ''
AND (:company IS NULL OR company = :company)
GROUP BY loc
ORDER BY moves DESC, loc
LIMIT :top_n
""")
rows = db.session.execute(sql, {
'since': since,
'company': company_name,
'top_n': top_n,
}).fetchall()
return [{
'location': r.location,
'moves': int(r.moves or 0),
'sku_count': int(r.sku_count or 0),
'last_move': r.last_move.strftime('%Y-%m-%d') if r.last_move else None,
} for r in rows]
def _get_session_scope(session_id):
"""读取会话的范围配置。
:return: (scope_type, locations) —— locations 仅当抽盘且配置了库位时非空
"""
if not session_id:
return STOCKTAKE_SCOPE_FULL, []
sess = StocktakeSession.query.filter_by(session_id=session_id).first()
if not sess:
return STOCKTAKE_SCOPE_FULL, []
cfg = sess.scope_config or {}
locs = cfg.get('locations') or []
if sess.scope_type != STOCKTAKE_SCOPE_ACTIVE or not locs:
return STOCKTAKE_SCOPE_FULL, []
return STOCKTAKE_SCOPE_ACTIVE, list(locs)
def find_stock_owner_company(uuid_or_barcode):
"""全局查找该条码/SKU 属于哪个公司 —— **不施加任何公司过滤**。
只在 get_stock_info 未命中时调用,用来区分两种「查不到」:
· 条码根本不存在 → 让工人核对是否扫错 / 敲错
· 条码存在但属于别家公司 → 明确告知跨公司,而不是笼统的「未找到物料」
前提SKU / 条码在全系统全局唯一。已核验三张库存表「表内重复、跨公司重复、
跨表重复」六项检查均为 0因此这里至多命中一条不存在归属歧义。
:return: 公司名;查不到则 None
"""
code = str(uuid_or_barcode).strip()
if not code:
return None
checks = [
(StockProduct, lambda m, c: db.or_(m.barcode == c, m.sku == c, m.serial_number == c)),
(StockSemi, lambda m, c: db.or_(m.barcode == c, m.sku == c, m.serial_number == c)),
(StockBuy, lambda m, c: db.or_(m.barcode == c, m.sku == c)),
]
for model, cond_fn in checks:
if not model:
continue
row = db.session.query(MaterialBase.company_name).join(
model, model.base_id == MaterialBase.id
).filter(cond_fn(model, code)).first()
if row and row[0]:
return row[0]
return None
def _classify_missing_stock(code, company_name):
"""把「查不到库存」细分为「不存在」与「跨公司」。
:return: (msg, http_status) —— 调用方按自己的响应结构包装
"""
if company_name and company_name != '__NO_COMPANY__':
owner = find_stock_owner_company(code)
if owner and owner != company_name:
return f'条码 [{code}] 属于【{owner}】,请勿跨公司盘点', 403
return f'未找到该物料库存: {code}', 404
def get_stock_info(uuid_or_barcode, company_name=None):
"""
根据 uuid 或 barcode 查询库存信息(★ 精确匹配优先,性能与准确性兼顾)
修复: 原来用 ilike %x% 全表模糊搜索 + .first()
在 SKU 前缀相同的场景会命中错误记录或漏匹配。
改为: 精确匹配(==)优先,命中即返回;无精确命中再回退模糊搜索。
★ 公司隔离: 不同公司可能存在相同 barcode/sku不加公司条件时先被查到的
记录会「吃掉」条码造成跨公司串货。company_name 由调用方经
get_current_company_filter() 取得。
返回: (item, source_table, stock_id) 或 (None, None, None)
"""
# 清洗输入:去掉前后空格和换行符
code = str(uuid_or_barcode).strip()
if not code:
return None, None, None
# ===== 精确匹配优先(走索引,快且准) =====
exact_checks = [
(StockProduct, lambda c: db.or_(
StockProduct.barcode == c,
StockProduct.sku == c,
StockProduct.serial_number == c
), 'stock_product'),
(StockSemi, lambda c: db.or_(
StockSemi.barcode == c,
StockSemi.sku == c,
StockSemi.serial_number == c
), 'stock_semi'),
(StockBuy, lambda c: db.or_(
StockBuy.barcode == c,
StockBuy.sku == c
), 'stock_buy'),
]
for model, cond_fn, table_name in exact_checks:
if not model:
continue
item = _filter_by_company(model.query, model, company_name).filter(cond_fn(code)).first()
if item:
return (item, table_name, item.id)
# ===== 精确未命中 → 回退模糊搜索(保留旧行为兜底) =====
fuzzy_checks = [
(StockProduct, lambda c: db.or_(
StockProduct.barcode.ilike(f"%{c}%"),
StockProduct.sku.ilike(f"%{c}%"),
StockProduct.serial_number.ilike(f"%{c}%")
), 'stock_product'),
(StockSemi, lambda c: db.or_(
StockSemi.barcode.ilike(f"%{c}%"),
StockSemi.sku.ilike(f"%{c}%"),
StockSemi.serial_number.ilike(f"%{c}%")
), 'stock_semi'),
(StockBuy, lambda c: db.or_(
StockBuy.barcode.ilike(f"%{c}%"),
StockBuy.sku.ilike(f"%{c}%")
), 'stock_buy'),
]
for model, cond_fn, table_name in fuzzy_checks:
if not model:
continue
item = _filter_by_company(model.query, model, company_name).filter(cond_fn(code)).first()
if item:
return (item, table_name, item.id)
return None, None, None
return (None, None, None)
@bp.route('/all', methods=['GET'])
@jwt_required()
def get_all_stock():
"""
获取所有库存 > 0 的物品
支持 AI 极简模式: ?ai_mode=true
- 只返回 name / spec / availableQuantity 三个字段
- 键名压缩为 n / s / c
★ Fail-Closed: 非 AI 模式自动剥离所有价格成本字段
"""
ai_mode = request.args.get('ai_mode', '').lower() == 'true'
_strip = _make_price_stripper('inventory_stocktake')
try:
all_items = []
# 1. 采购件
if StockBuy:
rows = StockBuy.query.filter(
StockBuy.stock_quantity > 0
).options(joinedload(StockBuy.base)).all()
for item in rows:
if ai_mode:
b = item.base
all_items.append({
'n': b.name if b else '',
's': b.spec_model if b else '',
'c': float(item.available_quantity or 0)
})
else:
d = item.to_dict()
_strip(d, 'buy')
all_items.append(d)
# 2. 半成品
if StockSemi:
try:
rows = StockSemi.query.filter(
StockSemi.stock_quantity > 0
).options(joinedload(StockSemi.base)).all()
for item in rows:
if ai_mode:
b = item.base
all_items.append({
'n': b.name if b else '',
's': b.spec_model if b else '',
'c': float(item.available_quantity or 0)
})
else:
d = item.to_dict()
_strip(d, 'semi')
all_items.append(d)
except Exception:
pass
# 3. 成品
if StockProduct:
try:
rows = StockProduct.query.filter(
StockProduct.stock_quantity > 0
).options(joinedload(StockProduct.base)).all()
for item in rows:
if ai_mode:
b = item.base
all_items.append({
'n': b.name if b else '',
's': b.spec_model if b else '',
'c': float(item.available_quantity or 0)
})
else:
d = item.to_dict()
_strip(d, 'product')
all_items.append(d)
except Exception:
pass
return jsonify(all_items), 200
except Exception as e:
print(f"Error: {e}")
return jsonify({"message": f"查询库存失败: {str(e)}"}), 500
# ==============================================================================
# 分页库存查询接口(服务端分页,出库/盘点/借用模块共用)
# ==============================================================================
def _make_price_stripper(permission_prefix=None):
"""
Fail-Closed 价格字段剥离器工厂。
选单/借用场景默认剥离所有价格字段;
仅当 permission_prefix 不在已知选单前缀列表中时放行(例如 material_list 场景由调用方自行处理)。
"""
# 已知的选单/借用前缀 → 无条件剥离所有价格成本
# ★ 新增前缀必须登记在此,否则该前缀的调用方会拿到价格/成本字段Fail-Closed 失效)
SELECTION_PREFIXES = {'outbound_selection', 'op_borrow_apply', 'inventory_stocktake', 'scrap_apply'}
def stripper(d, stock_category):
if permission_prefix is None or permission_prefix in SELECTION_PREFIXES:
if stock_category == 'buy':
for k in ('unit_price', 'post_tax_unit_price', 'total_price',
'tax_rate', 'currency', 'exchange_rate',
'pre_tax_unit_price', 'qty_inbound', 'in_quantity'):
d.pop(k, None)
elif stock_category == 'semi':
for k in ('raw_material_cost', 'manual_cost', 'unit_total_cost',
'total_price', 'unit_price'):
d.pop(k, None)
elif stock_category == 'product':
for k in ('raw_material_cost', 'manual_cost', 'unit_total_cost',
'sale_price', 'unit_price'):
d.pop(k, None)
return stripper
def _do_get_stock_list(permission_prefix=None):
"""
分页获取库存列表stock_quantity > 0 — 裸逻辑,供各模块复用
Args:
permission_prefix: 可选权限前缀(如 'outbound_selection' / 'op_borrow_apply')。
传入后自动剥离对应模块无权查看的价格/成本字段。
"""
try:
page = request.args.get('page', 1, type=int)
pageSize = request.args.get('pageSize', 20, type=int)
keyword = request.args.get('keyword', '', type=str).strip()
if page < 1:
page = 1
if pageSize < 1 or pageSize > 200:
pageSize = 20
# ★ Fail-Closed: 选单/借用场景默认剥离所有价格成本字段
_strip_price_fields = _make_price_stripper(permission_prefix)
# ★ 行级公司隔离:普通用户只能看到本公司的库存(超管/跨域不受限)
company_limit = get_current_company_filter()
all_items = []
# 1. 采购件
if StockBuy:
q = StockBuy.query.filter(StockBuy.stock_quantity > 0)
if company_limit is not None:
q = q.filter(StockBuy.base.has(MaterialBase.company_name == company_limit))
if keyword:
q = q.filter(
db.or_(
StockBuy.base.has(MaterialBase.name.ilike(f'%{keyword}%')),
StockBuy.base.has(MaterialBase.spec_model.ilike(f'%{keyword}%')),
StockBuy.sku.ilike(f'%{keyword}%')
)
)
rows = q.all()
for item in rows:
d = item.to_dict()
d['source_table'] = 'stock_buy' # ★ 报废等按单流程需要精准定位来源表
d['stock_type'] = 'material'
d['type'] = 'material'
d['typeLabel'] = '采购件'
d['name'] = d.get('material_name', d.get('name', ''))
d['standard'] = d.get('spec_model', d.get('standard', ''))
d['available_quantity'] = d.get('qty_available', d.get('available_quantity', 0))
_strip_price_fields(d, 'buy')
all_items.append(d)
# 2. 半成品
if StockSemi:
try:
q = StockSemi.query.filter(StockSemi.stock_quantity > 0)
if company_limit is not None:
q = q.filter(StockSemi.base.has(MaterialBase.company_name == company_limit))
if keyword:
q = q.filter(
db.or_(
StockSemi.base.has(MaterialBase.name.ilike(f'%{keyword}%')),
StockSemi.base.has(MaterialBase.spec_model.ilike(f'%{keyword}%')),
StockSemi.sku.ilike(f'%{keyword}%')
)
)
rows = q.all()
for item in rows:
d = item.to_dict()
d['source_table'] = 'stock_semi' # ★ 精准来源表
d['stock_type'] = 'semi'
d['type'] = 'semi'
d['typeLabel'] = '半成品'
d['name'] = d.get('material_name', d.get('name', ''))
d['standard'] = d.get('spec_model', d.get('standard', ''))
d['available_quantity'] = d.get('qty_available', d.get('available_quantity', 0))
_strip_price_fields(d, 'semi')
all_items.append(d)
except Exception:
pass
# 3. 成品
if StockProduct:
q = StockProduct.query.filter(StockProduct.stock_quantity > 0)
if company_limit is not None:
q = q.filter(StockProduct.base.has(MaterialBase.company_name == company_limit))
if keyword:
q = q.filter(
db.or_(
StockProduct.base.has(MaterialBase.name.ilike(f'%{keyword}%')),
StockProduct.base.has(MaterialBase.spec_model.ilike(f'%{keyword}%')),
StockProduct.sku.ilike(f'%{keyword}%'),
StockProduct.barcode.ilike(f'%{keyword}%'),
StockProduct.serial_number.ilike(f'%{keyword}%')
)
)
rows = q.all()
for item in rows:
d = item.to_dict()
d['source_table'] = 'stock_product' # ★ 精准来源表
d['stock_type'] = 'product'
d['type'] = 'product'
d['typeLabel'] = '成品'
d['name'] = d.get('material_name', d.get('name', ''))
d['standard'] = d.get('spec_model', d.get('standard', ''))
d['available_quantity'] = d.get('qty_available', d.get('available_quantity', 0))
_strip_price_fields(d, 'product')
all_items.append(d)
# ── 按规格+库位聚合(出库选单合并同类项)───────────────────────
is_aggregated = request.args.get('is_aggregated', 'false').lower() == 'true'
if is_aggregated:
grouped_dict = {}
for item in all_items:
# 核心聚合键:类型 + 规格型号 + 库位 + base_id含 base_id 防止不同物料被错误合并)
group_key = f"{item.get('type')}_{item.get('standard')}_{item.get('warehouse_location', '')}_{item.get('base_id', '')}"
if group_key in grouped_dict:
# 累加数量
existing = grouped_dict[group_key]
existing['available_quantity'] = float(existing.get('available_quantity', 0)) + float(item.get('available_quantity', 0))
existing['stock_quantity'] = float(existing.get('stock_quantity', 0)) + float(item.get('stock_quantity', 0))
# 保留 id 列表(出库提交时需用到)
existing_ids = existing.get('_ids', [])
existing_ids.append(item.get('id'))
existing['_ids'] = existing_ids
else:
# 存入代表项
grouped_dict[group_key] = item.copy()
# 强制统一数据类型以便前端处理
grouped_dict[group_key]['available_quantity'] = float(item.get('available_quantity', 0))
grouped_dict[group_key]['stock_quantity'] = float(item.get('stock_quantity', 0))
grouped_dict[group_key]['_ids'] = [item.get('id')]
# 替换原列表为聚合后的列表
all_items = list(grouped_dict.values())
# ── 手动切片分页 ────────────────────────────────────────────
total = len(all_items)
start = (page - 1) * pageSize
end = start + pageSize
paged = all_items[start:end]
return jsonify({
'msg': '获取成功',
'data': {
'list': paged,
'total': total,
'page': page,
'pageSize': pageSize
}
}), 200
except Exception as e:
current_app.logger.error(f"Get Stock List Failed: {str(e)}")
return jsonify({'msg': f'获取库存列表失败: {str(e)}'}), 500
@bp.route('/list', methods=['GET'])
@jwt_required()
@permission_required('outbound_selection')
def get_stock_list():
"""出库选单专用库存列表 — Fail-Closed: 剥离价格字段"""
return _do_get_stock_list(permission_prefix='outbound_selection')
# --------------------------------------------------------
# 盘库/出库/借库 扫码精确匹配接口
# GET /api/v1/inbound/stock/scan?barcode=xxx
# 精确匹配优先,替代前端 pageSize:10 模糊搜索 + find() 的漏匹配问题
# --------------------------------------------------------
@bp.route('/scan', methods=['GET'])
@jwt_required()
def scan_stock_by_barcode():
"""根据条码精确匹配库存记录(一次返回唯一命中,性能好且准确)"""
try:
barcode = request.args.get('barcode', '').strip()
if not barcode:
return jsonify({'code': 400, 'msg': 'barcode 不能为空'}), 400
company_limit = get_current_company_filter()
item, source_table, stock_id = get_stock_info(barcode, company_limit)
if not item:
# 区分「条码不存在」与「属于别家公司」,给现场人员可执行的提示
msg, status = _classify_missing_stock(barcode, company_limit)
return jsonify({'code': status, 'msg': msg}), status
d = item.to_dict()
d['stock_type'] = source_table.replace('stock_', '')
d['type'] = source_table.replace('stock_', '')
d['source_table'] = source_table
d['stock_id'] = stock_id
# 兼容前端字段
if hasattr(item, 'base') and item.base:
d['name'] = d.get('material_name') or item.base.name or ''
d['standard'] = d.get('spec_model') or item.base.spec_model or ''
d['stock_quantity'] = float(d.get('stock_quantity') or d.get('qty_stock') or 0)
d['available_quantity'] = float(d.get('available_quantity') or d.get('qty_available') or 0)
return jsonify({'code': 200, 'msg': 'success', 'data': d}), 200
except Exception as e:
traceback.print_exc()
return jsonify({'code': 500, 'msg': str(e)}), 500
# --- 草稿箱接口 ---
@bp.route('/draft/list', methods=['GET'])
@permission_required('inventory_stocktake')
def get_drafts():
"""
获取盘点草稿列表
支持分页、搜索SKU和排序
"""
# 获取分页参数
page = request.args.get('page', 1, type=int)
limit = request.args.get('limit', 20, type=int)
keyword = request.args.get('keyword', '', type=str)
session_id = request.args.get('session_id')
uuid = request.args.get('uuid', '', type=str)
# ★ 公司隔离:普通用户只能看到本公司的盘点草稿(超管/跨域不过滤)
company_name = get_current_company_filter()
# 防止 limit 过大(保持防御性上限,避免极端参数拖垮数据库)
limit = min(max(limit, 1), 500)
# ── 公共 JOIN 片段:草稿按 source_table 关联三张库存表 + material_base ──
join_sql = """
FROM stocktake_draft sd
LEFT JOIN stock_buy b ON sd.source_table = 'stock_buy' AND b.id = sd.stock_id
LEFT JOIN stock_semi s ON sd.source_table = 'stock_semi' AND s.id = sd.stock_id
LEFT JOIN stock_product p ON sd.source_table = 'stock_product' AND p.id = sd.stock_id
LEFT JOIN material_base mb ON mb.id = COALESCE(b.base_id, s.base_id, p.base_id)
"""
# ── 动态 WHERE 条件(全部参数绑定,防止 SQL 注入)──
conditions = []
params = {}
if company_name is not None:
conditions.append('sd.company_name = :company')
params['company'] = company_name
if session_id:
conditions.append('sd.session_id = :sid')
params['sid'] = session_id
if uuid:
conditions.append('sd.uuid = :uuid')
params['uuid'] = uuid
if keyword:
conditions.append("LOWER(COALESCE(b.sku, s.sku, p.sku, '')) LIKE :kw")
params['kw'] = f'%{keyword.lower()}%'
where_clause = ('WHERE ' + ' AND '.join(conditions)) if conditions else ''
# ── 总数COUNT 聚合,不再全量拉取到内存)──
count_sql = f'SELECT COUNT(*) {join_sql} {where_clause}'
total = db.session.execute(db.text(count_sql), params).scalar() or 0
# ── 真实已盘数(按库存维度去重)──
scanned_sql = f"""
SELECT COUNT(*) FROM (
SELECT 1 {join_sql} {where_clause}
GROUP BY sd.source_table, sd.stock_id
) t
"""
total_scanned = db.session.execute(db.text(scanned_sql), params).scalar() or 0
# ── 数据查询(真正的数据库级 LIMIT/OFFSET 分页,一次性连表取字段)──
offset = (page - 1) * limit
data_sql = f"""
SELECT
sd.id AS draft_id, sd.user_id, sd.uuid, sd.quantity, sd.scan_time,
sd.session_id, sd.source_table, sd.stock_id, sd.stock_qty, sd.diff_qty, sd.remark,
COALESCE(b.sku, s.sku, p.sku, '') AS sku,
mb.name AS material_name,
mb.spec_model AS spec_model
{join_sql}
{where_clause}
ORDER BY LOWER(COALESCE(b.sku, s.sku, p.sku, ''))
LIMIT :limit OFFSET :offset
"""
query_params = dict(params)
query_params['limit'] = limit
query_params['offset'] = offset
rows = db.session.execute(db.text(data_sql), query_params).fetchall()
# ── 组装返回结构(仅当前页数据,无 N+1──
items = []
for row in rows:
items.append({
'id': row.draft_id,
'user_id': row.user_id,
'uuid': row.uuid,
'quantity': float(row.quantity or 1),
'scan_time': row.scan_time.strftime('%Y-%m-%d %H:%M:%S') if row.scan_time else None,
'session_id': row.session_id,
'source_table': row.source_table,
'stock_id': row.stock_id,
'stock_qty': float(row.stock_qty or 0),
'diff_qty': float(row.diff_qty or 0),
'remark': row.remark,
'sku': row.sku or '',
'material_name': row.material_name or '',
'spec_model': row.spec_model or ''
})
return jsonify({
'items': items,
'total': total,
'total_scanned': total_scanned,
'page': page,
'limit': limit
}), 200
@bp.route('/draft/add', methods=['POST'])
@permission_required('inventory_stocktake:operation')
def add_draft():
"""
扫码同步 (支持更新数量)
如果 session_id 不存在则创建新的会话
差异计算逻辑调整:
- adjusted_stock_qty = 账面总库存 - 借出未还数量
- diff_qty = 实盘数量 - adjusted_stock_qty
"""
try:
data = request.json
user_id = _normalize_user_id()
uuid = data.get('uuid')
print(f"🚀 [SCAN DEBUG] 后端实际接收到的 UUID 原文: |{uuid}| (长度: {len(str(uuid)) if uuid else 0})")
quantity = float(data.get('quantity', 1))
session_id = data.get('session_id')
# ★ 新增: 提取备注字段
remark = data.get('remark')
if not uuid:
return jsonify({"message": "UUID不能为空"}), 400
# 如果没有 session_id创建新的
if not session_id:
session_id = f"STK-{datetime.now().strftime('%Y%m%d%H%M%S')}-{uuid_module.uuid4().hex[:6]}"
# ★ 公司隔离:按当前用户所属公司查库存,避免跨公司同码物料串货
company_name = get_current_company_filter()
# 获取库存信息
item, source_table, stock_id = get_stock_info(uuid, company_name)
if not item:
# 区分「条码不存在」与「属于别家公司」
msg, status = _classify_missing_stock(uuid, company_name)
return jsonify({"message": msg}), status
# 账面总库存
stock_qty = float(item.stock_quantity) if item.stock_quantity else 0
# 计算借出未还数量 (quantity - returned_quantity)
borrowed_result = db.session.query(
db.func.sum(db.func.coalesce(TransBorrow.quantity, 0) - db.func.coalesce(TransBorrow.returned_quantity, 0))
).filter(
TransBorrow.source_table == source_table,
TransBorrow.stock_id == stock_id,
TransBorrow.is_returned == False
).scalar()
total_borrowed = float(borrowed_result) if borrowed_result else 0
# 调整后的账面可用库存 = 账面总库存 - 借出未还数量
adjusted_stock_qty = stock_qty - total_borrowed
# ★ 查找是否已存在:按 (company_name, session_id, uuid) 去重,不按 user_id 隔离
# 修复: 同一盘点单多个用户(手机/平板不同账号)操作同一物料时,
# 之前按 user_id 匹配导致每个用户各建一条 → 重复记录
# ★ 公司隔离: 去重键必须含公司,否则跨公司同码物料会互相覆盖
draft = StocktakeDraft.query.filter_by(
uuid=uuid, session_id=session_id, company_name=company_name
).first()
if draft:
# 如果已存在,更新数量和时间
draft.quantity = quantity
draft.scan_time = datetime.now()
draft.stock_qty = adjusted_stock_qty
draft.diff_qty = quantity - adjusted_stock_qty
draft.source_table = source_table
draft.stock_id = stock_id
# 更新操作用户(最后操作者)
draft.user_id = user_id
# ★ 新增: 保存备注
if remark is not None:
draft.remark = remark.strip() if isinstance(remark, str) else remark
else:
# 如果不存在,创建新的
draft = StocktakeDraft(
user_id=user_id,
uuid=uuid,
quantity=quantity,
session_id=session_id,
stock_qty=adjusted_stock_qty,
diff_qty=quantity - adjusted_stock_qty,
source_table=source_table,
stock_id=stock_id,
# ★ 新增: 保存备注
remark=remark.strip() if isinstance(remark, str) and remark else (remark if remark else None),
# ★ 公司隔离: 记录该草稿所属公司
company_name=company_name
)
db.session.add(draft)
db.session.commit()
return jsonify({
"message": "Saved",
"session_id": session_id,
"draft_id": draft.id,
"adjusted_stock_qty": adjusted_stock_qty,
"total_borrowed": total_borrowed
}), 200
except Exception as e:
print(f"Add Draft Error: {e}")
db.session.rollback()
return jsonify({"message": str(e)}), 500
@bp.route('/draft/clear', methods=['POST'])
@permission_required('inventory_stocktake:operation')
def clear_draft():
"""
清除盘点草稿
★ 必须指定 session_id —— 禁止不传参全表清空(历史上会误删其他公司/用户的进度)。
清除范围同时受公司隔离约束。
"""
data = request.json or {}
session_id = data.get('session_id')
if not session_id or not str(session_id).strip():
return jsonify({"message": "session_id 不能为空"}), 400
company_name = get_current_company_filter()
try:
# 清除指定会话(且限于本公司)
query = StocktakeDraft.query.filter_by(session_id=session_id)
if company_name is not None:
query = query.filter_by(company_name=company_name)
# 改为对象级删除以触发审计事件
records = query.all()
count = len(records)
for rec in records:
db.session.delete(rec)
db.session.commit()
return jsonify({"message": f"已清除 {count} 条记录", "count": count}), 200
except Exception as e:
db.session.rollback()
return jsonify({"message": str(e)}), 500
@bp.route('/stocktake/companies', methods=['GET'])
@permission_required('inventory_stocktake')
def get_stocktake_companies():
"""
盘点页「公司选择器」的选项。
单独开接口而不复用 /inbound/buy/options原因有二
1. 后者要求 inbound_buy 权限,只做盘点的库管会 403
2. 后者返回全部公司名,而普通用户只需要看得到本公司(后端本来就会强制隔离)。
返回: { companies: [str] } —— 超管拿到全部公司,普通用户只拿到本公司。
"""
company_name = get_current_company_filter()
try:
query = db.session.query(MaterialBase.company_name).filter(
MaterialBase.company_name.isnot(None),
MaterialBase.company_name != ''
)
# '__NO_COMPANY__' 是 get_current_company_filter 的哨兵值(用户未绑定公司),
# 拿它去匹配只会得到空列表 —— 这正是期望行为。
if company_name is not None:
query = query.filter(MaterialBase.company_name == company_name)
companies = sorted({r[0] for r in query.distinct().all()})
return jsonify({'code': 200, 'data': {'companies': companies}}), 200
except Exception as e:
traceback.print_exc()
return jsonify({'code': 500, 'msg': str(e)}), 500
@bp.route('/stocktake/recommend-locations', methods=['GET'])
@permission_required('inventory_stocktake')
def recommend_locations():
"""
推荐近 N 天最活跃的库位 —— 供前端「活跃库位抽盘」的树形勾选做预选。
★ 只推荐、不落库。最终范围由用户在库位树上微调后,随 /draft/start-new
的 scope_config.locations 一并提交,决定权在前端 UI。
查询参数: days(默认30) / top_n(默认50)
返回: { locations: [full_path...], detail: [{location,moves,sku_count,last_move}], days, top_n }
"""
company_name = get_current_company_filter()
if not company_name or company_name == '__NO_COMPANY__':
return jsonify({'code': 400, 'msg': '请先选择公司,再获取推荐库位'}), 400
try:
days = int(request.args.get('days', 30))
top_n = int(request.args.get('top_n', 50))
except (TypeError, ValueError):
return jsonify({'code': 400, 'msg': 'days / top_n 必须是整数'}), 400
# 防御性上限,避免把整库扫进来
days = max(1, min(days, 365))
top_n = max(1, min(top_n, 500))
try:
locs = get_active_locations(company_name, days=days, top_n=top_n)
return jsonify({
'code': 200,
'data': {
'locations': [x['location'] for x in locs],
'detail': locs,
'days': days,
'top_n': top_n,
}
}), 200
except Exception as e:
traceback.print_exc()
return jsonify({'code': 500, 'msg': str(e)}), 500
@bp.route('/draft/active-session', methods=['GET'])
@permission_required('inventory_stocktake')
def get_active_session():
"""
获取当前公司正在进行的盘点会话。
多人多设备协同PDA 进入盘点页先问这个接口 —— 若本司已有同事开过的会话,
直接加入该 session_id而不是各开各的。
★ 改造:改为直接查 stocktake_sessioncompany_name + status='active')。
旧实现靠 max(scan_time) 从草稿行里猜「最近会话」,有两个先天缺陷:
1. 空会话(刚开启还没扫码)在草稿表里没有行,会被误判为「无盘点」,
其他 PDA 于是各开各的,会话分裂;
2. 已结束的会话没有状态标记generate-missing 跑完后仍被当成活跃。
现在活跃判定是精确查询,两个问题一并解决。
返回: {
session_id: str|null,
mode: 'open'|'blind'|null, # 明盘/盲盘
scope_type: 'full'|'active'|null,# 全仓/抽盘
total: int, # 该会话的草稿行数
scanned: int, # 实际扫到的件数(按物料去重,排除系统自动生成的漏盘记录)
last_scan_time: str|null,
initiator: str|null, # 发起人显示名(该会话最早一条扫码记录的操作人)
}
"""
company_name = get_current_company_filter()
empty = {
'session_id': None, 'mode': None, 'scope_type': None,
'total': 0, 'scanned': 0, 'last_scan_time': None, 'initiator': None
}
# 跨域角色未指定公司时无法确定「哪个公司的活跃会话」,直接返回无
if not company_name or company_name == '__NO_COMPANY__':
return jsonify({'code': 200, 'data': empty}), 200
try:
session = StocktakeSession.query.filter_by(
company_name=company_name, status=STOCKTAKE_STATUS_ACTIVE
).order_by(StocktakeSession.created_at.desc()).first()
if not session:
return jsonify({'code': 200, 'data': empty}), 200
session_id = session.session_id
def _scoped(q):
return q.filter(
StocktakeDraft.company_name == company_name,
StocktakeDraft.session_id == session_id
)
# 已扫件数:按 (source_table, stock_id) 去重。
# 排除 user_id == 'system' 的记录 —— 那是「结束盘点」时 generate-missing
# 自动生成的漏盘(实盘=0不是人扫到的计入会让进度虚高。
scanned = _scoped(
db.session.query(StocktakeDraft.source_table, StocktakeDraft.stock_id)
.filter(StocktakeDraft.user_id != 'system')
).distinct().count()
agg = _scoped(
db.session.query(
db.func.count(StocktakeDraft.id).label('total'),
db.func.max(StocktakeDraft.scan_time).label('last_scan_time')
)
).first()
# 发起人:该会话最早一条非系统扫码记录的操作人。
# 注:/draft/add 更新已有记录时会把 user_id 改写成「最后操作者」,
# 所以这里取到的是最早那条记录的最后操作人,是当前模型下最接近发起人的信号。
first_row = _scoped(
db.session.query(StocktakeDraft.user_id)
.filter(StocktakeDraft.user_id != 'system')
).order_by(StocktakeDraft.scan_time.asc()).first()
initiator = _resolve_user_name(first_row[0]) if first_row and first_row[0] else None
return jsonify({
'code': 200,
'data': {
'session_id': session_id,
'mode': session.mode,
'scope_type': session.scope_type,
'total': int(agg.total or 0) if agg else 0,
'scanned': int(scanned or 0),
'last_scan_time': agg.last_scan_time.strftime('%Y-%m-%d %H:%M:%S') if agg and agg.last_scan_time else None,
'initiator': initiator
}
}), 200
except Exception as e:
traceback.print_exc()
return jsonify({'code': 500, 'msg': str(e)}), 500
@bp.route('/draft/start-new', methods=['POST'])
@permission_required('inventory_stocktake:operation')
def start_new_session():
"""
开始新一轮盘点:签发 session_id 并落库一条 StocktakeSession(status='active')。
- 不删除任何历史草稿:新旧会话靠 session_id 隔离,清理走 /draft/clear
- 同一公司同时只允许一个活跃会话DB 唯一部分索引兜底),
因此先把该公司原有的活跃会话置为 finished再插入新记录同一事务
"""
data = request.get_json(silent=True) or {}
# 公司:普通用户强制取本公司;跨域角色(超管/WAREHOUSE_MGR必须显式指定
company_limit = get_current_company_filter()
if company_limit == '__NO_COMPANY__':
return jsonify({"message": "当前账号未绑定公司,无法开始盘点"}), 400
if company_limit is None:
company_name = (data.get('company_name') or '').strip()
if not company_name or company_name.upper() == 'ALL':
return jsonify({"message": "请先选择要盘点的公司"}), 400
else:
company_name = company_limit
mode = (data.get('mode') or STOCKTAKE_MODE_OPEN).strip().lower()
if mode not in (STOCKTAKE_MODE_OPEN, STOCKTAKE_MODE_BLIND):
return jsonify({"message": f"不支持的盘点模式: {mode}"}), 400
scope_type = (data.get('scope_type') or STOCKTAKE_SCOPE_FULL).strip().lower()
if scope_type not in (STOCKTAKE_SCOPE_FULL, STOCKTAKE_SCOPE_ACTIVE):
return jsonify({"message": f"不支持的盘点范围: {scope_type}"}), 400
scope_config = data.get('scope_config') or {}
if not isinstance(scope_config, dict):
return jsonify({"message": "scope_config 必须是对象"}), 400
# ★ 抽盘:开单时就把活跃库位**冻结**进 scope_config。
# 必须在创建会话这一刻算好并落库 —— all-items 与 generate-missing 稍后都要
# 读同一份范围;若各自实时计算,两次调用之间库位活跃度变化会导致范围漂移,
# 结束盘点时就会把「开单时在范围内、比对时已掉出范围」的库存误判成盘亏。
# ★ 抽盘:范围由**前端**决定(系统推荐 + 人工在库位树上微调后提交),
# 后端只做校验与归一化,不再自行计算 —— 否则用户手改的勾选会被覆盖。
if scope_type == STOCKTAKE_SCOPE_ACTIVE:
raw_locs = scope_config.get('locations')
if not isinstance(raw_locs, list):
return jsonify({
"message": "抽盘必须通过 scope_config.locations 提交库位列表(可先调 /stocktake/recommend-locations 拿推荐)"
}), 400
# 归一化:去空白、去重、保序
clean_locs, seen = [], set()
for x in raw_locs:
s = str(x).strip()
if s and s not in seen:
seen.add(s)
clean_locs.append(s)
if not clean_locs:
return jsonify({"message": "抽盘至少要勾选一个库位"}), 400
if len(clean_locs) > 2000:
return jsonify({
"message": f"单次抽盘最多 2000 个库位,当前 {len(clean_locs)}"
}), 400
scope_config = {
'locations': clean_locs,
'recommend_days': scope_config.get('recommend_days'),
'recommend_top_n': scope_config.get('recommend_top_n'),
'computed_at': beijing_time().strftime('%Y-%m-%d %H:%M:%S'),
}
new_session_id = f"STK-{datetime.now().strftime('%Y%m%d%H%M%S')}-{uuid_module.uuid4().hex[:6]}"
try:
# 先给本公司原有的活跃会话收尾,否则会撞 uq_stocktake_session_one_active
superseded = StocktakeSession.query.filter_by(
company_name=company_name, status=STOCKTAKE_STATUS_ACTIVE
).all()
for s in superseded:
s.status = STOCKTAKE_STATUS_FINISHED
s.finished_at = beijing_time()
db.session.add(StocktakeSession(
session_id=new_session_id,
company_name=company_name,
mode=mode,
scope_type=scope_type,
scope_config=scope_config,
status=STOCKTAKE_STATUS_ACTIVE,
created_by=_normalize_user_id(),
))
db.session.commit()
return jsonify({
"message": "已开启新盘点会话(历史草稿保留)",
"session_id": new_session_id,
"company_name": company_name,
"mode": mode,
"scope_type": scope_type,
"superseded_count": len(superseded),
"cleared_count": 0
}), 200
except Exception as e:
db.session.rollback()
traceback.print_exc()
return jsonify({"message": str(e)}), 500
# --- 盘点结束与差异报告 ---
@bp.route('/finish', methods=['POST'])
@permission_required('inventory_stocktake:operation')
def finish_stocktake():
"""
结束盘点
直接返回成功,前端会跳转到差异列表
草稿数据保留在表中
"""
return jsonify({
"message": "盘点已结束",
"code": 200
}), 200
@bp.route('/variance-report', methods=['GET'])
@permission_required('inventory_stocktake')
def get_variance_report():
"""
获取盘点差异报告
返回所有有差异的记录diff_qty != 0
"""
session_id = request.args.get('session_id')
# ★ 公司隔离:普通用户只能看到本公司的差异(超管/跨域不过滤)
company_name = get_current_company_filter()
try:
query = StocktakeDraft.query
if company_name is not None:
query = query.filter_by(company_name=company_name)
if session_id:
query = query.filter_by(session_id=session_id)
# 只返回有差异的记录
drafts = query.filter(StocktakeDraft.diff_qty != 0).order_by(
StocktakeDraft.scan_time.desc(),
StocktakeDraft.diff_qty.desc()
).all()
# 补充库存详情
result = []
for draft in drafts:
draft_dict = draft.to_dict()
# 获取库存详情
if draft.stock_id and draft.source_table:
stock = get_stock_record(draft.source_table, draft.stock_id)
if stock:
draft_dict['stock_name'] = getattr(stock, 'material_name', None) or \
getattr(stock, 'product_name', None) or ''
draft_dict['stock_spec'] = getattr(stock, 'spec_model', '') or \
getattr(stock, 'standard', '') or ''
draft_dict['stock_location'] = getattr(stock, 'warehouse_location', '') or ''
draft_dict['stock_unit'] = getattr(stock, 'unit', '')
result.append(draft_dict)
return jsonify({
"list": result,
"total": len(result)
}), 200
except Exception as e:
print(f"Error: {e}")
return jsonify({"message": str(e)}), 500
# --- 单条库存调整 (手动平账) ---
@bp.route('/adjust', methods=['POST'])
@permission_required('inventory_stocktake:operation')
def adjust_stock():
"""
单条库存调整接口
接收指定的草稿 ID执行以下操作
1. 根据 diff_qty 调整库存 (stock_quantity 和 available_quantity)
2. 生成流水账记录 (盘盈入库 / 盘亏出库)
3. 标记草稿为 is_processed=True
支持两种模式:
- 有草稿模式:通过 draft_id 或 stock_id+source_table 查找草稿
- 无草稿模式:直接传入 stock_id + diff_qty + source_table未扫码直接盘亏
"""
data = request.json
draft_id = data.get('draft_id')
stock_id = data.get('stock_id')
diff_qty = data.get('diff_qty')
source_table = data.get('source_table')
operator_name = data.get('operator_name', 'System')
remark = data.get('remark', '')
if not draft_id and not stock_id:
return jsonify({"message": "draft_id 或 stock_id 不能同时为空"}), 400
try:
# ★ 公司隔离:平账会写库存,绝不允许命中其他公司的草稿
company_name = get_current_company_filter()
# 1. 尝试获取草稿
draft = StocktakeDraft.query.get(draft_id) if draft_id else None
if draft is not None and company_name is not None and draft.company_name != company_name:
# draft_id 属于其他公司 → 视为未命中,走后续兜底/无草稿分支
draft = None
if not draft and stock_id and source_table:
draft_query = StocktakeDraft.query.filter_by(
stock_id=stock_id,
source_table=source_table
)
if company_name is not None:
draft_query = draft_query.filter_by(company_name=company_name)
draft = draft_query.first()
elif not draft and stock_id:
draft_query = StocktakeDraft.query.filter_by(stock_id=stock_id)
if company_name is not None:
draft_query = draft_query.filter_by(company_name=company_name)
draft = draft_query.first()
# 2. 核心逻辑分支
if draft:
# 有草稿模式
stock_id = draft.stock_id
source_table = draft.source_table
diff_qty = float(draft.diff_qty)
else:
# 无草稿模式(未扫码直接盘亏)
if diff_qty is None or source_table is None or not stock_id:
return jsonify({"message": "未扫码物资平账缺失必要参数(需提供 diff_qty 和 source_table"}), 400
diff_qty = float(diff_qty)
# 3. 获取并校验真实的库存记录 — ★ 修复并发:使用悲观锁防止 TOCTOU
stock = get_stock_record(source_table, stock_id, for_update=True)
if not stock:
return jsonify({"message": "平账失败:物理库存记录已不存在"}), 404
# 4. 计算调整
if diff_qty > 0:
# 盘盈:增加库存
new_stock_qty = float(stock.stock_quantity or 0) + diff_qty
new_avail_qty = float(stock.available_quantity or 0) + diff_qty
action_type = '盘盈入库'
elif diff_qty < 0:
# 盘亏:减少库存
abs_diff = abs(diff_qty)
current_avail = float(stock.available_quantity or 0)
if current_avail < abs_diff:
return jsonify({
"message": f"可用库存不足,当前可用: {current_avail},需要减少: {abs_diff}"
}), 400
new_stock_qty = float(stock.stock_quantity or 0) - abs_diff
new_avail_qty = current_avail - abs_diff
action_type = '盘亏出库'
else:
return jsonify({"message": "差异为0无需调整"}), 400
# 5. 执行库存调整
stock.stock_quantity = new_stock_qty
stock.available_quantity = new_avail_qty
# 6. 生成流水账记录
from app.models.outbound import TransOutbound
# 生成唯一单号
adj_no_suffix = draft.id if draft else f"MANUAL-{datetime.now().strftime('%Y%m%d%H%M%S')}"
trans_record = TransOutbound(
outbound_no=f"STKADJ-{datetime.now().strftime('%Y%m%d%H%M%S')}-{adj_no_suffix}",
sku=stock.sku,
source_table=source_table,
stock_id=stock_id,
barcode=getattr(stock, 'barcode', ''),
quantity=abs(diff_qty),
unit_price=getattr(stock, 'pre_tax_unit_price', 0) or getattr(stock, 'manual_cost', 0) or 0,
outbound_type=action_type, # 盘盈入库 / 盘亏出库
consumer_name='盘点调整',
operator_name=operator_name,
remark=f"{action_type} - 盘点差异调整,备注: {remark}",
outbound_time=datetime.now()
)
db.session.add(trans_record)
# 7. 删除草稿记录(仅在有草稿时)
if draft:
db.session.delete(draft)
db.session.commit()
return jsonify({
"message": f"{action_type}成功",
"action_type": action_type,
"diff_qty": diff_qty,
"new_stock_qty": new_stock_qty,
"new_avail_qty": new_avail_qty,
"draft_id": draft_id if draft_id else (draft.id if draft else None)
}), 200
except Exception as e:
db.session.rollback()
print(f"Adjust Stock Error: {e}")
return jsonify({"message": str(e)}), 500
@bp.route('/borrowed-quantities', methods=['POST'])
@permission_required('inventory_stocktake')
def get_borrowed_quantities():
"""批量获取借出未还数量"""
data = request.json.get('items', [])
result = {}
for item in data:
source = item.get('source_table')
stock_id = item.get('stock_id')
if source and stock_id is not None:
qty = TransBorrow.get_borrowed_quantity(source, stock_id)
result[f"{source}_{stock_id}"] = qty
return jsonify(result), 200
# --- 打印接口 ---
@bp.route('/print/selection', methods=['POST'])
@permission_required('inventory_stocktake:operation')
def print_selection():
try:
data = request.json
items = data.get('items', [])
if not items: return jsonify({"message": "未选择任何物品"}), 400
printer = NetworkPrintService()
success, msg = printer.print_outbound_selection(items)
return jsonify({"message": "打印指令已发送" if success else msg}), 200 if success else 500
except Exception as e:
return jsonify({"message": str(e)}), 500
@bp.route('/print/stocktake', methods=['POST'])
@permission_required('inventory_stocktake:operation')
def print_stocktake():
try:
data = request.json
printer = NetworkPrintService()
success, msg = printer.print_stocktake_report(data)
return jsonify({"message": "盘点报告已发送" if success else msg}), 200 if success else 500
except Exception as e:
return jsonify({"message": str(e)}), 500
@bp.route('/export-stocktake', methods=['GET'])
@permission_required('inventory_stocktake:operation')
def export_stocktake():
"""
导出盘点报告 Excel
包含3个Sheet:
1. 盘点差异明细 (diff_qty != 0)
2. 账实相符明细 (diff_qty == 0)
3. 外借在用资产明细 (未归还的借出记录)
4. 未盘点明细(疑似漏盘)
"""
try:
# ★ 获取 session_id 参数,用于过滤当前会话的扫描记录
session_id = request.args.get('session_id', '', type=str)
# ★ 公司隔离:导出内容限定在当前用户所属公司
company_name = get_current_company_filter()
# ★ 盲盘保护会话仍在进行中active且为盲盘时报表里不得出现账面数与差异。
# 会话一旦 finished 即解锁 —— 盘点结束后再核对差异是正常流程,
# 否则报告本身就没法用了。
_sess = StocktakeSession.query.filter_by(session_id=session_id).first() if session_id else None
blind_locked = bool(
_sess
and _sess.status == STOCKTAKE_STATUS_ACTIVE
and _sess.mode == STOCKTAKE_MODE_BLIND
)
BLIND_BLANK = ''
def _qty(v):
"""账面数/差异:盲盘锁定期间输出占位符,绝不落真实数值。"""
return BLIND_BLANK if blind_locked else float(v or 0)
# 创建工作簿
wb = Workbook()
wb.remove(wb.active)
# 定义样式
header_font = Font(bold=True, size=11, color="FFFFFF")
header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
header_alignment = Alignment(horizontal="center", vertical="center")
thin_border = Border(
left=Side(style='thin'),
right=Side(style='thin'),
top=Side(style='thin'),
bottom=Side(style='thin')
)
def get_material_info(source_table, stock_id):
"""获取物料基本信息"""
if source_table == 'stock_buy':
stock = StockBuy.query.get(stock_id)
elif source_table == 'stock_semi':
stock = StockSemi.query.get(stock_id) if StockSemi else None
elif source_table == 'stock_product':
stock = StockProduct.query.get(stock_id) if StockProduct else None
else:
return {'name': '-', 'sku': '-', 'spec': '-', 'unit': '-', 'location': '-'}
if not stock:
return {'name': '-', 'sku': '-', 'spec': '-', 'unit': '-', 'location': '-'}
# 安全获取 sku
stock_sku = getattr(stock, 'sku', None) or getattr(stock, 'SKU', None) or '-'
# 使用 base_id 关联查询物料基础表
material = None
base_id = getattr(stock, 'base_id', None)
if base_id:
material = MaterialBase.query.get(base_id)
# 规格型号:从 MaterialBase 的 spec_model 字段获取
spec = getattr(material, 'spec_model', None) if material else '-'
if not spec or spec == '-':
spec = getattr(stock, 'spec_model', None) or getattr(stock, 'standard', None) or '-'
# 库位:从库存表的 warehouse_location 字段获取
location = getattr(stock, 'warehouse_location', None) or '-'
return {
'name': material.name if material else stock_sku,
'sku': stock_sku,
'spec': spec,
'unit': getattr(stock, 'unit', None) or '',
'location': location
}
def get_user_name(user_id):
return _resolve_user_name(user_id)
def to_beijing_time(dt):
"""直接使用数据库中存储的标准时间(服务器时区已正确)"""
if not dt:
return ''
try:
if isinstance(dt, str):
return dt[:19]
return dt.strftime('%Y-%m-%d %H:%M:%S')
except:
return str(dt)[:19]
def set_header_row(ws, headers):
for col, header in enumerate(headers, 1):
cell = ws.cell(row=1, column=col, value=header)
cell.font = header_font
cell.fill = header_fill
cell.alignment = header_alignment
cell.border = thin_border
# ===== Sheet 1: 盘点全景汇总表 (放在最前面) =====
ws1 = wb.create_sheet("盘点全景汇总表", 0)
summary_headers = ["物料名称", "SKU", "规格型号", "库位", "调整后账面数", "实盘数", "差异数", "盘点状态", "盘点人", "盘点时间", "备注"]
set_header_row(ws1, summary_headers)
master_row_idx = 2 # 汇总表行计数器
# ===== Sheet 2: 盘点差异明细 =====
ws2 = wb.create_sheet("盘点差异明细")
diff_headers = ["物料名称", "SKU", "规格型号", "库位", "调整后账面数", "实盘数", "差异数", "盘点人", "盘点时间", "备注"]
set_header_row(ws2, diff_headers)
# 按 SKU 排序:先获取全部数据,再在 Python 中按 SKU 排序
# ★ 原依赖 /draft/start-new 清空整表,「全部草稿」恰好等价于当前会话;
# start-new 已改为不删数据,故必须显式按 session_id + 公司过滤,
# 否则会把历史会话、其他公司的盘点一起导出。
diff_query = StocktakeDraft.query.filter(StocktakeDraft.diff_qty != 0)
if session_id:
diff_query = diff_query.filter(StocktakeDraft.session_id == session_id)
if company_name is not None:
diff_query = diff_query.filter(StocktakeDraft.company_name == company_name)
diff_drafts = diff_query.all()
diff_drafts_with_sku = []
for draft in diff_drafts:
mat_info = get_material_info(draft.source_table, draft.stock_id)
diff_drafts_with_sku.append((mat_info.get('sku', ''), draft))
diff_drafts_with_sku.sort(key=lambda x: x[0] if x[0] else '')
diff_drafts = [d[1] for d in diff_drafts_with_sku]
for row_idx, draft in enumerate(diff_drafts, 2):
mat_info = get_material_info(draft.source_table, draft.stock_id)
# 写入 Sheet 2 (差异明细)
ws2.cell(row=row_idx, column=1, value=mat_info['name']).border = thin_border
ws2.cell(row=row_idx, column=2, value=mat_info['sku']).border = thin_border
ws2.cell(row=row_idx, column=3, value=mat_info['spec']).border = thin_border
ws2.cell(row=row_idx, column=4, value=mat_info['location']).border = thin_border
ws2.cell(row=row_idx, column=5, value=_qty(draft.stock_qty)).border = thin_border
ws2.cell(row=row_idx, column=6, value=float(draft.quantity or 0)).border = thin_border
ws2.cell(row=row_idx, column=7, value=_qty(draft.diff_qty)).border = thin_border
ws2.cell(row=row_idx, column=8, value=get_user_name(draft.user_id)).border = thin_border
ws2.cell(row=row_idx, column=9, value=to_beijing_time(draft.scan_time)).border = thin_border
ws2.cell(row=row_idx, column=10, value=draft.remark or '').border = thin_border
# 同时写入 Sheet 1 (汇总表)
ws1.cell(row=master_row_idx, column=1, value=mat_info['name']).border = thin_border
ws1.cell(row=master_row_idx, column=2, value=mat_info['sku']).border = thin_border
ws1.cell(row=master_row_idx, column=3, value=mat_info['spec']).border = thin_border
ws1.cell(row=master_row_idx, column=4, value=mat_info['location']).border = thin_border
ws1.cell(row=master_row_idx, column=5, value=_qty(draft.stock_qty)).border = thin_border
ws1.cell(row=master_row_idx, column=6, value=float(draft.quantity or 0)).border = thin_border
ws1.cell(row=master_row_idx, column=7, value=_qty(draft.diff_qty)).border = thin_border
ws1.cell(row=master_row_idx, column=8, value="有差异").border = thin_border
ws1.cell(row=master_row_idx, column=9, value=get_user_name(draft.user_id)).border = thin_border
ws1.cell(row=master_row_idx, column=10, value=to_beijing_time(draft.scan_time)).border = thin_border
ws1.cell(row=master_row_idx, column=11, value=draft.remark or '').border = thin_border
master_row_idx += 1
# ===== Sheet 3: 账实相符明细 =====
ws3 = wb.create_sheet("账实相符明细")
normal_headers = ["物料名称", "SKU", "规格型号", "库位", "调整后账面数", "实盘数", "差异数", "盘点人", "盘点时间", "备注"]
set_header_row(ws3, normal_headers)
# 按 SKU 排序
# ★ 同上:显式按 session_id + 公司过滤
normal_query = StocktakeDraft.query.filter(StocktakeDraft.diff_qty == 0)
if session_id:
normal_query = normal_query.filter(StocktakeDraft.session_id == session_id)
if company_name is not None:
normal_query = normal_query.filter(StocktakeDraft.company_name == company_name)
normal_drafts = normal_query.all()
normal_drafts_with_sku = []
for draft in normal_drafts:
mat_info = get_material_info(draft.source_table, draft.stock_id)
normal_drafts_with_sku.append((mat_info.get('sku', ''), draft))
normal_drafts_with_sku.sort(key=lambda x: x[0] if x[0] else '')
normal_drafts = [d[1] for d in normal_drafts_with_sku]
for row_idx, draft in enumerate(normal_drafts, 2):
mat_info = get_material_info(draft.source_table, draft.stock_id)
# 写入 Sheet 3 (账实相符)
ws3.cell(row=row_idx, column=1, value=mat_info['name']).border = thin_border
ws3.cell(row=row_idx, column=2, value=mat_info['sku']).border = thin_border
ws3.cell(row=row_idx, column=3, value=mat_info['spec']).border = thin_border
ws3.cell(row=row_idx, column=4, value=mat_info['location']).border = thin_border
ws3.cell(row=row_idx, column=5, value=_qty(draft.stock_qty)).border = thin_border
ws3.cell(row=row_idx, column=6, value=float(draft.quantity or 0)).border = thin_border
ws3.cell(row=row_idx, column=7, value=_qty(draft.diff_qty)).border = thin_border
ws3.cell(row=row_idx, column=8, value=get_user_name(draft.user_id)).border = thin_border
ws3.cell(row=row_idx, column=9, value=to_beijing_time(draft.scan_time)).border = thin_border
ws3.cell(row=row_idx, column=10, value=draft.remark or '').border = thin_border
# 同时写入 Sheet 1 (汇总表)
ws1.cell(row=master_row_idx, column=1, value=mat_info['name']).border = thin_border
ws1.cell(row=master_row_idx, column=2, value=mat_info['sku']).border = thin_border
ws1.cell(row=master_row_idx, column=3, value=mat_info['spec']).border = thin_border
ws1.cell(row=master_row_idx, column=4, value=mat_info['location']).border = thin_border
ws1.cell(row=master_row_idx, column=5, value=_qty(draft.stock_qty)).border = thin_border
ws1.cell(row=master_row_idx, column=6, value=float(draft.quantity or 0)).border = thin_border
ws1.cell(row=master_row_idx, column=7, value=_qty(draft.diff_qty)).border = thin_border
ws1.cell(row=master_row_idx, column=8, value="正常").border = thin_border
ws1.cell(row=master_row_idx, column=9, value=get_user_name(draft.user_id)).border = thin_border
ws1.cell(row=master_row_idx, column=10, value=to_beijing_time(draft.scan_time)).border = thin_border
ws1.cell(row=master_row_idx, column=11, value=draft.remark or '').border = thin_border
master_row_idx += 1
# ===== Sheet 4: 外借在用资产明细 =====
ws4 = wb.create_sheet("外借在用资产明细")
borrow_headers = ["借出单号", "借用人", "物料名称", "SKU", "规格型号", "借出总数", "已还数量", "待还数量", "借出时间", "预计归还时间"]
set_header_row(ws4, borrow_headers)
# 查询未归还的借出记录
unreturned_borrows = TransBorrow.query.filter(TransBorrow.is_returned == False).all()
for row_idx, borrow in enumerate(unreturned_borrows, 2):
mat_info = get_material_info(borrow.source_table, borrow.stock_id)
total_qty = float(borrow.quantity or 0)
returned_qty = float(borrow.returned_quantity or 0)
pending_qty = total_qty - returned_qty
ws4.cell(row=row_idx, column=1, value=borrow.borrow_no or '').border = thin_border
ws4.cell(row=row_idx, column=2, value=borrow.borrower_name or '').border = thin_border
ws4.cell(row=row_idx, column=3, value=mat_info['name']).border = thin_border
ws4.cell(row=row_idx, column=4, value=mat_info['sku']).border = thin_border
ws4.cell(row=row_idx, column=5, value=mat_info['spec']).border = thin_border
ws4.cell(row=row_idx, column=6, value=total_qty).border = thin_border
ws4.cell(row=row_idx, column=7, value=returned_qty).border = thin_border
ws4.cell(row=row_idx, column=8, value=pending_qty).border = thin_border
ws4.cell(row=row_idx, column=9, value=to_beijing_time(borrow.borrow_time)).border = thin_border
ws4.cell(row=row_idx, column=10, value='无限期' if not borrow.expected_return_time else to_beijing_time(borrow.expected_return_time)).border = thin_border
# ===== Sheet 5: 未盘点明细(疑似漏盘) =====
# 逻辑:获取已盘点的集合,遍历库存表,找出未盘点且有库存的物资
ws5 = wb.create_sheet("未盘点明细(疑似漏盘)")
unscanned_headers = ["物料名称", "SKU", "规格型号", "库位", "批号", "调整后账面数", "实盘数", "差异数", "状态"]
set_header_row(ws5, unscanned_headers)
# 获取已盘点的 (source_table, stock_id) 集合
# ★ 修复:只查询当前 session_id 的扫描记录,避免历史记录干扰
scanned_query = StocktakeDraft.query
if session_id:
scanned_query = scanned_query.filter_by(session_id=session_id)
# ★ 公司隔离:无论是否传 session_id都不得跨公司
if company_name is not None:
scanned_query = scanned_query.filter_by(company_name=company_name)
session_drafts = scanned_query.all()
scanned_set = {(d.source_table, d.stock_id) for d in session_drafts}
# ★ 性能优化:批量预取所有未还借用的聚合数量
# 单条 GROUP BY 查询替代循环内逐条 get_borrowed_qty() N+1
borrow_rows = db.session.query(
TransBorrow.source_table,
TransBorrow.stock_id,
db.func.sum(db.func.coalesce(TransBorrow.quantity, 0) - db.func.coalesce(TransBorrow.returned_quantity, 0)).label('pending')
).filter(
TransBorrow.is_returned == False
).group_by(
TransBorrow.source_table, TransBorrow.stock_id
).all()
borrow_map = {(r.source_table, r.stock_id): float(r.pending or 0) for r in borrow_rows}
unscanned_items = []
# ★ 修复 N+1 查询:使用 joinedload 预加载 base 关系,同时过滤 stock_quantity > 0
for stock in _filter_by_company(StockBuy.query, StockBuy, company_name).filter(StockBuy.stock_quantity > 0).options(joinedload(StockBuy.base)).all():
key = ('stock_buy', stock.id)
if key in scanned_set:
continue
# ★ 扣除外借数量O(1) 字典查找替代逐条 TransBorrow 查询
borrowed_qty = borrow_map.get(key, 0)
stock_qty = float(stock.stock_quantity or 0)
expected_qty = stock_qty - borrowed_qty
if expected_qty > 0:
# ★ 直接使用预加载的 base 关系,避免额外查询
material = stock.base
# ★ 安全提取批号/序列号:使用 getattr 降级链
batch_sn = getattr(stock, 'batch_number', None) or getattr(stock, 'sn', None) or getattr(stock, 'serial_number', None) or '-'
mat_info = {
'name': material.name if material else '-',
'sku': getattr(stock, 'sku', None) or '-',
'spec': getattr(material, 'spec_model', None) if material else '-',
'location': getattr(stock, 'warehouse_location', None) or '-',
'batch_no': batch_sn
}
unscanned_items.append({
'name': mat_info['name'],
'sku': mat_info['sku'],
'spec': mat_info['spec'],
'location': mat_info['location'],
'batch_no': mat_info['batch_no'],
'stock_qty': expected_qty,
'actual_qty': 0,
'diff_qty': -expected_qty,
'status': '未盘点'
})
# 遍历 StockSemi
if StockSemi:
for stock in _filter_by_company(StockSemi.query, StockSemi, company_name).filter(StockSemi.stock_quantity > 0).options(joinedload(StockSemi.base)).all():
key = ('stock_semi', stock.id)
if key in scanned_set:
continue
borrowed_qty = borrow_map.get(key, 0)
stock_qty = float(stock.stock_quantity or 0)
expected_qty = stock_qty - borrowed_qty
if expected_qty > 0:
# ★ 直接使用预加载的 base 关系,避免额外查询
material = stock.base
# ★ 安全提取批号/序列号:使用 getattr 降级链
batch_sn = getattr(stock, 'batch_number', None) or getattr(stock, 'sn', None) or getattr(stock, 'serial_number', None) or '-'
mat_info = {
'name': material.name if material else '-',
'sku': getattr(stock, 'sku', None) or '-',
'spec': getattr(material, 'spec_model', None) if material else '-',
'location': getattr(stock, 'warehouse_location', None) or '-',
'batch_no': batch_sn
}
unscanned_items.append({
'name': mat_info['name'],
'sku': mat_info['sku'],
'spec': mat_info['spec'],
'location': mat_info['location'],
'batch_no': mat_info['batch_no'],
'stock_qty': expected_qty,
'actual_qty': 0,
'diff_qty': -expected_qty,
'status': '未盘点'
})
# 遍历 StockProduct
if StockProduct:
for stock in _filter_by_company(StockProduct.query, StockProduct, company_name).filter(StockProduct.stock_quantity > 0).options(joinedload(StockProduct.base)).all():
key = ('stock_product', stock.id)
if key in scanned_set:
continue
stock_qty = float(stock.stock_quantity or 0)
if stock_qty <= 0:
continue
borrowed_qty = borrow_map.get(key, 0)
expected_qty = stock_qty - borrowed_qty
if expected_qty > 0:
# ★ 直接使用预加载的 base 关系,避免额外查询
material = stock.base
# ★ 安全提取批号/序列号:使用 getattr 降级链 (成品可能无此字段)
batch_sn = getattr(stock, 'batch_number', None) or getattr(stock, 'sn', None) or getattr(stock, 'serial_number', None) or '-'
mat_info = {
'name': material.name if material else '-',
'sku': getattr(stock, 'sku', None) or '-',
'spec': getattr(material, 'spec_model', None) if material else '-',
'location': getattr(stock, 'warehouse_location', None) or '-',
'batch_no': batch_sn
}
unscanned_items.append({
'name': mat_info['name'],
'sku': mat_info['sku'],
'spec': mat_info['spec'],
'location': mat_info['location'],
'batch_no': mat_info['batch_no'],
'stock_qty': expected_qty,
'actual_qty': 0,
'diff_qty': -expected_qty,
'status': '未盘点'
})
# 写入未盘点明细
for row_idx, item in enumerate(unscanned_items, 2):
# 写入 Sheet 5 (未盘点明细)
ws5.cell(row=row_idx, column=1, value=item['name']).border = thin_border
ws5.cell(row=row_idx, column=2, value=item['sku']).border = thin_border
ws5.cell(row=row_idx, column=3, value=item['spec']).border = thin_border
ws5.cell(row=row_idx, column=4, value=item['location']).border = thin_border
ws5.cell(row=row_idx, column=5, value=item.get('batch_no', '-')).border = thin_border # ★ 批号
ws5.cell(row=row_idx, column=6, value=_qty(item['stock_qty'])).border = thin_border
ws5.cell(row=row_idx, column=7, value=float(item['actual_qty'])).border = thin_border
ws5.cell(row=row_idx, column=8, value=_qty(item['diff_qty'])).border = thin_border
ws5.cell(row=row_idx, column=9, value=item['status']).border = thin_border
# 同时写入 Sheet 1 (汇总表) - 盘点人和时间留空
ws1.cell(row=master_row_idx, column=1, value=item['name']).border = thin_border
ws1.cell(row=master_row_idx, column=2, value=item['sku']).border = thin_border
ws1.cell(row=master_row_idx, column=3, value=item['spec']).border = thin_border
ws1.cell(row=master_row_idx, column=4, value=item['location']).border = thin_border
ws1.cell(row=master_row_idx, column=5, value=_qty(item['stock_qty'])).border = thin_border
ws1.cell(row=master_row_idx, column=6, value=float(item['actual_qty'])).border = thin_border
ws1.cell(row=master_row_idx, column=7, value=_qty(item['diff_qty'])).border = thin_border
ws1.cell(row=master_row_idx, column=8, value="未盘点").border = thin_border
ws1.cell(row=master_row_idx, column=9, value="").border = thin_border
ws1.cell(row=master_row_idx, column=10, value="").border = thin_border
master_row_idx += 1
# 调整列宽
for ws in [ws1, ws2, ws3, ws4, ws5]:
for col in ws.columns:
max_length = 0
col_letter = col[0].column_letter
for cell in col:
try:
if len(str(cell.value)) > max_length:
max_length = len(str(cell.value))
except:
pass
ws.column_dimensions[col_letter].width = min(max_length + 2, 30)
# 生成文件
output = io.BytesIO()
wb.save(output)
output.seek(0)
filename = f"盘点报告_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
return send_file(
output,
mimetype='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
as_attachment=True,
download_name=filename
)
except Exception as e:
print(f"Export Stocktake Error: {e}")
import traceback
traceback.print_exc()
return jsonify({"message": f"导出失败: {str(e)}"}), 500
# --------------------------------------------------------
# 生成漏盘数据 - 将未扫描的库存标记为全额盘亏
# POST /api/v1/inbound/stocktake/generate-missing
# --------------------------------------------------------
@bp.route('/stocktake/generate-missing', methods=['POST'])
@permission_required('inventory_stocktake:operation')
def generate_missing_stocktake():
"""
生成漏盘数据(幂等性):
找出所有真实库存 > 0但未被当前会话盘点扫描到的物料
自动生成盘点草稿,标记为盘亏(实盘=0差异=-库存数)
幂等性保护:在重新计算差集之前,先删除当前 session 下所有
由系统自动生成的漏盘记录quantity==0, user_id=='system'
保证该接口多次调用结果一致。
"""
try:
# ★ 获取 session_id 参数,用于隔离当前会话
data = request.get_json() or {}
session_id = data.get('session_id', '')
if not session_id:
return jsonify({'code': 400, 'msg': '缺少 session_id 参数'}), 400
# ★ 公司隔离:漏盘比对只在当前公司的库存与草稿范围内进行
company_name = get_current_company_filter()
# ★ 幂等性保护:先删除当前 session 下系统自动生成的漏盘记录
# 特征user_id == 'system' (表示由系统自动生成)
# 改为对象级删除以触发审计事件
system_query = StocktakeDraft.query.filter(
StocktakeDraft.session_id == session_id,
StocktakeDraft.user_id == 'system'
)
if company_name is not None:
system_query = system_query.filter(StocktakeDraft.company_name == company_name)
system_records = system_query.all()
deleted_count = len(system_records)
for rec in system_records:
db.session.delete(rec)
if deleted_count > 0:
db.session.commit()
print(f"[generate_missing] 已清理 {deleted_count} 条历史漏盘记录")
# 1. 获取当前会话已有盘点记录的 (source_table, stock_id) 集合
existing_query = db.session.query(
StocktakeDraft.source_table,
StocktakeDraft.stock_id
).filter(StocktakeDraft.session_id == session_id)
if company_name is not None:
existing_query = existing_query.filter(StocktakeDraft.company_name == company_name)
existing_records = existing_query.distinct().all()
scanned_keys = set()
for src_table, stock_id in existing_records:
if stock_id:
scanned_keys.add((src_table, stock_id))
# 2. 获取所有真实库存 > 0 的记录
#
# ★★★ 抽盘范围(本功能最关键的一处)★★★
# 漏盘比对只针对**范围内**的库存。范围取自会话 scope_config 里开单时
# 冻结的库位清单,与 all-items / merged-list 完全同源。
# 若此过滤缺失或与其他接口不同源,结束盘点会把抽盘范围外的全部未盘
# 库存批量生成为盘亏记录 —— 直接污染库存账。
_scope_type, scope_locations = _get_session_scope(session_id)
def _in_scope(query, model):
if scope_locations:
query = query.filter(model.warehouse_location.in_(scope_locations))
return query
all_stock = []
# 采购库存
for item in _in_scope(_filter_by_company(StockBuy.query, StockBuy, company_name), StockBuy).filter(StockBuy.stock_quantity > 0).all():
all_stock.append({
'source_table': 'stock_buy',
'stock_id': item.id,
'base_id': item.base_id,
'stock_qty': float(item.stock_quantity or 0)
})
# 半成品库存
if StockSemi:
for item in _in_scope(_filter_by_company(StockSemi.query, StockSemi, company_name), StockSemi).filter(StockSemi.stock_quantity > 0).all():
all_stock.append({
'source_table': 'stock_semi',
'stock_id': item.id,
'base_id': item.base_id,
'stock_qty': float(item.stock_quantity or 0)
})
# 成品库存
if StockProduct:
for item in _in_scope(_filter_by_company(StockProduct.query, StockProduct, company_name), StockProduct).filter(StockProduct.stock_quantity > 0).all():
all_stock.append({
'source_table': 'stock_product',
'stock_id': item.id,
'base_id': item.base_id,
'stock_qty': float(item.stock_quantity or 0)
})
# 3. 找出漏盘记录(库存中有但盘点中没有的)
missing_count = 0
for stock in all_stock:
key = (stock['source_table'], stock['stock_id'])
if key not in scanned_keys:
# 生成漏盘草稿
draft = StocktakeDraft(
user_id='system', # ★ 标记为系统自动生成,用于幂等性清理
uuid=f'MISSING-{stock["source_table"]}-{stock["stock_id"]}',
quantity=0, # 实盘数为0
scan_time=beijing_time(),
session_id=session_id, # ★ 使用传入的 session_id
source_table=stock['source_table'],
stock_id=stock['stock_id'],
stock_qty=stock['stock_qty'],
diff_qty=-stock['stock_qty'], # 差异 = 0 - 库存数 = 负数
remark='未盘点到,系统自动标记为盘亏',
company_name=company_name # ★ 公司隔离
)
db.session.add(draft)
missing_count += 1
db.session.commit()
# ★ 会话收尾:漏盘比对完成即视为该轮盘点结束。
# 不置为 finished 的话,/draft/active-session 会继续把它当活跃会话返回,
# 其他 PDA 会「加入」一个已经结束的盘点。
finished_session = StocktakeSession.query.filter_by(
session_id=session_id, status=STOCKTAKE_STATUS_ACTIVE
).first()
if finished_session:
finished_session.status = STOCKTAKE_STATUS_FINISHED
finished_session.finished_at = beijing_time()
db.session.commit()
return jsonify({
'code': 200,
'msg': f'成功生成 {missing_count} 条漏盘记录',
'data': {'count': missing_count, 'session_finished': bool(finished_session)}
})
except Exception as e:
db.session.rollback()
traceback.print_exc()
return jsonify({'code': 500, 'msg': f'生成漏盘数据失败: {str(e)}'}), 500
# --------------------------------------------------------
# 盘点物资合并列表(服务端 JOIN draft + stock
# GET /api/v1/inbound/stock/draft/merged-list
# --------------------------------------------------------
@bp.route('/draft/merged-list', methods=['GET'])
@permission_required('inventory_stocktake')
def get_draft_merged_list():
"""
数据库级分页合并UNION ALL + LEFT JOIN + LIMIT/OFFSET
不再加载全量库存到 Python 内存,由 PostgreSQL 完成分页。
"""
try:
session_id = request.args.get('session_id', '').strip()
if not session_id:
return jsonify({'code': 400, 'msg': 'session_id 不能为空'}), 400
keyword = request.args.get('keyword', '').strip().lower()
status_filter = request.args.get('status_filter', '').strip() # counted / uncounted
page = max(request.args.get('page', 1, type=int), 1)
page_size = min(request.args.get('pageSize', 20, type=int), 200)
# ★ 公司隔离:盘点基数(库存侧)与已扫草稿(草稿侧)都限定在当前公司。
# 超管/跨域用户 company_name 为 None不加任何公司条件。
company_name = get_current_company_filter()
company_cond_sd = ' AND sd.company_name = :company' if company_name is not None else ''
company_cond_bare = ' AND company_name = :company' if company_name is not None else ''
# ── 公共 CTE / 子查询片段 ──
# ★ 0 库存盘盈可见性:账面为 0、但本会话已被扫入的物料也要进明细列表。
# 否则工人扫到货架上账面为 0 的实物(盘盈)时,「已盘」计数会 +1
# 明细里却找不到这一行,看起来像系统丢了数据。
# 注意条件必须限定在**本会话**的草稿,不能是库容里存在过草稿就放行。
# ★ 批次/序列号stock_product 表**没有 batch_number 列**(只有 serial_number
# 所以第三个分支必须给 batch_number 补一个空串占位,否则 UNION 两边
# 列数不齐 / 报 column does not exist。前端据此显示为 '-'。
union_sql = """
SELECT id, 'stock_buy' AS source_table, sku,
stock_quantity AS stock_qty, warehouse_location, base_id,
COALESCE(batch_number, '') AS batch_number,
COALESCE(serial_number, '') AS serial_number
FROM stock_buy
WHERE stock_quantity > 0
OR id IN (SELECT sd.stock_id FROM stocktake_draft sd
WHERE sd.source_table = 'stock_buy' AND sd.session_id = :sid)
UNION ALL
SELECT id, 'stock_semi', sku,
stock_quantity, warehouse_location, base_id,
COALESCE(batch_number, ''),
COALESCE(serial_number, '')
FROM stock_semi
WHERE stock_quantity > 0
OR id IN (SELECT sd.stock_id FROM stocktake_draft sd
WHERE sd.source_table = 'stock_semi' AND sd.session_id = :sid)
UNION ALL
SELECT id, 'stock_product', sku,
stock_quantity, warehouse_location, base_id,
'' AS batch_number, -- stock_product 无 batch_number 列
COALESCE(serial_number, '')
FROM stock_product
WHERE stock_quantity > 0
OR id IN (SELECT sd.stock_id FROM stocktake_draft sd
WHERE sd.source_table = 'stock_product' AND sd.session_id = :sid)
"""
# ── 构建 WHERE 条件 ──
conditions = []
params = {'sid': session_id}
if company_name is not None:
conditions.append("mb.company_name = :company")
params['company'] = company_name
# ★ 抽盘范围:与 all-items / generate-missing 读同一份冻结的库位清单。
# 不加这一条的话,抽屉里仍会列出范围外的物资,与「总品项」对不上。
_scope_type, scope_locations = _get_session_scope(session_id)
if scope_locations:
conditions.append("cs.warehouse_location = ANY(:locations)")
params['locations'] = scope_locations
if keyword:
conditions.append("(LOWER(cs.sku) LIKE :kw OR LOWER(mb.name) LIKE :kw)")
params['kw'] = f'%{keyword}%'
if status_filter == 'counted':
conditions.append("COALESCE(sd.quantity, 0) > 0")
elif status_filter == 'uncounted':
conditions.append("COALESCE(sd.quantity, 0) <= 0")
where_clause = ('WHERE ' + ' AND '.join(conditions)) if conditions else ''
# ── COUNT 查询 ──
count_sql = f"""
SELECT COUNT(*) FROM (
{union_sql}
) cs
LEFT JOIN material_base mb ON cs.base_id = mb.id
LEFT JOIN stocktake_draft sd ON sd.source_table = cs.source_table
AND sd.stock_id = cs.id AND sd.session_id = :sid{company_cond_sd}
{where_clause}
"""
total = db.session.execute(db.text(count_sql), params).scalar()
# ── 已扫数量(去重) ──
scanned_sql = f"""
SELECT COUNT(DISTINCT (source_table, stock_id))
FROM stocktake_draft WHERE session_id = :sid{company_cond_bare}
"""
# 该语句只引用 :sid / :company单独构造参数避免夹带未引用的 :kw
scanned_params = {'sid': session_id}
if company_name is not None:
scanned_params['company'] = company_name
total_scanned = db.session.execute(db.text(scanned_sql), scanned_params).scalar() or 0
# ── 数据查询LIMIT/OFFSET ──
offset = (page - 1) * page_size
data_sql = f"""
SELECT
cs.id AS stock_id, cs.source_table, cs.sku,
cs.batch_number, cs.serial_number,
cs.stock_qty, cs.warehouse_location,
mb.name AS mat_name, mb.spec_model AS mat_spec,
sd.id AS draft_id, sd.quantity AS draft_qty,
sd.remark AS draft_remark
FROM (
{union_sql}
) cs
LEFT JOIN material_base mb ON cs.base_id = mb.id
LEFT JOIN stocktake_draft sd ON sd.source_table = cs.source_table
AND sd.stock_id = cs.id AND sd.session_id = :sid{company_cond_sd}
{where_clause}
ORDER BY cs.sku
LIMIT :limit OFFSET :offset
"""
params['limit'] = page_size
params['offset'] = offset
rows = db.session.execute(db.text(data_sql), params).fetchall()
# ── 明/盲盘控制:账面数与差异由后端强制决定是否下发 ──
# ★ Fail-Safe查不到会话配置时按盲盘处理。宁可让界面少显示两列
# 也不能因为会话行缺失就把真实账面数泄漏出去。
session = StocktakeSession.query.filter_by(session_id=session_id).first()
is_blind = True if session is None else (session.mode == STOCKTAKE_MODE_BLIND)
# ── 组装结果 ──
items = []
for row in rows:
draft_qty = float(row.draft_qty or 0)
stock_qty = float(row.stock_qty or 0)
diff_qty = (draft_qty - stock_qty) if row.draft_id else -stock_qty
items.append({
'draft_id': row.draft_id,
'stock_id': row.stock_id,
'source_table': row.source_table,
'uniqueKey': f"{row.source_table}_{row.stock_id}",
'sku': row.sku or '',
'batch_number': row.batch_number or '',
'serial_number': row.serial_number or '',
'material_name': row.mat_name or '',
'spec_model': row.mat_spec or '',
# ★ 盲盘:账面数置空,真实值不得出现在 HTTP 响应里
'stock_qty': None if is_blind else stock_qty,
'quantity': draft_qty,
# ★ 盲盘:差异同样置空 —— diff_qty 由账面数算出,
# 下发差异等于变相泄漏账面数
'diff_qty': None if is_blind else round(diff_qty, 4),
'remark': row.draft_remark or '',
'warehouse_location': row.warehouse_location or '',
})
return jsonify({
'code': 200, 'msg': '获取成功',
'data': {
'list': items, 'total': total,
'total_scanned': total_scanned,
'page': page, 'pageSize': page_size,
# 前端据此决定表头提示;真正的数据屏蔽已经在上面做完了
'mode': session.mode if session is not None else STOCKTAKE_MODE_BLIND,
}
}), 200
except Exception as e:
current_app.logger.error(f'合并列表查询失败: {str(e)}')
return jsonify({'code': 500, 'msg': str(e)}), 500
# --------------------------------------------------------
# 获取应盘物资清单(盘点基数)
# GET /api/v1/inbound/stock/stocktake/all-items
# --------------------------------------------------------
@bp.route('/stocktake/all-items', methods=['GET'])
@permission_required('inventory_stocktake')
def get_all_stocktake_items():
"""
获取应盘物资清单(库存 > 0 的物料)— ★ 分页返回,禁止全量
性能优化: 原来三次 .all() 全量加载 + 内存排序,库存量大时打开极慢。
改为: 分页返回 {items(当前页), total(总数), total_scanned(已盘数)}。
"""
try:
keyword = request.args.get('keyword', '', type=str).strip()
page = max(1, request.args.get('page', 1, type=int))
pageSize = min(200, max(1, request.args.get('pageSize', 50, type=int)))
# ★ 公司隔离:盘点基数只统计本公司库存(超管/跨域不过滤)
company_name = get_current_company_filter()
company_cond_bare = ' AND company_name = :company' if company_name is not None else ''
# ── 原生 SQLUNION ALL 三张库存表 + LEFT JOIN material_base数据库级分页 ──
# 注意stock_product 表没有 batch_number 列,故第三个分支直接用 serial_number。
union_sql = """
SELECT id, 'stock_buy' AS source_table, sku, barcode,
stock_quantity AS stock_qty, available_quantity, warehouse_location, base_id,
COALESCE(batch_number, serial_number, '') AS batch_no
FROM stock_buy WHERE stock_quantity > 0
UNION ALL
SELECT id, 'stock_semi' AS source_table, sku, barcode,
stock_quantity, available_quantity, warehouse_location, base_id,
COALESCE(batch_number, serial_number, '') AS batch_no
FROM stock_semi WHERE stock_quantity > 0
UNION ALL
SELECT id, 'stock_product' AS source_table, sku, barcode,
stock_quantity, available_quantity, warehouse_location, base_id,
serial_number AS batch_no
FROM stock_product WHERE stock_quantity > 0
"""
# ── 动态 WHERESKU / 物料名 / 规格 模糊搜索)──
conditions = []
params = {}
if company_name is not None:
conditions.append("mb.company_name = :company")
params['company'] = company_name
# ★ 抽盘范围:范围在开单时已冻结进会话 scope_config这里只读不算
# 确保与 generate-missing 用的是同一份库位清单
_scope_type, scope_locations = _get_session_scope(
request.args.get('session_id', '', type=str)
)
if scope_locations:
conditions.append("cs.warehouse_location = ANY(:locations)")
params['locations'] = scope_locations
if keyword:
conditions.append("(LOWER(cs.sku) LIKE :kw OR LOWER(mb.name) LIKE :kw OR LOWER(mb.spec_model) LIKE :kw)")
params['kw'] = f'%{keyword.lower()}%'
where_clause = ('WHERE ' + ' AND '.join(conditions)) if conditions else ''
# 总数COUNT 聚合,不再全量加载)
count_sql = f"""
SELECT COUNT(*) FROM ( {union_sql} ) cs
LEFT JOIN material_base mb ON mb.id = cs.base_id
{where_clause}
"""
total = db.session.execute(db.text(count_sql), params).scalar() or 0
# 数据查询(真正的数据库级 LIMIT/OFFSET 分页)
offset = (page - 1) * pageSize
data_sql = f"""
SELECT cs.id, cs.source_table, cs.sku, cs.barcode, cs.batch_no,
cs.stock_qty, cs.available_quantity, cs.warehouse_location,
mb.name AS material_name, mb.spec_model AS spec_model
FROM ( {union_sql} ) cs
LEFT JOIN material_base mb ON mb.id = cs.base_id
{where_clause}
ORDER BY LOWER(cs.sku)
LIMIT :limit OFFSET :offset
"""
query_params = dict(params)
query_params['limit'] = pageSize
query_params['offset'] = offset
rows = db.session.execute(db.text(data_sql), query_params).fetchall()
# ── 组装当前页数据(无 item.base 懒加载 N+1──
paged = []
for row in rows:
paged.append({
'id': row.id,
'sku': row.sku or '',
'barcode': row.barcode or '',
'batch_no': row.batch_no or '',
'material_name': row.material_name or '',
'spec_model': row.spec_model or '',
'stock_qty': float(row.stock_qty or 0),
'available_qty': float(row.available_quantity or 0),
'source_table': row.source_table,
'warehouse_location': row.warehouse_location or ''
})
# 统计已盘数量(该 session 下已扫的SQL COUNT 聚合)
session_id = request.args.get('session_id', '', type=str)
total_scanned = 0
if session_id:
scanned_sql = f"""
SELECT COUNT(DISTINCT (source_table, stock_id))
FROM stocktake_draft WHERE session_id = :sid{company_cond_bare}
"""
scanned_params = {'sid': session_id}
if company_name is not None:
scanned_params['company'] = company_name
total_scanned = db.session.execute(
db.text(scanned_sql), scanned_params
).scalar() or 0
return jsonify({
'code': 200,
'data': {
'items': paged,
'total': total,
'total_scanned': total_scanned,
'page': page,
'pageSize': pageSize
}
}), 200
except Exception as e:
import traceback
traceback.print_exc()
return jsonify({'code': 500, 'msg': f'获取应盘清单失败: {str(e)}'}), 500
# --------------------------------------------------------
# 更新盘点实盘数(手动修改)
# POST /api/v1/inbound/stock/stocktake/update-quantity
# --------------------------------------------------------
@bp.route('/stocktake/update-quantity', methods=['POST'])
@permission_required('inventory_stocktake:operation')
def update_stocktake_quantity():
"""
更新盘点实盘数
用于手动修改盘点数量
"""
try:
data = request.json
stock_id = data.get('stock_id')
source_table = data.get('source_table')
session_id = data.get('session_id') # ★ 修复:按会话隔离,防止跨会话误改
quantity = float(data.get('quantity', 0))
# 备注为**可选**字段:只有显式传入时才覆盖,避免改实盘数时把备注清空
remark = data.get('remark', None)
if not stock_id or not source_table:
return jsonify({'code': 400, 'msg': '缺少必要参数'}), 400
# ★ 公司隔离:只能改本公司的盘点记录
company_name = get_current_company_filter()
# 查找对应的盘点记录(★ 修复:限定 session_id避免不同盘点会话相互覆盖
query = StocktakeDraft.query.filter_by(
stock_id=stock_id,
source_table=source_table
)
if session_id:
query = query.filter_by(session_id=session_id)
if company_name is not None:
query = query.filter_by(company_name=company_name)
draft = query.first()
if not draft:
# 该物料在本会话尚未扫码 → 没有草稿行可改。
# 前端已把未扫行的实盘数置为不可编辑,这里是兜底(例如页面未刷新、
# 或有人直接调接口)。刻意**不**在此处补建草稿 —— 一旦允许不扫码
# 就能填数,扫码这道工序就形同虚设,盘盈盘亏也无从查起。
return jsonify({
'code': 404,
'msg': '该物料尚未扫码,请先扫描条码后再修改实盘数'
}), 404
# 更新数量
draft.quantity = quantity
draft.scan_time = beijing_time()
# 计算差异(★ 这是写库计算,不是报表输出,绝不能用 _qty —— 那会在盲盘时
# 返回占位符 '—' 导致减法报错。账面数在这里始终按真实值参与计算。)
draft.diff_qty = quantity - float(draft.stock_qty or 0)
if remark is not None:
draft.remark = remark.strip() if isinstance(remark, str) else remark
db.session.commit()
return jsonify({'code': 200, 'msg': '更新成功'}), 200
except Exception as e:
db.session.rollback()
import traceback
traceback.print_exc()
return jsonify({'code': 500, 'msg': f'更新失败: {str(e)}'}), 500
# ==============================================================================
# 库存状态变更接口(逆向物流的基础能力)
# ==============================================================================
# ★ 为什么需要它
# 分配器已引入「仅 status='在库' 方能被分配」的硬隔离
# (见 app/services/inventory_reservation.py::allocatable_filter
# 但在此之前,**全系统没有任何入口**能改写 stock 行的 status —— 状态只能靠
# 手工改库,于是「坏件退回 / 送修 / 冻结」在系统里没有落点,逆向物流无从谈起。
# 本接口补上这一环。
#
# ★ 阶段二接入预告
# 退回 / 送修 / 报废流程落地时,应**复用本接口背后的同一条变更路径**
# (而不是各自复制一份赋值逻辑),以保证「什么状态算可出货」在全系统
# 只有一处定义。届时可考虑抽成 service 层函数,本路由只做鉴权与解析。
#
# ★ 审计留痕
# stock_buy / stock_semi / stock_product 均在 audit_listener 的白名单内
# app/core/audit_listener.py:44-48因此 status / quality_status 的
# 变更会由 SQLAlchemy 事件监听器**自动**写入 audit_logs记录操作人
# (取自 JWT、IP、变更前后的值 —— 本接口刻意不手工记账,避免双写。
# 注意:监听器要求 HTTP 请求上下文,故状态变更必须在请求内直接落库,
# 不可丢给后台任务,否则会静默失去审计痕迹。
@bp.route('/<int:stock_id>/change-status', methods=['POST'])
@permission_required('inventory_stocktake:operation')
def change_stock_status(stock_id):
"""
变更单条库存行的状态(在库 / 冻结 / 不良品)。
Body(JSON):
{
"source_table": "stock_buy" | "stock_semi" | "stock_product", # 必填
"status": "在库" | "冻结" | "不良品", # 必填
"quality_status": "合格" | "不合格" | "待检" # 可选
}
典型用法:
· 发现坏件 → status='不良品',从此不再被分配出货
· 争议/盘点待查 → status='冻结'
· 维修完成放回池子 → status='在库'
"""
data = request.get_json(silent=True) or {}
source_table = (data.get('source_table') or '').strip()
new_status = (data.get('status') or '').strip()
quality_status = data.get('quality_status')
# ---- 1. 参数校验(脏值一律挡在入口,不让它进库)----
if not source_table or not new_status:
return jsonify({'code': 400, 'msg': 'source_table 与 status 均为必填'}), 400
model = get_stock_model(source_table)
if model is None:
return jsonify({
'code': 400,
'msg': f'不支持的 source_table: {source_table}'
f'仅支持 stock_buy / stock_semi / stock_product',
}), 400
if new_status not in VALID_STOCK_STATUSES:
return jsonify({
'code': 400,
'msg': f'不支持的 status: {new_status}'
f'仅支持 {"".join(VALID_STOCK_STATUSES)}',
}), 400
try:
# ---- 2. 行锁 + 取行 ----
# ★ with_for_update 是必须的:本接口会与出库/借库的预占、报废的扣减
# 并发。不加锁的话「冻结」可能与「扣减」交错 —— 冻完之后该行仍被
# 扣走并发货,冻结形同虚设。
row = model.query.with_for_update().get(stock_id)
if not row:
return jsonify({
'code': 404,
'msg': f'库存记录不存在: {source_table}#{stock_id}',
}), 404
# ---- 3. 多租户隔离:非跨域用户只能动本公司的库存 ----
# 与分配器的口径一致(分配器按 MaterialBase.company_name 过滤候选行),
# 否则普通用户可越权冻结他司库存,等于一种拒绝服务。
company_limit = get_current_company_filter()
if company_limit is not None:
base = row.base
if (company_limit == '__NO_COMPANY__' or base is None
or (base.company_name or '') != company_limit):
return jsonify({'code': 403, 'msg': '无权操作其他公司的库存'}), 403
# ---- 4. 质量列:按表实际拥有的列写入,缺列时明确报错而非静默丢弃 ----
# ★ 实测 stock_buy 没有 quality_status 列(只有 inspection_status
# 静默忽略会让调用方以为写成功了。
if quality_status is not None:
if not hasattr(row, 'quality_status'):
return jsonify({
'code': 400,
'msg': f'{source_table} 没有 quality_status 字段;'
f'采购件的检验状态请通过入库单的 inspection_status 维护',
}), 400
row.quality_status = quality_status
old_status = row.status
old_quality = getattr(row, 'quality_status', None)
row.status = new_status
# 提交后由 audit_listener 自动记录本次变更(含操作人与前后值)
db.session.commit()
return jsonify({
'code': 200,
'msg': '状态变更成功',
'data': {
'source_table': source_table,
'stock_id': stock_id,
'sku': row.sku,
'old_status': old_status,
'status': row.status,
'old_quality_status': old_quality,
'quality_status': getattr(row, 'quality_status', None),
},
}), 200
except Exception as e:
db.session.rollback()
traceback.print_exc()
return jsonify({'code': 500, 'msg': f'状态变更失败: {str(e)}'}), 500
# ==============================================================================
# 原单退回 & 不良品回库(逆向物流二期)
# ==============================================================================
# 架构要点(详见 app/models/transaction.py 的模块注释与
# db_migrations/phase2_return_and_defective_goods.sql
#
# · 良品退回 → 加回原库存行stock_quantity 与 available_quantity 同步加回,
# 与出库时 restore_then_deduct 的扣减口径严格对称)
# · 不良品退回 → **完全不动库存表**,转入独立的 trans_defective_goods 在管
# 台账;修好后按 remaining_qty 部分/整批回库
#
# 为什么坏件不进库存表status 是**行级**属性,质量是**件级**属性。把坏件
# 加回原行只能整行打不良,而实测 stock_buy 单行最大 4789 件、中位 8 件 ——
# 退 1 件坏件会让整行良品一起被隔离,是静默的大规模库存损失。
def _lock_source_stock_row(source_table, stock_id):
"""
解析并锁定退回目标的**原库存行**。业务不满足即抛 ValueError。
三条 Fail-Closed 规则:
1. source_table 必须是三张库存表之一 —— 维修单等非库存来源没有可退回的行;
2. 库存行必须仍然存在 —— 入库模块会物理删除库存行(见
buy/semi/product_service 的 db.session.delete(stock)),实测 1077 条
出库记录中已有 7 条指向不存在的行;
3. 调用方拿到行后还需自行做公司隔离与状态校验(见 _assert_company_owns
★ 为什么必须加锁:本行随后会被加减数量,且与出库/报废/状态变更并发。
不加锁会出现「读-改-写」丢失更新lost update
"""
model = get_stock_model(source_table)
if model is None:
raise ValueError(
f'来源「{source_table or "(空)"}」不支持退回,'
f'仅支持 stock_buy / stock_semi / stock_product'
)
row = model.query.with_for_update().get(stock_id) if stock_id else None
if not row:
raise ValueError(
f'原库存行已不存在({source_table}#{stock_id}),无法自动退回,'
f'请改走入库流程手工登记这批实物'
)
return row
def _assert_company_owns(row):
"""
行级多租户隔离:非跨域用户只能操作本公司库存。不满足即抛 PermissionError。
口径与扫码出库OutboundService.get_stock_by_barcode、状态变更接口完全一致
—— 都走 MaterialBase.company_name避免三处隔离逻辑分叉。
"""
company_limit = get_current_company_filter()
if company_limit is None:
return
base = getattr(row, 'base', None)
if (company_limit == '__NO_COMPANY__' or base is None
or (base.company_name or '') != company_limit):
raise PermissionError('无权操作其他公司的库存')
@bp.route('/defective', methods=['GET'])
@permission_required('inventory_stocktake')
def list_defective_goods():
"""
不良品在管台账分页查询(供「不良品在管台账」看板页使用)。
Query:
page 页码,默认 1
page_size 每页条数,默认 20
status 状态精确过滤(待处理/处理中/已回库/已报废/已闭环);'全部' 或空 = 不过滤
keyword 模糊匹配 物料名称 / SKU / 规格型号
start_date / end_date 按退回时间created_at过滤
行级隔离:本表自带 company_name 快照,直接按它过滤。刻意不联表
MaterialBase —— 坏件的原库存行可能已被删除,联表会让记录整批消失。
"""
page = request.args.get('page', 1, type=int) or 1
page_size = request.args.get('page_size', 20, type=int) or 20
page_size = min(max(page_size, 1), 200) # 防超大分页拖垮库
status = (request.args.get('status') or '').strip()
keyword = (request.args.get('keyword') or '').strip()
start_date = (request.args.get('start_date') or '').strip()
end_date = (request.args.get('end_date') or '').strip()
try:
query = TransDefectiveGoods.query
# 行级隔离(超管/跨域 company_limit 为 None不受限
company_limit = get_current_company_filter()
if company_limit is not None:
query = query.filter(TransDefectiveGoods.company_name == company_limit)
if status and status not in ('全部', 'all'):
if status not in VALID_DEFECTIVE_STATUSES:
return jsonify({
'code': 400,
'msg': f'不支持的状态:{status}'
f'仅支持 {"".join(VALID_DEFECTIVE_STATUSES)}',
}), 400
query = query.filter(TransDefectiveGoods.status == status)
if keyword:
like = f'%{keyword}%'
query = query.filter(db.or_(
TransDefectiveGoods.material_name.ilike(like),
TransDefectiveGoods.sku.ilike(like),
TransDefectiveGoods.spec_model.ilike(like),
))
# 日期边界补全时分秒,避免 10 位日期被当成零点截断(与全系统口径一致)
if start_date and len(start_date) == 10:
start_date = f'{start_date} 00:00:00'
if end_date and len(end_date) == 10:
end_date = f'{end_date} 23:59:59'
if start_date:
query = query.filter(TransDefectiveGoods.created_at >= start_date)
if end_date:
query = query.filter(TransDefectiveGoods.created_at <= end_date)
# 默认按退回时间倒序:最新的坏件最需要处理
query = query.order_by(TransDefectiveGoods.created_at.desc(),
TransDefectiveGoods.id.desc())
pg = query.paginate(page=page, per_page=page_size, error_out=False)
# 汇总卡:在管总量(剩余待处理合计),供看板顶部展示
pending_total = db.session.query(
db.func.coalesce(db.func.sum(TransDefectiveGoods.remaining_qty), 0)
)
if company_limit is not None:
pending_total = pending_total.filter(
TransDefectiveGoods.company_name == company_limit)
return jsonify({
'code': 200,
'msg': 'success',
'data': {
'list': [g.to_dict() for g in pg.items],
'total': pg.total,
'page': page,
'page_size': page_size,
'pending_total': float(pending_total.scalar() or 0),
},
}), 200
except Exception as e:
traceback.print_exc()
return jsonify({'code': 500, 'msg': f'查询失败: {str(e)}'}), 500
def _defective_unit_cost(goods):
"""
取坏件单价,用于报废台账的 cost_at_scrap / total_lossbest-effort
取价口径与既有报废模块一致:成品取 sale_price采购件取 pre_tax_unit_price
半成品无价(返回 0
★ 原库存行可能已被删除(入库模块会物理删除库存行),故取不到时返回 0 ——
与借库转报废TransService.scrap_borrow 里 cost_at_scrap=0/total_loss=0
口径一致。刻意**不**因缺行而中断报废:实物已经销毁,台账必须先记上,
成本缺失是可接受的降级,记录丢失不是。
"""
model = get_stock_model(goods.source_table)
if model is None or not goods.stock_id:
return 0.0
row = model.query.get(goods.stock_id)
if not row:
return 0.0
if goods.source_table == 'stock_product':
return float(getattr(row, 'sale_price', 0) or 0)
if goods.source_table == 'stock_buy':
return float(getattr(row, 'pre_tax_unit_price', 0) or 0)
return 0.0
@bp.route('/return-from-outbound', methods=['POST'])
@permission_required('inventory_stocktake:operation')
# ★ 幂等锁置于 permission_required 内层(理由见 restock_defective_goods
@prevent_double_submit(lock_timeout=5)
def return_from_outbound():
"""
通用原单退回。
Body(JSON):
{
"outbound_id": 123, # 必填trans_outbound.id出库**明细行**,非单号)
"return_qty": 2, # 必填,本次退回数量
"is_defective": false, # 必填true=不良品退回false=良品退回
"reason": "错领退回" # 可选
}
两条分支的差异:
· 良品 → 加回原库存行的 stock_quantity 与 available_quantity
· 不良品 → 库存表分毫不动,转 trans_defective_goods 在管台账
"""
data = request.get_json(silent=True) or {}
operator_name = _normalize_user_id()
outbound_id = data.get('outbound_id')
is_defective = data.get('is_defective')
reason = (data.get('reason') or '').strip() or None
# ---- 1. 入参校验(脏值一律挡在入口)----
if not outbound_id:
return jsonify({'code': 400, 'msg': 'outbound_id 为必填'}), 400
if is_defective is None:
return jsonify({
'code': 400,
'msg': 'is_defective 为必填true=不良品退回false=良品退回)',
}), 400
is_defective = bool(is_defective)
try:
return_qty = float(data.get('return_qty') or 0)
except (TypeError, ValueError):
return jsonify({'code': 400, 'msg': 'return_qty 无效'}), 400
if return_qty <= 0:
return jsonify({'code': 400, 'msg': '退回数量必须大于 0'}), 400
try:
# ---- 2. 锁定原出库明细并校验退回额度 ----
# ★ 行锁不可省:并发两笔退回若各自读到相同的 returned_quantity会双双
# 通过额度校验,合计退回量超过出库量 —— 凭空多出库存。
outbound = TransOutbound.query.with_for_update().get(outbound_id)
if not outbound:
raise ValueError(f'出库记录不存在ID: {outbound_id}')
shipped = float(outbound.quantity or 0)
returned = float(outbound.returned_quantity or 0)
returnable = shipped - returned
if return_qty > returnable:
raise ValueError(
f'退回数量({return_qty})超出可退额度({returnable})'
f'原出库 {shipped},已退回 {returned}'
)
# ---- 3. 锁定原库存行 + 多租户隔离 ----
stock_row = _lock_source_stock_row(outbound.source_table, outbound.stock_id)
_assert_company_owns(stock_row)
goods = None
if is_defective:
# ================= 不良品分支 =================
# ★ 原库存表**分毫不动**:坏件全程存放于独立在管台账,既不占用库存
# 数量、也不改库存行 status从根上杜绝「坏件混进可分配池」。
base = getattr(stock_row, 'base', None)
goods = TransDefectiveGoods(
outbound_id=outbound.id,
source_table=outbound.source_table,
stock_id=outbound.stock_id,
base_id=getattr(stock_row, 'base_id', None),
sku=getattr(stock_row, 'sku', '') or '',
material_name=(base.name if base else '') or '',
spec_model=(base.spec_model if base else '') or '',
quantity=return_qty,
remaining_qty=return_qty,
status=DEFECTIVE_STATUS_PENDING,
company_name=(base.company_name if base else '') or '',
reason=reason,
operator=operator_name,
)
db.session.add(goods)
outcome = '不良品已转入在管台账'
else:
# ================= 良品分支 =================
# ★ 状态防呆:把良品加回一个已冻结/不良品的行,会让良品被该行的状态
# 连带隔离status 是行级属性)—— 静默造成良品不可用。宁可报错让
# 人先决定该行的归属。
current = (stock_row.status or '').strip()
if current != STOCK_STATUS_IN_STOCK:
raise ValueError(
f'原库存行当前状态为「{current or "未设置"}」,'
f'良品退回要求该行处于「{STOCK_STATUS_IN_STOCK}」状态'
)
stock_row.stock_quantity = float(stock_row.stock_quantity or 0) + return_qty
stock_row.available_quantity = float(stock_row.available_quantity or 0) + return_qty
outcome = '良品已加回原库存'
# ---- 4. 累加退回额度 + 写退回流水 ----
outbound.returned_quantity = returned + return_qty
ledger = TransReturn(
outbound_id=outbound.id,
stock_id=outbound.stock_id,
source_table=outbound.source_table,
sku=outbound.sku,
return_qty=return_qty,
return_type=RETURN_TYPE_DEFECTIVE if is_defective else RETURN_TYPE_GOOD,
reason=reason,
operator=operator_name,
)
db.session.add(ledger)
db.session.flush() # 先拿到 ledger.id供在管台账回填
# 在管台账回填来源流水 id形成「出库 → 退回流水 → 在管台账」的追溯闭环
if goods is not None:
goods.return_id = ledger.id
db.session.commit()
return jsonify({
'code': 200,
'msg': f'退回成功,{outcome}',
'data': {
'outbound_id': outbound.id,
'return_id': ledger.id,
'return_type': ledger.return_type,
'return_qty': return_qty,
'returned_quantity': float(outbound.returned_quantity),
'returnable_quantity': shipped - float(outbound.returned_quantity),
'defective_goods_id': goods.id if goods is not None else None,
},
}), 200
except PermissionError as e:
db.session.rollback()
return jsonify({'code': 403, 'msg': str(e)}), 403
except ValueError as e:
db.session.rollback()
return jsonify({'code': 400, 'msg': str(e)}), 400
except Exception as e:
db.session.rollback()
traceback.print_exc()
return jsonify({'code': 500, 'msg': f'退回失败: {str(e)}'}), 500
@bp.route('/defective/<int:goods_id>/restock', methods=['POST'])
@permission_required('inventory_stocktake:operation')
# ★ 幂等锁必须置于 permission_required **内层**prevent_double_submit 依赖
# get_jwt_identity(),若放在外层则 JWT 尚未验证 → 抛错 → 被其 except 捕获
# 后 fail-open 降级放行,锁形同虚设。
@prevent_double_submit(lock_timeout=5)
def restock_defective_goods(goods_id):
"""
不良品修好后一键回库。
Body(JSON):
{
"restock_qty": 2, # 可选,缺省 = 全部剩余在管量
"remark": "已更换主板" # 可选
}
★ 与 trans_repair维修模块完全解耦本接口操作的是 trans_defective_goods
在管台账。trans_repair 是 SN 单台粒度且无任何数量列,承载不了「一批坏件」。
"""
data = request.get_json(silent=True) or {}
operator_name = _normalize_user_id()
try:
# ---- 1. 锁定在管记录 ----
# ★ 行锁不可省:并发两次回库若各自读到相同的 remaining_qty会双双通过
# 校验,合计回库量超过在管量 —— 凭空多出库存。
goods = TransDefectiveGoods.query.with_for_update().get(goods_id)
if not goods:
raise ValueError(f'不良品在管记录不存在ID: {goods_id}')
# ---- 2. 状态守门Fail-Closed----
# 已回库 → 再回库就是凭空多一份库存;已报废 → 实物已销毁。
if goods.status not in RESTOCKABLE_DEFECTIVE_STATUSES:
raise ValueError(
f'当前状态为「{goods.status}」,不可回库'
f'(仅 {"".join(RESTOCKABLE_DEFECTIVE_STATUSES)} 可回库)'
)
remaining = float(goods.remaining_qty or 0)
if remaining <= 0:
raise ValueError('该记录在管数量为 0无可回库数量')
raw = data.get('restock_qty')
if raw is None or raw == '':
restock_qty = remaining # 缺省:整批剩余一次回库
else:
try:
restock_qty = float(raw)
except (TypeError, ValueError):
raise ValueError('restock_qty 无效')
if restock_qty <= 0:
raise ValueError('回库数量必须大于 0')
if restock_qty > remaining:
raise ValueError(f'回库数量({restock_qty})超过在管数量({remaining})')
# ---- 3. 回到原库存行 ----
stock_row = _lock_source_stock_row(goods.source_table, goods.stock_id)
_assert_company_owns(stock_row)
current = (stock_row.status or '').strip()
if current != STOCK_STATUS_IN_STOCK:
raise ValueError(
f'原库存行当前状态为「{current or "未设置"}」,'
f'请先将其恢复为「{STOCK_STATUS_IN_STOCK}」再回库'
)
stock_row.stock_quantity = float(stock_row.stock_quantity or 0) + restock_qty
stock_row.available_quantity = float(stock_row.available_quantity or 0) + restock_qty
# ---- 4. 递减在管量、累加回库量并推进状态机 ----
new_remaining = remaining - restock_qty
goods.remaining_qty = new_remaining
goods.restocked_qty = float(goods.restocked_qty or 0) + restock_qty
# ★ 终态由「累计去向」推导而非「最后一次动作」:本批可能既回库过、
# 又报废过,按最后一次动作定状态会产生误导(见 defective_close_status
goods.status = (
defective_close_status(goods.restocked_qty, goods.scrapped_qty)
if new_remaining <= 0 else DEFECTIVE_STATUS_IN_PROGRESS
)
# ★ 刻意**不覆盖** goods.operator该字段记录的是「谁退回来的」
# 覆盖会丢掉退回环节的责任人。本次回库人由 audit_listener 自动
# 写入 audit_logstrans_defective_goods 已在审计白名单内)。
if data.get('remark'):
goods.remark = str(data['remark']).strip()
db.session.commit()
return jsonify({
'code': 200,
'msg': '回库成功',
'data': {
'id': goods.id,
'restock_qty': restock_qty,
'remaining_qty': float(goods.remaining_qty),
'status': goods.status,
'source_table': goods.source_table,
'stock_id': goods.stock_id,
},
}), 200
except PermissionError as e:
db.session.rollback()
return jsonify({'code': 403, 'msg': str(e)}), 403
except ValueError as e:
db.session.rollback()
return jsonify({'code': 400, 'msg': str(e)}), 400
except Exception as e:
db.session.rollback()
traceback.print_exc()
return jsonify({'code': 500, 'msg': f'回库失败: {str(e)}'}), 500
@bp.route('/defective/<int:goods_id>/scrap', methods=['POST'])
@permission_required('inventory_stocktake:operation')
# ★ 幂等锁置于 permission_required 内层(理由见 restock_defective_goods
@prevent_double_submit(lock_timeout=5)
def scrap_defective_goods(goods_id):
"""
在管坏件报废(鉴定后确认无法维修,直接销毁)。
Body(JSON):
{
"scrap_qty": 2, # 可选,缺省 = 全部剩余在管量
"reason": "主板烧毁无法修复" # 必填
}
★ 与库存表的关系:坏件从未进入库存表(二期设计),因此本接口**不动任何
库存行**——它只做两件事:
1. 递减在管台账的 remaining_qty、累加 scrapped_qty、推进状态机
2. 往 trans_scrap 写一条报废台账,保证报废报表口径完整。
这与「库存行报废」(扣 stock_quantity / available_quantity是两条
互不重叠的路径,不会重复扣减。
"""
data = request.get_json(silent=True) or {}
operator_name = _normalize_user_id()
reason = (data.get('reason') or '').strip()
if not reason:
return jsonify({'code': 400, 'msg': '报废原因必填'}), 400
try:
# ---- 1. 锁定在管记录(并发下防超报废)----
goods = TransDefectiveGoods.query.with_for_update().get(goods_id)
if not goods:
raise ValueError(f'不良品在管记录不存在ID: {goods_id}')
# ---- 2. 状态守门Fail-Closed----
if goods.status not in SCRAPPABLE_DEFECTIVE_STATUSES:
raise ValueError(
f'当前状态为「{goods.status}」,不可报废'
f'(仅 {"".join(SCRAPPABLE_DEFECTIVE_STATUSES)} 可报废)'
)
remaining = float(goods.remaining_qty or 0)
if remaining <= 0:
raise ValueError('该记录在管数量为 0无可报废数量')
raw = data.get('scrap_qty')
if raw is None or raw == '':
scrap_qty = remaining # 缺省:整批剩余一次报废
else:
try:
scrap_qty = float(raw)
except (TypeError, ValueError):
raise ValueError('scrap_qty 无效')
if scrap_qty <= 0:
raise ValueError('报废数量必须大于 0')
if scrap_qty > remaining:
raise ValueError(f'报废数量({scrap_qty})超过在管数量({remaining})')
# ---- 3. 多租户隔离 ----
# 直接比对台账自身的 company_name 快照 —— 坏件的原库存行可能已被删除,
# 不能依赖联表取公司(那会让这类记录绕过隔离)。
company_limit = get_current_company_filter()
if company_limit is not None:
if (company_limit == '__NO_COMPANY__'
or (goods.company_name or '') != company_limit):
raise PermissionError('无权操作其他公司的不良品')
# ---- 4. 递减在管量、累加报废量、推进状态机 ----
new_remaining = remaining - scrap_qty
goods.remaining_qty = new_remaining
goods.scrapped_qty = float(goods.scrapped_qty or 0) + scrap_qty
goods.status = (
defective_close_status(goods.restocked_qty, goods.scrapped_qty)
if new_remaining <= 0 else DEFECTIVE_STATUS_IN_PROGRESS
)
# ---- 5. 写报废台账 ----
# ★ source_table 用 'trans_defective_goods'、stock_id 存台账自身主键,
# 与既有约定一致trans_borrow / trans_repair 作为来源时同样存各自主键)。
# 刻意**不**写原始库存表名 —— 那批坏件从未计入原库存行,若冒充库存
# 来源会让报废台账与库存表对不上账。
unit_price = _defective_unit_cost(goods)
db.session.add(TransScrap(
sku=goods.sku or '',
source_table='trans_defective_goods',
stock_id=goods.id,
quantity=scrap_qty,
reason=f"[不良品在管报废] {reason}",
operator_name=operator_name,
approval_status='approved', # 在管坏件鉴定后直接销毁,不走审批流
cost_at_scrap=unit_price,
total_loss=round(unit_price * scrap_qty, 2),
))
db.session.commit()
return jsonify({
'code': 200,
'msg': '报废成功',
'data': {
'id': goods.id,
'scrap_qty': scrap_qty,
'remaining_qty': float(goods.remaining_qty),
'restocked_qty': float(goods.restocked_qty),
'scrapped_qty': float(goods.scrapped_qty),
'status': goods.status,
},
}), 200
except PermissionError as e:
db.session.rollback()
return jsonify({'code': 403, 'msg': str(e)}), 403
except ValueError as e:
db.session.rollback()
return jsonify({'code': 400, 'msg': str(e)}), 400
except Exception as e:
db.session.rollback()
traceback.print_exc()
return jsonify({'code': 500, 'msg': f'报废失败: {str(e)}'}), 500