331 lines
12 KiB
Python
331 lines
12 KiB
Python
"""运输通用地址库 API"""
|
|
from datetime import datetime
|
|
from typing import Any, List, Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
|
|
from pydantic import BaseModel
|
|
from sqlalchemy import case, func, or_
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.utils.excel_export import (
|
|
EXPORT_MAX_ROWS,
|
|
ExportColumn,
|
|
ExportValidationError,
|
|
export_rows_to_xlsx_response,
|
|
filter_export_columns,
|
|
)
|
|
|
|
from app.core.database import get_db
|
|
from app.core.db_auth import DbUserContext, get_db_current_user
|
|
from app.core.delete_guard import assert_row_deletable_by_creator, user_info_from_context
|
|
from app.core.tenant_context import get_tenant_no
|
|
from app.core.utils import generate_uuid
|
|
from app.models.pl_models import PlTransportAddress
|
|
from app.schemas.pl.transport_address import (
|
|
TransportAddressCreate,
|
|
TransportAddressListResponse,
|
|
TransportAddressRead,
|
|
TransportAddressUpdate,
|
|
)
|
|
from app.services.pl.transport_address_list_filters import (
|
|
build_transport_address_column_filters,
|
|
)
|
|
|
|
router = APIRouter(tags=["运输地址库"])
|
|
|
|
|
|
def _merge_query(request: Request, **known) -> dict:
|
|
out = {k: v for k, v in known.items() if v not in (None, "")}
|
|
for k, v in request.query_params.multi_items():
|
|
if k in ("page", "pageSize"):
|
|
continue
|
|
if k not in out and v not in (None, ""):
|
|
out[k] = v
|
|
return out
|
|
|
|
|
|
def _param_get(params: dict, *keys: str) -> Any:
|
|
for key in keys:
|
|
if key in params and params[key] not in (None, ""):
|
|
return params[key]
|
|
return None
|
|
|
|
|
|
def _body_to_params(body: BaseModel) -> dict:
|
|
data = body.model_dump(by_alias=True, exclude_none=True)
|
|
for key in ("ids", "columns"):
|
|
data.pop(key, None)
|
|
extra = getattr(body, "model_extra", None) or {}
|
|
for key, value in extra.items():
|
|
if key not in data and value not in (None, ""):
|
|
data[key] = value
|
|
return data
|
|
|
|
|
|
def _address_to_read(row: PlTransportAddress) -> TransportAddressRead:
|
|
return TransportAddressRead.model_validate(row)
|
|
|
|
|
|
def _get_address(
|
|
db: Session, tenant_no: str, address_id: str, user: DbUserContext
|
|
) -> PlTransportAddress:
|
|
row = (
|
|
db.query(PlTransportAddress)
|
|
.filter(
|
|
PlTransportAddress.id == address_id,
|
|
PlTransportAddress.tenant_no == tenant_no,
|
|
PlTransportAddress.del_flag == 0,
|
|
)
|
|
.first()
|
|
)
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="地址不存在")
|
|
return row
|
|
|
|
|
|
def _build_address_query(db: Session, tenant_no: str, user: DbUserContext):
|
|
# 运输地址为主数据选项，不做 create_by 数据范围过滤
|
|
_ = user
|
|
query = db.query(PlTransportAddress).filter(
|
|
PlTransportAddress.tenant_no == tenant_no,
|
|
PlTransportAddress.del_flag == 0,
|
|
)
|
|
return query
|
|
|
|
|
|
@router.get("/address", response_model=TransportAddressListResponse)
|
|
def list_transport_addresses(
|
|
request: Request,
|
|
q: Optional[str] = Query(default=None, description="搜索名称或地址"),
|
|
status: Optional[str] = Query(default=None, description="状态 1-启用 0-停用"),
|
|
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),
|
|
) -> TransportAddressListResponse:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
params = _merge_query(request, q=q, status=status)
|
|
query = _build_address_query(db, tenant_no, user)
|
|
|
|
qv = _param_get(params, "q")
|
|
if qv:
|
|
keyword = str(qv).strip().lower()
|
|
query = query.filter(
|
|
or_(
|
|
func.lower(PlTransportAddress.address_name).contains(keyword),
|
|
func.lower(PlTransportAddress.full_address).contains(keyword),
|
|
func.lower(PlTransportAddress.contact_name).contains(keyword),
|
|
PlTransportAddress.contact_phone.contains(str(qv).strip()),
|
|
)
|
|
)
|
|
|
|
column_filters = build_transport_address_column_filters(
|
|
address_name=_param_get(params, "addressName", "address_name"),
|
|
full_address=_param_get(params, "fullAddress", "full_address"),
|
|
status=_param_get(params, "status"),
|
|
status_in=_param_get(params, "statusIn", "status_in"),
|
|
)
|
|
if column_filters:
|
|
query = query.filter(*column_filters)
|
|
|
|
query = query.order_by(
|
|
case((PlTransportAddress.order_num.is_(None), 1), else_=0).asc(),
|
|
PlTransportAddress.create_time.desc(),
|
|
)
|
|
|
|
total = query.count()
|
|
offset = (page - 1) * page_size
|
|
rows = query.offset(offset).limit(page_size).all()
|
|
return TransportAddressListResponse(
|
|
total=total,
|
|
list=[_address_to_read(row) for row in rows],
|
|
)
|
|
|
|
|
|
@router.get("/address/all", response_model=TransportAddressListResponse)
|
|
def list_all_transport_addresses(
|
|
q: Optional[str] = Query(default=None, description="模糊检索：仅地址名称、详细地址"),
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> TransportAddressListResponse:
|
|
"""选择器用：仅返回启用中的地址（不分页）。q 仅匹配 address_name、full_address。"""
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
query = _build_address_query(db, tenant_no, user).filter(PlTransportAddress.status == "1")
|
|
|
|
if q:
|
|
keyword = q.strip().lower()
|
|
query = query.filter(
|
|
or_(
|
|
func.lower(PlTransportAddress.address_name).contains(keyword),
|
|
func.lower(PlTransportAddress.full_address).contains(keyword),
|
|
)
|
|
)
|
|
|
|
query = query.order_by(
|
|
case((PlTransportAddress.order_num.is_(None), 1), else_=0).asc(),
|
|
PlTransportAddress.create_time.desc(),
|
|
)
|
|
rows = query.limit(200).all()
|
|
return TransportAddressListResponse(total=len(rows), list=[_address_to_read(row) for row in rows])
|
|
|
|
|
|
|
|
ADDRESS_EXPORT_COLUMNS = [
|
|
ExportColumn(title="地址名称", field="addressName"),
|
|
ExportColumn(title="详细地址", field="fullAddress"),
|
|
ExportColumn(title="联系人", field="contactName"),
|
|
ExportColumn(title="联系电话", field="contactPhone"),
|
|
ExportColumn(title="注意事项", field="notes"),
|
|
ExportColumn(title="状态", field="statusLabel"),
|
|
ExportColumn(title="备注", field="remark"),
|
|
]
|
|
|
|
|
|
class TransportAddressExportBody(BaseModel):
|
|
q: Optional[str] = None
|
|
status: Optional[str] = None
|
|
ids: Optional[List[str]] = None
|
|
columns: Optional[List[str]] = None
|
|
|
|
model_config = {"extra": "allow"}
|
|
|
|
|
|
@router.post("/address/export")
|
|
def export_transport_addresses(
|
|
body: TransportAddressExportBody,
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
):
|
|
"""导出运输地址列表（当前筛选全量，或 ids 已选）。"""
|
|
try:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
params = _body_to_params(body)
|
|
query = _build_address_query(db, tenant_no, user)
|
|
qv = _param_get(params, "q")
|
|
if qv:
|
|
keyword = str(qv).strip().lower()
|
|
query = query.filter(
|
|
or_(
|
|
func.lower(PlTransportAddress.address_name).contains(keyword),
|
|
func.lower(PlTransportAddress.full_address).contains(keyword),
|
|
func.lower(PlTransportAddress.contact_name).contains(keyword),
|
|
PlTransportAddress.contact_phone.contains(str(qv).strip()),
|
|
)
|
|
)
|
|
column_filters = build_transport_address_column_filters(
|
|
address_name=_param_get(params, "addressName", "address_name"),
|
|
full_address=_param_get(params, "fullAddress", "full_address"),
|
|
status=_param_get(params, "status"),
|
|
status_in=_param_get(params, "statusIn", "status_in"),
|
|
)
|
|
if column_filters:
|
|
query = query.filter(*column_filters)
|
|
if body.ids:
|
|
query = query.filter(PlTransportAddress.id.in_(body.ids))
|
|
query = query.order_by(
|
|
case((PlTransportAddress.order_num.is_(None), 1), else_=0).asc(),
|
|
PlTransportAddress.create_time.desc(),
|
|
)
|
|
rows = query.limit(EXPORT_MAX_ROWS + 1).all()
|
|
items = []
|
|
for row in rows:
|
|
data = _address_to_read(row).model_dump(by_alias=True)
|
|
st = str(data.get("status") or "")
|
|
data["statusLabel"] = "启用" if st == "1" else ("停用" if st == "0" else st)
|
|
items.append(data)
|
|
requested = None
|
|
if body.columns:
|
|
requested = []
|
|
for key in body.columns:
|
|
if key == "status":
|
|
requested.append("statusLabel")
|
|
else:
|
|
requested.append(key)
|
|
cols = filter_export_columns(ADDRESS_EXPORT_COLUMNS, requested)
|
|
date_str = datetime.now().strftime("%Y%m%d")
|
|
return export_rows_to_xlsx_response(
|
|
db,
|
|
tenant_no=tenant_no,
|
|
sheet_title="运输地址",
|
|
columns=cols,
|
|
rows=items,
|
|
filename_stem=f"运输地址_{date_str}",
|
|
)
|
|
except ExportValidationError as e:
|
|
raise HTTPException(status_code=400, detail=str(e)) from e
|
|
|
|
|
|
|
|
@router.post("/address", response_model=TransportAddressRead, status_code=status.HTTP_201_CREATED)
|
|
def create_transport_address(
|
|
payload: TransportAddressCreate,
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> TransportAddressRead:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
row = PlTransportAddress(
|
|
id=generate_uuid(),
|
|
tenant_no=tenant_no,
|
|
address_name=payload.address_name.strip(),
|
|
full_address=payload.full_address.strip(),
|
|
lng=payload.lng,
|
|
lat=payload.lat,
|
|
contact_name=payload.contact_name,
|
|
contact_phone=payload.contact_phone,
|
|
notes=payload.notes,
|
|
status=payload.status or "1",
|
|
remark=payload.remark,
|
|
create_by=user.id,
|
|
create_dept_id=user.dept_id,
|
|
)
|
|
db.add(row)
|
|
db.commit()
|
|
db.refresh(row)
|
|
return _address_to_read(row)
|
|
|
|
|
|
@router.get("/address/{id}", response_model=TransportAddressRead)
|
|
def get_transport_address(
|
|
id: str,
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> TransportAddressRead:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
row = _get_address(db, tenant_no, id, user)
|
|
return _address_to_read(row)
|
|
|
|
|
|
@router.put("/address/{id}", response_model=TransportAddressRead)
|
|
def update_transport_address(
|
|
id: str,
|
|
payload: TransportAddressUpdate,
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> TransportAddressRead:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
row = _get_address(db, tenant_no, id, user)
|
|
|
|
for key, value in payload.model_dump(exclude_unset=True).items():
|
|
if key in ("address_name", "full_address") and isinstance(value, str):
|
|
value = value.strip()
|
|
setattr(row, key, value)
|
|
|
|
row.update_by = user.id
|
|
db.commit()
|
|
db.refresh(row)
|
|
return _address_to_read(row)
|
|
|
|
|
|
@router.delete("/address/{id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_transport_address(
|
|
id: str,
|
|
db: Session = Depends(get_db),
|
|
user: DbUserContext = Depends(get_db_current_user),
|
|
) -> None:
|
|
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
|
|
row = _get_address(db, tenant_no, id, user)
|
|
assert_row_deletable_by_creator(row, user_info_from_context(user))
|
|
row.del_flag = 1
|
|
row.update_by = user.id
|
|
db.commit()
|