"""Lot, serial, expiry, quality and traceability rules.""" from datetime import date, timedelta from decimal import Decimal from fastapi import HTTPException from sqlalchemy import text def calculate_expiry(produced_on: date, shelf_life_days: int) -> date: return produced_on + timedelta(days=int(shelf_life_days)) def validate_move_tracking(tracking_type: str, qty: Decimal, *, serial_numbers: list[str] | None = None, batch_no: str | None = None) -> None: serials = [value.strip() for value in (serial_numbers or []) if value and value.strip()] quantity = Decimal(str(qty)) if tracking_type == "serial": if quantity != quantity.to_integral_value() or quantity <= 0: raise HTTPException(status_code=400, detail="序列号商品数量必须是正整数") if len(serials) != int(quantity) or len(set(serials)) != len(serials): raise HTTPException(status_code=400, detail="序列号商品必须每单位提供一个不重复序列号") elif serials: raise HTTPException(status_code=400, detail="非序列号商品不能提交序列号") if tracking_type == "lot" and not batch_no: raise HTTPException(status_code=400, detail="批次商品必须填写批次号") async def load_product_profile(db, tenant_id: int, product_id: int, variant_id: int) -> dict: row = (await db.execute( text("SELECT tracking_type,qc_required,shelf_life_days,removal_strategy " "FROM inventory_product_profiles WHERE tenant_id=:tid AND product_id=:pid " "AND variant_id IN (:vid,0) ORDER BY variant_id DESC LIMIT 1"), {"tid": tenant_id, "pid": product_id, "vid": variant_id}, )).fetchone() return { "tracking_type": row[0] if row else "none", "qc_required": bool(row[1]) if row else False, "shelf_life_days": int(row[2]) if row and row[2] is not None else None, "removal_strategy": row[3] if row else "fifo", } async def register_received_serials(db, tenant_id: int, *, serial_numbers: list[str], product_id: int, variant_id: int, warehouse_id: int, location_id: int, batch_id: int, stock_state: str, operation_id: int) -> None: for serial_no in serial_numbers: try: await db.execute( text("INSERT INTO inventory_serials " "(tenant_id,product_id,variant_id,serial_no,stock_state,location_id,warehouse_id,batch_id," "source_operation_id,created_at,updated_at) " "VALUES (:tid,:pid,:vid,:serial,:state,:location,:wh,:batch,:op,NOW(3),NOW(3))"), {"tid": tenant_id, "pid": product_id, "vid": variant_id or None, "serial": serial_no, "state": stock_state, "location": location_id, "wh": warehouse_id, "batch": batch_id or None, "op": operation_id}, ) except Exception: raise HTTPException(status_code=409, detail=f"序列号 {serial_no} 已存在") serial_id = int((await db.execute(text("SELECT LAST_INSERT_ID()"))).scalar()) await db.execute( text("INSERT INTO inventory_serial_events " "(tenant_id,serial_id,operation_id,event_type,to_warehouse_id,to_location_id,to_state,created_at,updated_at) " "VALUES (:tid,:serial,:op,'receipt',:wh,:location,:state,NOW(3),NOW(3))"), {"tid": tenant_id, "serial": serial_id, "op": operation_id, "wh": warehouse_id, "location": location_id, "state": stock_state}, ) async def move_serials(db, tenant_id: int, *, serial_numbers: list[str], product_id: int, variant_id: int, operation_id: int, event_type: str, from_warehouse_id: int, to_warehouse_id: int, from_location_id: int, to_location_id: int, from_state: str, to_state: str, batch_id: int = 0) -> None: if not serial_numbers: return rows = (await db.execute( text("SELECT id,serial_no FROM inventory_serials WHERE tenant_id=:tid AND product_id=:pid " "AND COALESCE(variant_id,0)=:vid AND warehouse_id=:wh AND location_id=:location " "AND stock_state=:state AND COALESCE(batch_id,0)=:batch FOR UPDATE"), {"tid": tenant_id, "pid": product_id, "vid": variant_id, "wh": from_warehouse_id, "location": from_location_id, "state": from_state, "batch": batch_id}, )).fetchall() available = {row[1]: int(row[0]) for row in rows} missing = [serial for serial in serial_numbers if serial not in available] if missing: raise HTTPException(status_code=400, detail=f"序列号不可用或位置/状态不匹配: {', '.join(missing)}") for serial_no in serial_numbers: serial_id = available[serial_no] await db.execute( text("UPDATE inventory_serials SET warehouse_id=:to_wh,location_id=:to_location,stock_state=:to_state," "updated_at=NOW(3) WHERE id=:id AND tenant_id=:tid"), {"to_wh": to_warehouse_id, "to_location": to_location_id, "to_state": to_state, "id": serial_id, "tid": tenant_id}, ) await db.execute( text("INSERT INTO inventory_serial_events " "(tenant_id,serial_id,operation_id,event_type,from_warehouse_id,to_warehouse_id," "from_location_id,to_location_id,from_state,to_state,created_at,updated_at) " "VALUES (:tid,:serial,:op,:event,:from_wh,:to_wh,:from_location,:to_location," ":from_state,:to_state,NOW(3),NOW(3))"), {"tid": tenant_id, "serial": serial_id, "op": operation_id, "event": event_type, "from_wh": from_warehouse_id, "to_wh": to_warehouse_id, "from_location": from_location_id, "to_location": to_location_id, "from_state": from_state, "to_state": to_state}, ) async def apply_batch_quality(db, tenant_id: int, batch_id: int, decision: str, operator_id: int) -> int | None: from app.plugins.inventory.operations import _insert_operation, _write_operation_leg batch = (await db.execute( text("SELECT product_id,COALESCE(variant_id,0),qc_state FROM inventory_batches " "WHERE id=:id AND tenant_id=:tid FOR UPDATE"), {"id": batch_id, "tid": tenant_id}, )).fetchone() if batch is None: raise HTTPException(status_code=404, detail="批次不存在") if decision == "hold": await db.execute(text("UPDATE inventory_batches SET qc_state='hold',updated_at=NOW(3) " "WHERE id=:id AND tenant_id=:tid"), {"id": batch_id, "tid": tenant_id}) return None if decision not in ("passed", "rejected"): raise HTTPException(status_code=400, detail="质检结果无效") balances = (await db.execute( text("SELECT warehouse_id,location_id,qty FROM inventory_balances WHERE tenant_id=:tid " "AND batch_id=:batch AND stock_state='qc' AND qty>0 FOR UPDATE"), {"tid": tenant_id, "batch": batch_id}, )).fetchall() if not balances: raise HTTPException(status_code=409, detail="批次没有待质检库存") operation_id = await _insert_operation( db, tenant_id, operation_type="quality", source_doc_type="batch", source_doc_id=batch_id, warehouse_id=int(balances[0][0]), reason=decision, ) destination_state = "sellable" if decision == "passed" else "defective" for index, row in enumerate(balances, 1): qty = Decimal(str(row[2])) await db.execute( text("INSERT INTO inventory_moves " "(tenant_id,operation_id,product_id,variant_id,source_warehouse_id,destination_warehouse_id," "source_location_id,destination_location_id,planned_qty,done_qty,batch_id,source_state," "destination_state,created_at,updated_at) VALUES " "(:tid,:op,:pid,:vid,:wh,:wh,:location,:location,:qty,:qty,:batch,'qc',:destination,NOW(3),NOW(3))"), {"tid": tenant_id, "op": operation_id, "pid": batch[0], "vid": batch[1], "wh": row[0], "location": row[1], "qty": qty, "batch": batch_id, "destination": destination_state}, ) move_id = int((await db.execute(text("SELECT LAST_INSERT_ID()"))).scalar()) common = dict(db=db, tenant_id=tenant_id, operation_id=operation_id, move_id=move_id, product_id=int(batch[0]), variant_id=int(batch[1]), warehouse_id=int(row[0]), location_id=int(row[1]), batch_id=batch_id, unit_cost=None, operator_id=operator_id, reason="批次质检") await _write_operation_leg(**common, qty_delta=-qty, stock_state="qc", leg_suffix=f"qc_out_{index}") await _write_operation_leg(**common, qty_delta=qty, stock_state=destination_state, leg_suffix=f"qc_in_{index}") for row in balances: serials = (await db.execute( text("SELECT serial_no FROM inventory_serials WHERE tenant_id=:tid AND batch_id=:batch " "AND warehouse_id=:wh AND location_id=:location AND stock_state='qc'"), {"tid": tenant_id, "batch": batch_id, "wh": row[0], "location": row[1]}, )).fetchall() await move_serials( db, tenant_id, serial_numbers=[serial[0] for serial in serials], product_id=int(batch[0]), variant_id=int(batch[1]), operation_id=operation_id, event_type="quality", from_warehouse_id=int(row[0]), to_warehouse_id=int(row[0]), from_location_id=int(row[1]), to_location_id=int(row[1]), from_state="qc", to_state=destination_state, batch_id=batch_id, ) await db.execute(text("UPDATE inventory_batches SET qc_state=:state,updated_at=NOW(3) " "WHERE id=:id AND tenant_id=:tid"), {"state": decision, "id": batch_id, "tid": tenant_id}) await db.execute(text("UPDATE inventory_operations SET state='done',actual_at=NOW(3),updated_at=NOW(3) " "WHERE id=:id AND tenant_id=:tid"), {"id": operation_id, "tid": tenant_id}) return operation_id async def trace_lot(db, tenant_id: int, batch_id: int) -> list[dict]: rows = (await db.execute( text("SELECT id,doc_type,doc_id,warehouse_id,location_id,stock_state,qty_delta,created_at " "FROM inventory_transactions WHERE tenant_id=:tid AND batch_id=:batch ORDER BY id"), {"tid": tenant_id, "batch": batch_id}, )).fetchall() return [{"transaction_id": row[0], "doc_type": row[1], "doc_id": row[2], "warehouse_id": row[3], "location_id": row[4], "stock_state": row[5], "qty_delta": str(row[6]), "created_at": str(row[7])} for row in rows] async def trace_serial(db, tenant_id: int, serial_no: str) -> list[dict]: rows = (await db.execute( text("SELECT e.id,e.operation_id,e.event_type,e.from_warehouse_id,e.to_warehouse_id," "e.from_location_id,e.to_location_id,e.from_state,e.to_state,e.created_at " "FROM inventory_serial_events e JOIN inventory_serials s ON s.id=e.serial_id " "WHERE e.tenant_id=:tid AND s.tenant_id=:tid AND s.serial_no=:serial ORDER BY e.id"), {"tid": tenant_id, "serial": serial_no}, )).fetchall() return [{"event_id": row[0], "operation_id": row[1], "event_type": row[2], "from_warehouse_id": row[3], "to_warehouse_id": row[4], "from_location_id": row[5], "to_location_id": row[6], "from_state": row[7], "to_state": row[8], "created_at": str(row[9])} for row in rows]