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

库存资产excel文件导出

Browse Source
This commit is contained in:
dxc
2026-02-25 09:55:25 +08:00
parent 447b1890ab
commit 63a3cf269d
5 changed files with 356 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -4,11 +4,16 @@ from app.extensions import db
from app.models.base import MaterialBase
from app.models.inbound.buy import StockBuy
from app.models.inbound.semi import StockSemi
# from app.models.inbound.product import StockProduct
from app.models.inbound.product import StockProduct
# from app.models.inbound.service import StockService
from sqlalchemy import or_
from sqlalchemy import or_, and_
import traceback
import json
import io
import datetime
# 需要 pip install openpyxl
from openpyxl import Workbook
from openpyxl.styles import Font, Alignment, Border, Side, PatternFill
class MaterialBaseService:
@ -91,7 +96,7 @@ class MaterialBaseService:
for x in items:
# 1. 获取库存数 (兼容不同字段名)
q = getattr(x, 'stock_quantity', getattr(x, 'actual_quantity', getattr(x, 'quantity', 0)))
q = getattr(x, 'stock_quantity', getattr(x, 'in_quantity', 0)) # 优先取库存，其次入库
# 2. 获取可用数
a = getattr(x, 'available_quantity', q)
@ -137,7 +142,6 @@ class MaterialBaseService:
query = query.filter_by(is_enabled=is_active)
# [修改3] 默认排序方式改为按 spec_model 排序
# 如果需要更复杂的“/前内容”排序，通常直接按字符串排序也能满足前缀分组的需求
pagination = query.order_by(MaterialBase.spec_model.asc()).paginate(page=page, per_page=limit,
error_out=False)
@ -280,14 +284,16 @@ class MaterialBaseService:
buy_usage_count = StockBuy.query.filter_by(base_id=m_id).count()
semi_usage_count = StockSemi.query.filter_by(base_id=m_id).count()
prod_usage_count = StockProduct.query.filter_by(base_id=m_id).count()
total_usage = buy_usage_count + semi_usage_count
total_usage = buy_usage_count + semi_usage_count + prod_usage_count
if total_usage > 0:
raise ValueError(
f"无法删除：该基础物料正被使用中。\n"
f"- 采购库存记录: {buy_usage_count} 条\n"
f"- 半成品库存记录: {semi_usage_count} 条\n"
f"- 成品库存记录: {prod_usage_count} 条\n"
f"请先清理相关库存或仅‘禁用’此条目。"
)
@ -298,4 +304,235 @@ class MaterialBaseService:
except Exception as e:
db.session.rollback()
print(f"删除基础信息失败: {e}")
raise e
# ==============================================================================
# [核心修改] 统一资产统计导出
# ==============================================================================
@staticmethod
def export_excel(filters=None):
"""
全口径资产统计报表：
逻辑：先查库存表 (Buy, Semi, Product)，关联基础信息，只导出实际存在的库存记录。
"""
try:
# 1. 构造基础信息的筛选条件 (用于过滤库存)
filter_conditions = []
if filters:
if filters.get('keyword'):
kw = f"%{filters['keyword']}%"
filter_conditions.append(or_(
MaterialBase.name.ilike(kw),
MaterialBase.common_name.ilike(kw),
MaterialBase.spec_model.ilike(kw),
MaterialBase.company_name.ilike(kw)
))
if filters.get('company'):
filter_conditions.append(MaterialBase.company_name == filters['company'])
if filters.get('category'):
filter_conditions.append(MaterialBase.category == filters['category'])
if filters.get('type'):
filter_conditions.append(MaterialBase.material_type == filters['type'])
if filters.get('isEnabled') is not None:
is_active = bool(int(filters['isEnabled']))
filter_conditions.append(MaterialBase.is_enabled == is_active)
# 2. 分别查询三个库存表，并 Join MaterialBase 进行筛选
# 2.1 采购库存 (StockBuy)
query_buy = db.session.query(StockBuy, MaterialBase).join(
MaterialBase, StockBuy.base_id == MaterialBase.id
)
for cond in filter_conditions:
query_buy = query_buy.filter(cond)
list_buy = query_buy.all()
# 2.2 半成品库存 (StockSemi)
query_semi = db.session.query(StockSemi, MaterialBase).join(
MaterialBase, StockSemi.base_id == MaterialBase.id
)
for cond in filter_conditions:
query_semi = query_semi.filter(cond)
list_semi = query_semi.all()
# 2.3 成品库存 (StockProduct)
query_product = db.session.query(StockProduct, MaterialBase).join(
MaterialBase, StockProduct.base_id == MaterialBase.id
)
for cond in filter_conditions:
query_product = query_product.filter(cond)
list_product = query_product.all()
# 3. 数据整合
all_rows = []
# 处理采购件
for stock, base in list_buy:
# 价格计算
unit_price = float(stock.unit_price or 0)
tax_rate = float(stock.tax_rate or 0)
price_incl = unit_price * (1 + tax_rate / 100.0)
qty = float(stock.stock_quantity or 0)
# 计算不含税总价 = 数量 * 不含税单价
total_val_excl = qty * unit_price
# 计算含税总价 = 数量 * 含税单价
total_val_incl = qty * price_incl
ident = stock.batch_number or stock.serial_number or stock.barcode or stock.sku
all_rows.append({
"base": base,
"type_name": "采购件",
"ident": ident,
"loc": stock.warehouse_location,
"source": stock.supplier_name,
"date": stock.in_date,
"qty": qty,
"avail": float(stock.available_quantity or 0),
"price_excl": unit_price,
"total_val_excl": total_val_excl, # [新增]
"tax": tax_rate,
"price_incl": price_incl,
"total_val": total_val_incl
})
# 处理半成品
for stock, base in list_semi:
cost = float(stock.raw_material_cost or 0) + float(stock.manual_cost or 0)
qty = float(stock.stock_quantity or 0)
# 半成品不含税总价 = 数量 * 成本
total_val_excl = qty * cost
# 含税总价同上 (税率0)
total_val_incl = qty * cost
ident = stock.batch_number or stock.serial_number or stock.barcode or stock.sku
all_rows.append({
"base": base,
"type_name": "半成品",
"ident": ident,
"loc": stock.warehouse_location,
"source": stock.production_manager,
"date": stock.production_date,
"qty": qty,
"avail": float(stock.available_quantity or 0),
"price_excl": cost,
"total_val_excl": total_val_excl, # [新增]
"tax": 0.0,
"price_incl": cost,
"total_val": total_val_incl
})
# 处理成品
for stock, base in list_product:
cost = float(stock.raw_material_cost or 0) + float(stock.manual_cost or 0)
qty = float(stock.stock_quantity or 0)
total_val_excl = qty * cost
total_val_incl = qty * cost
ident = stock.serial_number or stock.barcode or stock.sku
all_rows.append({
"base": base,
"type_name": "成品",
"ident": ident,
"loc": stock.warehouse_location,
"source": stock.production_manager,
"date": stock.production_date,
"qty": qty,
"avail": float(stock.available_quantity or 0),
"price_excl": cost,
"total_val_excl": total_val_excl, # [新增]
"tax": 0.0,
"price_incl": cost,
"total_val": total_val_incl
})
# 4. 排序：按公司 -> 规格型号 -> 基础ID -> 批号 排序
all_rows.sort(key=lambda x: (
x['base'].company_name or "",
x['base'].spec_model or "",
x['base'].id,
x['ident'] or ""
))
# 5. 生成 Excel
wb = Workbook()
ws = wb.active
ws.title = "库存统计"
# 表头 [修改] 增加 "资产总额 (不含税)"
headers = [
"所属公司", "资产名称", "规格型号", "物料类型",
"类别一级", "类别二级", "类别三级", "类别四级", "类别五级",
"计量单位",
"库存性质", "唯一标识码 (批号/SN)", "仓库位置",
"资产来源", "入库/生产日期",
"库存数量", "可用数量",
"单价/成本 (不含税)", "资产总额 (不含税)", "税率 (%)", "单价/成本 (含税)", "资产总额 (含税)"
]
ws.append(headers)
# 样式
header_fill = PatternFill(start_color="D7E4BC", end_color="D7E4BC", fill_type="solid")
border_style = Border(left=Side(style='thin'), right=Side(style='thin'), top=Side(style='thin'),
bottom=Side(style='thin'))
for cell in ws[1]:
cell.font = Font(bold=True, name='微软雅黑')
cell.alignment = Alignment(horizontal='center', vertical='center')
cell.fill = header_fill
cell.border = border_style
# 写入数据
for r in all_rows:
base = r['base']
# 类别拆分
cat_parts = (base.category or "").split('/')
while len(cat_parts) < 5:
cat_parts.append("")
# 日期格式化
date_str = r['date'].strftime('%Y-%m-%d') if isinstance(r['date'], datetime.date) else ""
row_val = [
base.company_name,
base.name,
base.spec_model,
base.material_type,
cat_parts[0], cat_parts[1], cat_parts[2], cat_parts[3], cat_parts[4],
base.unit,
r['type_name'],
r['ident'],
r['loc'],
r['source'],
date_str,
r['qty'],
r['avail'],
r['price_excl'],
r['total_val_excl'], # [新增] 对应列
r['tax'],
r['price_incl'],
r['total_val']
]
ws.append(row_val)
# 列宽调整
dims = {}
for row in ws.rows:
for cell in row:
if cell.value:
dims[cell.column_letter] = max((dims.get(cell.column_letter, 0), len(str(cell.value))))
for col, value in dims.items():
ws.column_dimensions[col].width = min(value + 2, 30)
output = io.BytesIO()
wb.save(output)
output.seek(0)
return output
except Exception as e:
traceback.print_exc()
raise e