shanxia2.0/backend/app/api/pl/transport_address.py

"""运输通用地址库 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()