"""仓库库存路由：材料 CRUD + 出入库 + Excel 导入导出。""" from fastapi import APIRouter, UploadFile from pydantic import BaseModel, Field from sqlalchemy import desc, select from huamenchuang.core.deps import CurrentUser, DbSession from huamenchuang.core.response import fail, ok, paginated from huamenchuang.models.contract import StockMove, WarehouseItem router = APIRouter(prefix="/warehouse", tags=["仓库"]) class ItemBody(BaseModel): name: str = Field(min_length=1, max_length=128) bom_code: str = "" material: str = Field(default="profile", pattern="^(profile|glass|hardware|addon)$") color: str = "" spec: str = "" unit: str = "根" cost: float = 0 price: float = 0 count: float = 0 warning: float = 0 length: int = 0 note: str = "" class MoveBody(BaseModel): item_id: int direction: int = Field(default=1, ge=-1, le=1) # 1入库 -1出库 count: float = Field(gt=0) note: str = "" def _out(i: WarehouseItem) -> dict: return { "id": i.id, "bom_code": i.bom_code, "name": i.name, "material": i.material, "color": i.color, "spec": i.spec, "unit": i.unit, "cost": float(i.cost), "price": float(i.price), "count": float(i.count), "warning": float(i.warning), "length": i.length, "note": i.note, "updated_at": str(i.updated_at), } @router.get("/list") async def list_items(user: CurrentUser, db: DbSession, page: int = 1, page_size: int = 15, keyword: str = "", material: str | None = None): """材料库存列表（分页 + 筛选，对齐目标站 warehouse_list 语义）。""" from sqlalchemy import func, or_ conds = [WarehouseItem.company_id == user.company_id] if material: conds.append(WarehouseItem.material == material) if kw := keyword.strip(): like = f"%{kw}%" conds.append(or_(WarehouseItem.name.like(like), WarehouseItem.bom_code.like(like))) where = *conds, total = (await db.execute(select(func.count()).select_from(WarehouseItem).where(*where))).scalar_one() stmt = (select(WarehouseItem).where(*where).order_by(WarehouseItem.id.desc()) .offset((page - 1) * page_size).limit(page_size)) items = list((await db.execute(stmt)).scalars()) return ok(data=paginated([_out(i) for i in items], page, page_size, total)) @router.post("") async def create_item(body: ItemBody, user: CurrentUser, db: DbSession): """新增材料。""" item = WarehouseItem(company_id=user.company_id, **body.model_dump()) db.add(item) await db.flush() return ok(data={"id": item.id}, message="创建成功") # ---------------- Excel 导入导出 ---------------- @router.get("/excel-out") async def excel_out(user: CurrentUser, db: DbSession): """库存全量导出（对齐目标站 warehouse_excel_out）。""" from fastapi.responses import Response from huamenchuang.services.excel_service import simple_table_workbook items = list((await db.execute( select(WarehouseItem).where(WarehouseItem.company_id == user.company_id) .order_by(WarehouseItem.id))).scalars()) mat_label = {"profile": "型材", "glass": "玻璃", "hardware": "五金", "addon": "配件"} data = [[i.bom_code, i.name, mat_label.get(i.material, i.material), i.spec, i.color, float(i.price), float(i.cost), float(i.count), i.unit, i.length, float(i.warning)] for i in items] content = simple_table_workbook("库存列表", ["编码", "名称", "类别", "规格", "颜色", "售价", "成本", "库存数量", "单位", "单根长度mm", "预警线"], data, widths=[12, 20, 8, 14, 10, 10, 10, 10, 8, 12, 10]) return Response(content, media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", headers={"Content-Disposition": "attachment; filename=warehouse.xlsx"}) @router.post("/excel-in") async def excel_in(file: UploadFile, user: CurrentUser, db: DbSession): """材料批量导入（对齐目标站 materials_config_excel：同名材料按名称匹配更新数量，否则新增）。""" if not file.filename or not file.filename.lower().endswith((".xlsx", ".xls")): return fail("请上传 xlsx 文件") from huamenchuang.services.excel_service import pick, read_rows try: rows = read_rows(await file.read()) except Exception: return fail("文件解析失败") if not rows: return fail("文件中没有数据行") mat_map = {"型材": "profile", "玻璃": "glass", "五金": "hardware", "配件": "addon", "profile": "profile", "glass": "glass", "hardware": "hardware", "addon": "addon"} existing = {i.name: i for i in (await db.execute( select(WarehouseItem).where(WarehouseItem.company_id == user.company_id))).scalars()} created, updated, skipped = 0, 0, 0 for i, row in enumerate(rows, 2): name = pick(row, "名称", "材料名称", "name") if not name: skipped += 1 continue name = str(name).strip() mat = mat_map.get(str(pick(row, "类别", "材料", "material", default="型材") or "型材"), "profile") values = dict( bom_code=str(pick(row, "编码", "bom_code", default="") or ""), material=mat, spec=str(pick(row, "规格", "spec", default="") or ""), color=str(pick(row, "颜色", "color", default="") or ""), price=float(pick(row, "售价", "price", default=0) or 0), cost=float(pick(row, "成本", "cost", default=0) or 0), count=float(pick(row, "库存数量", "数量", "count", default=0) or 0), unit=str(pick(row, "单位", "unit", default="根") or "根"), length=int(float(pick(row, "单根长度mm", "长度", "length", default=0) or 0)), warning=float(pick(row, "预警线", "warning", default=0) or 0)) if name in existing: # 同名更新（数量累加） item = existing[name] for k, v in values.items(): if k == "count": continue setattr(item, k, v) item.count = float(item.count) + values["count"] updated += 1 else: item = WarehouseItem(company_id=user.company_id, name=name, **values) db.add(item) existing[name] = item created += 1 await db.flush() return ok(data={"created": created, "updated": updated, "skipped": skipped}, message=f"导入完成：新增 {created} 条，更新 {updated} 条，跳过 {skipped} 条") @router.put("/{item_id}") async def update_item(item_id: int, body: ItemBody, user: CurrentUser, db: DbSession): """编辑材料。""" item = await db.get(WarehouseItem, item_id) if item is None or item.company_id != user.company_id: return fail("材料不存在") for k, v in body.model_dump().items(): setattr(item, k, v) return ok(message="更新成功") @router.post("/move") async def stock_move(body: MoveBody, user: CurrentUser, db: DbSession): """出入库：direction=1 入库 / -1 出库（出库不允许负库存）。""" item = await db.get(WarehouseItem, body.item_id) if item is None or item.company_id != user.company_id: return fail("材料不存在") delta = body.count if body.direction == 1 else -body.count if float(item.count) + delta < 0: return fail(f"库存不足: 当前 {float(item.count)}{item.unit}") item.count = float(item.count) + delta db.add(StockMove(item_id=item.id, direction=body.direction, count=body.count, note=body.note, user_id=user.id)) return ok(data={"count": float(item.count)}, message="操作成功") @router.get("/moves") async def move_logs(user: CurrentUser, db: DbSession, page: int = 1, page_size: int = 20): """出入库流水。""" total_q = select(func.count()).select_from(StockMove) total = (await db.execute(total_q)).scalar_one() stmt = (select(StockMove).order_by(desc(StockMove.id)) .offset((page - 1) * page_size).limit(page_size)) moves = list((await db.execute(stmt)).scalars()) return ok(data=paginated([{ "id": m.id, "item_id": m.item_id, "direction": m.direction, "count": float(m.count), "note": m.note, "created_at": str(m.created_at), } for m in moves], page, page_size, total))