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

"""运输通用地址库 API"""
from datetime import datetime
from typing import List, Optional

from fastapi import APIRouter, Depends, HTTPException, Query, 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,
)

router = APIRouter(tags=["运输地址库"])


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(
    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())
    query = _build_address_query(db, tenant_no, user)

    if status:
        query = query.filter(PlTransportAddress.status == status)

    if q:
        keyword = q.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(q.strip()),
            )
        )

    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


@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())
        query = _build_address_query(db, tenant_no, user)
        if body.status:
            query = query.filter(PlTransportAddress.status == body.status)
        if body.q:
            keyword = body.q.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(body.q.strip()),
                )
            )
        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()