846 lines
31 KiB
Python
846 lines
31 KiB
Python
"""运营支撑 API（洁净室、进出记录、仓储设施、储位、展厅预约）"""
|
|
from datetime import datetime
|
|
from typing import Any, Optional
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
from sqlalchemy import or_
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.data_scope import apply_data_scope
|
|
from app.core.database import get_db
|
|
from app.core.menu_guard import require_admin_or_super_admin
|
|
from app.core.response import error_response, success_response
|
|
from app.crud import clean_room as clean_room_svc
|
|
from app.crud import exhibition_appointment as exhibition_svc
|
|
from app.crud import storage_facility as storage_facility_svc
|
|
from app.crud import storage_occupancy as occupancy_svc
|
|
from app.core.utils import generate_uuid
|
|
from app.models.models import BizCleanRoom, BizCleanRoomAccessLog, BizExhibitionAppointment, BizStorageFacility, BizStorageFacilitySlot, SysUser
|
|
|
|
router = APIRouter(prefix="/web/admin", tags=["运营支撑"])
|
|
|
|
|
|
def _get_tenant_no(db: Session, user: dict) -> str:
|
|
user_id = user.get("user_id")
|
|
if not user_id:
|
|
raise PermissionError("用户身份信息缺失，无法确定租户")
|
|
sys_user = db.query(SysUser).filter(SysUser.id == user_id).first()
|
|
if not sys_user or not sys_user.tenant_no:
|
|
raise PermissionError("用户租户信息缺失")
|
|
return sys_user.tenant_no
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 洁净室
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class CleanRoomCreateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
room_code: str = Field(..., alias="roomCode")
|
|
room_name: str = Field(..., alias="roomName")
|
|
room_grade: Optional[str] = Field(None, alias="roomGrade")
|
|
room_status: Optional[str] = Field("zc", alias="roomStatus")
|
|
area_sqm: Optional[float] = Field(None, alias="areaSqm")
|
|
order_num: Optional[int] = Field(0, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class CleanRoomUpdateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
room_code: Optional[str] = Field(None, alias="roomCode")
|
|
room_name: Optional[str] = Field(None, alias="roomName")
|
|
room_grade: Optional[str] = Field(None, alias="roomGrade")
|
|
room_status: Optional[str] = Field(None, alias="roomStatus")
|
|
area_sqm: Optional[float] = Field(None, alias="areaSqm")
|
|
order_num: Optional[int] = Field(None, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
@router.get("/clean-rooms")
|
|
async def list_clean_rooms(
|
|
page: int = Query(1, ge=1),
|
|
pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
|
|
keyword: Optional[str] = None,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
query = db.query(BizCleanRoom).filter(
|
|
BizCleanRoom.del_flag == 0,
|
|
BizCleanRoom.tenant_no == tenant_no,
|
|
)
|
|
kw = (keyword or "").strip()
|
|
if kw:
|
|
like = f"%{kw}%"
|
|
query = query.filter(or_(BizCleanRoom.room_code.like(like), BizCleanRoom.room_name.like(like)))
|
|
query = apply_data_scope(query, BizCleanRoom, user, db, tenant_no)
|
|
total = query.count()
|
|
items = query.order_by(BizCleanRoom.create_time.desc()).offset((page - 1) * pageSize).limit(pageSize).all()
|
|
result_list = [clean_room_svc._row_to_dict(r) for r in items]
|
|
tp = clean_room_svc.total_pages(total, pageSize)
|
|
return success_response({"list": result_list, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
@router.post("/clean-rooms")
|
|
async def create_clean_room(
|
|
body: CleanRoomCreateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
obj = clean_room_svc.create_clean_room(
|
|
db,
|
|
room_code=body.room_code,
|
|
room_name=body.room_name,
|
|
room_grade=body.room_grade,
|
|
room_status=body.room_status or "zc",
|
|
area_sqm=body.area_sqm,
|
|
tenant_no=tenant_no,
|
|
order_num=body.order_num,
|
|
remark=body.remark,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(clean_room_svc._row_to_dict(obj), message="创建成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/clean-rooms/{room_id}")
|
|
async def update_clean_room(
|
|
room_id: str,
|
|
body: CleanRoomUpdateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
|
obj = clean_room_svc.update_clean_room(
|
|
db,
|
|
room_id,
|
|
room_code=payload.get("room_code"),
|
|
room_name=payload.get("room_name"),
|
|
room_grade=payload.get("room_grade"),
|
|
room_status=payload.get("room_status"),
|
|
area_sqm=payload.get("area_sqm"),
|
|
order_num=payload.get("order_num"),
|
|
remark=payload.get("remark"),
|
|
user_id=user.get("user_id"),
|
|
tenant_no=tenant_no,
|
|
)
|
|
return success_response(clean_room_svc._row_to_dict(obj), message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.delete("/clean-rooms/{room_id}")
|
|
async def delete_clean_room(
|
|
room_id: str,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
clean_room_svc.soft_delete_clean_room(db, room_id, tenant_no=tenant_no)
|
|
return success_response(None, message="删除成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 进出记录
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class AccessLogCreateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
clean_room_id: str = Field(..., alias="cleanRoomId")
|
|
person_name: Optional[str] = Field(None, alias="personName")
|
|
dept_name: Optional[str] = Field(None, alias="deptName")
|
|
enter_time: Optional[str] = Field(None, alias="enterTime")
|
|
purpose: Optional[str] = None
|
|
leave_time: Optional[str] = Field(None, alias="leaveTime")
|
|
body_temp: Optional[str] = Field(None, alias="bodyTemp")
|
|
order_num: Optional[int] = Field(0, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class AccessLogUpdateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
person_name: Optional[str] = Field(None, alias="personName")
|
|
dept_name: Optional[str] = Field(None, alias="deptName")
|
|
enter_time: Optional[str] = Field(None, alias="enterTime")
|
|
purpose: Optional[str] = None
|
|
leave_time: Optional[str] = Field(None, alias="leaveTime")
|
|
body_temp: Optional[str] = Field(None, alias="bodyTemp")
|
|
order_num: Optional[int] = Field(None, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
def _parse_optional_dt(value: Optional[str]) -> Optional[datetime]:
|
|
if not value:
|
|
return None
|
|
return datetime.fromisoformat(value)
|
|
|
|
|
|
@router.get("/clean-room-access-logs")
|
|
async def list_access_logs(
|
|
page: int = Query(1, ge=1),
|
|
pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
|
|
cleanRoomId: str = Query(..., alias="cleanRoomId"),
|
|
keyword: Optional[str] = None,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
query = db.query(BizCleanRoomAccessLog).filter(
|
|
BizCleanRoomAccessLog.del_flag == 0,
|
|
BizCleanRoomAccessLog.tenant_no == tenant_no,
|
|
BizCleanRoomAccessLog.clean_room_id == cleanRoomId,
|
|
)
|
|
if keyword:
|
|
kw = f"%{keyword}%"
|
|
query = query.filter(
|
|
BizCleanRoomAccessLog.person_name.like(kw)
|
|
| BizCleanRoomAccessLog.dept_name.like(kw)
|
|
)
|
|
query = apply_data_scope(query, BizCleanRoomAccessLog, user, db, tenant_no)
|
|
total = query.count()
|
|
items = query.order_by(BizCleanRoomAccessLog.create_time.desc()).offset((page - 1) * pageSize).limit(pageSize).all()
|
|
result_list = [clean_room_svc._access_log_to_dict(r) for r in items]
|
|
tp = clean_room_svc.total_pages(total, pageSize)
|
|
return success_response({"list": result_list, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
@router.post("/clean-room-access-logs")
|
|
async def create_access_log(
|
|
body: AccessLogCreateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
obj = clean_room_svc.create_access_log(
|
|
db,
|
|
clean_room_id=body.clean_room_id,
|
|
person_name=body.person_name,
|
|
dept_name=body.dept_name,
|
|
enter_time=_parse_optional_dt(body.enter_time),
|
|
purpose=body.purpose,
|
|
leave_time=_parse_optional_dt(body.leave_time),
|
|
body_temp=body.body_temp,
|
|
tenant_no=tenant_no,
|
|
order_num=body.order_num,
|
|
remark=body.remark,
|
|
user_id_param=user.get("user_id"),
|
|
)
|
|
return success_response(clean_room_svc._access_log_to_dict(obj), message="创建成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/clean-room-access-logs/{log_id}")
|
|
async def update_access_log(
|
|
log_id: str,
|
|
body: AccessLogUpdateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
|
obj = clean_room_svc.update_access_log(
|
|
db,
|
|
log_id,
|
|
person_name=payload.get("person_name"),
|
|
dept_name=payload.get("dept_name"),
|
|
enter_time=_parse_optional_dt(payload.get("enter_time")) if "enter_time" in payload else None,
|
|
purpose=payload.get("purpose"),
|
|
leave_time=_parse_optional_dt(payload.get("leave_time")) if "leave_time" in payload else None,
|
|
body_temp=payload.get("body_temp"),
|
|
order_num=payload.get("order_num"),
|
|
remark=payload.get("remark"),
|
|
user_id_param=user.get("user_id"),
|
|
tenant_no=tenant_no,
|
|
)
|
|
return success_response(clean_room_svc._access_log_to_dict(obj), message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.delete("/clean-room-access-logs/{log_id}")
|
|
async def delete_access_log(
|
|
log_id: str,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
clean_room_svc.soft_delete_access_log(db, log_id, tenant_no=tenant_no)
|
|
return success_response(None, message="删除成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 仓储设施
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class StorageFacilityCreateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
parent_id: str = Field("0", alias="parentId")
|
|
node_name: str = Field(..., alias="nodeName")
|
|
node_type: Optional[str] = Field(None, alias="nodeType")
|
|
is_leaf: Optional[int] = Field(0, alias="isLeaf")
|
|
order_num: Optional[int] = Field(0, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class StorageFacilityUpdateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
node_name: Optional[str] = Field(None, alias="nodeName")
|
|
node_type: Optional[str] = Field(None, alias="nodeType")
|
|
is_leaf: Optional[int] = Field(None, alias="isLeaf")
|
|
order_num: Optional[int] = Field(None, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
@router.get("/storage-facilities")
|
|
async def list_storage_facilities(
|
|
keyword: Optional[str] = None,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
query = db.query(BizStorageFacility).filter(
|
|
BizStorageFacility.del_flag == 0,
|
|
BizStorageFacility.tenant_no == tenant_no,
|
|
)
|
|
if keyword:
|
|
query = query.filter(BizStorageFacility.node_name.like(f"%{keyword}%"))
|
|
query = apply_data_scope(query, BizStorageFacility, user, db, tenant_no)
|
|
items = query.order_by(
|
|
BizStorageFacility.order_num.asc(),
|
|
BizStorageFacility.create_time.asc(),
|
|
).all()
|
|
result_list = [storage_facility_svc._row_to_dict(r) for r in items]
|
|
return success_response(result_list)
|
|
|
|
|
|
@router.post("/storage-facilities")
|
|
async def create_storage_facility(
|
|
body: StorageFacilityCreateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
obj = storage_facility_svc.create_facility(
|
|
db,
|
|
parent_id=body.parent_id,
|
|
node_name=body.node_name,
|
|
node_type=body.node_type,
|
|
is_leaf=body.is_leaf,
|
|
tenant_no=tenant_no,
|
|
order_num=body.order_num,
|
|
remark=body.remark,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(storage_facility_svc._row_to_dict(obj), message="创建成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/storage-facilities/{node_id}")
|
|
async def update_storage_facility(
|
|
node_id: str,
|
|
body: StorageFacilityUpdateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
|
obj = storage_facility_svc.update_facility(
|
|
db,
|
|
node_id,
|
|
node_name=payload.get("node_name"),
|
|
node_type=payload.get("node_type"),
|
|
is_leaf=payload.get("is_leaf"),
|
|
order_num=payload.get("order_num"),
|
|
remark=payload.get("remark"),
|
|
user_id=user.get("user_id"),
|
|
tenant_no=tenant_no,
|
|
)
|
|
return success_response(storage_facility_svc._row_to_dict(obj), message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.delete("/storage-facilities/{node_id}")
|
|
async def delete_storage_facility(
|
|
node_id: str,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
storage_facility_svc.soft_delete_facility(db, node_id, tenant_no=tenant_no)
|
|
return success_response(None, message="删除成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 储位
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class StorageSlotCreateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
facility_node_id: str = Field(..., alias="facilityNodeId")
|
|
slot_code: Optional[str] = Field(None, alias="slotCode")
|
|
slot_status: Optional[str] = Field(None, alias="slotStatus")
|
|
order_num: Optional[int] = Field(0, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class StorageSlotBatchCreateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
facility_node_id: str = Field(..., alias="facilityNodeId")
|
|
rows: int = Field(..., ge=1, le=50)
|
|
cols: int = Field(..., ge=1, le=50)
|
|
code_prefix: Optional[str] = Field("", alias="codePrefix")
|
|
|
|
|
|
class StorageSlotUpdateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
slot_code: Optional[str] = Field(None, alias="slotCode")
|
|
slot_status: Optional[str] = Field(None, alias="slotStatus")
|
|
order_num: Optional[int] = Field(None, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class StorageSlotClearBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
facility_node_id: str = Field(..., alias="facilityNodeId")
|
|
|
|
|
|
class StorageSlotInsertRowBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
facility_node_id: str = Field(..., alias="facilityNodeId")
|
|
after_row_num: int = Field(..., ge=0, alias="afterRowNum")
|
|
cols: int = Field(..., ge=1)
|
|
code_prefix: Optional[str] = Field("", alias="codePrefix")
|
|
|
|
|
|
@router.get("/storage-facility-slots")
|
|
async def list_storage_slots(
|
|
facilityNodeId: str = Query(..., alias="facilityNodeId"),
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
query = db.query(BizStorageFacilitySlot).filter(
|
|
BizStorageFacilitySlot.del_flag == 0,
|
|
BizStorageFacilitySlot.tenant_no == tenant_no,
|
|
BizStorageFacilitySlot.facility_node_id == facilityNodeId,
|
|
)
|
|
query = apply_data_scope(query, BizStorageFacilitySlot, user, db, tenant_no)
|
|
items = query.order_by(BizStorageFacilitySlot.order_num.asc(), BizStorageFacilitySlot.create_time.asc()).all()
|
|
result_list = [storage_facility_svc._slot_to_dict(r) for r in items]
|
|
return success_response(result_list)
|
|
|
|
|
|
@router.post("/storage-facility-slots/batch")
|
|
async def batch_create_storage_slots(
|
|
body: StorageSlotBatchCreateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
items = storage_facility_svc.batch_create_slots(
|
|
db,
|
|
facility_node_id=body.facility_node_id,
|
|
rows=body.rows,
|
|
cols=body.cols,
|
|
code_prefix=body.code_prefix,
|
|
tenant_no=tenant_no,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(items, message=f"批量创建 {len(items)} 个格子成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.post("/storage-facility-slots")
|
|
async def create_storage_slot(
|
|
body: StorageSlotCreateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
obj = storage_facility_svc.create_slot(
|
|
db,
|
|
facility_node_id=body.facility_node_id,
|
|
slot_code=body.slot_code,
|
|
slot_status=body.slot_status,
|
|
tenant_no=tenant_no,
|
|
order_num=body.order_num,
|
|
remark=body.remark,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(storage_facility_svc._slot_to_dict(obj), message="创建成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/storage-facility-slots/{slot_id}")
|
|
async def update_storage_slot(
|
|
slot_id: str,
|
|
body: StorageSlotUpdateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
|
obj = storage_facility_svc.update_slot(
|
|
db,
|
|
slot_id,
|
|
slot_code=payload.get("slot_code"),
|
|
slot_status=payload.get("slot_status"),
|
|
order_num=payload.get("order_num"),
|
|
remark=payload.get("remark"),
|
|
user_id=user.get("user_id"),
|
|
tenant_no=tenant_no,
|
|
)
|
|
return success_response(storage_facility_svc._slot_to_dict(obj), message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.post("/storage-facility-slots/insert-row")
|
|
async def insert_storage_slot_row(
|
|
body: StorageSlotInsertRowBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
result = storage_facility_svc.insert_row_and_shift_slots(
|
|
db,
|
|
facility_node_id=body.facility_node_id,
|
|
after_row_num=body.after_row_num,
|
|
cols=body.cols,
|
|
code_prefix=body.code_prefix or "",
|
|
tenant_no=tenant_no,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(result, message="新增行成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.post("/storage-facility-slots/clear")
|
|
async def clear_storage_slots(
|
|
body: StorageSlotClearBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
cleared_count = storage_facility_svc.clear_slots_by_facility_node(
|
|
db,
|
|
facility_node_id=body.facility_node_id,
|
|
tenant_no=tenant_no,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(
|
|
{"clearedCount": cleared_count},
|
|
message=f"已清空 {cleared_count} 个格子",
|
|
)
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.delete("/storage-facility-slots/{slot_id}")
|
|
async def delete_storage_slot(
|
|
slot_id: str,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
storage_facility_svc.soft_delete_slot(db, slot_id, tenant_no=tenant_no)
|
|
return success_response(None, message="删除成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 库位占用记录
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class OccupancyCreateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
slot_id: str = Field(..., alias="slotId")
|
|
item_type: str = Field("other", alias="itemType")
|
|
item_code: str = Field(..., alias="itemCode")
|
|
item_name: str = Field(..., alias="itemName")
|
|
item_source: str = Field("manual", alias="itemSource")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
@router.get("/storage-slot-occupancies")
|
|
async def list_occupancies(
|
|
slotId: str = Query(..., alias="slotId"),
|
|
occupancyStatus: Optional[str] = Query("stored", alias="occupancyStatus"),
|
|
itemType: Optional[str] = Query(None, alias="itemType"),
|
|
page: int = Query(1),
|
|
pageSize: int = Query(20, alias="pageSize"),
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
result = occupancy_svc.list_occupancies(
|
|
db,
|
|
slot_id=slotId,
|
|
occupancy_status=occupancyStatus,
|
|
item_type=itemType,
|
|
tenant_no=tenant_no,
|
|
page=page,
|
|
page_size=pageSize,
|
|
)
|
|
return success_response(result)
|
|
|
|
|
|
@router.get("/storage-slot-occupancies/history")
|
|
async def list_occupancy_history(
|
|
slotId: Optional[str] = Query(None, alias="slotId"),
|
|
facilityNodeId: Optional[str] = Query(None, alias="facilityNodeId"),
|
|
itemType: Optional[str] = Query(None, alias="itemType"),
|
|
startDate: Optional[str] = Query(None, alias="startDate"),
|
|
endDate: Optional[str] = Query(None, alias="endDate"),
|
|
page: int = Query(1),
|
|
pageSize: int = Query(20, alias="pageSize"),
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
result = occupancy_svc.list_occupancy_history(
|
|
db,
|
|
slot_id=slotId,
|
|
facility_node_id=facilityNodeId,
|
|
item_type=itemType,
|
|
start_date=startDate,
|
|
end_date=endDate,
|
|
tenant_no=tenant_no,
|
|
page=page,
|
|
page_size=pageSize,
|
|
)
|
|
return success_response(result)
|
|
|
|
|
|
@router.post("/storage-slot-occupancies")
|
|
async def create_occupancy(
|
|
body: OccupancyCreateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
item_id = generate_uuid()
|
|
obj = occupancy_svc.create_occupancy(
|
|
db,
|
|
slot_id=body.slot_id,
|
|
item_type=body.item_type,
|
|
item_id=item_id,
|
|
item_code=body.item_code,
|
|
item_name=body.item_name,
|
|
item_source=body.item_source,
|
|
tenant_no=tenant_no,
|
|
user_id=user.get("user_id"),
|
|
remark=body.remark,
|
|
)
|
|
return success_response(occupancy_svc._row_to_dict(obj), message="入库成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/storage-slot-occupancies/{occupancy_id}/remove")
|
|
async def remove_occupancy(
|
|
occupancy_id: str,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
obj = occupancy_svc.remove_occupancy(
|
|
db,
|
|
occupancy_id,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(occupancy_svc._row_to_dict(obj), message="出库成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 展厅预约
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class ExhibitionAppointmentCreateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
appointment_no: Optional[str] = Field(None, alias="appointmentNo")
|
|
visit_org: Optional[str] = Field(None, alias="visitOrg")
|
|
contact_name: Optional[str] = Field(None, alias="contactName")
|
|
visit_date: Optional[str] = Field(None, alias="visitDate")
|
|
visitor_count: Optional[int] = Field(None, alias="visitorCount")
|
|
reception_user_id: Optional[str] = Field(None, alias="receptionUserId")
|
|
order_num: Optional[int] = Field(0, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
class ExhibitionAppointmentUpdateBody(BaseModel):
|
|
model_config = ConfigDict(populate_by_name=True)
|
|
|
|
visit_org: Optional[str] = Field(None, alias="visitOrg")
|
|
contact_name: Optional[str] = Field(None, alias="contactName")
|
|
visit_date: Optional[str] = Field(None, alias="visitDate")
|
|
visitor_count: Optional[int] = Field(None, alias="visitorCount")
|
|
reception_user_id: Optional[str] = Field(None, alias="receptionUserId")
|
|
order_num: Optional[int] = Field(None, alias="orderNum")
|
|
remark: Optional[str] = None
|
|
|
|
|
|
def _normalize_visit_date_bound(value: str, *, end_of_day: bool = False) -> str:
|
|
val = (value or "").strip()
|
|
if len(val) == 10:
|
|
return f"{val} 23:59:59" if end_of_day else f"{val} 00:00:00"
|
|
return val
|
|
|
|
|
|
@router.get("/exhibition-appointments")
|
|
async def list_exhibition_appointments(
|
|
page: int = Query(1, ge=1),
|
|
pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
|
|
keyword: Optional[str] = None,
|
|
visitDateStart: Optional[str] = Query(None, alias="visitDateStart"),
|
|
visitDateEnd: Optional[str] = Query(None, alias="visitDateEnd"),
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
tenant_no = _get_tenant_no(db, user)
|
|
query = db.query(BizExhibitionAppointment).filter(
|
|
BizExhibitionAppointment.del_flag == 0,
|
|
BizExhibitionAppointment.tenant_no == tenant_no,
|
|
)
|
|
kw = (keyword or "").strip()
|
|
if kw:
|
|
like = f"%{kw}%"
|
|
query = query.filter(
|
|
or_(
|
|
BizExhibitionAppointment.appointment_no.like(like),
|
|
BizExhibitionAppointment.contact_name.like(like),
|
|
BizExhibitionAppointment.visit_org.like(like),
|
|
)
|
|
)
|
|
if visitDateStart:
|
|
query = query.filter(
|
|
BizExhibitionAppointment.visit_date >= _normalize_visit_date_bound(visitDateStart)
|
|
)
|
|
if visitDateEnd:
|
|
query = query.filter(
|
|
BizExhibitionAppointment.visit_date <= _normalize_visit_date_bound(visitDateEnd, end_of_day=True)
|
|
)
|
|
query = apply_data_scope(query, BizExhibitionAppointment, user, db, tenant_no)
|
|
total = query.count()
|
|
items = query.order_by(BizExhibitionAppointment.create_time.desc()).offset((page - 1) * pageSize).limit(pageSize).all()
|
|
result_list = [exhibition_svc._row_to_dict(r) for r in items]
|
|
tp = exhibition_svc.total_pages(total, pageSize)
|
|
return success_response({"list": result_list, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})
|
|
|
|
|
|
@router.post("/exhibition-appointments")
|
|
async def create_exhibition_appointment(
|
|
body: ExhibitionAppointmentCreateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
obj = exhibition_svc.create_exhibition_appointment(
|
|
db,
|
|
appointment_no=body.appointment_no,
|
|
visit_org=body.visit_org,
|
|
contact_name=body.contact_name,
|
|
visit_date=body.visit_date,
|
|
visitor_count=body.visitor_count,
|
|
reception_user_id=body.reception_user_id,
|
|
tenant_no=tenant_no,
|
|
order_num=body.order_num,
|
|
remark=body.remark,
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(exhibition_svc._row_to_dict(obj), message="创建成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.put("/exhibition-appointments/{appointment_id}")
|
|
async def update_exhibition_appointment(
|
|
appointment_id: str,
|
|
body: ExhibitionAppointmentUpdateBody,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
tenant_no = _get_tenant_no(db, user)
|
|
payload: dict[str, Any] = body.model_dump(exclude_unset=True, by_alias=False)
|
|
obj = exhibition_svc.update_exhibition_appointment(
|
|
db,
|
|
appointment_id,
|
|
visit_org=payload.get("visit_org"),
|
|
contact_name=payload.get("contact_name"),
|
|
visit_date=payload.get("visit_date"),
|
|
visitor_count=payload.get("visitor_count"),
|
|
reception_user_id=payload.get("reception_user_id"),
|
|
order_num=payload.get("order_num"),
|
|
remark=payload.get("remark"),
|
|
user_id=user.get("user_id"),
|
|
)
|
|
return success_response(exhibition_svc._row_to_dict(obj), message="更新成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|
|
|
|
|
|
@router.delete("/exhibition-appointments/{appointment_id}")
|
|
async def delete_exhibition_appointment(
|
|
appointment_id: str,
|
|
user: dict = Depends(require_admin_or_super_admin),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
try:
|
|
exhibition_svc.soft_delete_exhibition_appointment(db, appointment_id)
|
|
return success_response(None, message="删除成功")
|
|
except ValueError as e:
|
|
return error_response(400, str(e))
|