from __future__ import annotations from io import BytesIO from datetime import date, datetime from decimal import Decimal from fastapi import HTTPException from openpyxl import Workbook, load_workbook from sqlalchemy import func, select from sqlalchemy.orm import Session from app.models.master_data import Item, Warehouse from app.models.operations import StockLot, Stocktake, StocktakeAdjustment, StocktakeLine, StocktakeWarehouse, WarehouseLocation from app.services.operations import create_inventory_txn, to_decimal, upsert_stock_balance ACTIVE_STOCKTAKE_STATUSES = {"LOCKED", "EXPORTED", "IMPORTED"} STOCKTAKE_EXCEL_HEADERS = [ "盘库单号", "盘库校验码", "仓库ID", "快照行ID", "仓库", "库位", "物料编码", "物料名称", "批次号", "来源库存批次号", "系统数量", "系统重量(kg)", "系统单价", "盘点数量", "盘点重量(kg)", "盘点备注", ] STOCKTAKE_QUANTITY_HEADERS = {"系统数量", "盘点数量"} EMPTY_STOCKTAKE_MARKER_REMARK = "空库盘点校验行，请保留此行用于校验盘库单号" def build_stocktake_no(db: Session) -> str: date_part = date.today().strftime("%Y%m%d") prefix = f"PK-{date_part}" count = db.scalar(select(func.count(Stocktake.id)).where(Stocktake.stocktake_no.like(f"{prefix}-%"))) or 0 return f"{prefix}-{count + 1:03d}" def stocktake_warehouse_type_label(value: str | None) -> str: return { "RAW": "原材料库", "SEMI": "半成品库", "FINISHED": "成品库", "AUX": "辅料库", "SCRAP": "废料库", "RETURN": "退货库", }.get(str(value or "").upper(), str(value or "未知仓库")) def stocktake_validation_token(stocktake_no: str, warehouse_id: int, warehouse_type: str | None) -> str: return f"{stocktake_no}|{int(warehouse_id)}|{str(warehouse_type or '').upper()}" def stocktake_weight_only(warehouse_type: str | None, item_type: str | None) -> bool: _ = item_type return str(warehouse_type or "").upper() in {"RAW", "SCRAP"} def stocktake_excel_headers_for_type(warehouse_type: str | None) -> list[str]: if stocktake_weight_only(warehouse_type, None): return [header for header in STOCKTAKE_EXCEL_HEADERS if header not in STOCKTAKE_QUANTITY_HEADERS] return list(STOCKTAKE_EXCEL_HEADERS) def stocktake_excel_row_for_headers(payload: dict[str, object], headers: list[str]) -> list[object]: return [payload.get(header, "") for header in headers] def get_locked_warehouse_ids(db: Session, warehouse_ids: list[int] | set[int]) -> set[int]: ids = {int(item) for item in warehouse_ids if item} if not ids: return set() rows = db.scalars( select(StocktakeWarehouse.warehouse_id) .join(Stocktake, Stocktake.id == StocktakeWarehouse.stocktake_id) .where( StocktakeWarehouse.warehouse_id.in_(ids), StocktakeWarehouse.status == "LOCKED", Stocktake.status.in_(ACTIVE_STOCKTAKE_STATUSES), ) ).all() return {int(row) for row in rows} def ensure_warehouses_unlocked(db: Session, warehouse_ids: list[int] | set[int], action_label: str) -> None: locked_ids = get_locked_warehouse_ids(db, warehouse_ids) if not locked_ids: return rows = db.scalars(select(Warehouse).where(Warehouse.id.in_(locked_ids)).order_by(Warehouse.id.asc())).all() labels = "、".join(row.warehouse_name for row in rows) raise HTTPException(status_code=423, detail=f"{labels}正在盘库，暂不能执行{action_label}") def decimal_value(value: object, places: str = "0.000001") -> Decimal: return Decimal(str(value or 0)).quantize(Decimal(places)) def current_time() -> datetime: return datetime.now() def start_stocktake(db: Session, *, warehouse_ids: list[int], user_id: int | None, remark: str | None = None) -> Stocktake: ids = [int(item) for item in warehouse_ids if item] if not ids: raise HTTPException(status_code=400, detail="请选择需要盘库的仓库") if len(set(ids)) != 1: raise HTTPException(status_code=400, detail="一次只能盘一个仓库，请先在仓库栏选中单个仓库") ensure_warehouses_unlocked(db, ids, "开始盘库") warehouses = db.scalars(select(Warehouse).where(Warehouse.id.in_(ids)).order_by(Warehouse.id.asc())).all() if len(warehouses) != len(set(ids)): raise HTTPException(status_code=404, detail="存在无效仓库，无法开始盘库") now = current_time() stocktake = Stocktake( stocktake_no=build_stocktake_no(db), status="LOCKED", started_by_user_id=user_id, started_at=now, remark=remark, ) db.add(stocktake) db.flush() warehouses_by_id = {warehouse.id: warehouse for warehouse in warehouses} stocktake_warehouses_by_id: dict[int, StocktakeWarehouse] = {} for warehouse in warehouses: row = StocktakeWarehouse( stocktake_id=stocktake.id, warehouse_id=warehouse.id, warehouse_type=str(warehouse.warehouse_type or "").upper(), warehouse_name=warehouse.warehouse_name, status="LOCKED", ) db.add(row) db.flush() stocktake_warehouses_by_id[warehouse.id] = row lot_rows = db.execute( select(StockLot, Item) .join(Item, Item.id == StockLot.item_id) .where( StockLot.warehouse_id.in_(ids), StockLot.status == "AVAILABLE", StockLot.remaining_weight_kg > 0, ) .order_by(StockLot.warehouse_id.asc(), Item.item_code.asc(), StockLot.lot_no.asc(), StockLot.id.asc()) ).all() line_no = 1 for lot, item in lot_rows: warehouse = warehouses_by_id.get(lot.warehouse_id) if not warehouse: continue weight_only = stocktake_weight_only(warehouse.warehouse_type, item.item_type) location_name = None if lot.location_id: location_name = db.scalar(select(WarehouseLocation.location_name).where(WarehouseLocation.id == lot.location_id)) db.add( StocktakeLine( stocktake_id=stocktake.id, stocktake_warehouse_id=stocktake_warehouses_by_id[warehouse.id].id, line_no=line_no, warehouse_id=warehouse.id, warehouse_name=warehouse.warehouse_name, warehouse_type=str(warehouse.warehouse_type or "").upper(), location_id=lot.location_id, location_name=location_name, item_id=item.id, item_code=item.item_code, item_name=item.item_name, item_type=item.item_type, lot_id=lot.id, lot_no=lot.lot_no, source_material_sub_batch_no=lot.source_material_sub_batch_no or lot.material_sub_batch_no, snapshot_qty=Decimal("0") if weight_only else to_decimal(lot.remaining_qty), snapshot_weight_kg=to_decimal(lot.remaining_weight_kg), snapshot_unit_cost=to_decimal(lot.unit_cost, "0.0001"), counted_qty=None, counted_weight_kg=None, diff_qty=Decimal("0"), diff_weight_kg=Decimal("0"), diff_amount=Decimal("0"), diff_type="MATCH", row_status="SNAPSHOT", remark=None, ) ) line_no += 1 db.commit() db.refresh(stocktake) return stocktake def _stocktake_lines(db: Session, stocktake_id: int) -> list[StocktakeLine]: return db.scalars( select(StocktakeLine) .where(StocktakeLine.stocktake_id == stocktake_id) .order_by(StocktakeLine.warehouse_id.asc(), StocktakeLine.line_no.asc()) ).all() def _single_stocktake_warehouse(db: Session, stocktake_id: int) -> StocktakeWarehouse: rows = db.scalars( select(StocktakeWarehouse) .where(StocktakeWarehouse.stocktake_id == stocktake_id) .order_by(StocktakeWarehouse.id.asc()) ).all() if len(rows) != 1: raise HTTPException(status_code=400, detail="当前盘库单不是单仓库盘库，无法继续处理") return rows[0] def build_stocktake_workbook(db: Session, stocktake_id: int) -> Workbook: stocktake = db.get(Stocktake, stocktake_id) if not stocktake: raise HTTPException(status_code=404, detail="盘库单不存在") lines = _stocktake_lines(db, stocktake_id) stocktake_warehouses = db.scalars( select(StocktakeWarehouse).where(StocktakeWarehouse.stocktake_id == stocktake_id).order_by(StocktakeWarehouse.id.asc()) ).all() workbook = Workbook() default_sheet = workbook.active workbook.remove(default_sheet) grouped: dict[int, list[StocktakeLine]] = {} for line in lines: grouped.setdefault(int(line.warehouse_id), []).append(line) for warehouse in stocktake_warehouses: grouped.setdefault(int(warehouse.warehouse_id), []) warehouses_by_id = {int(warehouse.warehouse_id): warehouse for warehouse in stocktake_warehouses} for warehouse_id, sheet_lines in grouped.items(): warehouse = warehouses_by_id.get(int(warehouse_id)) if not warehouse: continue sheet_name = stocktake_warehouse_type_label(warehouse.warehouse_type) worksheet = workbook.create_sheet(title=sheet_name[:31]) worksheet.freeze_panes = "A2" headers = stocktake_excel_headers_for_type(warehouse.warehouse_type) worksheet.append(headers) if not sheet_lines: token = stocktake_validation_token(stocktake.stocktake_no, warehouse.warehouse_id, warehouse.warehouse_type) worksheet.append( stocktake_excel_row_for_headers( { "盘库单号": stocktake.stocktake_no, "盘库校验码": token, "仓库ID": warehouse.warehouse_id, "快照行ID": "", "仓库": sheet_name, "库位": "", "物料编码": "", "物料名称": "", "批次号": "", "来源库存批次号": "", "系统数量": 0, "系统重量(kg)": 0, "系统单价": 0, "盘点数量": 0, "盘点重量(kg)": 0, "盘点备注": EMPTY_STOCKTAKE_MARKER_REMARK, }, headers, ) ) for line in sheet_lines: worksheet.append( stocktake_excel_row_for_headers( { "盘库单号": stocktake.stocktake_no, "盘库校验码": stocktake_validation_token(stocktake.stocktake_no, line.warehouse_id, line.warehouse_type), "仓库ID": line.warehouse_id, "快照行ID": line.id, "仓库": line.warehouse_name, "库位": line.location_name or "", "物料编码": line.item_code, "物料名称": line.item_name, "批次号": line.lot_no or "", "来源库存批次号": line.source_material_sub_batch_no or "", "系统数量": float(line.snapshot_qty or 0), "系统重量(kg)": float(line.snapshot_weight_kg or 0), "系统单价": float(line.snapshot_unit_cost or 0), "盘点数量": float(line.snapshot_qty or 0), "盘点重量(kg)": float(line.snapshot_weight_kg or 0), "盘点备注": "", }, headers, ) ) widths_by_header = { "盘库单号": 18, "盘库校验码": 30, "仓库ID": 10, "快照行ID": 12, "仓库": 14, "库位": 14, "物料编码": 18, "物料名称": 28, "批次号": 22, "来源库存批次号": 24, "系统数量": 14, "系统重量(kg)": 16, "系统单价": 12, "盘点数量": 14, "盘点重量(kg)": 16, "盘点备注": 24, } for index, header in enumerate(headers, start=1): width = widths_by_header.get(header, 14) worksheet.column_dimensions[worksheet.cell(row=1, column=index).column_letter].width = width stocktake.exported_at = current_time() stocktake.status = "EXPORTED" db.add(stocktake) db.commit() return workbook def _clean_text(value: object) -> str: return str(value or "").strip() def _read_workbook_rows(content: bytes, warehouse_type: str | None = None) -> list[dict[str, object]]: try: workbook = load_workbook(BytesIO(content), read_only=True, data_only=True) except Exception as exc: raise ValueError("Excel 文件读取失败，请确认上传的是盘库清单 xlsx 文件") from exc rows: list[dict[str, object]] = [] for worksheet in workbook.worksheets: values = list(worksheet.iter_rows(values_only=True)) if not values: continue headers = [_clean_text(item) for item in values[0]] required_headers = stocktake_excel_headers_for_type(warehouse_type) missing = [header for header in required_headers if header not in headers] if missing: raise ValueError(f"{worksheet.title} 缺少必要列：{'、'.join(missing)}") for row in values[1:]: if not any(_clean_text(item) for item in row): continue payload = {headers[index]: value for index, value in enumerate(row) if index < len(headers)} payload["_sheet_name"] = worksheet.title rows.append(payload) return rows def _is_empty_stocktake_marker_row(row: dict[str, object]) -> bool: remark = _clean_text(row.get("盘点备注")) if "空库" not in remark: return False if _clean_text(row.get("快照行ID")): return False if any(_clean_text(row.get(label)) for label in ["物料编码", "物料名称", "批次号", "来源库存批次号"]): return False return all( decimal_value(row.get(label)) == 0 for label in ["系统数量", "系统重量(kg)", "系统单价", "盘点数量", "盘点重量(kg)"] ) def _validate_stocktake_row_marker(row: dict[str, object], stocktake: Stocktake, warehouse: StocktakeWarehouse) -> None: row_stocktake_no = _clean_text(row.get("盘库单号")) if row_stocktake_no != stocktake.stocktake_no: raise ValueError(f"存在不属于当前盘库单的行：{row_stocktake_no or '空盘库单号'}") row_warehouse_id = _clean_text(row.get("仓库ID")) if row_warehouse_id and int(float(row_warehouse_id)) != int(warehouse.warehouse_id): raise ValueError("导入清单不属于当前仓库，请确认仓库栏选中项和盘库清单一致") row_warehouse_name = _clean_text(row.get("仓库")) if row_warehouse_name and row_warehouse_name != warehouse.warehouse_name: raise ValueError("导入清单不属于当前仓库，请确认仓库栏选中项和盘库清单一致") expected_token = stocktake_validation_token(stocktake.stocktake_no, warehouse.warehouse_id, warehouse.warehouse_type) row_token = _clean_text(row.get("盘库校验码")) if row_token != expected_token: raise ValueError("盘库校验码不匹配，请导入本次盘库导出的仓库清单") def _reset_stocktake_preview_for_reimport(db: Session, stocktake_id: int, warehouse_id: int) -> dict[int, StocktakeLine]: lines = db.scalars( select(StocktakeLine) .where(StocktakeLine.stocktake_id == stocktake_id, StocktakeLine.warehouse_id == warehouse_id) .order_by(StocktakeLine.line_no.asc()) ).all() existing_lines: dict[int, StocktakeLine] = {} for line in lines: if line.lot_id is None and line.diff_type == "NEW_LOT": db.delete(line) continue line.counted_qty = None line.counted_weight_kg = None line.diff_qty = Decimal("0") line.diff_weight_kg = Decimal("0") line.diff_amount = Decimal("0") line.diff_type = "MATCH" line.row_status = "SNAPSHOT" line.error_message = None line.remark = None db.add(line) existing_lines[line.id] = line db.flush() return existing_lines def _line_diff_type(qty_diff: Decimal, weight_diff: Decimal, *, is_new: bool = False, is_missing: bool = False) -> str: if is_new: return "NEW_LOT" if is_missing: return "MISSING" basis = weight_diff if weight_diff != 0 else qty_diff if basis > 0: return "GAIN" if basis < 0: return "LOSS" return "MATCH" def _stocktake_summary(lines: list[StocktakeLine]) -> dict[str, int | float]: summary = { "match_count": 0, "gain_count": 0, "loss_count": 0, "new_lot_count": 0, "missing_count": 0, "error_count": 0, "diff_amount": 0.0, } for line in lines: key = { "MATCH": "match_count", "GAIN": "gain_count", "LOSS": "loss_count", "NEW_LOT": "new_lot_count", "MISSING": "missing_count", }.get(line.diff_type) if key: summary[key] += 1 if line.row_status == "ERROR": summary["error_count"] += 1 summary["diff_amount"] += float(line.diff_amount or 0) return summary def import_stocktake_workbook(db: Session, stocktake_id: int, content: bytes, user_id: int | None) -> dict[str, object]: stocktake = db.get(Stocktake, stocktake_id) if not stocktake: raise HTTPException(status_code=404, detail="盘库单不存在") if stocktake.status not in {"LOCKED", "EXPORTED", "IMPORTED"}: raise HTTPException(status_code=400, detail="当前盘库单状态不允许导入") stocktake_warehouse = _single_stocktake_warehouse(db, stocktake_id) rows = _read_workbook_rows(content, stocktake_warehouse.warehouse_type) existing_lines = _reset_stocktake_preview_for_reimport(db, stocktake_id, stocktake_warehouse.warehouse_id) seen_line_ids: set[int] = set() next_line_no = max((line.line_no for line in existing_lines.values()), default=0) + 1 for row in rows: _validate_stocktake_row_marker(row, stocktake, stocktake_warehouse) if _is_empty_stocktake_marker_row(row): continue line_id_text = _clean_text(row.get("快照行ID")) counted_qty = decimal_value(row.get("盘点数量")) counted_weight = decimal_value(row.get("盘点重量(kg)")) remark = _clean_text(row.get("盘点备注")) or None if line_id_text: line_id = int(float(line_id_text)) line = existing_lines.get(line_id) if not line: raise ValueError(f"快照行ID {line_id} 不存在于当前盘库单") line.error_message = None if line_id in seen_line_ids: line.error_message = "导入清单中该快照行重复" seen_line_ids.add(line_id) snapshot_qty = decimal_value(line.snapshot_qty) snapshot_weight = decimal_value(line.snapshot_weight_kg) weight_only = stocktake_weight_only(line.warehouse_type, line.item_type) qty_diff = Decimal("0") if weight_only else counted_qty - snapshot_qty weight_diff = counted_weight - snapshot_weight line.counted_qty = Decimal("0") if weight_only else counted_qty line.counted_weight_kg = counted_weight line.diff_qty = qty_diff line.diff_weight_kg = weight_diff line.diff_amount = (weight_diff * decimal_value(line.snapshot_unit_cost, "0.0001")).quantize(Decimal("0.01")) line.diff_type = _line_diff_type(qty_diff, weight_diff) line.row_status = "ERROR" if line.error_message else "READY" line.remark = remark db.add(line) continue item_code = _clean_text(row.get("物料编码")) item = db.scalar(select(Item).where(Item.item_code == item_code).limit(1)) warehouse = db.get(Warehouse, stocktake_warehouse.warehouse_id) if not item or not warehouse: raise ValueError(f"新增批次行物料或仓库不存在：{stocktake_warehouse.warehouse_name} / {item_code}") unit_cost = decimal_value(row.get("系统单价"), "0.0001") weight_only = stocktake_weight_only(warehouse.warehouse_type, item.item_type) db.add( StocktakeLine( stocktake_id=stocktake.id, stocktake_warehouse_id=stocktake_warehouse.id, line_no=next_line_no, warehouse_id=warehouse.id, warehouse_name=warehouse.warehouse_name, warehouse_type=str(warehouse.warehouse_type or "").upper(), location_id=None, location_name=_clean_text(row.get("库位")) or None, item_id=item.id, item_code=item.item_code, item_name=item.item_name, item_type=item.item_type, lot_id=None, lot_no=_clean_text(row.get("批次号")) or None, source_material_sub_batch_no=_clean_text(row.get("来源库存批次号")) or None, snapshot_qty=Decimal("0"), snapshot_weight_kg=Decimal("0"), snapshot_unit_cost=unit_cost, counted_qty=Decimal("0") if weight_only else counted_qty, counted_weight_kg=counted_weight, diff_qty=Decimal("0") if weight_only else counted_qty, diff_weight_kg=counted_weight, diff_amount=(counted_weight * unit_cost).quantize(Decimal("0.01")), diff_type="NEW_LOT", row_status="READY", remark=remark, ) ) next_line_no += 1 for line_id, line in existing_lines.items(): if line_id in seen_line_ids: continue line.counted_qty = Decimal("0") line.counted_weight_kg = Decimal("0") line.diff_qty = -decimal_value(line.snapshot_qty) line.diff_weight_kg = -decimal_value(line.snapshot_weight_kg) line.diff_amount = (decimal_value(line.diff_weight_kg) * decimal_value(line.snapshot_unit_cost, "0.0001")).quantize(Decimal("0.01")) line.diff_type = "MISSING" line.row_status = "READY" line.error_message = None line.remark = "导入清单缺失，按盘点为0处理" db.add(line) stocktake.status = "IMPORTED" stocktake.imported_by_user_id = user_id stocktake.imported_at = current_time() db.add(stocktake) db.commit() lines = db.scalars( select(StocktakeLine) .where(StocktakeLine.stocktake_id == stocktake_id, StocktakeLine.warehouse_id == stocktake_warehouse.warehouse_id) .order_by(StocktakeLine.line_no.asc()) ).all() return {"stocktake": stocktake, "lines": lines, "summary": _stocktake_summary(lines)} def _stocktake_txn_type(diff_type: str) -> str: return { "GAIN": "STOCKTAKE_GAIN", "LOSS": "STOCKTAKE_LOSS", "MISSING": "STOCKTAKE_LOSS", "NEW_LOT": "STOCKTAKE_NEW_LOT", }.get(diff_type, "STOCKTAKE_GAIN") def confirm_stocktake( db: Session, stocktake_id: int, *, user_id: int | None, confirm_text: str | None = None, remark: str | None = None, ) -> Stocktake: _ = confirm_text stocktake = db.get(Stocktake, stocktake_id) if not stocktake: raise HTTPException(status_code=404, detail="盘库单不存在") lines = db.scalars(select(StocktakeLine).where(StocktakeLine.stocktake_id == stocktake_id).order_by(StocktakeLine.line_no.asc())).all() now = current_time() if stocktake.status != "IMPORTED": if stocktake.status in {"LOCKED", "EXPORTED"} and not lines: stocktake.status = "IMPORTED" stocktake.imported_by_user_id = user_id stocktake.imported_at = now db.add(stocktake) db.flush() else: raise HTTPException(status_code=400, detail="请先导入盘点清单并确认差异后再过账") error_lines = [line for line in lines if line.row_status == "ERROR"] if error_lines: raise HTTPException(status_code=400, detail=f"存在 {len(error_lines)} 行异常，不能确认盘库") for line in lines: if line.diff_type == "MATCH": continue weight_only = stocktake_weight_only(line.warehouse_type, line.item_type) unit_cost = decimal_value(line.snapshot_unit_cost, "0.0001") qty_change = decimal_value(line.diff_qty) weight_change = decimal_value(line.diff_weight_kg) if line.diff_type == "NEW_LOT": counted_qty = Decimal("0") if weight_only else max(Decimal("0"), decimal_value(line.counted_qty)) counted_weight = max(Decimal("0"), decimal_value(line.counted_weight_kg)) lot = StockLot( lot_no=line.lot_no or f"{stocktake.stocktake_no}-NEW-{line.line_no:04d}", lot_role="STOCKTAKE_NEW", item_id=line.item_id, warehouse_id=line.warehouse_id, location_id=line.location_id, source_doc_type="STOCKTAKE", source_doc_id=stocktake.id, source_line_id=line.id, source_material_sub_batch_no=line.source_material_sub_batch_no, source_material_summary="盘库新增批次", inbound_qty=counted_qty, inbound_weight_kg=counted_weight, remaining_qty=counted_qty, remaining_weight_kg=counted_weight, locked_qty=Decimal("0"), locked_weight_kg=Decimal("0"), unit_cost=unit_cost, production_date=now.date(), quality_status="PASS", status="AVAILABLE", remark=line.remark or "盘库新增", ) db.add(lot) db.flush() lot_id = lot.id before_qty = Decimal("0") before_weight = Decimal("0") after_qty = counted_qty after_weight = counted_weight qty_change = after_qty weight_change = after_weight else: lot = db.get(StockLot, line.lot_id) if not lot: raise HTTPException(status_code=400, detail=f"盘库行 {line.line_no} 对应批次不存在") before_qty = Decimal("0") if weight_only else decimal_value(lot.remaining_qty) before_weight = decimal_value(lot.remaining_weight_kg) after_qty = Decimal("0") if weight_only else max(Decimal("0"), decimal_value(line.counted_qty)) after_weight = max(Decimal("0"), decimal_value(line.counted_weight_kg)) qty_change = after_qty - before_qty weight_change = after_weight - before_weight lot.remaining_qty = after_qty lot.remaining_weight_kg = after_weight lot.status = "AVAILABLE" if after_qty > 0 or after_weight > 0 else "DEPLETED" db.add(lot) lot_id = lot.id upsert_stock_balance( db, item_id=line.item_id, warehouse_id=line.warehouse_id, location_id=line.location_id, qty_delta=qty_change, weight_delta=weight_change, available_qty_delta=qty_change, available_weight_delta=weight_change, unit_cost=unit_cost, ) txn = create_inventory_txn( db, txn_type=_stocktake_txn_type(line.diff_type), item_id=line.item_id, warehouse_id=line.warehouse_id, location_id=line.location_id, lot_id=lot_id, qty_change=qty_change, weight_change=weight_change, unit_cost=unit_cost, source_doc_type="STOCKTAKE", source_doc_id=stocktake.id, source_line_id=line.id, biz_time=now, operator_user_id=user_id, remark=line.remark or remark or "盘库调整", amount_basis="WEIGHT", ) db.flush() db.add( StocktakeAdjustment( stocktake_id=stocktake.id, stocktake_line_id=line.id, warehouse_id=line.warehouse_id, item_id=line.item_id, lot_id=line.lot_id, adjustment_lot_id=lot_id, inventory_txn_id=txn.id, adjustment_type=line.diff_type, before_qty=before_qty, after_qty=after_qty, before_weight_kg=before_weight, after_weight_kg=after_weight, qty_change=qty_change, weight_change_kg=weight_change, unit_cost=unit_cost, amount=(abs(weight_change) * unit_cost).quantize(Decimal("0.01")), ) ) for warehouse in db.scalars(select(StocktakeWarehouse).where(StocktakeWarehouse.stocktake_id == stocktake_id)).all(): warehouse.status = "UNLOCKED" db.add(warehouse) stocktake.status = "CONFIRMED" stocktake.confirmed_by_user_id = user_id stocktake.confirmed_at = now if remark: stocktake.remark = remark db.add(stocktake) db.commit() db.refresh(stocktake) return stocktake def cancel_stocktake(db: Session, stocktake_id: int, *, user_id: int | None, remark: str | None = None) -> Stocktake: stocktake = db.get(Stocktake, stocktake_id) if not stocktake: raise HTTPException(status_code=404, detail="盘库单不存在") if stocktake.status not in ACTIVE_STOCKTAKE_STATUSES: raise HTTPException(status_code=400, detail="当前盘库单不能取消") now = current_time() for warehouse in db.scalars(select(StocktakeWarehouse).where(StocktakeWarehouse.stocktake_id == stocktake_id)).all(): warehouse.status = "UNLOCKED" db.add(warehouse) stocktake.status = "CANCELED" stocktake.canceled_by_user_id = user_id stocktake.canceled_at = now stocktake.remark = remark or stocktake.remark db.add(stocktake) db.commit() db.refresh(stocktake) return stocktake def stocktake_lines_for_read(db: Session, stocktake_id: int) -> list[StocktakeLine]: return db.scalars( select(StocktakeLine) .where(StocktakeLine.stocktake_id == stocktake_id) .order_by(StocktakeLine.warehouse_id.asc(), StocktakeLine.line_no.asc()) ).all() def stocktake_warehouses_for_read(db: Session, stocktake_id: int) -> list[StocktakeWarehouse]: return db.scalars( select(StocktakeWarehouse) .where(StocktakeWarehouse.stocktake_id == stocktake_id) .order_by(StocktakeWarehouse.id.asc()) ).all()