Files
shanxia2.0/backend/app/api/pl/plan.py
T

3381 lines
125 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""运输计划管理 - 计划 API"""
import re
from collections import defaultdict
from datetime import date, datetime
from typing import Any, List, Optional, Sequence
from fastapi import APIRouter, Body, Depends, HTTPException, Query, Request, status
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import func, or_
from sqlalchemy.orm import Session
from app.core.config import settings
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.menu_guard import SUPER_ADMIN_ROLE, get_user_menu_codes
from app.services.pl_data_scope import apply_pl_scope, assert_pl_row_in_scope
from app.services.pl.list_sort import order_plan, parse_sort_params
from app.services.pl_plan_arrange_copy_flags import load_plan_arrange_copy_flags
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.crm_models import CrmContract, CrmContractProduct, CrmProduct
from app.models.db_models import DbInboundOutbound
from app.models.pl_models import (
PlCarrier,
PlDriver,
PlPlan,
PlPlanContract,
PlPlanDelivery,
PlPlanLoad,
PlPlanProduct,
PlPlanTransport,
PlTransport,
PlTransportDbLink,
PlVehicle,
)
from app.schemas.pl import PlanProductCreate
from app.schemas.pl.cancel_reason import CancelReasonBody, normalize_optional_reason
from app.schemas.pl.plan import (
CreateFromContractRequest,
DeliverySlipPrintRead,
PlanArchiveBody,
PlanArchiveBrandQtyItem,
PlanContractGapListResponse,
PlanContractGapRead,
PlanContractLink,
PlanCreate,
PlanCreateFull,
PlanDriverFeedbackUpdateBody,
HazardCertReadyBody,
ErpEnteredBody,
PlanForceFollowStatusBody,
PlanInboundSlipTemplateSave,
PlanDeliverySlipTemplateSave,
PlanSubpackSlipTemplateSave,
PlanDeliveryBody,
PlanLoadBody,
PlanListQuery,
PlanListResponse,
PlanProductCertsUpdate,
PlanProductRead,
PlanRead,
PlanSideContractsUpdateBody,
PlanSiteFeedbackUpdateBody,
PlanStopContractsUpdateBody,
PlanUpdate,
PlanWithProducts,
)
from app.services.pl.plan_column_facet_context import strip_plan_facet_self_params
from app.services.pl_plan_contract_gap import (
apply_contract_gap_keyword_filter,
apply_contract_gap_type_filter,
list_plan_contract_gaps,
plan_has_contract_gap_exists,
update_plan_side_contracts,
update_plan_stop_contracts,
)
from app.services.pl_brand_archive import load_brand_qtys_by_plan_product_ids
from app.services.pl_product_display import (
load_crm_product_alias_maps,
resolve_product_display_names,
)
from app.schemas.pl.transport import (
PlanArrangeInfoRead,
TransportInventoryLinkRead,
TransportIoOrderBrief,
TransportSiteLinkRead,
)
from app.services.pl_dibang_bridge_service import (
StopDibangOption,
revoke_regeneratable_dibang_for_transport,
sync_linked_dibang_dates_for_plan,
upsert_stop_dibang_links,
)
from app.services.pl_inventory_link_read import build_transport_inventory_link_read
from app.services.pl_io_order_service import get_active_io_order_for_plan, void_io_orders_for_plan
from app.services.pl_internal_customer import plan_requires_site_link
from app.services.pl_plan_site_link_service import build_site_records_for_arranged_plan
from app.services.pl_transport_site_service import revoke_site_links_for_transport
from app.services.pl_inventory_service import (
apply_archive_inventory_for_plan,
apply_archive_inventory_for_plan_stops,
)
from app.services.pl_plan_archive_stop_service import (
apply_archive_stop_quantities,
sum_delivery_outbound_qty,
sync_plan_products_from_stop_archive,
)
from app.services.pl_plan_archive_export import build_plan_archive_export_rows
from app.services.pl_plan_price_mask import (
mask_plan_with_products_prices,
user_can_archive_plan,
user_can_view_archive_prices,
)
from app.services.pl_plan_list_attachments import load_plan_document_attachments_by_plan_ids
from app.services.pl_plan_print_service import get_delivery_slip_print_context
from app.services.pl_plan_stop_service import (
attach_plan_stops_to_item,
collect_stop_qty_warnings,
collect_unique_delivery_date_strings,
load_plan_stops_grouped,
replace_plan_stops,
sync_plan_legacy_from_stops,
uses_multi_stop_payload,
validate_plan_stops_write,
)
from app.services.pl_customer_stop_pref_service import safe_upsert_prefs_from_plan_stops
from app.services.pl.plan_list_filters import (
apply_stock_role_filter,
build_plan_column_filters,
parse_single_stop_kind,
parse_stop_customer_keywords,
plan_plate_match,
plan_product_name_exists,
)
from app.services.pl_plan_service import (
apply_archive_brand_qtys_only,
apply_archive_freight,
apply_archive_product_quantities,
apply_archive_stop_feedback_attachments,
apply_plan_driver_feedback,
apply_plan_site_feedback,
apply_vehicle_prefill_to_plan,
assert_contract_selectable_for_plan,
apply_plan_product_fields,
build_plan_product,
resolve_plan_product_id_for_upsert,
calc_plan_total_amount,
clear_draft_transport_for_plan,
clear_hazard_cert_ready,
collect_hazard_product_ids,
determine_plan_type,
fetch_hazard_flags_map,
load_plan_contracts_grouped,
plan_has_hazard_product_exists,
record_plan_change,
replace_plan_contracts,
resolve_hazard_flags_from_crm,
set_hazard_cert_ready,
set_plan_erp_entered,
sync_transport_prefill_for_plan,
update_plan_product_certs,
assert_can_save_as_draft,
resolve_follow_status_on_save,
validate_create_full,
validate_draft_min_fields,
validate_hazard_cert_ready_for_transport,
validate_plan_party_contracts,
)
router = APIRouter(prefix="/plan", tags=["运输计划-计划管理"])
_PLAN_NO_SEQ_WIDTH = 3
def _resolve_list_product_display(
product: PlPlanProduct,
*,
standard_by_id: dict[str, str],
alias_options_by_id: dict[str, list[str]],
) -> tuple[str, Optional[str]]:
"""列表展示用:优先 CRM 标准名,并返回反推别名(详情仍返回库内 product_name)。"""
display_name, alias = resolve_product_display_names(
saved_product_name=product.product_name,
product_id=product.product_id,
standard_by_id=standard_by_id,
alias_options_by_id=alias_options_by_id,
)
return display_name or (product.product_name or ""), alias
def _resolve_create_date(value: Optional[date]) -> date:
return value or date.today()
def _generate_plan_no(db: Session) -> str:
"""生成计划编号 POYYYYMMDD###(当日 3 位序号,每日重置)"""
today = date.today()
prefix = f"PO{today.strftime('%Y%m%d')}"
# uk_plan_no 对全表生效(含软删),取号时不能过滤 del_flag
rows = (
db.query(PlPlan.plan_no)
.filter(PlPlan.plan_no.like(f"{prefix}%"))
.all()
)
max_seq = 0
for (plan_no,) in rows:
suffix = (plan_no or "")[len(prefix) :]
if suffix.isdigit():
max_seq = max(max_seq, int(suffix))
seq = max_seq + 1
return f"{prefix}{seq:0{_PLAN_NO_SEQ_WIDTH}d}"
_SUBPACK_SLIP_NO_SEQ_WIDTH = 3
_SUBPACK_SLIP_NO_CANONICAL_PATTERN = re.compile(r"^NO:SXFZ(\d{8})-(\d+)$")
_SUBPACK_SLIP_NO_BARE_PATTERN = re.compile(r"^SXFZ(\d{8})-(\d+)$")
_SUBPACK_SLIP_NO_LEGACY_PATTERNS = (
re.compile(r"^NO:SXFZ-(\d{4})-(\d{2})-(\d{2})-(\d+)$"),
re.compile(r"^SXFZ-(\d{4})-(\d{2})-(\d{2})-(\d+)$"),
)
def _format_subpack_slip_no(day: str, seq: int) -> str:
return f"NO:SXFZ{day}-{int(seq):0{_SUBPACK_SLIP_NO_SEQ_WIDTH}d}"
def _normalize_subpack_slip_no(slip_no: str) -> Optional[str]:
"""将历史分装单号统一为 NO:SXFZYYYYMMDD-###"""
raw = (slip_no or "").strip()
if not raw:
return None
matched = _SUBPACK_SLIP_NO_CANONICAL_PATTERN.match(raw)
if matched:
day, seq = matched.groups()
return _format_subpack_slip_no(day, int(seq))
matched = _SUBPACK_SLIP_NO_BARE_PATTERN.match(raw)
if matched:
day, seq = matched.groups()
return _format_subpack_slip_no(day, int(seq))
for pattern in _SUBPACK_SLIP_NO_LEGACY_PATTERNS:
matched = pattern.match(raw)
if matched:
year, month, day, seq = matched.groups()
return _format_subpack_slip_no(f"{year}{month}{day}", int(seq))
return None
def _subpack_slip_no_seq(slip_no: str, day_prefix: str) -> int:
normalized = _normalize_subpack_slip_no(slip_no)
if not normalized or not normalized.startswith(day_prefix):
return 0
suffix = normalized[len(day_prefix) :]
return int(suffix) if suffix.isdigit() else 0
def _ensure_subpack_slip_no(db: Session, tenant_no: str, plan: PlPlan) -> str:
"""分配分装单号 NO:SXFZYYYYMMDD-###(按租户每日递增)"""
existing = (plan.subpack_slip_no or "").strip()
if existing:
normalized = _normalize_subpack_slip_no(existing)
if normalized:
if normalized != existing:
plan.subpack_slip_no = normalized
return normalized
today = date.today()
prefix = f"NO:SXFZ{today.strftime('%Y%m%d')}-"
rows = (
db.query(PlPlan.subpack_slip_no)
.filter(
PlPlan.tenant_no == tenant_no,
PlPlan.subpack_slip_no.isnot(None),
PlPlan.subpack_slip_no != "",
)
.all()
)
max_seq = 0
for (slip_no,) in rows:
max_seq = max(max_seq, _subpack_slip_no_seq(slip_no or "", prefix))
next_no = f"{prefix}{max_seq + 1:0{_SUBPACK_SLIP_NO_SEQ_WIDTH}d}"
plan.subpack_slip_no = next_no
return next_no
def _determine_plan_type(supplier_name: Optional[str], destination_name: Optional[str]) -> str:
"""根据供应商和目的地自动判断单据类型"""
return determine_plan_type(supplier_name, destination_name)
def _get_plan_type_label(plan_type: str) -> str:
labels = {
"direct": "直销单",
"inbound": "入库单",
"outbound": "出库单",
}
return labels.get(plan_type, plan_type)
def _generate_notify_message(plan: PlPlan, products: list) -> str:
"""生成通知信息"""
product_names = "、".join([p.product_name for p in products[:3]])
if len(products) > 3:
product_names += "等"
return f"""计划号:{plan.plan_no}
企业名称:{plan.customer_name or ''}
提货单位:{plan.destination_name or ''}
到货时间:{plan.contract_end_date or ''}
药品名称:{product_names}
到货吨位:{sum(float(p.inbound_estimate_qty or p.outbound_estimate_qty or 0) for p in products)}吨
备注:"""
# =============================================================================
# 计划列表
# =============================================================================
def _parse_status_csv(value: Optional[str]) -> List[str]:
if not value:
return []
return [item.strip() for item in value.split(",") if item.strip()]
def _apply_follow_status_filter(query, follow_status: Optional[str]):
statuses = _parse_status_csv(follow_status)
if len(statuses) == 1:
return query.filter(PlPlan.follow_status == statuses[0])
if len(statuses) > 1:
return query.filter(PlPlan.follow_status.in_(statuses))
return query
def _apply_exclude_follow_status_filter(query, exclude_follow_status: Optional[str]):
statuses = _parse_status_csv(exclude_follow_status)
if len(statuses) == 1:
return query.filter(PlPlan.follow_status != statuses[0])
if len(statuses) > 1:
return query.filter(~PlPlan.follow_status.in_(statuses))
return query
def _merge_query(request: Request, **known) -> dict:
out = {k: v for k, v in known.items() if v is not None and v != ""}
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 _filters_from_params(params: dict, *, db: Session, tenant_no: str) -> list:
return build_plan_column_filters(
db=db,
tenant_no=tenant_no,
plan_no=_param_get(params, "planNo", "plan_no"),
plan_no_in=_param_get(params, "planNoIn", "plan_no_in"),
belongs_company=_param_get(params, "belongsCompany", "belongs_company"),
belongs_company_in=_param_get(
params, "belongsCompanyIn", "belongs_company_in"
),
plan_type=_param_get(params, "planType", "plan_type"),
plan_type_in=_param_get(params, "planTypeIn", "plan_type_in"),
transport_status=_param_get(params, "transportStatus", "transport_status"),
transport_status_in=_param_get(params, "transportStatusIn", "transport_status_in"),
archive_status=_param_get(params, "archiveStatus", "archive_status"),
archive_status_in=_param_get(params, "archiveStatusIn", "archive_status_in"),
customer_name=_param_get(params, "customerName", "customer_name"),
product_name=_param_get(params, "productName", "product_name"),
product_name_in=_param_get(params, "productNameIn", "product_name_in"),
package_spec=_param_get(params, "packageSpec", "package_spec"),
package_spec_in=_param_get(params, "packageSpecIn", "package_spec_in"),
license_plate=_param_get(params, "licensePlate", "license_plate"),
license_plate_in=_param_get(params, "licensePlateIn", "license_plate_in"),
stop_kind=_param_get(params, "stopKind", "stop_kind"),
stop_kind_in=_param_get(params, "stopKindIn", "stop_kind_in"),
stop_customers=_param_get(params, "stopCustomers", "stop_customers"),
load_customers_in=_param_get(params, "loadCustomersIn", "load_customers_in"),
delivery_customers_in=_param_get(params, "deliveryCustomersIn", "delivery_customers_in"),
plan_date_from=_param_get(params, "planDateFrom", "plan_date_from"),
plan_date_to=_param_get(params, "planDateTo", "plan_date_to"),
delivery_time_from=_param_get(params, "deliveryTimeFrom", "delivery_time_from"),
delivery_time_to=_param_get(params, "deliveryTimeTo", "delivery_time_to"),
cert_upload_status=_param_get(params, "certUploadStatus", "cert_upload_status"),
hazard_cert_ready=_param_get(params, "hazardCertReady", "hazard_cert_ready"),
hazard_cert_ready_in=_param_get(
params, "hazardCertReadyIn", "hazard_cert_ready_in"
),
supplier_name=_param_get(params, "supplierName", "supplier_name"),
destination_name=_param_get(params, "destinationName", "destination_name"),
erp_entered=_param_get(params, "erpEntered", "erp_entered"),
erp_entered_in=_param_get(params, "erpEnteredIn", "erp_entered_in"),
carrier_name=_param_get(params, "carrierName", "carrier_name"),
carrier_name_in=_param_get(params, "carrierNameIn", "carrier_name_in"),
personnel_summary=_param_get(params, "personnelSummary", "personnel_summary"),
follow_user_name=_param_get(params, "followUserName", "follow_user_name"),
stop_address=_param_get(params, "stopAddress", "stop_address"),
contract_no=_param_get(params, "contractNo", "contract_no"),
contract_no_in=_param_get(params, "contractNoIn", "contract_no_in"),
product_spec=_param_get(params, "productSpec", "product_spec"),
product_spec_in=_param_get(params, "productSpecIn", "product_spec_in"),
product_unit=_param_get(params, "productUnit", "product_unit"),
product_unit_in=_param_get(params, "productUnitIn", "product_unit_in"),
product_plan_qty_op=_param_get(params, "productPlanQtyOp", "product_plan_qty_op"),
product_plan_qty=_param_get(params, "productPlanQty", "product_plan_qty"),
product_plan_qty_min=_param_get(
params, "productPlanQtyMin", "product_plan_qty_min"
),
product_plan_qty_max=_param_get(
params, "productPlanQtyMax", "product_plan_qty_max"
),
product_actual_qty_op=_param_get(
params, "productActualQtyOp", "product_actual_qty_op"
),
product_actual_qty=_param_get(params, "productActualQty", "product_actual_qty"),
product_actual_qty_min=_param_get(
params, "productActualQtyMin", "product_actual_qty_min"
),
product_actual_qty_max=_param_get(
params, "productActualQtyMax", "product_actual_qty_max"
),
product_inbound_qty_op=_param_get(
params, "productInboundQtyOp", "product_inbound_qty_op"
),
product_inbound_qty=_param_get(
params, "productInboundQty", "product_inbound_qty"
),
product_inbound_qty_min=_param_get(
params, "productInboundQtyMin", "product_inbound_qty_min"
),
product_inbound_qty_max=_param_get(
params, "productInboundQtyMax", "product_inbound_qty_max"
),
product_outbound_qty_op=_param_get(
params, "productOutboundQtyOp", "product_outbound_qty_op"
),
product_outbound_qty=_param_get(
params, "productOutboundQty", "product_outbound_qty"
),
product_outbound_qty_min=_param_get(
params, "productOutboundQtyMin", "product_outbound_qty_min"
),
product_outbound_qty_max=_param_get(
params, "productOutboundQtyMax", "product_outbound_qty_max"
),
product_unit_price_op=_param_get(
params, "productUnitPriceOp", "product_unit_price_op"
),
product_unit_price=_param_get(params, "productUnitPrice", "product_unit_price"),
product_unit_price_min=_param_get(
params, "productUnitPriceMin", "product_unit_price_min"
),
product_unit_price_max=_param_get(
params, "productUnitPriceMax", "product_unit_price_max"
),
product_amount_op=_param_get(params, "productAmountOp", "product_amount_op"),
product_amount=_param_get(params, "productAmount", "product_amount"),
product_amount_min=_param_get(params, "productAmountMin", "product_amount_min"),
product_amount_max=_param_get(params, "productAmountMax", "product_amount_max"),
stop_dibang=_param_get(params, "stopDibang", "stop_dibang"),
stop_dibang_in=_param_get(params, "stopDibangIn", "stop_dibang_in"),
)
def _body_to_params(body: BaseModel) -> dict:
data = body.model_dump(by_alias=True, exclude_none=True)
for key in ("ids", "columns", "exportFormat", "export_format"):
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
class PlanColumnFacetBody(BaseModel):
"""运输计划列头可选值(去重)。额外字段与列表过滤同名,用于 Excel 级联上下文。"""
model_config = ConfigDict(extra="allow")
field: str = Field(
...,
description="planNo | licensePlate | carrierName | loadCustomers | deliveryCustomers | contractNo | productName | productSpec | packageSpec | productUnit",
)
keyword: Optional[str] = Field(default=None, description="可选值关键字")
limit: Optional[int] = Field(default=500, ge=1, le=500, description="返回条数上限")
def _truthy_param(value: Any) -> bool:
if value is True:
return True
if value in (1, "1", "true", "True", "yes", "YES"):
return True
return False
def _build_plan_facet_base_query(
db: Session,
*,
tenant_no: str,
user: DbUserContext,
params: dict,
):
"""与列表同源的计划上下文查询(已排除当前 facet 列自身条件)。"""
query = db.query(PlPlan).filter(
PlPlan.tenant_no == tenant_no,
PlPlan.del_flag == 0,
)
contract_gap_only = _truthy_param(
_param_get(params, "contractGapOnly", "contract_gap_only")
)
q_kw = (_param_get(params, "q") or "")
if isinstance(q_kw, str):
q_kw = q_kw.strip()
else:
q_kw = str(q_kw).strip() if q_kw else ""
if contract_gap_only:
query = apply_contract_gap_keyword_filter(query, tenant_no, q_kw)
elif q_kw:
q_lower = q_kw.lower()
query = query.filter(
or_(
PlPlan.plan_no.contains(q_kw),
func.lower(PlPlan.supplier_name).contains(q_lower),
func.lower(PlPlan.customer_name).contains(q_lower),
func.lower(PlPlan.destination_name).contains(q_lower),
plan_product_name_exists(db, tenant_no, q_kw),
plan_plate_match(q_kw),
)
)
follow_status = _param_get(params, "followStatus", "follow_status")
query = _apply_follow_status_filter(query, follow_status)
exclude_follow_status = _param_get(
params, "excludeFollowStatus", "exclude_follow_status"
)
query = _apply_exclude_follow_status_filter(query, exclude_follow_status)
has_hazard = _param_get(params, "hasHazardProduct", "has_hazard_product")
if _truthy_param(has_hazard):
query = query.filter(plan_has_hazard_product_exists(tenant_no))
if contract_gap_only:
query = query.filter(
PlPlan.follow_status != "cancelled",
plan_has_contract_gap_exists(tenant_no),
)
gap_type = _param_get(params, "gapType", "gap_type")
if gap_type:
query = apply_contract_gap_type_filter(query, tenant_no, str(gap_type))
customer_keyword = _param_get(params, "customerKeyword", "customer_keyword")
if contract_gap_only:
query = apply_contract_gap_keyword_filter(
query, tenant_no, str(customer_keyword or ""), include_plan_no=False
)
elif customer_keyword and str(customer_keyword).strip():
kw = str(customer_keyword).strip()
kw_lower = kw.lower()
query = query.filter(
or_(
func.lower(PlPlan.supplier_name).contains(kw_lower),
func.lower(PlPlan.customer_name).contains(kw_lower),
)
)
column_filters = _filters_from_params(params, db=db, tenant_no=tenant_no)
if column_filters:
query = query.filter(*column_filters)
stock_role = _param_get(params, "stockRole", "stock_role")
try:
query = apply_stock_role_filter(
query, stock_role=stock_role, tenant_no=tenant_no
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
query = apply_pl_scope(query, PlPlan, user, db, tenant_no)
return query
@router.post("/column-facet-values")
def plan_column_facet_values(
body: PlanColumnFacetBody,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
):
"""运输计划列头可选值:在列表上下文(排除本列)下去重,对齐 Excel 级联。"""
from sqlalchemy import func as sa_func
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
lim = min(max(1, int(body.limit or 500)), 500)
kw = (body.keyword or "").strip()
field = str(body.field or "").strip()
raw_params = _body_to_params(body)
params = strip_plan_facet_self_params(raw_params, field)
base = _build_plan_facet_base_query(
db, tenant_no=tenant_no, user=user, params=params
)
plan_id_sq = base.with_entities(PlPlan.id).subquery()
def _finish(rows):
out = []
for row in rows:
value = str(row.value).strip() if row.value is not None else ""
if not value:
continue
out.append({"value": value, "label": value, "count": int(row.cnt or 0)})
return out
def _plan_col_facet(col):
q = (
db.query(col.label("value"), sa_func.count(PlPlan.id).label("cnt"))
.select_from(PlPlan)
.filter(
PlPlan.id.in_(plan_id_sq),
col.isnot(None),
col != "",
)
)
if kw:
q = q.filter(col.like(f"%{kw}%"))
return _finish(q.group_by(col).order_by(col.asc()).limit(lim).all())
def _sub_facet(model, col):
q = (
db.query(col.label("value"), sa_func.count(PlPlan.id).label("cnt"))
.select_from(PlPlan)
.join(model, model.plan_id == PlPlan.id)
.filter(
PlPlan.id.in_(plan_id_sq),
model.tenant_no == tenant_no,
model.del_flag == 0,
col.isnot(None),
col != "",
)
)
if kw:
q = q.filter(col.like(f"%{kw}%"))
return _finish(q.group_by(col).order_by(col.asc()).limit(lim).all())
if field == "planNo":
return {"list": _plan_col_facet(PlPlan.plan_no)}
if field == "licensePlate":
return {"list": _plan_col_facet(PlPlan.license_plate)}
if field in ("loadCustomers", "loadCustomer"):
return {"list": _sub_facet(PlPlanLoad, PlPlanLoad.customer_name)}
if field in ("deliveryCustomers", "deliveryCustomer"):
return {"list": _sub_facet(PlPlanDelivery, PlPlanDelivery.customer_name)}
if field == "productName":
return {"list": _sub_facet(PlPlanProduct, PlPlanProduct.product_name)}
if field == "productSpec":
return {"list": _sub_facet(PlPlanProduct, PlPlanProduct.model)}
if field in ("carrierName",):
col = PlTransport.carrier_name
q = (
db.query(col.label("value"), sa_func.count(PlPlan.id).label("cnt"))
.select_from(PlPlan)
.join(PlTransport, PlTransport.id == PlPlan.transport_id)
.filter(
PlPlan.id.in_(plan_id_sq),
PlTransport.tenant_no == tenant_no,
PlTransport.del_flag == 0,
col.isnot(None),
col != "",
)
)
if kw:
q = q.filter(col.like(f"%{kw}%"))
return {"list": _finish(q.group_by(col).order_by(col.asc()).limit(lim).all())}
if field in ("packageSpec", "productPackageSpec"):
return {"list": _sub_facet(PlPlanProduct, PlPlanProduct.package_spec)}
if field == "productUnit":
return {"list": _sub_facet(PlPlanProduct, PlPlanProduct.unit)}
if field in ("contractNo", "productContract"):
link_q = (
db.query(
PlPlanContract.contract_no.label("value"),
sa_func.count(PlPlan.id).label("cnt"),
)
.select_from(PlPlan)
.join(PlPlanContract, PlPlanContract.plan_id == PlPlan.id)
.filter(
PlPlan.id.in_(plan_id_sq),
PlPlanContract.tenant_no == tenant_no,
PlPlanContract.del_flag == 0,
PlPlanContract.contract_no.isnot(None),
PlPlanContract.contract_no != "",
)
)
header_q = (
db.query(
PlPlan.contract_no.label("value"),
sa_func.count(PlPlan.id).label("cnt"),
)
.filter(
PlPlan.id.in_(plan_id_sq),
PlPlan.contract_no.isnot(None),
PlPlan.contract_no != "",
)
)
if kw:
link_q = link_q.filter(PlPlanContract.contract_no.like(f"%{kw}%"))
header_q = header_q.filter(PlPlan.contract_no.like(f"%{kw}%"))
merged: dict = {}
for row in link_q.group_by(PlPlanContract.contract_no).all():
value = str(row.value).strip()
if value:
merged[value] = merged.get(value, 0) + int(row.cnt or 0)
for row in header_q.group_by(PlPlan.contract_no).all():
value = str(row.value).strip()
if value:
merged[value] = merged.get(value, 0) + int(row.cnt or 0)
items = sorted(merged.items(), key=lambda kv: kv[0])[:lim]
return {"list": [{"value": v, "label": v, "count": c} for v, c in items]}
raise HTTPException(status_code=400, detail="不支持的列")
@router.get("/", response_model=PlanListResponse)
def list_plans(
request: Request,
q: Optional[str] = Query(
default=None,
description="模糊搜索(计划编号/供方名称/需方名称/产品名称/车牌号)",
),
plan_no: Optional[str] = Query(default=None, description="计划编号"),
plan_type: Optional[str] = Query(default=None, alias="planType", description="单据类型"),
follow_status: Optional[str] = Query(
default=None,
alias="followStatus",
description="跟进状态,多个以英文逗号分隔",
),
exclude_follow_status: Optional[str] = Query(
default=None,
alias="excludeFollowStatus",
description="排除的跟进状态,多个以英文逗号分隔",
),
transport_status: Optional[str] = Query(default=None, alias="transportStatus", description="运输状态"),
archive_status: Optional[str] = Query(default=None, alias="archiveStatus", description="归档状态"),
customer_name: Optional[str] = Query(default=None, alias="customerName", description="客户名称"),
product_name: Optional[str] = Query(default=None, alias="productName", description="产品名称"),
package_spec: Optional[str] = Query(default=None, alias="packageSpec", description="包装规格"),
license_plate: Optional[str] = Query(default=None, alias="licensePlate", description="车牌号"),
plan_date_from: Optional[date] = Query(default=None, alias="planDateFrom", description="计划日期开始"),
plan_date_to: Optional[date] = Query(default=None, alias="planDateTo", description="计划日期结束"),
delivery_time_from: Optional[date] = Query(
default=None, alias="deliveryTimeFrom", description="送货时间开始"
),
delivery_time_to: Optional[date] = Query(
default=None, alias="deliveryTimeTo", description="送货时间结束"
),
has_hazard_product: Optional[int] = Query(
default=None,
alias="hasHazardProduct",
description="1=仅含易制毒/易制爆产品的计划",
),
cert_upload_status: Optional[str] = Query(
default=None,
alias="certUploadStatus",
description="incomplete=至少一张必需危化证书未上传",
),
hazard_cert_ready: Optional[int] = Query(
default=None,
alias="hazardCertReady",
description="0=未确认齐备 1=已确认齐备",
),
contract_id: Optional[str] = Query(
default=None,
alias="contractId",
description="CRM合同ID:返回关联该合同的运输计划(含 pl_plan_contract 与遗留主表 contract_id)",
),
include_document_attachments: Optional[int] = Query(
default=0,
alias="includeDocumentAttachments",
description="1=聚合单据附件(归档列表需要;默认关闭以降低列表耗时)",
),
include_arrange_copy_flags: Optional[int] = Query(
default=0,
alias="includeArrangeCopyFlags",
description="1=计算复制给司机/现场/磅单标志(运输计划列表需要;默认关闭)",
),
stock_role: Optional[str] = Query(
default=None,
alias="stockRole",
description="inbound=送货段山峡/云栋;outbound=装货段山峡/云栋",
),
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),
) -> PlanListResponse:
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
query = db.query(PlPlan).filter(
PlPlan.tenant_no == tenant_no,
PlPlan.del_flag == 0,
)
if contract_id and str(contract_id).strip():
from app.services.pl_contract_plan_qty import plan_ids_linked_to_contract
linked_ids = plan_ids_linked_to_contract(
db, tenant_no=tenant_no, contract_id=str(contract_id).strip()
)
if not linked_ids:
return PlanListResponse(total=0, list=[])
query = query.filter(PlPlan.id.in_(linked_ids))
if q:
q_kw = q.strip()
if q_kw:
q_lower = q_kw.lower()
query = query.filter(
or_(
PlPlan.plan_no.contains(q_kw),
func.lower(PlPlan.supplier_name).contains(q_lower),
func.lower(PlPlan.customer_name).contains(q_lower),
func.lower(PlPlan.destination_name).contains(q_lower),
plan_product_name_exists(db, tenant_no, q_kw),
plan_plate_match(q_kw),
)
)
query = _apply_follow_status_filter(query, follow_status)
query = _apply_exclude_follow_status_filter(query, exclude_follow_status)
if has_hazard_product == 1:
query = query.filter(plan_has_hazard_product_exists(tenant_no))
params = _merge_query(
request,
planNo=plan_no,
planType=plan_type,
transportStatus=transport_status,
archiveStatus=archive_status,
customerName=customer_name,
productName=product_name,
packageSpec=package_spec,
licensePlate=license_plate,
planDateFrom=plan_date_from,
planDateTo=plan_date_to,
deliveryTimeFrom=delivery_time_from,
deliveryTimeTo=delivery_time_to,
certUploadStatus=cert_upload_status,
hazardCertReady=hazard_cert_ready,
)
column_filters = _filters_from_params(params, db=db, tenant_no=tenant_no)
if column_filters:
query = query.filter(*column_filters)
try:
query = apply_stock_role_filter(
query, stock_role=stock_role, tenant_no=tenant_no
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
query = apply_pl_scope(query, PlPlan, user, db, tenant_no)
# 统计总数
total = query.count()
# 分页
offset = (page - 1) * page_size
sf, so = parse_sort_params(params)
plans = (
query.order_by(order_plan(sf, so), PlPlan.id.desc())
.offset(offset)
.limit(page_size)
.all()
)
plan_ids = [plan.id for plan in plans]
products_by_plan: dict[str, list[PlPlanProduct]] = {}
all_products: list[PlPlanProduct] = []
if plan_ids:
all_products = (
db.query(PlPlanProduct)
.filter(
PlPlanProduct.plan_id.in_(plan_ids),
PlPlanProduct.del_flag == 0,
)
.order_by(PlPlanProduct.order_num)
.all()
)
for product in all_products:
products_by_plan.setdefault(product.plan_id, []).append(product)
product_ids = [product.product_id for product in all_products if product.product_id]
hazard_map = fetch_hazard_flags_map(
db,
tenant_no,
product_ids,
)
standard_by_id, alias_options_by_id = load_crm_product_alias_maps(
db,
tenant_no=tenant_no,
product_ids=product_ids,
)
brand_qtys_by_pp = load_brand_qtys_by_plan_product_ids(
db,
tenant_no=tenant_no,
plan_product_ids=[p.id for p in all_products],
)
contracts_by_plan = load_plan_contracts_grouped(db, tenant_no, plan_ids)
need_copy_flags = int(include_arrange_copy_flags or 0) == 1
need_document_attachments = int(include_document_attachments or 0) == 1
arrange_copy_flags = (
load_plan_arrange_copy_flags(db, tenant_no, plans) if need_copy_flags else {}
)
document_attachments_by_plan = (
load_plan_document_attachments_by_plan_ids(
db,
tenant_no=tenant_no,
plan_ids=plan_ids,
)
if need_document_attachments
else {}
)
stops_by_plan = load_plan_stops_grouped(db, tenant_no, plan_ids)
can_price = user_can_view_archive_prices(db, user)
result_list: list[PlanWithProducts] = []
for plan in plans:
enriched_products: list[PlanProductRead] = []
for product in products_by_plan.get(plan.id, []):
is_toxic, is_explosive = resolve_hazard_flags_from_crm(
db,
tenant_no,
product.product_id,
hazard_map=hazard_map,
)
item = PlanProductRead.model_validate(product)
raw_brand_qtys = brand_qtys_by_pp.get(product.id) or None
brand_qtys = (
[PlanArchiveBrandQtyItem.model_validate(x) for x in raw_brand_qtys]
if raw_brand_qtys
else None
)
list_product_name, list_alias_name = _resolve_list_product_display(
product,
standard_by_id=standard_by_id,
alias_options_by_id=alias_options_by_id,
)
item = item.model_copy(
update={
"is_toxic": is_toxic,
"is_explosive": is_explosive,
"brand_qtys": brand_qtys,
"product_name": list_product_name,
"alias_name": list_alias_name,
}
)
enriched_products.append(item)
plan_item = PlanWithProducts.model_validate(plan)
plan_item.products = enriched_products
supplier_contracts, demander_contracts = contracts_by_plan.get(plan.id, ([], []))
plan_item.supplier_contracts = supplier_contracts
plan_item.demander_contracts = demander_contracts
flags = arrange_copy_flags.get(plan.id)
if flags:
plan_item.can_copy_to_driver = flags.can_copy_to_driver
plan_item.can_copy_to_site = flags.can_copy_to_site
plan_item.can_copy_pound_qrcode = flags.can_copy_pound_qrcode
if not (plan_item.personnel_summary or "").strip():
snapshot = plan_item.resource_snapshot
if snapshot and snapshot.personnel_summary:
plan_item.personnel_summary = snapshot.personnel_summary
elif plan_item.driver_name:
plan_item.personnel_summary = f"{plan_item.driver_name}(驾)"
plan_item.document_attachments = document_attachments_by_plan.get(plan.id, [])
attach_plan_stops_to_item(
plan_item,
plan,
stops_by_plan.get(plan.id, (None, [])),
)
plan_item = mask_plan_with_products_prices(plan_item, enabled=not can_price)
result_list.append(plan_item)
return PlanListResponse(total=total, list=result_list)
PLAN_EXPORT_COLUMNS = [
ExportColumn(title="计划编号", field="planNo"),
ExportColumn(title="跟进状态", field="followStatus"),
ExportColumn(title="计划日期", field="planDate"),
ExportColumn(title="送货时间", field="deliveryTimes"),
ExportColumn(title="单据类型", field="planType"),
ExportColumn(title="供方", field="supplierName"),
ExportColumn(title="需方", field="customerName"),
ExportColumn(title="产品名称", field="productName"),
ExportColumn(title="品牌", field="brandName"),
ExportColumn(title="别名", field="aliasName"),
ExportColumn(title="车牌号", field="licensePlate"),
ExportColumn(title="司机", field="driverName"),
ExportColumn(title="运输状态", field="transportStatus"),
ExportColumn(title="归档状态", field="archiveStatus"),
ExportColumn(title="ERP", field="erpEntered"),
ExportColumn(title="备注", field="remark"),
]
# 列表可见列 prop → 导出 field(与归档导出字段对齐)
_PLAN_EXPORT_COLUMN_ALIASES = {
"productBrand": "brandName",
"productAlias": "aliasName",
}
def _normalize_plan_export_columns(
requested: Optional[Sequence[str]],
) -> Optional[list[str]]:
if not requested:
return None
return [
_PLAN_EXPORT_COLUMN_ALIASES.get(str(key), str(key)) for key in requested
]
def _format_plan_list_export_brand(product: PlPlanProduct) -> str:
"""与列表/表单一致:仅指定品牌时导出品牌名。"""
if (product.brand_mode or "").strip() != "required":
return ""
return (product.brand_name or "").strip()
def _build_plan_list_export_rows(
db: Session,
*,
tenant_no: str,
plans: list[PlPlan],
stops_by_plan: dict,
) -> list[dict]:
"""计划列表导出:一计划多产品时按产品展开,并解析 CRM 标准名/别名。"""
plan_ids = [plan.id for plan in plans]
products_by_plan: dict[str, list[PlPlanProduct]] = defaultdict(list)
product_ids: set[str] = set()
if plan_ids:
product_rows = (
db.query(PlPlanProduct)
.filter(
PlPlanProduct.tenant_no == tenant_no,
PlPlanProduct.plan_id.in_(plan_ids),
PlPlanProduct.del_flag == 0,
)
.order_by(PlPlanProduct.order_num.asc(), PlPlanProduct.id.asc())
.all()
)
for product in product_rows:
products_by_plan[product.plan_id].append(product)
if product.product_id:
product_ids.add(product.product_id)
standard_by_id, alias_options_by_id = load_crm_product_alias_maps(
db, tenant_no=tenant_no, product_ids=product_ids
)
items: list[dict] = []
for plan in plans:
data = PlanWithProducts.model_validate(plan).model_dump(by_alias=True)
data.pop("products", None)
if data.get("planDate") is not None:
data["planDate"] = str(data["planDate"])[:10]
if data.get("erpEntered") is not None:
data["erpEntered"] = "是" if int(data["erpEntered"]) else "否"
_load_read, delivery_reads = stops_by_plan.get(plan.id, (None, []))
times = collect_unique_delivery_date_strings(delivery_reads)
data["deliveryTimes"] = " / ".join(times) if times else ""
plan_products = products_by_plan.get(plan.id) or []
if not plan_products:
data["productName"] = ""
data["brandName"] = ""
data["aliasName"] = ""
items.append(data)
continue
for product in plan_products:
row = dict(data)
display_name, alias = resolve_product_display_names(
saved_product_name=product.product_name,
product_id=product.product_id,
standard_by_id=standard_by_id,
alias_options_by_id=alias_options_by_id,
)
row["productName"] = display_name or (product.product_name or "")
row["brandName"] = _format_plan_list_export_brand(product)
row["aliasName"] = (alias or "").strip()
items.append(row)
return items
PLAN_ARCHIVE_EXPORT_COLUMNS = [
ExportColumn(title="序号", field="rowNo"),
ExportColumn(title="计划编号", field="planNo"),
ExportColumn(title="合同编号", field="contractNo"),
ExportColumn(title="计划日期", field="planDate"),
ExportColumn(title="归档状态", field="archiveStatus"),
ExportColumn(title="跟进状态", field="followStatus"),
ExportColumn(title="单据类型", field="planType"),
ExportColumn(title="装货客户", field="loadCustomer"),
ExportColumn(title="装货地址", field="loadAddress"),
ExportColumn(title="装货时间", field="loadTime"),
ExportColumn(title="产品名称", field="productName"),
ExportColumn(title="品牌", field="brandName"),
ExportColumn(title="别名", field="aliasName"),
ExportColumn(title="型号/规格", field="model"),
ExportColumn(title="包装", field="packageSpec"),
ExportColumn(title="单位", field="unit", dict_type="crm_product_unit"),
ExportColumn(title="装货计划数量", field="loadPlanQty"),
ExportColumn(title="装货实际数量", field="loadActualQty"),
ExportColumn(title="装货入库数量", field="loadInboundQty"),
ExportColumn(title="装货出库数量", field="loadOutboundQty"),
ExportColumn(title="装货单价(元)", field="loadUnitPrice"),
ExportColumn(title="装货金额(元)", field="loadAmount"),
ExportColumn(title="送货段次", field="deliveryStopKind"),
ExportColumn(title="送货客户", field="deliveryCustomer"),
ExportColumn(title="送货地址", field="deliveryAddress"),
ExportColumn(title="送货时间", field="deliveryTime"),
ExportColumn(title="送货计划数量", field="deliveryPlanQty"),
ExportColumn(title="送货实际数量", field="deliveryActualQty"),
ExportColumn(title="送货入库数量", field="deliveryInboundQty"),
ExportColumn(title="送货出库数量", field="deliveryOutboundQty"),
ExportColumn(title="送货单价(元)", field="deliveryUnitPrice"),
ExportColumn(title="送货金额(元)", field="deliveryAmount"),
ExportColumn(title="车牌号", field="licensePlate"),
ExportColumn(title="运费单位", field="freightType"),
ExportColumn(title="运费单价", field="freightPrice"),
ExportColumn(title="运费金额", field="freightAmount"),
ExportColumn(title="ERP", field="erpEntered"),
ExportColumn(title="备注", field="remark"),
]
def _load_transports_by_id(db: Session, tenant_no: str, transport_ids: list[str]) -> dict[str, PlTransport]:
if not transport_ids:
return {}
rows = (
db.query(PlTransport)
.filter(
PlTransport.tenant_no == tenant_no,
PlTransport.del_flag == 0,
PlTransport.id.in_(transport_ids),
)
.all()
)
return {row.id: row for row in rows}
class PlanExportBody(BaseModel):
q: Optional[str] = None
plan_no: Optional[str] = Field(default=None, alias="planNo")
plan_type: Optional[str] = Field(default=None, alias="planType")
follow_status: Optional[str] = Field(default=None, alias="followStatus")
exclude_follow_status: Optional[str] = Field(default=None, alias="excludeFollowStatus")
transport_status: Optional[str] = Field(default=None, alias="transportStatus")
archive_status: Optional[str] = Field(default=None, alias="archiveStatus")
customer_name: Optional[str] = Field(default=None, alias="customerName")
product_name: Optional[str] = Field(default=None, alias="productName")
package_spec: Optional[str] = Field(default=None, alias="packageSpec")
license_plate: Optional[str] = Field(default=None, alias="licensePlate")
plan_date_from: Optional[date] = Field(default=None, alias="planDateFrom")
plan_date_to: Optional[date] = Field(default=None, alias="planDateTo")
delivery_time_from: Optional[date] = Field(default=None, alias="deliveryTimeFrom")
delivery_time_to: Optional[date] = Field(default=None, alias="deliveryTimeTo")
has_hazard_product: Optional[int] = Field(default=None, alias="hasHazardProduct")
cert_upload_status: Optional[str] = Field(default=None, alias="certUploadStatus")
hazard_cert_ready: Optional[int] = Field(default=None, alias="hazardCertReady")
stock_role: Optional[str] = Field(default=None, alias="stockRole")
ids: Optional[List[str]] = None
columns: Optional[List[str]] = None
export_format: Optional[str] = Field(
default=None,
alias="exportFormat",
description="archive=运输计划归档 Excel;archive_zip=归档数据及附件 ZIP",
)
model_config = {"populate_by_name": True, "extra": "allow"}
@router.post("/export")
def export_plans(
body: PlanExportBody,
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 = db.query(PlPlan).filter(
PlPlan.tenant_no == tenant_no,
PlPlan.del_flag == 0,
)
if body.q:
q_kw = body.q.strip()
if q_kw:
q_lower = q_kw.lower()
query = query.filter(
or_(
PlPlan.plan_no.contains(q_kw),
func.lower(PlPlan.supplier_name).contains(q_lower),
func.lower(PlPlan.customer_name).contains(q_lower),
func.lower(PlPlan.destination_name).contains(q_lower),
plan_product_name_exists(db, tenant_no, q_kw),
plan_plate_match(q_kw),
)
)
query = _apply_follow_status_filter(query, body.follow_status)
query = _apply_exclude_follow_status_filter(query, body.exclude_follow_status)
if body.has_hazard_product == 1:
query = query.filter(plan_has_hazard_product_exists(tenant_no))
column_filters = _filters_from_params(
_body_to_params(body), db=db, tenant_no=tenant_no
)
if column_filters:
query = query.filter(*column_filters)
try:
query = apply_stock_role_filter(
query, stock_role=body.stock_role, tenant_no=tenant_no
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
if body.ids:
query = query.filter(PlPlan.id.in_(body.ids))
query = apply_pl_scope(query, PlPlan, user, db, tenant_no)
export_params = _body_to_params(body)
sf, so = parse_sort_params(export_params)
plans = (
query.order_by(order_plan(sf, so), PlPlan.id.desc())
.limit(EXPORT_MAX_ROWS + 1)
.all()
)
date_str = datetime.now().strftime("%Y%m%d")
export_format = (body.export_format or "").strip().lower()
is_archive_export = export_format == "archive"
is_archive_zip = export_format == "archive_zip"
if len(plans) > EXPORT_MAX_ROWS:
raise ExportValidationError(f"导出超过 {EXPORT_MAX_ROWS} 条,请缩小筛选范围")
if is_archive_export or is_archive_zip:
if not plans:
raise ExportValidationError("没有可导出的数据")
transport_ids = [plan.transport_id for plan in plans if plan.transport_id]
transports_by_id = _load_transports_by_id(db, tenant_no, transport_ids)
items = build_plan_archive_export_rows(
db,
tenant_no,
plans,
transports_by_id,
stop_kind=parse_single_stop_kind(
_param_get(export_params, "stopKind", "stop_kind"),
_param_get(export_params, "stopKindIn", "stop_kind_in"),
),
stop_customer_keywords=parse_stop_customer_keywords(
_param_get(export_params, "stopCustomers", "stop_customers")
),
)
cols = filter_export_columns(
PLAN_ARCHIVE_EXPORT_COLUMNS, _normalize_plan_export_columns(body.columns)
)
if not user_can_view_archive_prices(db, user):
cols = [c for c in cols if c.field not in ARCHIVE_PRICE_EXPORT_FIELDS]
if is_archive_zip:
from app.services.pl_plan_archive_zip_export import (
build_archive_zip_streaming_response,
collect_hazard_cert_items,
load_plan_products_by_plan_ids,
merge_plan_attachment_items,
)
plan_ids = [p.id for p in plans]
docs = load_plan_document_attachments_by_plan_ids(
db, tenant_no=tenant_no, plan_ids=plan_ids
)
products_by_plan = load_plan_products_by_plan_ids(
db, tenant_no=tenant_no, plan_ids=plan_ids
)
attachments_by_plan = {
pid: merge_plan_attachment_items(
docs.get(pid, []),
collect_hazard_cert_items(products_by_plan.get(pid, [])),
)
for pid in plan_ids
}
stamp = datetime.now().strftime("%Y%m%d%H%M")
return build_archive_zip_streaming_response(
db=db,
tenant_no=tenant_no,
columns=cols,
rows=items,
attachments_by_plan=attachments_by_plan,
stamp=stamp,
)
sheet_title = "计划归档"
filename_stem = f"计划归档_{date_str}"
else:
plan_ids = [plan.id for plan in plans]
stops_by_plan = load_plan_stops_grouped(db, tenant_no, plan_ids)
items = _build_plan_list_export_rows(
db,
tenant_no=tenant_no,
plans=plans,
stops_by_plan=stops_by_plan,
)
cols = filter_export_columns(
PLAN_EXPORT_COLUMNS, _normalize_plan_export_columns(body.columns)
)
sheet_title = "计划列表"
filename_stem = f"计划列表_{date_str}"
return export_rows_to_xlsx_response(
db,
tenant_no=tenant_no,
sheet_title=sheet_title,
columns=cols,
rows=items,
filename_stem=filename_stem,
)
except ExportValidationError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
@router.get("/count-by-contract")
def count_plans_by_contract(
contract_ids: str = Query(..., alias="contractIds", description="合同ID,逗号分隔"),
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> dict:
from app.models.pl_models import PlPlanContract
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
ids = [item.strip() for item in contract_ids.split(",") if item.strip()]
if not ids:
return {"counts": {}}
# 旧单:主表 contract_id;新单:pl_plan_contract 分侧关联(按计划去重计数)
legacy_rows = (
db.query(PlPlan.contract_id, func.count(PlPlan.id))
.filter(
PlPlan.tenant_no == tenant_no,
PlPlan.del_flag == 0,
PlPlan.contract_id.in_(ids),
)
)
legacy_rows = apply_pl_scope(legacy_rows, PlPlan, user, db, tenant_no).group_by(PlPlan.contract_id).all()
counts: dict[str, int] = {contract_id: count for contract_id, count in legacy_rows}
link_q = (
db.query(PlPlanContract.contract_id, PlPlan.id)
.join(PlPlan, PlPlan.id == PlPlanContract.plan_id)
.filter(
PlPlanContract.tenant_no == tenant_no,
PlPlanContract.del_flag == 0,
PlPlanContract.contract_id.in_(ids),
PlPlan.tenant_no == tenant_no,
PlPlan.del_flag == 0,
)
)
link_q = apply_pl_scope(link_q, PlPlan, user, db, tenant_no)
seen: set[tuple[str, str]] = set()
for contract_id, plan_id in link_q.all():
key = (contract_id, plan_id)
if key in seen:
continue
seen.add(key)
# 若旧主表已计过同一计划+合同,不再重复加(新单主表 contract_id 应为空)
counts[contract_id] = counts.get(contract_id, 0) + 1
return {"counts": counts}
@router.get("/contract-product-arrange-qty")
def get_contract_product_arrange_qty(
contract_id: str = Query(..., alias="contractId"),
product_id: str = Query(default="", alias="productId"),
package_spec: Optional[str] = Query(default=None, alias="packageSpec"),
exclude_plan_id: Optional[str] = Query(default=None, alias="excludePlanId"),
contract_product_id: Optional[str] = Query(default=None, alias="contractProductId"),
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> dict:
"""表单可安排数量 + 问号弹窗:合同用量与已安排计划列表。"""
from app.services.pl_contract_plan_qty import get_contract_product_arrange_qty_detail
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
return get_contract_product_arrange_qty_detail(
db,
tenant_no=tenant_no,
contract_id=contract_id,
product_id=product_id,
package_spec=package_spec,
exclude_plan_id=exclude_plan_id,
contract_product_id=contract_product_id,
)
@router.post("/contract-product-arrange-qty/batch")
def batch_contract_product_arrange_qty(
body: dict,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> dict:
"""
批量可安排数量摘要(不含已安排计划明细)。
仅供运输计划新增/编辑/调整运输表单打开时预取,问号弹窗仍走单条详情接口。
"""
from app.services.pl_contract_plan_qty import batch_contract_product_arrange_qty_summaries
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
items = body.get("items") if isinstance(body, dict) else None
if not isinstance(items, list):
items = []
rows = batch_contract_product_arrange_qty_summaries(
db, tenant_no=tenant_no, items=items
)
return {"items": rows}
@router.get("/contract-gaps", response_model=PlanContractGapListResponse)
def list_contract_gaps(
request: Request,
page: int = Query(default=1, ge=1),
page_size: int = Query(default=20, ge=1, le=500, alias="pageSize"),
plan_no: Optional[str] = Query(default=None, alias="planNo", description="计划编号"),
gap_type: Optional[str] = Query(
default=None,
alias="gapType",
description="缺口类型:supplier | demander | both,多个以英文逗号分隔",
),
customer_keyword: Optional[str] = Query(
default=None,
alias="customerKeyword",
description="供方或需方名称关键词",
),
plan_date_from: Optional[date] = Query(default=None, alias="planDateFrom"),
plan_date_to: Optional[date] = Query(default=None, alias="planDateTo"),
delivery_time_from: Optional[date] = Query(default=None, alias="deliveryTimeFrom"),
delivery_time_to: Optional[date] = Query(default=None, alias="deliveryTimeTo"),
plan_type: Optional[str] = Query(default=None, alias="planType"),
follow_status: Optional[str] = Query(default=None, alias="followStatus"),
q: Optional[str] = Query(
default=None,
description="模糊搜索(计划编号/供方名称/需方名称)",
),
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> PlanContractGapListResponse:
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
params = _merge_query(
request,
planNo=plan_no,
gapType=gap_type,
customerKeyword=customer_keyword,
planDateFrom=plan_date_from,
planDateTo=plan_date_to,
deliveryTimeFrom=delivery_time_from,
deliveryTimeTo=delivery_time_to,
planType=plan_type,
followStatus=follow_status,
)
column_filters = _filters_from_params(params, db=db, tenant_no=tenant_no)
total, items = list_plan_contract_gaps(
db,
tenant_no=tenant_no,
user=user,
page=page,
page_size=page_size,
gap_type=_param_get(params, "gapType", "gapTypeIn", "gap_type"),
customer_keyword=_param_get(params, "customerKeyword", "customer_keyword"),
follow_status=_param_get(params, "followStatus", "followStatusIn", "follow_status"),
q=q,
sort_field=_param_get(params, "sortField", "sort_field"),
sort_order=_param_get(params, "sortOrder", "sort_order"),
column_filters=column_filters,
)
return PlanContractGapListResponse(total=total, list=items)
CONTRACT_GAP_EXPORT_COLUMNS = [
ExportColumn(title="计划编号", field="planNo"),
ExportColumn(title="计划日期", field="planDate"),
ExportColumn(title="送货时间", field="deliveryTimes"),
ExportColumn(title="单据类型", field="planType"),
ExportColumn(title="跟进状态", field="followStatus"),
ExportColumn(title="供方", field="supplierName"),
ExportColumn(title="需方", field="customerName"),
ExportColumn(title="缺口类型", field="gapType"),
]
class ContractGapExportBody(BaseModel):
plan_no: Optional[str] = Field(default=None, alias="planNo")
gap_type: Optional[str] = Field(default=None, alias="gapType")
customer_keyword: Optional[str] = Field(default=None, alias="customerKeyword")
plan_date_from: Optional[date] = Field(default=None, alias="planDateFrom")
plan_date_to: Optional[date] = Field(default=None, alias="planDateTo")
delivery_time_from: Optional[date] = Field(default=None, alias="deliveryTimeFrom")
delivery_time_to: Optional[date] = Field(default=None, alias="deliveryTimeTo")
q: Optional[str] = None
ids: Optional[List[str]] = None
columns: Optional[List[str]] = None
model_config = {"populate_by_name": True, "extra": "allow"}
@router.post("/contract-gaps/export")
def export_contract_gaps(
body: ContractGapExportBody,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
):
try:
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
params = _body_to_params(body)
column_filters = _filters_from_params(params, db=db, tenant_no=tenant_no)
_, items = list_plan_contract_gaps(
db,
tenant_no=tenant_no,
user=user,
page=1,
page_size=EXPORT_MAX_ROWS,
gap_type=_param_get(params, "gapType", "gapTypeIn", "gap_type"),
customer_keyword=_param_get(params, "customerKeyword", "customer_keyword"),
follow_status=_param_get(
params, "followStatus", "followStatusIn", "follow_status"
),
q=_param_get(params, "q"),
ids=body.ids,
max_page_size=EXPORT_MAX_ROWS,
sort_field=_param_get(params, "sortField", "sort_field"),
sort_order=_param_get(params, "sortOrder", "sort_order"),
column_filters=column_filters,
)
plan_type_labels = {"direct": "直销单", "inbound": "入库单", "outbound": "出库单"}
gap_labels = {"supplier": "仅缺供方合同", "demander": "仅缺需方合同", "both": "双侧缺口"}
rows = []
for item in items:
data = item.model_dump(by_alias=True)
data["planType"] = plan_type_labels.get(str(data.get("planType") or ""), data.get("planType"))
sides = set(data.get("gapSides") or [])
if "supplier" in sides and "demander" in sides:
data["gapType"] = "双侧缺口"
elif "supplier" in sides:
data["gapType"] = "仅缺供方合同"
elif "demander" in sides:
data["gapType"] = "仅缺需方合同"
else:
data["gapType"] = ""
times = data.get("deliveryTimes") or []
if isinstance(times, list):
data["deliveryTimes"] = " / ".join(str(t) for t in times) if times else ""
for k, v in list(data.items()):
if hasattr(v, "isoformat"):
data[k] = str(v)[:19]
rows.append(data)
cols = filter_export_columns(CONTRACT_GAP_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.put("/{id}/stop-contracts", response_model=PlanContractGapRead)
def put_plan_stop_contracts(
id: str,
payload: PlanStopContractsUpdateBody,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> PlanContractGapRead:
"""保存某装货/送货段合同关联(合同补全)。"""
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
plan = _get_plan(db, tenant_no, id, user)
result = update_plan_stop_contracts(
db,
tenant_no=tenant_no,
plan=plan,
stop_type=payload.stop_type,
stop_id=payload.stop_id,
contracts=payload.contracts,
user_id=user.id,
dept_id=user.dept_id,
)
plan.update_by = user.id
db.commit()
return result
@router.put("/{id}/contracts/{side}", response_model=PlanContractGapRead)
def put_plan_side_contracts(
id: str,
side: str,
payload: PlanSideContractsUpdateBody,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> PlanContractGapRead:
"""单侧保存计划合同关联(合同补全)。"""
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
plan = _get_plan(db, tenant_no, id, user)
result = update_plan_side_contracts(
db,
tenant_no=tenant_no,
plan=plan,
side=side,
contracts=payload.contracts,
user_id=user.id,
dept_id=user.dept_id,
)
plan.update_by = user.id
db.commit()
return result
# =============================================================================
# 创建计划
# =============================================================================
@router.post("/", response_model=PlanRead, status_code=status.HTTP_201_CREATED)
def create_plan(
payload: PlanCreate,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> PlanRead:
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
# 生成计划编号
plan_no = _generate_plan_no(db)
# 单据类型由前端显式选择;未选则存空,不再自动推断为直销单
plan_type = (payload.plan_type or "").strip()
plan = PlPlan(
id=generate_uuid(),
tenant_no=tenant_no,
plan_no=plan_no,
create_date=_resolve_create_date(payload.create_date),
plan_date=payload.plan_date,
belongs_company=payload.belongs_company,
plan_type=plan_type,
follow_status="pending_arrange",
transport_status=payload.transport_status or "pending",
archive_status=payload.archive_status or "unarchived",
complete_status=payload.complete_status or "no",
contract_id=payload.contract_id,
contract_no=payload.contract_no,
supplier_id=payload.supplier_id,
supplier_name=payload.supplier_name,
destination_id=payload.destination_id,
destination_name=payload.destination_name,
customer_id=payload.customer_id,
customer_name=payload.customer_name,
contact_name=payload.contact_name,
contact_phone=payload.contact_phone,
party_a=payload.party_a,
party_b=payload.party_b,
contract_start_date=payload.contract_start_date,
contract_end_date=payload.contract_end_date,
delivery_method=payload.delivery_method,
settlement_method=payload.settlement_method,
has_freight=payload.has_freight,
pickup_address=payload.pickup_address,
pickup_lng=payload.pickup_lng,
pickup_lat=payload.pickup_lat,
pickup_time=payload.pickup_time,
pickup_contact=payload.pickup_contact,
pickup_notes=payload.pickup_notes,
destination_address=payload.destination_address,
destination_lng=payload.destination_lng,
destination_lat=payload.destination_lat,
delivery_time=payload.delivery_time,
delivery_contact=payload.delivery_contact,
destination_notes=payload.destination_notes,
follow_user_id=payload.follow_user_id or user.id,
follow_user_name=payload.follow_user_name or user.nick_name,
remark=payload.remark,
create_by=user.id,
create_dept_id=user.dept_id,
)
db.add(plan)
# 添加产品明细
if payload.products:
for idx, prod in enumerate(payload.products):
pp = build_plan_product(
tenant_no=tenant_no,
plan_id=plan.id,
plan_type=plan_type,
prod=prod,
order_num=idx,
user_id=user.id,
dept_id=user.dept_id,
)
db.add(pp)
plan.total_amount = calc_plan_total_amount(payload.products or [])
db.commit()
db.refresh(plan)
return plan
@router.post("/create-full", response_model=PlanWithProducts, status_code=status.HTTP_201_CREATED)
def create_plan_full(
payload: PlanCreateFull,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> PlanWithProducts:
"""创建运输计划(完整表单:供方/需方 + 产品库产品 + 可选预选车辆,不自动安排运输)"""
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
save_mode = (getattr(payload, "save_mode", None) or "formal").strip() or "formal"
is_draft = save_mode == "draft"
multi_stop = uses_multi_stop_payload(payload.load, payload.deliveries)
qty_warnings: list[str] = []
load_body = None
delivery_bodies: list = []
supplier_contracts: list[PlanContractLink] = []
demander_contracts: list[PlanContractLink] = []
if is_draft:
validate_draft_min_fields(
belongs_company=payload.belongs_company,
create_date=payload.create_date,
)
if multi_stop and payload.load is not None:
try:
load_body, delivery_bodies = validate_plan_stops_write(
db,
tenant_no,
payload.load,
payload.deliveries,
payload.products,
)
qty_warnings = collect_stop_qty_warnings(
payload.products, load_body, delivery_bodies
)
except HTTPException:
load_body, delivery_bodies = None, []
multi_stop = False
elif multi_stop:
load_body, delivery_bodies = validate_plan_stops_write(
db,
tenant_no,
payload.load,
payload.deliveries,
payload.products,
)
qty_warnings = collect_stop_qty_warnings(payload.products, load_body, delivery_bodies)
else:
validate_create_full(payload)
supplier_contracts, demander_contracts = validate_plan_party_contracts(
db,
tenant_no,
supplier_id=payload.supplier_id,
supplier_name=payload.supplier_name,
customer_id=payload.customer_id,
customer_name=payload.customer_name,
supplier_contracts=payload.supplier_contracts,
demander_contracts=payload.demander_contracts,
require_products=payload.products,
)
load_body = None
delivery_bodies = []
plan_no = _generate_plan_no(db)
# 单据类型由前端显式选择;未选则存空,不再自动推断为直销单
plan_type = (payload.plan_type or "").strip()
has_vehicle = bool(payload.transport and payload.transport.vehicle_id)
demander_name = (payload.customer_name or "").strip() or None
plan = PlPlan(
id=generate_uuid(),
tenant_no=tenant_no,
plan_no=plan_no,
create_date=_resolve_create_date(payload.create_date),
plan_date=payload.plan_date,
belongs_company=payload.belongs_company,
plan_type=plan_type,
follow_status=resolve_follow_status_on_save(
save_mode=save_mode, current=None, is_create=True
),
transport_status="pending",
archive_status="unarchived",
complete_status="no",
contract_id=None,
contract_no=None,
supplier_id=payload.supplier_id,
supplier_name=payload.supplier_name,
destination_id=payload.customer_id or payload.destination_id,
destination_name=demander_name or payload.destination_name,
customer_id=payload.customer_id,
customer_name=payload.customer_name,
contact_name=payload.contact_name,
contact_phone=payload.contact_phone,
party_a=payload.party_a,
party_b=payload.party_b,
contract_start_date=payload.contract_start_date,
contract_end_date=payload.contract_end_date,
delivery_method=payload.delivery_method,
settlement_method=payload.settlement_method,
has_freight=payload.has_freight,
freight_remark=payload.freight_remark,
pickup_address=payload.pickup_address,
pickup_lng=payload.pickup_lng,
pickup_lat=payload.pickup_lat,
pickup_time=payload.pickup_time,
pickup_contact=payload.pickup_contact,
pickup_notes=payload.pickup_notes,
destination_address=payload.destination_address,
destination_lng=payload.destination_lng,
destination_lat=payload.destination_lat,
delivery_time=payload.delivery_time,
delivery_contact=payload.delivery_contact,
destination_notes=payload.destination_notes,
total_amount=calc_plan_total_amount(payload.products or []),
follow_user_id=payload.follow_user_id or user.id,
follow_user_name=payload.follow_user_name or user.nick_name,
remark=payload.remark,
create_by=user.id,
create_dept_id=user.dept_id,
)
db.add(plan)
db.flush()
if multi_stop and load_body is not None:
sync_plan_legacy_from_stops(plan, load_body, delivery_bodies)
plan_products: list[PlPlanProduct] = []
for idx, prod in enumerate(payload.products or []):
is_toxic, is_explosive = resolve_hazard_flags_from_crm(
db,
tenant_no,
prod.product_id,
)
prod_payload = prod.model_copy(update={"is_toxic": is_toxic, "is_explosive": is_explosive})
pp = build_plan_product(
tenant_no=tenant_no,
plan_id=plan.id,
plan_type=plan_type,
prod=prod_payload,
order_num=idx,
user_id=user.id,
dept_id=user.dept_id,
)
db.add(pp)
plan_products.append(pp)
if multi_stop and load_body is not None:
replace_plan_stops(
db,
tenant_no=tenant_no,
plan_id=plan.id,
load=load_body,
deliveries=delivery_bodies,
user_id=user.id,
dept_id=user.dept_id,
plan_products=plan_products,
)
safe_upsert_prefs_from_plan_stops(
db,
tenant_no=tenant_no,
load=load_body,
deliveries=delivery_bodies,
user_id=user.id,
dept_id=user.dept_id,
)
else:
replace_plan_contracts(
db,
tenant_no=tenant_no,
plan_id=plan.id,
supplier_contracts=supplier_contracts,
demander_contracts=demander_contracts,
user_id=user.id,
dept_id=user.dept_id,
)
if has_vehicle and payload.transport:
apply_vehicle_prefill_to_plan(
db,
tenant_no=tenant_no,
plan=plan,
transport_payload=payload.transport,
products=plan_products,
user_id=user.id,
dept_id=user.dept_id,
)
db.commit()
db.refresh(plan)
result = PlanWithProducts.model_validate(plan)
result.products = plan_products
result.supplier_contracts = supplier_contracts
result.demander_contracts = demander_contracts
result.warnings = qty_warnings
stops = load_plan_stops_grouped(db, tenant_no, [plan.id]).get(plan.id, (None, []))
attach_plan_stops_to_item(result, plan, stops)
return result
# =============================================================================
# 从合同创建计划
# =============================================================================
@router.post("/create-from-contract", response_model=PlanRead, status_code=status.HTTP_201_CREATED)
def create_from_contract(
payload: CreateFromContractRequest,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> PlanRead:
"""从 CRM 合同发起计划"""
from app.models.crm_models import CrmContract, CrmContractProduct
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
# 查询合同信息
contract = (
db.query(CrmContract)
.filter(CrmContract.id == payload.contract_id, CrmContract.del_flag == 0)
.first()
)
if not contract:
raise HTTPException(status_code=404, detail="合同不存在")
assert_contract_selectable_for_plan(db, contract.id, tenant_no)
# 生成计划编号
plan_no = _generate_plan_no(db)
# 自动判断单据类型
plan_type = _determine_plan_type(contract.supplier_name, contract.customer_name)
plan = PlPlan(
id=generate_uuid(),
tenant_no=tenant_no,
plan_no=plan_no,
create_date=_resolve_create_date(payload.create_date),
plan_date=payload.plan_date,
belongs_company=contract.belongs_company or "shanxia",
plan_type=plan_type,
follow_status="pending_arrange",
transport_status="pending",
archive_status="unarchived",
complete_status="no",
contract_id=contract.id,
contract_no=contract.contract_no,
supplier_id=contract.supplier_id,
supplier_name=(contract.party_a or contract.supplier_name or "").strip() or None,
destination_id=contract.customer_id,
# 客户名称严格跟合同「客户名称」;目的地名称与客户保持一致(不再取需方可章名 party_b)
destination_name=(contract.customer_name or "").strip() or None,
customer_id=contract.customer_id,
customer_name=(contract.customer_name or "").strip() or None,
contact_name=contract.contact_name,
contact_phone=contract.contact_phone,
party_a=contract.party_a,
party_b=contract.party_b,
contract_start_date=contract.start_date,
contract_end_date=contract.end_date,
delivery_method=contract.delivery_method,
settlement_method=contract.settlement_method,
has_freight=1 if contract.settlement_method and "含运" in contract.settlement_method else 0,
follow_user_id=payload.follow_user_id or user.id,
follow_user_name=payload.follow_user_name or user.nick_name,
create_by=user.id,
create_dept_id=user.dept_id,
)
db.add(plan)
# 查询合同产品明细
contract_products = (
db.query(CrmContractProduct)
.filter(
CrmContractProduct.contract_id == contract.id,
CrmContractProduct.del_flag == 0,
)
.all()
)
# 添加产品明细
for idx, cp in enumerate(contract_products):
is_toxic, is_explosive = resolve_hazard_flags_from_crm(
db,
tenant_no,
cp.product_id,
)
pp = PlPlanProduct(
id=generate_uuid(),
tenant_no=tenant_no,
plan_id=plan.id,
product_id=cp.product_id,
product_no=cp.product_no,
product_name=cp.product_name,
brand_mode=cp.brand_mode,
brand_id=cp.brand_id,
brand_name=cp.brand_name,
model=cp.model,
unit=cp.unit or "吨",
is_explosive=is_explosive,
is_toxic=is_toxic,
inbound_estimate_qty=cp.quantity,
outbound_estimate_qty=cp.quantity,
order_num=idx,
create_by=user.id,
create_dept_id=user.dept_id,
)
db.add(pp)
db.commit()
db.refresh(plan)
return plan
# =============================================================================
# 获取计划详情
# =============================================================================
@router.get("/{id}", response_model=PlanWithProducts)
def get_plan(
id: str,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> PlanWithProducts:
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
plan = _get_plan(db, tenant_no, id, user)
products = (
db.query(PlPlanProduct)
.filter(
PlPlanProduct.plan_id == id,
PlPlanProduct.del_flag == 0,
)
.order_by(PlPlanProduct.order_num)
.all()
)
hazard_map = fetch_hazard_flags_map(
db,
tenant_no,
[product.product_id for product in products if product.product_id],
)
brand_qtys_by_pp = load_brand_qtys_by_plan_product_ids(
db,
tenant_no=tenant_no,
plan_product_ids=[p.id for p in products],
)
enriched_products = []
for product in products:
is_toxic, is_explosive = resolve_hazard_flags_from_crm(
db,
tenant_no,
product.product_id,
hazard_map=hazard_map,
)
item = PlanProductRead.model_validate(product)
raw_brand_qtys = brand_qtys_by_pp.get(product.id) or None
brand_qtys = (
[PlanArchiveBrandQtyItem.model_validate(x) for x in raw_brand_qtys]
if raw_brand_qtys
else None
)
item = item.model_copy(
update={
"is_toxic": is_toxic,
"is_explosive": is_explosive,
"brand_qtys": brand_qtys,
}
)
enriched_products.append(item)
result = PlanWithProducts.model_validate(plan)
result.products = enriched_products
supplier_contracts, demander_contracts = load_plan_contracts_grouped(
db, tenant_no, [id]
).get(id, ([], []))
result.supplier_contracts = supplier_contracts
result.demander_contracts = demander_contracts
stops = load_plan_stops_grouped(db, tenant_no, [id]).get(id, (None, []))
attach_plan_stops_to_item(result, plan, stops)
can_price = user_can_view_archive_prices(db, user)
return mask_plan_with_products_prices(result, enabled=not can_price)
# =============================================================================
# 更新计划
# =============================================================================
@router.put("/{id}", response_model=PlanWithProducts)
def update_plan(
id: str,
payload: PlanUpdate,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> PlanWithProducts:
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
plan = _get_plan(db, tenant_no, id, user)
if plan.follow_status == "cancelled":
raise HTTPException(status_code=400, detail="已作废的计划不能修改")
save_mode = (getattr(payload, "save_mode", None) or "formal").strip() or "formal"
is_draft = save_mode == "draft"
if is_draft:
assert_can_save_as_draft(plan.follow_status)
validate_draft_min_fields(
belongs_company=(
payload.belongs_company
if payload.belongs_company is not None
else plan.belongs_company
),
create_date=(
payload.create_date if payload.create_date is not None else plan.create_date
),
)
multi_stop = uses_multi_stop_payload(payload.load, payload.deliveries)
qty_warnings: list[str] = []
load_body: Optional[PlanLoadBody] = None
delivery_bodies: list[PlanDeliveryBody] = []
# 更新主表字段(跟进状态仅由业务动作 / saveMode 决议维护,忽略客户端 follow_status)
update_data = payload.model_dump(
exclude_unset=True,
exclude={
"products",
"transport",
"follow_status",
"save_mode",
"supplier_contracts",
"demander_contracts",
"load",
"deliveries",
},
)
for key, value in update_data.items():
setattr(plan, key, value)
# 新模型:单一合同字段不再作为写入源
if "supplier_contracts" in payload.model_fields_set or "demander_contracts" in payload.model_fields_set:
plan.contract_id = None
plan.contract_no = None
if payload.customer_id is not None or payload.customer_name is not None:
plan.destination_id = plan.customer_id
plan.destination_name = plan.customer_name
supplier_id = plan.supplier_id
supplier_name = plan.supplier_name
customer_id = plan.customer_id
customer_name = plan.customer_name
if is_draft:
supplier_contracts: list[PlanContractLink] = []
demander_contracts: list[PlanContractLink] = []
if multi_stop and payload.load is not None:
try:
load_body, delivery_bodies = validate_plan_stops_write(
db,
tenant_no,
payload.load,
payload.deliveries,
payload.products,
)
products_for_warn = payload.products if payload.products is not None else None
qty_warnings = collect_stop_qty_warnings(
products_for_warn, load_body, delivery_bodies
)
sync_plan_legacy_from_stops(plan, load_body, delivery_bodies)
except HTTPException:
# 跳过段更新,保留库内原段
load_body, delivery_bodies = None, []
multi_stop = False
if not multi_stop:
supplier_contracts, demander_contracts = load_plan_contracts_grouped(
db, tenant_no, [id]
).get(id, ([], []))
elif multi_stop:
load_body, delivery_bodies = validate_plan_stops_write(
db,
tenant_no,
payload.load,
payload.deliveries,
payload.products,
)
products_for_warn = payload.products if payload.products is not None else None
qty_warnings = collect_stop_qty_warnings(products_for_warn, load_body, delivery_bodies)
sync_plan_legacy_from_stops(plan, load_body, delivery_bodies)
supplier_contracts = []
demander_contracts = []
else:
supplier_contracts_in = (
payload.supplier_contracts
if "supplier_contracts" in payload.model_fields_set
else load_plan_contracts_grouped(db, tenant_no, [id]).get(id, ([], []))[0]
)
demander_contracts_in = (
payload.demander_contracts
if "demander_contracts" in payload.model_fields_set
else load_plan_contracts_grouped(db, tenant_no, [id]).get(id, ([], []))[1]
)
# 编辑保存若带了产品或任一侧合同字段,按新规则完整校验供方/需方
if (
payload.products is not None
or "supplier_contracts" in payload.model_fields_set
or "demander_contracts" in payload.model_fields_set
or payload.supplier_id is not None
or payload.customer_id is not None
):
supplier_contracts, demander_contracts = validate_plan_party_contracts(
db,
tenant_no,
supplier_id=supplier_id,
supplier_name=supplier_name,
customer_id=customer_id,
customer_name=customer_name,
supplier_contracts=supplier_contracts_in,
demander_contracts=demander_contracts_in,
require_products=payload.products if payload.products is not None else None,
)
replace_plan_contracts(
db,
tenant_no=tenant_no,
plan_id=id,
supplier_contracts=supplier_contracts,
demander_contracts=demander_contracts,
user_id=user.id,
dept_id=user.dept_id,
)
else:
supplier_contracts, demander_contracts = load_plan_contracts_grouped(db, tenant_no, [id]).get(
id, ([], [])
)
plan_type = payload.plan_type or plan.plan_type
# 更新产品明细(按 id upsert,保持 pl_plan_product.id 稳定)
if payload.products is not None:
existing_all = (
db.query(PlPlanProduct)
.filter(
PlPlanProduct.plan_id == id,
PlPlanProduct.tenant_no == tenant_no,
)
.all()
)
existing_all_by_id = {row.id: row for row in existing_all}
existing_products = [row for row in existing_all if row.del_flag == 0]
old_hazard_product_ids = collect_hazard_product_ids(db, tenant_no, existing_products)
cert_by_product_id = {
row.product_id: (
row.purchase_cert_url,
row.transport_cert_url,
row.legal_use_cert_url,
)
for row in existing_products
}
kept_ids: set[str] = set()
plan_products: list[PlPlanProduct] = []
for idx, prod in enumerate(payload.products):
is_toxic, is_explosive = resolve_hazard_flags_from_crm(
db,
tenant_no,
prod.product_id,
)
purchase_cert = prod.purchase_cert_url
transport_cert = prod.transport_cert_url
legal_use_cert = prod.legal_use_cert_url
old_certs = cert_by_product_id.get(prod.product_id, (None, None, None))
old_purchase, old_transport, old_legal_use = old_certs
purchase_cert = purchase_cert or old_purchase
transport_cert = transport_cert or old_transport
legal_use_cert = legal_use_cert or old_legal_use
prod_payload = prod.model_copy(
update={
"is_toxic": is_toxic,
"is_explosive": is_explosive,
"purchase_cert_url": purchase_cert,
"transport_cert_url": transport_cert,
"legal_use_cert_url": legal_use_cert,
}
)
row_id, existing = resolve_plan_product_id_for_upsert(
plan_id=id,
payload_id=prod.id,
existing_by_id=existing_all_by_id,
)
if existing is not None:
apply_plan_product_fields(
existing,
plan_type=plan_type,
prod=prod_payload,
order_num=idx,
user_id=user.id,
)
kept_ids.add(existing.id)
plan_products.append(existing)
else:
pp = build_plan_product(
tenant_no=tenant_no,
plan_id=id,
plan_type=plan_type,
prod=prod_payload,
order_num=idx,
user_id=user.id,
dept_id=user.dept_id,
product_row_id=row_id,
)
db.add(pp)
kept_ids.add(pp.id)
plan_products.append(pp)
for row in existing_products:
if row.id not in kept_ids:
row.del_flag = 1
row.update_by = user.id
new_hazard_product_ids = collect_hazard_product_ids(db, tenant_no, payload.products)
if old_hazard_product_ids != new_hazard_product_ids:
clear_hazard_cert_ready(plan)
plan.total_amount = calc_plan_total_amount(payload.products)
else:
plan_products = (
db.query(PlPlanProduct)
.filter(
PlPlanProduct.plan_id == id,
PlPlanProduct.del_flag == 0,
)
.all()
)
# 同步运输预选 / 关联运输单运费
if payload.transport is not None:
if payload.transport.vehicle_id:
sync_transport_prefill_for_plan(
db,
tenant_no=tenant_no,
plan=plan,
transport_payload=payload.transport,
products=plan_products,
user_id=user.id,
dept_id=user.dept_id,
)
else:
clear_draft_transport_for_plan(
db,
tenant_no=tenant_no,
plan=plan,
user_id=user.id,
)
if multi_stop and load_body is not None:
replace_plan_stops(
db,
tenant_no=tenant_no,
plan_id=id,
load=load_body,
deliveries=delivery_bodies,
user_id=user.id,
dept_id=user.dept_id,
plan_products=plan_products,
)
safe_upsert_prefs_from_plan_stops(
db,
tenant_no=tenant_no,
load=load_body,
deliveries=delivery_bodies,
user_id=user.id,
dept_id=user.dept_id,
)
plan.update_by = user.id
plan.follow_status = resolve_follow_status_on_save(
save_mode=save_mode,
current=plan.follow_status,
is_create=False,
)
dibang_date_warnings = sync_linked_dibang_dates_for_plan(
db,
tenant_no=tenant_no,
plan=plan,
user_id=user.id,
dept_id=user.dept_id or "",
)
dibang_upsert_warnings: list[str] = []
if (
plan.transport_status == "arranged"
and plan.transport_id
and multi_stop
and load_body is not None
):
transport_row = (
db.query(PlTransport)
.filter(
PlTransport.id == plan.transport_id,
PlTransport.tenant_no == tenant_no,
PlTransport.del_flag == 0,
)
.first()
)
if transport_row:
stop_options: list[StopDibangOption] = []
load_row = (
db.query(PlPlanLoad)
.filter(
PlPlanLoad.plan_id == id,
PlPlanLoad.tenant_no == tenant_no,
PlPlanLoad.del_flag == 0,
)
.first()
)
if load_row:
stop_options.append(
StopDibangOption(
stop_type="load",
stop_id=load_row.id,
generate_dibang=bool(load_row.generate_dibang),
)
)
delivery_rows = (
db.query(PlPlanDelivery)
.filter(
PlPlanDelivery.plan_id == id,
PlPlanDelivery.tenant_no == tenant_no,
PlPlanDelivery.del_flag == 0,
)
.order_by(PlPlanDelivery.seq.asc(), PlPlanDelivery.create_time.asc())
.all()
)
for delivery_row in delivery_rows:
stop_options.append(
StopDibangOption(
stop_type="delivery",
stop_id=delivery_row.id,
generate_dibang=bool(delivery_row.generate_dibang),
)
)
if stop_options:
try:
_link_results, dibang_upsert_warnings = upsert_stop_dibang_links(
db,
tenant_no=tenant_no,
plan=plan,
transport=transport_row,
stop_options=stop_options,
user_id=user.id,
dept_id=user.dept_id or "",
)
except ValueError as exc:
db.rollback()
raise HTTPException(status_code=400, detail=str(exc)) from exc
db.commit()
db.refresh(plan)
result = PlanWithProducts.model_validate(plan)
result.products = plan_products
result.supplier_contracts = supplier_contracts
result.demander_contracts = demander_contracts
result.warnings = [*qty_warnings, *dibang_date_warnings, *dibang_upsert_warnings]
stops = load_plan_stops_grouped(db, tenant_no, [id]).get(id, (None, []))
attach_plan_stops_to_item(result, plan, stops)
return result
@router.patch("/{id}/product-certs", response_model=PlanWithProducts)
def patch_plan_product_certs(
id: str,
payload: PlanProductCertsUpdate,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> PlanWithProducts:
"""更新计划危化品产品证书(购买证/运输证/合法使用证明,不触发安排运输)"""
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
_get_plan(db, tenant_no, id, user)
update_plan_product_certs(
db,
tenant_no=tenant_no,
plan_id=id,
items=payload.products,
user_id=user.id,
)
plan = _get_plan(db, tenant_no, id, user)
products = (
db.query(PlPlanProduct)
.filter(
PlPlanProduct.plan_id == id,
PlPlanProduct.del_flag == 0,
)
.order_by(PlPlanProduct.order_num)
.all()
)
result = PlanWithProducts.model_validate(plan)
result.products = products
return result
@router.patch("/{id}/hazard-cert-ready", response_model=PlanWithProducts)
def patch_plan_hazard_cert_ready(
id: str,
payload: HazardCertReadyBody,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> PlanWithProducts:
"""标记或取消危化品证书齐备(人工确认)"""
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
_get_plan(db, tenant_no, id, user)
set_hazard_cert_ready(
db,
tenant_no=tenant_no,
plan_id=id,
ready=payload.ready,
user_id=user.id,
)
plan = _get_plan(db, tenant_no, id, user)
products = (
db.query(PlPlanProduct)
.filter(
PlPlanProduct.plan_id == id,
PlPlanProduct.del_flag == 0,
)
.order_by(PlPlanProduct.order_num)
.all()
)
result = PlanWithProducts.model_validate(plan)
result.products = products
return result
@router.patch("/{id}/erp-entered", response_model=PlanWithProducts)
def patch_plan_erp_entered(
id: str,
payload: ErpEnteredBody,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> PlanWithProducts:
"""标记或取消运输计划已录入 ERP。"""
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
set_plan_erp_entered(
db,
tenant_no=tenant_no,
plan_id=id,
entered=payload.entered,
user_id=user.id,
)
plan = _get_plan(db, tenant_no, id, user)
products = (
db.query(PlPlanProduct)
.filter(
PlPlanProduct.plan_id == id,
PlPlanProduct.del_flag == 0,
)
.order_by(PlPlanProduct.order_num)
.all()
)
result = PlanWithProducts.model_validate(plan)
result.products = products
result.document_attachments = load_plan_document_attachments_by_plan_ids(
db,
tenant_no=tenant_no,
plan_ids=[id],
).get(id, [])
return result
def _build_driver_url(token: str) -> str:
origin = (settings.PUBLIC_WEB_ORIGIN or "").rstrip("/")
return f"{origin}/#/public/pl/transport/{token}"
@router.get("/{id}/arrange-info", response_model=PlanArrangeInfoRead)
def get_plan_arrange_info(
id: str,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> PlanArrangeInfoRead:
"""获取计划已安排运输后的司机链接与地磅出入库关联(供查看弹窗复制简述/二维码)。"""
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
plan = _get_plan(db, tenant_no, id, user)
empty = PlanArrangeInfoRead(
driver_url=None,
license_plate=None,
site_records=[],
inventory_records=[],
requires_site_link=False,
)
if plan.follow_status == "cancelled" or plan.transport_status != "arranged" or not plan.transport_id:
return empty
transport = (
db.query(PlTransport)
.filter(
PlTransport.id == plan.transport_id,
PlTransport.tenant_no == tenant_no,
PlTransport.del_flag == 0,
)
.first()
)
if not transport or transport.transport_status == "cancelled":
return empty
token = (transport.driver_access_token or "").strip()
driver_url = _build_driver_url(token) if token else None
requires_site_link = plan_requires_site_link(db, tenant_no=tenant_no, plan=plan)
site_records = build_site_records_for_arranged_plan(
db,
tenant_no=tenant_no,
plan=plan,
transport=transport,
)
links = (
db.query(PlTransportDbLink)
.filter(
PlTransportDbLink.tenant_no == tenant_no,
PlTransportDbLink.plan_id == id,
PlTransportDbLink.transport_id == transport.id,
PlTransportDbLink.del_flag == 0,
)
.all()
)
inventory_records: list[TransportInventoryLinkRead] = []
for link in links:
product = (
db.query(PlPlanProduct)
.filter(
PlPlanProduct.id == link.plan_product_id,
PlPlanProduct.del_flag == 0,
)
.first()
)
inventory = (
db.query(DbInboundOutbound)
.filter(
DbInboundOutbound.id == link.db_inventory_id,
DbInboundOutbound.tenant_no == tenant_no,
DbInboundOutbound.del_flag == 0,
)
.first()
)
inventory_records.append(
build_transport_inventory_link_read(
db,
plan_product_id=link.plan_product_id,
product_name=(product.product_name if product else "") or "",
db_purchase_plan_id=link.db_purchase_plan_id,
db_inventory_id=link.db_inventory_id,
batch_no=(inventory.batch_no if inventory else "") or "",
plan_no=plan.plan_no or "",
stop_type=link.stop_type,
stop_id=link.stop_id,
business_type=(inventory.business_type if inventory else None),
partner_company=(inventory.partner_company if inventory else None),
)
)
io_order_brief = None
active_io = get_active_io_order_for_plan(db, tenant_no, plan)
if active_io:
kind, header = active_io
from app.services.pl_io_order_service import build_io_order_site_url
token = (header.site_access_token or "").strip()
io_order_brief = TransportIoOrderBrief(
kind=kind,
order_id=header.id,
order_no=header.order_no,
site_access_token=token or None,
site_url=build_io_order_site_url(kind, token) if token else None,
status=header.status,
)
return PlanArrangeInfoRead(
driver_url=driver_url,
license_plate=transport.license_plate or plan.license_plate,
site_records=site_records,
inventory_records=inventory_records,
io_order=io_order_brief,
requires_site_link=requires_site_link,
)
# =============================================================================
# 跟进状态:等通知 / 取消等通知
# =============================================================================
@router.post("/{id}/wait-notice", response_model=PlanRead)
def mark_plan_wait_notice(
id: str,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> PlanRead:
"""将跟进状态从「待安排」设为「等通知」。"""
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
plan = _get_plan(db, tenant_no, id, user)
if plan.follow_status == "cancelled":
raise HTTPException(status_code=400, detail="已取消的计划不能标记等通知")
if plan.archive_status == "archived":
raise HTTPException(status_code=400, detail="已归档的计划不能标记等通知")
if plan.follow_status not in ("pending_arrange", "drafting"):
raise HTTPException(status_code=400, detail="仅待安排或草稿中状态可标记为等通知")
plan.follow_status = "wait_notice"
plan.update_by = user.id
db.commit()
db.refresh(plan)
return plan
@router.post("/{id}/cancel-wait-notice", response_model=PlanRead)
def cancel_plan_wait_notice(
id: str,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> PlanRead:
"""将跟进状态从「等通知」改回「待安排」。"""
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
plan = _get_plan(db, tenant_no, id, user)
if plan.follow_status != "wait_notice":
raise HTTPException(status_code=400, detail="仅等通知状态可取消等通知")
if plan.archive_status == "archived":
raise HTTPException(status_code=400, detail="已归档的计划不能取消等通知")
plan.follow_status = "pending_arrange"
plan.update_by = user.id
db.commit()
db.refresh(plan)
return plan
FORCE_FOLLOW_STATUS_PERM = "pl:plan:force-follow-status"
ARCHIVE_PRICE_EXPORT_FIELDS = frozenset(
{
"unitPrice",
"amount",
"loadUnitPrice",
"loadAmount",
"deliveryUnitPrice",
"deliveryAmount",
"freightPrice",
"freightAmount",
}
)
ALLOWED_FORCE_FOLLOW_STATUSES = frozenset(
{"drafting", "wait_notice", "pending_arrange", "arranged", "completed", "cancelled"}
)
@router.post("/{id}/force-follow-status", response_model=PlanRead)
def force_plan_follow_status(
id: str,
payload: PlanForceFollowStatusBody,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> PlanRead:
"""强制修改跟进状态(仅改字段,不作废联动、不补运输安排)。需权限 pl:plan:force-follow-status。"""
if user.role != SUPER_ADMIN_ROLE:
codes = get_user_menu_codes(db, user.id)
if FORCE_FOLLOW_STATUS_PERM not in codes:
raise HTTPException(status_code=403, detail="无强制修改跟进状态权限")
next_status = (payload.follow_status or "").strip()
if next_status not in ALLOWED_FORCE_FOLLOW_STATUSES:
raise HTTPException(
status_code=400,
detail="跟进状态仅支持:草稿中/等通知/待安排/已安排/已完成/已取消",
)
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
plan = _get_plan(db, tenant_no, id, user)
plan.follow_status = next_status
plan.update_by = user.id
db.commit()
db.refresh(plan)
return plan
# =============================================================================
# 取消计划
# =============================================================================
@router.post("/{id}/cancel", response_model=PlanRead)
def cancel_plan(
id: str,
body: Optional[CancelReasonBody] = Body(default=None),
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> PlanRead:
"""作废运输计划:取消跟进状态、作废关联地磅出入库单、取消运输单。"""
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
plan = _get_plan(db, tenant_no, id, user)
if plan.follow_status == "cancelled":
raise HTTPException(status_code=400, detail="计划已作废")
if plan.archive_status == "archived":
raise HTTPException(status_code=400, detail="已归档的计划不能作废")
if plan.transport_id:
transport = (
db.query(PlTransport)
.filter(
PlTransport.id == plan.transport_id,
PlTransport.tenant_no == tenant_no,
PlTransport.del_flag == 0,
)
.first()
)
if transport:
try:
revoke_regeneratable_dibang_for_transport(
db,
tenant_no=tenant_no,
transport_id=transport.id,
user_id=user.id,
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
revoke_site_links_for_transport(
db,
tenant_no=tenant_no,
transport_id=transport.id,
user_id=user.id,
)
transport.transport_status = "cancelled"
transport.update_by = user.id
void_io_orders_for_plan(
db, tenant_no=tenant_no, plan_id=plan.id, user_id=user.id
)
now = datetime.now()
reason = normalize_optional_reason(body.cancel_reason if body else None)
plan.follow_status = "cancelled"
plan.transport_status = "pending"
plan.cancel_reason = reason
plan.cancel_time = now
plan.cancel_by = user.id
plan.update_by = user.id
db.commit()
db.refresh(plan)
return plan
# =============================================================================
# 归档计划
# =============================================================================
@router.post("/{id}/archive-draft", response_model=PlanWithProducts)
def archive_plan_draft(
id: str,
payload: PlanArchiveBody = Body(...),
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> PlanWithProducts:
"""暂存归档填写内容:不改库存、不标已归档。"""
if not user_can_archive_plan(db, user):
raise HTTPException(status_code=403, detail="无运输计划归档操作权限")
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
plan = _get_plan(db, tenant_no, id, user)
if plan.follow_status == "cancelled":
raise HTTPException(status_code=400, detail="已作废的计划不能暂存归档")
if plan.archive_status == "archived":
raise HTTPException(status_code=400, detail="已归档的计划请使用调整保存,不能暂存")
use_stop_archive = payload.load is not None and payload.deliveries is not None
if use_stop_archive:
apply_archive_stop_quantities(
db,
tenant_no=tenant_no,
plan=plan,
load_block=payload.load,
delivery_blocks=payload.deliveries or [],
user_id=user.id,
strict=False,
)
sync_plan_products_from_stop_archive(
db, tenant_no=tenant_no, plan_id=plan.id, user_id=user.id
)
if payload.products:
apply_archive_brand_qtys_only(
db,
tenant_no=tenant_no,
plan=plan,
items=payload.products,
user_id=user.id,
)
elif payload.products:
apply_archive_product_quantities(
db, tenant_no=tenant_no, plan=plan, items=payload.products, user_id=user.id
)
if payload.freight is not None:
apply_archive_freight(
db,
tenant_no=tenant_no,
plan=plan,
freight=payload.freight,
user_id=user.id,
strict=False,
)
if payload.feedbacks is not None and plan.transport_id:
apply_archive_stop_feedback_attachments(
db,
tenant_no=tenant_no,
plan=plan,
feedbacks=payload.feedbacks,
user_id=user.id,
dept_id=user.dept_id or "",
)
plan.archive_status = "draft"
plan.update_by = user.id
record_plan_change(
db,
tenant_no=tenant_no,
plan_id=plan.id,
change_type="archive_draft",
change_content="暂存归档内容",
user_id=user.id,
user_name=user.name or None,
dept_id=user.dept_id,
)
db.commit()
return get_plan(id, db, user)
@router.post("/{id}/archive", response_model=PlanWithProducts)
def archive_plan(
id: str,
payload: Optional[PlanArchiveBody] = Body(default=None),
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> PlanWithProducts:
if not user_can_archive_plan(db, user):
raise HTTPException(status_code=403, detail="无运输计划归档操作权限")
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
plan = _get_plan(db, tenant_no, id, user)
if plan.follow_status == "cancelled":
raise HTTPException(status_code=400, detail="已作废的计划不能归档")
already_archived = plan.archive_status == "archived"
if not already_archived:
products = (
db.query(PlPlanProduct)
.filter(
PlPlanProduct.plan_id == plan.id,
PlPlanProduct.tenant_no == tenant_no,
PlPlanProduct.del_flag == 0,
)
.order_by(PlPlanProduct.order_num)
.all()
)
validate_hazard_cert_ready_for_transport(db, tenant_no, plan, products)
if payload is None:
if already_archived:
raise HTTPException(status_code=400, detail="计划已归档,请提交归档表单以调整内容")
plan.archive_status = "archived"
plan.complete_status = "yes"
plan.follow_status = "completed"
plan.update_by = user.id
db.commit()
return get_plan(id, db, user)
use_stop_archive = payload.load is not None and payload.deliveries is not None
if use_stop_archive:
if not payload.deliveries:
raise HTTPException(status_code=400, detail="请填写全部送货段归档")
from app.services.pl_contract_plan_qty import (
ARCHIVE_LINE_QTY_OVERAGE_CODE,
collect_archive_line_overages,
)
overages = collect_archive_line_overages(
db,
tenant_no=tenant_no,
plan_id=plan.id,
load_block=payload.load,
delivery_blocks=payload.deliveries,
)
if overages and not payload.confirm_overage:
raise HTTPException(
status_code=422,
detail={
"code": ARCHIVE_LINE_QTY_OVERAGE_CODE,
"message": "归档实际数量超过合同行剩余,请确认超交",
"items": overages,
},
)
apply_archive_stop_quantities(
db,
tenant_no=tenant_no,
plan=plan,
load_block=payload.load,
delivery_blocks=payload.deliveries,
user_id=user.id,
)
sync_plan_products_from_stop_archive(
db, tenant_no=tenant_no, plan_id=plan.id, user_id=user.id
)
if payload.products:
apply_archive_brand_qtys_only(
db,
tenant_no=tenant_no,
plan=plan,
items=payload.products,
user_id=user.id,
)
apply_archive_inventory_for_plan_stops(
db,
tenant_no=tenant_no,
plan=plan,
user_id=user.id,
dept_id=user.dept_id or "",
)
else:
if not payload.products:
raise HTTPException(status_code=400, detail="请提交产品归档明细")
apply_archive_product_quantities(
db, tenant_no=tenant_no, plan=plan, items=payload.products, user_id=user.id
)
apply_archive_inventory_for_plan(
db,
tenant_no=tenant_no,
plan=plan,
user_id=user.id,
dept_id=user.dept_id or "",
)
if payload.freight is not None:
apply_archive_freight(
db, tenant_no=tenant_no, plan=plan, freight=payload.freight, user_id=user.id
)
elif int(plan.has_freight or 0) == 1 and not already_archived:
if not plan.transport_id:
raise HTTPException(status_code=400, detail="计划有运输费用但未安排运输,请先安排运输")
if payload.feedbacks is not None and plan.transport_id:
apply_archive_stop_feedback_attachments(
db,
tenant_no=tenant_no,
plan=plan,
feedbacks=payload.feedbacks,
user_id=user.id,
dept_id=user.dept_id or "",
)
if not already_archived:
plan.archive_status = "archived"
plan.complete_status = "yes"
plan.follow_status = "completed"
plan.update_by = user.id
record_plan_change(
db,
tenant_no=tenant_no,
plan_id=plan.id,
change_type="archive_adjust" if already_archived else "archive",
change_content="调整归档内容" if already_archived else "运输计划归档",
user_id=user.id,
user_name=user.name or None,
dept_id=user.dept_id,
)
db.commit()
return get_plan(id, db, user)
@router.patch("/{id}/driver-feedback", response_model=PlanWithProducts)
def update_plan_driver_feedback(
id: str,
payload: PlanDriverFeedbackUpdateBody,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> PlanWithProducts:
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
plan = _get_plan(db, tenant_no, id, user)
apply_plan_driver_feedback(
db,
tenant_no=tenant_no,
plan=plan,
payload=payload,
user_id=user.id,
)
record_plan_change(
db,
tenant_no=tenant_no,
plan_id=plan.id,
change_type="feedback_adjust",
change_content="归档人员更新司机回传信息",
user_id=user.id,
user_name=user.name or None,
dept_id=user.dept_id,
)
plan.update_by = user.id
db.commit()
return get_plan(id, db, user)
@router.patch("/{id}/site-feedback/{link_id}", response_model=PlanWithProducts)
def update_plan_site_feedback(
id: str,
link_id: str,
payload: PlanSiteFeedbackUpdateBody,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> PlanWithProducts:
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
plan = _get_plan(db, tenant_no, id, user)
link = apply_plan_site_feedback(
db,
tenant_no=tenant_no,
plan=plan,
link_id=link_id,
payload=payload,
user_id=user.id,
)
product_label = (link.product or "现场").strip()
record_plan_change(
db,
tenant_no=tenant_no,
plan_id=plan.id,
change_type="feedback_adjust",
change_content=f"归档人员更新现场回传信息({product_label})",
user_id=user.id,
user_name=user.name or None,
dept_id=user.dept_id,
)
plan.update_by = user.id
db.commit()
return get_plan(id, db, user)
# =============================================================================
# 入库单打印模板
# =============================================================================
@router.get("/{id}/inbound-slip-template")
def get_inbound_slip_template(
id: str,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> str:
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
plan = _get_plan(db, tenant_no, id, user)
return plan.slip_template_content or ""
@router.put("/{id}/inbound-slip-template", status_code=status.HTTP_204_NO_CONTENT)
def save_inbound_slip_template(
id: str,
payload: PlanInboundSlipTemplateSave,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> None:
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
plan = _get_plan(db, tenant_no, id, user)
if plan.plan_type != "inbound":
raise HTTPException(status_code=400, detail="仅入库单计划支持保存入库单打印模板")
if plan.transport_status != "arranged":
raise HTTPException(status_code=400, detail="请先完成运输安排后再保存入库单打印模板")
plan.slip_template_content = payload.slip_template_content or ""
plan.update_by = user.id
db.commit()
# =============================================================================
# 送(提)货单打印模板
# =============================================================================
@router.get("/{id}/print-delivery-slip", response_model=DeliverySlipPrintRead)
def get_print_delivery_slip(
id: str,
customerId: Optional[str] = Query(default=None, description="送货客户ID(多客户时必填)"),
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
):
"""按送货客户聚合送(提)货单打印数据;不传 customerId 时仅返回客户选项。"""
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
plan = _get_plan(db, tenant_no, id, user)
plan_date = plan.plan_date.isoformat() if plan.plan_date else None
try:
return get_delivery_slip_print_context(
db,
tenant_no=tenant_no,
plan_id=plan.id,
plan_no=plan.plan_no or "",
plan_date=plan_date,
customer_id=customerId,
)
except ValueError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
@router.get("/{id}/delivery-slip-template")
def get_delivery_slip_template(
id: str,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> str:
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
plan = _get_plan(db, tenant_no, id, user)
return plan.delivery_slip_template_content or ""
@router.put("/{id}/delivery-slip-template", status_code=status.HTTP_204_NO_CONTENT)
def save_delivery_slip_template(
id: str,
payload: PlanDeliverySlipTemplateSave,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> None:
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
plan = _get_plan(db, tenant_no, id, user)
if plan.plan_type == "inbound":
raise HTTPException(status_code=400, detail="入库单计划请使用入库单打印模板")
if plan.transport_status != "arranged":
raise HTTPException(status_code=400, detail="请先完成运输安排后再保存送(提)货单打印模板")
plan.delivery_slip_template_content = payload.delivery_slip_template_content or ""
plan.update_by = user.id
db.commit()
# =============================================================================
# 分装单打印模板
# =============================================================================
@router.get("/{id}/subpack-slip-template")
def get_subpack_slip_template(
id: str,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> str:
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
plan = _get_plan(db, tenant_no, id, user)
return plan.subpack_slip_template_content or ""
@router.put("/{id}/subpack-slip-template", status_code=status.HTTP_204_NO_CONTENT)
def save_subpack_slip_template(
id: str,
payload: PlanSubpackSlipTemplateSave,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> None:
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
plan = _get_plan(db, tenant_no, id, user)
if plan.plan_type == "inbound":
raise HTTPException(status_code=400, detail="入库单计划请使用入库单打印模板")
plan.subpack_slip_template_content = payload.subpack_slip_template_content or ""
plan.update_by = user.id
db.commit()
@router.post("/{id}/subpack-slip-no")
def ensure_subpack_slip_no(
id: str,
db: Session = Depends(get_db),
user: DbUserContext = Depends(get_db_current_user),
) -> str:
tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
plan = _get_plan(db, tenant_no, id, user)
if plan.plan_type == "inbound":
raise HTTPException(status_code=400, detail="入库单计划不支持分装单号")
slip_no = _ensure_subpack_slip_no(db, tenant_no, plan)
plan.update_by = user.id
db.commit()
return slip_no
# =============================================================================
# 删除计划
# =============================================================================
@router.delete("/{id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_plan(
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())
plan = _get_plan(db, tenant_no, id, user)
if plan.follow_status not in ("pending_arrange", "wait_notice", "drafting"):
raise HTTPException(status_code=400, detail="只能删除待安排、草稿中或等通知且未安排运输的计划")
if plan.transport_status != "pending":
raise HTTPException(status_code=400, detail="已安排运输的计划不能删除")
if plan.archive_status == "archived":
raise HTTPException(status_code=400, detail="已归档的计划不能删除")
assert_row_deletable_by_creator(plan, user_info_from_context(user))
plan.del_flag = 1
plan.update_by = user.id
db.query(PlPlanProduct).filter(
PlPlanProduct.plan_id == id,
PlPlanProduct.tenant_no == tenant_no,
PlPlanProduct.del_flag == 0,
).update({"del_flag": 1, "update_by": user.id}, synchronize_session=False)
db.query(PlPlanTransport).filter(
PlPlanTransport.plan_id == id,
PlPlanTransport.tenant_no == tenant_no,
PlPlanTransport.del_flag == 0,
).update({"del_flag": 1, "update_by": user.id}, synchronize_session=False)
db.commit()
# =============================================================================
# 辅助函数
# =============================================================================
def _get_plan(db: Session, tenant_no: str, plan_id: str, user: DbUserContext) -> PlPlan:
plan = (
db.query(PlPlan)
.filter(
PlPlan.id == plan_id,
PlPlan.tenant_no == tenant_no,
PlPlan.del_flag == 0,
)
.first()
)
if not plan:
raise HTTPException(status_code=404, detail="计划不存在")
assert_pl_row_in_scope(db, plan, PlPlan, user, tenant_no)
return plan