zlwuandCursor
8f1f08b284
Allow filtered and selected row export on inventory and shared IO order pages. Co-authored-by: Cursor <cursoragent@cursor.com>
454 lines
16 KiB
Python
454 lines
16 KiB
Python
"""运输计划管理 - 盘存统计 API"""
|
|
from datetime import date, datetime
|
|
from decimal import Decimal
|
|
from typing import List, Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from pydantic import BaseModel, Field
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.database import get_db
|
|
from app.core.db_auth import DbUserContext, get_db_current_user
|
|
from app.services.pl_data_scope import apply_pl_scope, assert_pl_row_in_scope
|
|
from app.core.tenant_context import get_tenant_no
|
|
from app.utils.excel_export import (
|
|
EXPORT_MAX_ROWS,
|
|
ExportColumn,
|
|
ExportValidationError,
|
|
export_rows_to_xlsx_response,
|
|
filter_export_columns,
|
|
)
|
|
from app.core.utils import generate_uuid
|
|
from app.models.pl_models import (
|
|
PlDocumentReminder,
|
|
PlInventory,
|
|
PlInventoryTransaction,
|
|
PlInventoryWarning,
|
|
)
|
|
from app.schemas.pl.inventory import (
|
|
DocumentReminderRead,
|
|
DocumentReminderStatsRead,
|
|
DocumentReminderUpdate,
|
|
InventoryLedgerQuery,
|
|
InventoryLedgerResponse,
|
|
InventoryRead,
|
|
InventorySummaryItem,
|
|
InventorySummaryResponse,
|
|
InventoryTransactionQuery,
|
|
InventoryTransactionRead,
|
|
InventoryTransactionResponse,
|
|
InventoryWarningRead,
|
|
InventoryWarningThreshold,
|
|
InventoryWarningUpdate,
|
|
TransportStatsItem,
|
|
TransportStatsResponse,
|
|
)
|
|
from app.services.pl_cert_service import scan_cert_expiry
|
|
|
|
router = APIRouter(prefix="/inventory", tags=["运输计划-盘存统计"])
|
|
|
|
|
|
# =============================================================================
|
|
# 库存台账
|
|
# =============================================================================
|
|
|
|
|
|
@router.get("/ledger", response_model=InventoryLedgerResponse)
|
|
def get_inventory_ledger(
|
|
product_name: Optional[str] = Query(default=None, alias="productName", description="产品名称"),
|
|
page: int = Query(default=1, ge=1),
|
|
page_size: int = Query(default=20, ge=1, le=500, alias="pageSize"),
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> InventoryLedgerResponse:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
query = db.query(PlInventory).filter(
|
|
PlInventory.tenant_no == tenant_no,
|
|
PlInventory.del_flag == 0,
|
|
)
|
|
|
|
if product_name:
|
|
query = query.filter(PlInventory.product_name.contains(product_name))
|
|
|
|
query = apply_pl_scope(query, PlInventory, user, db, tenant_no)
|
|
|
|
total = query.count()
|
|
offset = (page - 1) * page_size
|
|
items = query.order_by(PlInventory.product_name).offset(offset).limit(page_size).all()
|
|
|
|
return InventoryLedgerResponse(total=total, list=items)
|
|
|
|
|
|
# =============================================================================
|
|
# 库存预警阈值设置
|
|
# =============================================================================
|
|
|
|
|
|
|
|
LEDGER_EXPORT_COLUMNS = [
|
|
ExportColumn(title="产品名称", field="productName"),
|
|
ExportColumn(title="品牌", field="brandName"),
|
|
ExportColumn(title="产品编号", field="productNo"),
|
|
ExportColumn(title="规格型号", field="model"),
|
|
ExportColumn(title="单位", field="unit"),
|
|
ExportColumn(title="库存数量", field="stockQty"),
|
|
ExportColumn(title="预警阈值", field="warnThreshold"),
|
|
ExportColumn(title="状态", field="status"),
|
|
]
|
|
|
|
|
|
class InventoryLedgerExportBody(BaseModel):
|
|
product_name: Optional[str] = Field(default=None, alias="productName")
|
|
ids: Optional[List[str]] = None
|
|
columns: Optional[List[str]] = None
|
|
model_config = {"populate_by_name": True}
|
|
|
|
|
|
@router.post("/ledger/export")
|
|
def export_inventory_ledger(
|
|
body: InventoryLedgerExportBody,
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
):
|
|
try:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
query = db.query(PlInventory).filter(
|
|
PlInventory.tenant_no == tenant_no,
|
|
PlInventory.del_flag == 0,
|
|
)
|
|
if body.product_name:
|
|
query = query.filter(PlInventory.product_name.contains(body.product_name))
|
|
if body.ids:
|
|
query = query.filter(PlInventory.id.in_(body.ids))
|
|
query = apply_pl_scope(query, PlInventory, user, db, tenant_no)
|
|
items = query.order_by(PlInventory.product_name).limit(EXPORT_MAX_ROWS + 1).all()
|
|
rows = []
|
|
for x in items:
|
|
data = InventoryRead.model_validate(x).model_dump(by_alias=True)
|
|
low = data.get("lowStockWarning")
|
|
high = data.get("highStockWarning")
|
|
data["warnThreshold"] = f"{low or '-'} ~ {high or '-'}"
|
|
stock = data.get("stockQty") or 0
|
|
try:
|
|
stock_f = float(stock)
|
|
low_f = float(low or 0)
|
|
except (TypeError, ValueError):
|
|
stock_f, low_f = 0.0, 0.0
|
|
data["status"] = "库存不足" if stock_f < low_f else "正常"
|
|
rows.append(data)
|
|
cols = filter_export_columns(LEDGER_EXPORT_COLUMNS, body.columns)
|
|
date_str = datetime.now().strftime("%Y%m%d")
|
|
return export_rows_to_xlsx_response(
|
|
db, tenant_no=tenant_no, sheet_title="库存台账", columns=cols, rows=rows,
|
|
filename_stem=f"库存台账_{date_str}",
|
|
)
|
|
except ExportValidationError as e:
|
|
raise HTTPException(status_code=400, detail=str(e)) from e
|
|
|
|
|
|
@router.post("/warning-threshold")
|
|
def set_warning_threshold(
|
|
payload: InventoryWarningThreshold,
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> dict:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
|
|
brand_id = payload.brand_id or ""
|
|
inventory = (
|
|
db.query(PlInventory)
|
|
.filter(
|
|
PlInventory.product_id == payload.product_id,
|
|
PlInventory.brand_id == brand_id,
|
|
PlInventory.tenant_no == tenant_no,
|
|
PlInventory.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
|
|
if inventory:
|
|
assert_pl_row_in_scope(db, inventory, PlInventory, user, tenant_no)
|
|
inventory.low_stock_warning = payload.low_stock_warning
|
|
inventory.high_stock_warning = payload.high_stock_warning
|
|
if payload.brand_name is not None:
|
|
inventory.brand_name = payload.brand_name
|
|
else:
|
|
# 创建库存记录
|
|
inventory = PlInventory(
|
|
id=generate_uuid(),
|
|
tenant_no=tenant_no,
|
|
product_id=payload.product_id,
|
|
brand_id=brand_id,
|
|
brand_name=payload.brand_name,
|
|
product_no="",
|
|
product_name="",
|
|
unit="吨",
|
|
stock_qty=0,
|
|
stock_amount=0,
|
|
low_stock_warning=payload.low_stock_warning,
|
|
high_stock_warning=payload.high_stock_warning,
|
|
create_by=user.id,
|
|
create_dept_id=user.dept_id,
|
|
)
|
|
db.add(inventory)
|
|
|
|
db.commit()
|
|
return {"success": True}
|
|
|
|
|
|
# =============================================================================
|
|
# 库存预警列表
|
|
# =============================================================================
|
|
|
|
|
|
@router.get("/warnings", response_model=list[InventoryWarningRead])
|
|
def list_warnings(
|
|
status: Optional[str] = Query(default=None, description="状态"),
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> list[InventoryWarningRead]:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
query = db.query(PlInventoryWarning).filter(
|
|
PlInventoryWarning.tenant_no == tenant_no,
|
|
PlInventoryWarning.del_flag == 0,
|
|
)
|
|
|
|
if status:
|
|
query = query.filter(PlInventoryWarning.status == status)
|
|
|
|
query = apply_pl_scope(query, PlInventoryWarning, user, db, tenant_no)
|
|
return query.order_by(PlInventoryWarning.create_time.desc()).all()
|
|
|
|
|
|
@router.put("/warnings/{id}", response_model=InventoryWarningRead)
|
|
def update_warning(
|
|
id: str,
|
|
payload: InventoryWarningUpdate,
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> InventoryWarningRead:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
|
|
warning = (
|
|
db.query(PlInventoryWarning)
|
|
.filter(
|
|
PlInventoryWarning.id == id,
|
|
PlInventoryWarning.tenant_no == tenant_no,
|
|
PlInventoryWarning.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
|
|
if not warning:
|
|
raise HTTPException(status_code=404, detail="预警不存在")
|
|
assert_pl_row_in_scope(db, warning, PlInventoryWarning, user, tenant_no)
|
|
|
|
if payload.status == "resolved":
|
|
warning.status = "resolved"
|
|
from datetime import datetime
|
|
|
|
warning.resolve_time = datetime.now()
|
|
warning.resolve_remark = payload.resolve_remark
|
|
|
|
db.commit()
|
|
db.refresh(warning)
|
|
return warning
|
|
|
|
|
|
# =============================================================================
|
|
# 出入库明细
|
|
# =============================================================================
|
|
|
|
|
|
@router.get("/transactions", response_model=InventoryTransactionResponse)
|
|
def list_transactions(
|
|
product_id: Optional[str] = Query(default=None, alias="productId", description="产品ID"),
|
|
transaction_type: Optional[str] = Query(default=None, alias="transactionType", description="事务类型"),
|
|
date_from: Optional[date] = Query(default=None, alias="dateFrom", description="开始日期"),
|
|
date_to: Optional[date] = Query(default=None, alias="dateTo", description="结束日期"),
|
|
page: int = Query(default=1, ge=1),
|
|
page_size: int = Query(default=20, ge=1, le=500, alias="pageSize"),
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> InventoryTransactionResponse:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
query = db.query(PlInventoryTransaction).filter(
|
|
PlInventoryTransaction.tenant_no == tenant_no,
|
|
PlInventoryTransaction.del_flag == 0,
|
|
)
|
|
|
|
if product_id:
|
|
query = query.filter(PlInventoryTransaction.product_id == product_id)
|
|
|
|
if transaction_type:
|
|
query = query.filter(PlInventoryTransaction.transaction_type == transaction_type)
|
|
|
|
if date_from:
|
|
query = query.filter(PlInventoryTransaction.create_time >= date_from)
|
|
|
|
if date_to:
|
|
query = query.filter(PlInventoryTransaction.create_time <= date_to)
|
|
|
|
query = apply_pl_scope(query, PlInventoryTransaction, user, db, tenant_no)
|
|
|
|
total = query.count()
|
|
offset = (page - 1) * page_size
|
|
items = query.order_by(PlInventoryTransaction.create_time.desc()).offset(offset).limit(page_size).all()
|
|
|
|
return InventoryTransactionResponse(total=total, list=items)
|
|
|
|
|
|
# =============================================================================
|
|
# 出入库汇总
|
|
# =============================================================================
|
|
|
|
|
|
@router.get("/summary", response_model=InventorySummaryResponse)
|
|
def get_summary(
|
|
group_by: str = Query(default="month", description="汇总维度 month/week"),
|
|
date_from: date = Query(..., alias="dateFrom", description="开始日期"),
|
|
date_to: date = Query(..., alias="dateTo", description="结束日期"),
|
|
product_id: Optional[str] = Query(default=None, alias="productId", description="产品ID"),
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> InventorySummaryResponse:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
|
|
query = db.query(PlInventoryTransaction).filter(
|
|
PlInventoryTransaction.tenant_no == tenant_no,
|
|
PlInventoryTransaction.del_flag == 0,
|
|
PlInventoryTransaction.create_time >= date_from,
|
|
PlInventoryTransaction.create_time <= date_to,
|
|
)
|
|
query = apply_pl_scope(query, PlInventoryTransaction, user, db, tenant_no)
|
|
|
|
if product_id:
|
|
query = query.filter(PlInventoryTransaction.product_id == product_id)
|
|
|
|
transactions = query.all()
|
|
|
|
# 按产品分组汇总
|
|
summary_map = {}
|
|
for t in transactions:
|
|
key = t.product_id
|
|
|
|
if key not in summary_map:
|
|
summary_map[key] = {
|
|
"product_id": t.product_id,
|
|
"product_name": t.product_name,
|
|
"inbound_qty": Decimal("0"),
|
|
"outbound_qty": Decimal("0"),
|
|
}
|
|
|
|
if t.transaction_type == "inbound":
|
|
summary_map[key]["inbound_qty"] += t.qty
|
|
else:
|
|
summary_map[key]["outbound_qty"] += abs(t.qty)
|
|
|
|
items = []
|
|
for key, data in summary_map.items():
|
|
net = data["inbound_qty"] - data["outbound_qty"]
|
|
items.append(InventorySummaryItem(
|
|
period="汇总",
|
|
product_id=data["product_id"],
|
|
product_name=data["product_name"],
|
|
inbound_qty=data["inbound_qty"],
|
|
outbound_qty=data["outbound_qty"],
|
|
net_change=net,
|
|
))
|
|
|
|
return InventorySummaryResponse(list=items)
|
|
|
|
|
|
# =============================================================================
|
|
# 证件到期提醒
|
|
# =============================================================================
|
|
|
|
|
|
@router.get("/document-reminders/stats", response_model=DocumentReminderStatsRead)
|
|
def document_reminder_stats(
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> DocumentReminderStatsRead:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
scan_cert_expiry(db, tenant_no)
|
|
db.commit()
|
|
|
|
query = db.query(PlDocumentReminder).filter(
|
|
PlDocumentReminder.tenant_no == tenant_no,
|
|
PlDocumentReminder.del_flag == 0,
|
|
PlDocumentReminder.status == "pending",
|
|
PlDocumentReminder.days_remaining <= 0,
|
|
)
|
|
query = apply_pl_scope(query, PlDocumentReminder, user, db, tenant_no)
|
|
expired_pending_count = query.count()
|
|
return DocumentReminderStatsRead(expired_pending_count=expired_pending_count)
|
|
|
|
|
|
@router.get("/document-reminders", response_model=list[DocumentReminderRead])
|
|
def list_document_reminders(
|
|
status: Optional[str] = Query(default=None, description="状态"),
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> list[DocumentReminderRead]:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
scan_cert_expiry(db, tenant_no)
|
|
db.commit()
|
|
|
|
query = db.query(PlDocumentReminder).filter(
|
|
PlDocumentReminder.tenant_no == tenant_no,
|
|
PlDocumentReminder.del_flag == 0,
|
|
)
|
|
|
|
if status:
|
|
query = query.filter(PlDocumentReminder.status == status)
|
|
|
|
query = apply_pl_scope(query, PlDocumentReminder, user, db, tenant_no)
|
|
return query.order_by(PlDocumentReminder.expire_date.asc()).all()
|
|
|
|
|
|
@router.put("/document-reminders/{id}", response_model=DocumentReminderRead)
|
|
def update_document_reminder(
|
|
id: str,
|
|
payload: DocumentReminderUpdate,
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> DocumentReminderRead:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
|
|
reminder = (
|
|
db.query(PlDocumentReminder)
|
|
.filter(
|
|
PlDocumentReminder.id == id,
|
|
PlDocumentReminder.tenant_no == tenant_no,
|
|
PlDocumentReminder.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
|
|
if not reminder:
|
|
raise HTTPException(status_code=404, detail="提醒不存在")
|
|
assert_pl_row_in_scope(db, reminder, PlDocumentReminder, user, tenant_no)
|
|
|
|
reminder.status = payload.status
|
|
if payload.remark:
|
|
reminder.remark = payload.remark
|
|
|
|
db.commit()
|
|
db.refresh(reminder)
|
|
return reminder
|
|
|
|
|
|
# =============================================================================
|
|
# 生成证件到期提醒（定时任务调用）
|
|
# =============================================================================
|
|
|
|
|
|
@router.post("/check-document-expiry")
|
|
def check_document_expiry(db: Session = Depends(get_db)) -> dict:
|
|
"""检查证件到期，生成提醒（供定时任务调用）"""
|
|
created_count = scan_cert_expiry(db)
|
|
db.commit()
|
|
return {"created_count": created_count}
|