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

refactor: unify variance calculation and implement backend Excel export with borrowed assets sheet

Browse Source
This commit is contained in:
DXC
2026-03-18 11:10:54 +08:00
parent 1e38696e68
commit 54ea476206
2 changed files with 184 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

185
inventory-backend/app/api/v1/inbound/stock.py
View File

@ -1,14 +1,18 @@
from flask import Blueprint, jsonify, request
from flask import Blueprint, jsonify, request, send_file
from app.extensions import db
# ★★★ 修复点：必须引入 datetime，否则下方更新时间时会报错 500 ★★★
from datetime import datetime
from app.utils.decorators import permission_required
import uuid as uuid_module
import io
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
from app.models.transaction import TransBorrow
from app.models.base import MaterialBase
def _normalize_user_id(user_id):
@ -144,6 +148,10 @@ def add_draft():
"""
扫码同步 (支持更新数量)
如果 session_id 不存在则创建新的会话
差异计算逻辑调整：
- adjusted_stock_qty = 账面总库存 - 借出未还数量
- diff_qty = 实盘数量 - adjusted_stock_qty
"""
try:
data = request.json
@ -164,7 +172,21 @@ def add_draft():
if not item:
return jsonify({"message": "未找到对应的库存记录"}), 404
# 账面总库存
stock_qty = float(item.stock_quantity) if item.stock_quantity else 0
# 计算借出未还数量 (quantity - returned_quantity)
borrowed_result = db.session.query(
db.func.sum(TransBorrow.quantity - TransBorrow.returned_quantity)
).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
# 查找是否已存在
draft = StocktakeDraft.query.filter_by(user_id=user_id, uuid=uuid, session_id=session_id).first()
@ -173,8 +195,8 @@ def add_draft():
# 如果已存在，更新数量和时间
draft.quantity = quantity
draft.scan_time = datetime.now()
draft.stock_qty = stock_qty
draft.diff_qty = quantity - stock_qty
draft.stock_qty = adjusted_stock_qty
draft.diff_qty = quantity - adjusted_stock_qty
draft.source_table = source_table
draft.stock_id = stock_id
else:
@ -184,8 +206,8 @@ def add_draft():
uuid=uuid,
quantity=quantity,
session_id=session_id,
stock_qty=stock_qty,
diff_qty=quantity - stock_qty,
stock_qty=adjusted_stock_qty,
diff_qty=quantity - adjusted_stock_qty,
source_table=source_table,
stock_id=stock_id
)
@ -195,7 +217,9 @@ def add_draft():
return jsonify({
"message": "Saved",
"session_id": session_id,
"draft_id": draft.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}")
@ -460,3 +484,152 @@ def print_stocktake():
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. 外借在用资产明细 (未归还的借出记录)
"""
try:
# 创建工作簿
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': '-'}
if not stock:
return {'name': '-', 'sku': '-', 'spec': '-', 'unit': '-'}
# 尝试从MaterialBase获取名称
material = MaterialBase.query.filter_by(sku=stock.sku).first()
return {
'name': material.name if material else stock.sku,
'sku': stock.sku,
'spec': stock.spec_model or stock.standard or '-',
'unit': stock.unit or '个'
}
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("盘点差异明细")
diff_headers = ["物料名称", "SKU", "规格型号", "库位", "调整后账面数", "实盘数", "差异数", "盘点人", "盘点时间"]
set_header_row(ws1, diff_headers)
diff_drafts = StocktakeDraft.query.filter(StocktakeDraft.diff_qty != 0).all()
for row_idx, draft in enumerate(diff_drafts, 2):
mat_info = get_material_info(draft.source_table, draft.stock_id)
ws1.cell(row=row_idx, column=1, value=mat_info['name']).border = thin_border
ws1.cell(row=row_idx, column=2, value=mat_info['sku']).border = thin_border
ws1.cell(row=row_idx, column=3, value=mat_info['spec']).border = thin_border
ws1.cell(row=row_idx, column=4, value=draft.source_table).border = thin_border
ws1.cell(row=row_idx, column=5, value=float(draft.stock_qty or 0)).border = thin_border
ws1.cell(row=row_idx, column=6, value=float(draft.quantity or 0)).border = thin_border
ws1.cell(row=row_idx, column=7, value=float(draft.diff_qty or 0)).border = thin_border
ws1.cell(row=row_idx, column=8, value=draft.user_id or '').border = thin_border
ws1.cell(row=row_idx, column=9, value=str(draft.scan_time)[:19] if draft.scan_time else '').border = thin_border
# ===== Sheet 2: 账实相符明细 =====
ws2 = wb.create_sheet("账实相符明细")
normal_headers = ["物料名称", "SKU", "规格型号", "库位", "调整后账面数", "实盘数", "差异数", "盘点人", "盘点时间"]
set_header_row(ws2, normal_headers)
normal_drafts = StocktakeDraft.query.filter(StocktakeDraft.diff_qty == 0).all()
for row_idx, draft in enumerate(normal_drafts, 2):
mat_info = get_material_info(draft.source_table, draft.stock_id)
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=draft.source_table).border = thin_border
ws2.cell(row=row_idx, column=5, value=float(draft.stock_qty or 0)).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=float(draft.diff_qty or 0)).border = thin_border
ws2.cell(row=row_idx, column=8, value=draft.user_id or '').border = thin_border
ws2.cell(row=row_idx, column=9, value=str(draft.scan_time)[:19] if draft.scan_time else '').border = thin_border
# ===== Sheet 3: 外借在用资产明细 =====
ws3 = wb.create_sheet("外借在用资产明细")
borrow_headers = ["借出单号", "借用人", "物料名称", "SKU", "规格型号", "借出总数", "已还数量", "待还数量", "借出时间", "预计归还时间"]
set_header_row(ws3, 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
ws3.cell(row=row_idx, column=1, value=borrow.borrow_no or '').border = thin_border
ws3.cell(row=row_idx, column=2, value=borrow.borrower_name or '').border = thin_border
ws3.cell(row=row_idx, column=3, value=mat_info['name']).border = thin_border
ws3.cell(row=row_idx, column=4, value=mat_info['sku']).border = thin_border
ws3.cell(row=row_idx, column=5, value=mat_info['spec']).border = thin_border
ws3.cell(row=row_idx, column=6, value=total_qty).border = thin_border
ws3.cell(row=row_idx, column=7, value=returned_qty).border = thin_border
ws3.cell(row=row_idx, column=8, value=pending_qty).border = thin_border
ws3.cell(row=row_idx, column=9, value=str(borrow.borrow_time)[:19] if borrow.borrow_time else '').border = thin_border
ws3.cell(row=row_idx, column=10, value='无限期' if not borrow.expected_return_time else str(borrow.expected_return_time)[:10]).border = thin_border
# 调整列宽
for ws in [ws1, ws2, ws3]:
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