1137 lines
36 KiB
Python
1137 lines
36 KiB
Python
"""合同产品在运输计划上的安排数量汇总。"""
|
||
from __future__ import annotations
|
||
|
||
from collections import defaultdict
|
||
from typing import Any, DefaultDict, Dict, Iterable, Optional, Set, Tuple
|
||
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models.pl_models import (
|
||
PlPlan,
|
||
PlPlanContract,
|
||
PlPlanDelivery,
|
||
PlPlanDeliveryProduct,
|
||
PlPlanLoad,
|
||
PlPlanLoadProduct,
|
||
)
|
||
from app.services.pl_plan_service import (
|
||
PLAN_CONTRACT_SIDE_DEMANDER,
|
||
PLAN_CONTRACT_SIDE_SUPPLIER,
|
||
)
|
||
from app.services.pl_plan_stop_service import (
|
||
PLAN_CONTRACT_SIDE_DELIVERY,
|
||
PLAN_CONTRACT_SIDE_LOAD,
|
||
)
|
||
|
||
# 待安排 / 等通知 / 作废：不计入「已安排」（见 _bucket_for_plan）
|
||
|
||
ProductPackageKey = Tuple[str, str]
|
||
|
||
|
||
def normalize_package_spec(value: Any) -> str:
|
||
if value is None:
|
||
return ""
|
||
return str(value).strip()
|
||
|
||
|
||
def product_package_key(product_id: Any, package_spec: Any = None) -> Optional[ProductPackageKey]:
|
||
pid = str(product_id or "").strip()
|
||
if not pid:
|
||
return None
|
||
return (pid, normalize_package_spec(package_spec))
|
||
|
||
|
||
def _to_float(value: Any) -> float:
|
||
if value is None:
|
||
return 0.0
|
||
try:
|
||
return float(value)
|
||
except (TypeError, ValueError):
|
||
return 0.0
|
||
|
||
|
||
def _bucket_for_plan(*, archive_status: Optional[str], follow_status: Optional[str], transport_status: Optional[str]) -> Optional[str]:
|
||
"""返回 'archived' | 'arranged_unarchived' | None（不计入）。"""
|
||
follow = (follow_status or "").strip()
|
||
if follow == "cancelled" or follow in ("wait_notice", "pending_arrange"):
|
||
return None
|
||
archive = (archive_status or "").strip() or "unarchived"
|
||
if archive == "archived":
|
||
return "archived"
|
||
# 已安排未归档（含 archive draft 暂存）
|
||
transport = (transport_status or "").strip()
|
||
if transport == "arranged" or follow in ("arranged", "completed"):
|
||
return "arranged_unarchived"
|
||
return None
|
||
|
||
|
||
def merge_product_qty_maps(
|
||
*,
|
||
archived: Dict[ProductPackageKey, float],
|
||
arranged_unarchived: Dict[ProductPackageKey, float],
|
||
) -> Dict[ProductPackageKey, Dict[str, float]]:
|
||
"""(product_id, package_spec) -> {archivedQty, arrangedUnarchivedQty, arrangedTotalQty}"""
|
||
keys = set(archived) | set(arranged_unarchived)
|
||
out: Dict[ProductPackageKey, Dict[str, float]] = {}
|
||
for key in keys:
|
||
a = round(float(archived.get(key, 0.0)), 3)
|
||
u = round(float(arranged_unarchived.get(key, 0.0)), 3)
|
||
out[key] = {
|
||
"archivedQty": a,
|
||
"arrangedUnarchivedQty": u,
|
||
"arrangedTotalQty": round(a + u, 3),
|
||
}
|
||
return out
|
||
|
||
|
||
# plan：计划数量；archive_actual：装货 inbound / 送货 outbound，null 回退 plan_qty
|
||
QTY_MODE_PLAN = "plan"
|
||
QTY_MODE_ARCHIVE_ACTUAL = "archive_actual"
|
||
|
||
|
||
def _resolve_stop_product_qty(
|
||
row: Any,
|
||
*,
|
||
stop_kind: str,
|
||
qty_mode: str,
|
||
) -> float:
|
||
"""按模式解析段产品计入量。"""
|
||
if qty_mode == QTY_MODE_ARCHIVE_ACTUAL:
|
||
attr = "inbound_qty" if stop_kind == "load" else "outbound_qty"
|
||
raw = getattr(row, attr, None)
|
||
if raw is not None:
|
||
return _to_float(raw)
|
||
return _to_float(getattr(row, "plan_qty", None))
|
||
return _to_float(getattr(row, "plan_qty", None))
|
||
|
||
|
||
def accumulate_stop_products(
|
||
target: DefaultDict[ProductPackageKey, float],
|
||
products: Iterable[Any],
|
||
*,
|
||
stop_kind: str = "load",
|
||
qty_mode: str = QTY_MODE_PLAN,
|
||
) -> None:
|
||
for row in products or []:
|
||
key = product_package_key(
|
||
getattr(row, "product_id", None),
|
||
getattr(row, "package_spec", None),
|
||
)
|
||
if not key:
|
||
continue
|
||
qty = _resolve_stop_product_qty(row, stop_kind=stop_kind, qty_mode=qty_mode)
|
||
if qty:
|
||
target[key] += qty
|
||
|
||
|
||
def resolve_linked_plan_scopes(
|
||
link_rows: Iterable[PlPlanContract],
|
||
legacy_plan_ids: Iterable[str],
|
||
) -> Dict[str, Dict[str, Any]]:
|
||
"""
|
||
plan_id -> {
|
||
load_ids: set[str] | None, # None = 该侧全部装货段
|
||
delivery_ids: set[str] | None,
|
||
use_all_stops: bool, # 遗留主表合同：优先送货段，否则装货段，各产品只计一次
|
||
}
|
||
"""
|
||
scopes: Dict[str, Dict[str, Any]] = {}
|
||
|
||
def _ensure(plan_id: str) -> Dict[str, Any]:
|
||
if plan_id not in scopes:
|
||
scopes[plan_id] = {
|
||
"load_ids": set(),
|
||
"delivery_ids": set(),
|
||
"use_all_stops": False,
|
||
"has_explicit": False,
|
||
}
|
||
return scopes[plan_id]
|
||
|
||
for row in link_rows or []:
|
||
plan_id = str(getattr(row, "plan_id", None) or "").strip()
|
||
if not plan_id:
|
||
continue
|
||
scope = _ensure(plan_id)
|
||
side = (getattr(row, "side", None) or "").strip()
|
||
stop_id = str(getattr(row, "stop_id", None) or "").strip()
|
||
scope["has_explicit"] = True
|
||
if side in (PLAN_CONTRACT_SIDE_LOAD, PLAN_CONTRACT_SIDE_SUPPLIER):
|
||
if stop_id:
|
||
scope["load_ids"].add(stop_id)
|
||
else:
|
||
scope["load_ids"] = None # type: ignore[assignment]
|
||
elif side in (PLAN_CONTRACT_SIDE_DELIVERY, PLAN_CONTRACT_SIDE_DEMANDER):
|
||
if stop_id:
|
||
scope["delivery_ids"].add(stop_id)
|
||
else:
|
||
scope["delivery_ids"] = None # type: ignore[assignment]
|
||
elif stop_id:
|
||
# 未知 side 但有 stop：两边都尝试匹配时由调用方按 id 过滤
|
||
scope["load_ids"].add(stop_id)
|
||
scope["delivery_ids"].add(stop_id)
|
||
|
||
for plan_id in legacy_plan_ids or []:
|
||
pid = str(plan_id or "").strip()
|
||
if not pid:
|
||
continue
|
||
scope = _ensure(pid)
|
||
if not scope["has_explicit"]:
|
||
scope["use_all_stops"] = True
|
||
|
||
return scopes
|
||
|
||
|
||
def sum_qtys_for_scope(
|
||
*,
|
||
load_products_by_load: Dict[str, list],
|
||
delivery_products_by_delivery: Dict[str, list],
|
||
all_load_ids: Set[str],
|
||
all_delivery_ids: Set[str],
|
||
scope: Dict[str, Any],
|
||
qty_mode: str = QTY_MODE_PLAN,
|
||
) -> Dict[ProductPackageKey, float]:
|
||
acc: DefaultDict[ProductPackageKey, float] = defaultdict(float)
|
||
|
||
if scope.get("use_all_stops"):
|
||
# 遗留：优先送货段数量；无送货则用装货，避免装+送双计
|
||
if all_delivery_ids:
|
||
for did in all_delivery_ids:
|
||
accumulate_stop_products(
|
||
acc,
|
||
delivery_products_by_delivery.get(did, []),
|
||
stop_kind="delivery",
|
||
qty_mode=qty_mode,
|
||
)
|
||
else:
|
||
for lid in all_load_ids:
|
||
accumulate_stop_products(
|
||
acc,
|
||
load_products_by_load.get(lid, []),
|
||
stop_kind="load",
|
||
qty_mode=qty_mode,
|
||
)
|
||
return dict(acc)
|
||
|
||
load_ids = scope.get("load_ids")
|
||
delivery_ids = scope.get("delivery_ids")
|
||
|
||
if load_ids is None:
|
||
for lid in all_load_ids:
|
||
accumulate_stop_products(
|
||
acc,
|
||
load_products_by_load.get(lid, []),
|
||
stop_kind="load",
|
||
qty_mode=qty_mode,
|
||
)
|
||
elif load_ids:
|
||
for lid in load_ids:
|
||
accumulate_stop_products(
|
||
acc,
|
||
load_products_by_load.get(lid, []),
|
||
stop_kind="load",
|
||
qty_mode=qty_mode,
|
||
)
|
||
|
||
if delivery_ids is None:
|
||
for did in all_delivery_ids:
|
||
accumulate_stop_products(
|
||
acc,
|
||
delivery_products_by_delivery.get(did, []),
|
||
stop_kind="delivery",
|
||
qty_mode=qty_mode,
|
||
)
|
||
elif delivery_ids:
|
||
for did in delivery_ids:
|
||
accumulate_stop_products(
|
||
acc,
|
||
delivery_products_by_delivery.get(did, []),
|
||
stop_kind="delivery",
|
||
qty_mode=qty_mode,
|
||
)
|
||
|
||
return dict(acc)
|
||
|
||
|
||
def aggregate_contract_product_plan_qtys(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
contract_id: str,
|
||
) -> Dict[ProductPackageKey, Dict[str, float]]:
|
||
"""返回 (productId, packageSpec) -> 三数量字段。"""
|
||
cid = (contract_id or "").strip()
|
||
if not cid:
|
||
return {}
|
||
|
||
link_rows = (
|
||
db.query(PlPlanContract)
|
||
.filter(
|
||
PlPlanContract.tenant_no == tenant_no,
|
||
PlPlanContract.del_flag == 0,
|
||
PlPlanContract.contract_id == cid,
|
||
)
|
||
.all()
|
||
)
|
||
link_plan_ids = {str(r.plan_id) for r in link_rows if r.plan_id}
|
||
|
||
legacy_plans = (
|
||
db.query(PlPlan)
|
||
.filter(
|
||
PlPlan.tenant_no == tenant_no,
|
||
PlPlan.del_flag == 0,
|
||
PlPlan.contract_id == cid,
|
||
)
|
||
.all()
|
||
)
|
||
legacy_only_ids = {p.id for p in legacy_plans if p.id not in link_plan_ids}
|
||
|
||
scopes = resolve_linked_plan_scopes(link_rows, legacy_only_ids)
|
||
plan_ids = list(scopes.keys())
|
||
if not plan_ids:
|
||
return {}
|
||
|
||
plans = (
|
||
db.query(PlPlan)
|
||
.filter(
|
||
PlPlan.tenant_no == tenant_no,
|
||
PlPlan.del_flag == 0,
|
||
PlPlan.id.in_(plan_ids),
|
||
PlPlan.follow_status != "cancelled",
|
||
)
|
||
.all()
|
||
)
|
||
plan_by_id = {p.id: p for p in plans}
|
||
|
||
loads = (
|
||
db.query(PlPlanLoad)
|
||
.filter(
|
||
PlPlanLoad.tenant_no == tenant_no,
|
||
PlPlanLoad.del_flag == 0,
|
||
PlPlanLoad.plan_id.in_(plan_ids),
|
||
)
|
||
.all()
|
||
)
|
||
deliveries = (
|
||
db.query(PlPlanDelivery)
|
||
.filter(
|
||
PlPlanDelivery.tenant_no == tenant_no,
|
||
PlPlanDelivery.del_flag == 0,
|
||
PlPlanDelivery.plan_id.in_(plan_ids),
|
||
)
|
||
.all()
|
||
)
|
||
load_ids_by_plan: DefaultDict[str, Set[str]] = defaultdict(set)
|
||
delivery_ids_by_plan: DefaultDict[str, Set[str]] = defaultdict(set)
|
||
for row in loads:
|
||
load_ids_by_plan[row.plan_id].add(row.id)
|
||
for row in deliveries:
|
||
delivery_ids_by_plan[row.plan_id].add(row.id)
|
||
|
||
all_load_ids = {row.id for row in loads}
|
||
all_delivery_ids = {row.id for row in deliveries}
|
||
|
||
load_products_by_load: DefaultDict[str, list] = defaultdict(list)
|
||
if all_load_ids:
|
||
for prod in (
|
||
db.query(PlPlanLoadProduct)
|
||
.filter(
|
||
PlPlanLoadProduct.tenant_no == tenant_no,
|
||
PlPlanLoadProduct.del_flag == 0,
|
||
PlPlanLoadProduct.load_id.in_(all_load_ids),
|
||
)
|
||
.all()
|
||
):
|
||
load_products_by_load[prod.load_id].append(prod)
|
||
|
||
delivery_products_by_delivery: DefaultDict[str, list] = defaultdict(list)
|
||
if all_delivery_ids:
|
||
for prod in (
|
||
db.query(PlPlanDeliveryProduct)
|
||
.filter(
|
||
PlPlanDeliveryProduct.tenant_no == tenant_no,
|
||
PlPlanDeliveryProduct.del_flag == 0,
|
||
PlPlanDeliveryProduct.delivery_id.in_(all_delivery_ids),
|
||
)
|
||
.all()
|
||
):
|
||
delivery_products_by_delivery[prod.delivery_id].append(prod)
|
||
|
||
archived: DefaultDict[ProductPackageKey, float] = defaultdict(float)
|
||
arranged_unarchived: DefaultDict[ProductPackageKey, float] = defaultdict(float)
|
||
|
||
for plan_id, scope in scopes.items():
|
||
plan = plan_by_id.get(plan_id)
|
||
if not plan:
|
||
continue
|
||
bucket = _bucket_for_plan(
|
||
archive_status=plan.archive_status,
|
||
follow_status=plan.follow_status,
|
||
transport_status=plan.transport_status,
|
||
)
|
||
if not bucket:
|
||
continue
|
||
qty_mode = (
|
||
QTY_MODE_ARCHIVE_ACTUAL if bucket == "archived" else QTY_MODE_PLAN
|
||
)
|
||
qtys = sum_qtys_for_scope(
|
||
load_products_by_load=load_products_by_load,
|
||
delivery_products_by_delivery=delivery_products_by_delivery,
|
||
all_load_ids=load_ids_by_plan.get(plan_id, set()),
|
||
all_delivery_ids=delivery_ids_by_plan.get(plan_id, set()),
|
||
scope=scope,
|
||
qty_mode=qty_mode,
|
||
)
|
||
target = archived if bucket == "archived" else arranged_unarchived
|
||
for pid, qty in qtys.items():
|
||
target[pid] += qty
|
||
|
||
return merge_product_qty_maps(archived=dict(archived), arranged_unarchived=dict(arranged_unarchived))
|
||
|
||
|
||
def attach_plan_qty_fields_to_products(
|
||
products: list,
|
||
qty_by_product_id: Dict[ProductPackageKey, Dict[str, float]],
|
||
) -> None:
|
||
for item in products or []:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
key = product_package_key(
|
||
item.get("productId"),
|
||
item.get("packageSpec") or item.get("package_spec"),
|
||
)
|
||
stats = (qty_by_product_id.get(key) if key else None) or {
|
||
"archivedQty": 0.0,
|
||
"arrangedUnarchivedQty": 0.0,
|
||
"arrangedTotalQty": 0.0,
|
||
}
|
||
item["archivedQty"] = stats["archivedQty"]
|
||
item["arrangedUnarchivedQty"] = stats["arrangedUnarchivedQty"]
|
||
item["arrangedTotalQty"] = stats["arrangedTotalQty"]
|
||
|
||
|
||
def plan_ids_linked_to_contract(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
contract_id: str,
|
||
) -> Set[str]:
|
||
cid = (contract_id or "").strip()
|
||
if not cid:
|
||
return set()
|
||
ids: Set[str] = set()
|
||
for (plan_id,) in (
|
||
db.query(PlPlanContract.plan_id)
|
||
.filter(
|
||
PlPlanContract.tenant_no == tenant_no,
|
||
PlPlanContract.del_flag == 0,
|
||
PlPlanContract.contract_id == cid,
|
||
)
|
||
.distinct()
|
||
.all()
|
||
):
|
||
if plan_id:
|
||
ids.add(str(plan_id))
|
||
for (plan_id,) in (
|
||
db.query(PlPlan.id)
|
||
.filter(
|
||
PlPlan.tenant_no == tenant_no,
|
||
PlPlan.del_flag == 0,
|
||
PlPlan.contract_id == cid,
|
||
)
|
||
.all()
|
||
):
|
||
if plan_id:
|
||
ids.add(str(plan_id))
|
||
return ids
|
||
|
||
|
||
def contract_ids_linked_to_plan(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
plan: PlPlan,
|
||
) -> Set[str]:
|
||
ids: Set[str] = set()
|
||
if plan.contract_id:
|
||
ids.add(str(plan.contract_id).strip())
|
||
for (cid,) in (
|
||
db.query(PlPlanContract.contract_id)
|
||
.filter(
|
||
PlPlanContract.tenant_no == tenant_no,
|
||
PlPlanContract.del_flag == 0,
|
||
PlPlanContract.plan_id == plan.id,
|
||
)
|
||
.distinct()
|
||
.all()
|
||
):
|
||
if cid:
|
||
ids.add(str(cid).strip())
|
||
return {x for x in ids if x}
|
||
|
||
|
||
def auto_start_execute_approved_contracts(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
plan: PlPlan,
|
||
user_id: Optional[str] = None,
|
||
) -> list[str]:
|
||
"""将关联且状态为 approved 的合同置为 executing（不 commit）。返回已推进的合同 id。"""
|
||
from datetime import datetime
|
||
|
||
from app.models.crm_models import CrmContract
|
||
|
||
started: list[str] = []
|
||
for cid in contract_ids_linked_to_plan(db, tenant_no=tenant_no, plan=plan):
|
||
row = (
|
||
db.query(CrmContract)
|
||
.filter(
|
||
CrmContract.id == cid,
|
||
CrmContract.tenant_no == tenant_no,
|
||
CrmContract.del_flag == 0,
|
||
)
|
||
.first()
|
||
)
|
||
if not row or row.status != "approved":
|
||
continue
|
||
row.status = "executing"
|
||
row.update_by = user_id
|
||
row.update_time = datetime.now()
|
||
started.append(cid)
|
||
return started
|
||
|
||
|
||
# =============================================================================
|
||
# 安排运输：合同数量硬拦 / 成功后提醒
|
||
# =============================================================================
|
||
|
||
FRAMEWORK_CONTRACT_TYPES = frozenset({"framework", "framework_sale", "framework_purchase"})
|
||
|
||
|
||
def effective_arranged_total(
|
||
*,
|
||
arranged_total: float,
|
||
old_plan_qty: float,
|
||
new_plan_qty: float,
|
||
) -> float:
|
||
"""已安排合计 − 本计划旧量 + 本计划新量。"""
|
||
return round(
|
||
_to_float(arranged_total) - _to_float(old_plan_qty) + _to_float(new_plan_qty),
|
||
3,
|
||
)
|
||
|
||
|
||
def contract_product_has_fixed_qty(
|
||
*,
|
||
contract_type: Optional[str],
|
||
quantity: Any,
|
||
quantity_mode: Optional[str],
|
||
) -> bool:
|
||
"""合同产品行是否按固定数量做超量硬拦。"""
|
||
ctype = (contract_type or "").strip()
|
||
mode = (quantity_mode or "").strip() or "fixed"
|
||
if ctype in FRAMEWORK_CONTRACT_TYPES and mode == "customer_demand":
|
||
return False
|
||
qty = _to_float(quantity)
|
||
return qty > 0
|
||
|
||
|
||
def evaluate_product_qty_for_arrange(
|
||
*,
|
||
has_fixed_qty: bool,
|
||
contract_qty: Optional[float],
|
||
arranged_total: float,
|
||
old_plan_qty: float,
|
||
new_plan_qty: float,
|
||
) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
返回:
|
||
- None: 跳过（本次无量）
|
||
- kind=violation | remaining | framework_cumulative
|
||
"""
|
||
new_qty = _to_float(new_plan_qty)
|
||
if new_qty <= 0:
|
||
return None
|
||
|
||
old_qty = _to_float(old_plan_qty)
|
||
arranged = _to_float(arranged_total)
|
||
effective = effective_arranged_total(
|
||
arranged_total=arranged,
|
||
old_plan_qty=old_qty,
|
||
new_plan_qty=new_qty,
|
||
)
|
||
|
||
if not has_fixed_qty:
|
||
return {
|
||
"kind": "framework_cumulative",
|
||
"arrangedTotalQty": effective,
|
||
"thisPlanQty": round(new_qty, 3),
|
||
"effectiveTotal": effective,
|
||
}
|
||
|
||
cq = _to_float(contract_qty)
|
||
overflow = round(effective - cq, 3)
|
||
base = {
|
||
"contractQty": round(cq, 3),
|
||
"arrangedTotalQty": effective,
|
||
"thisPlanQty": round(new_qty, 3),
|
||
"oldPlanQty": round(old_qty, 3),
|
||
"newPlanQty": round(new_qty, 3),
|
||
"effectiveTotal": effective,
|
||
}
|
||
if overflow > 0:
|
||
return {**base, "kind": "violation", "overflowQty": overflow}
|
||
return {
|
||
**base,
|
||
"kind": "remaining",
|
||
"remainingQty": round(cq - effective, 3),
|
||
}
|
||
|
||
|
||
CONTRACT_QTY_VIOLATION_CODE = "CONTRACT_QTY_OVERFLOW"
|
||
|
||
|
||
def _stop_kind_from_scope(scope: Dict[str, Any]) -> str:
|
||
load_ids = scope.get("load_ids")
|
||
delivery_ids = scope.get("delivery_ids")
|
||
has_load = load_ids is None or bool(load_ids)
|
||
has_delivery = delivery_ids is None or bool(delivery_ids)
|
||
if scope.get("use_all_stops"):
|
||
return "delivery"
|
||
if has_load and not has_delivery:
|
||
return "load"
|
||
if has_delivery and not has_load:
|
||
return "delivery"
|
||
if has_load:
|
||
return "load"
|
||
return "delivery"
|
||
|
||
|
||
def _plan_scope_for_contract(
|
||
*,
|
||
plan_id: str,
|
||
contract_id: str,
|
||
link_rows: list[PlPlanContract],
|
||
legacy_contract_id: Optional[str],
|
||
) -> Dict[str, Any]:
|
||
cid = (contract_id or "").strip()
|
||
links = [r for r in link_rows if str(r.contract_id or "").strip() == cid]
|
||
legacy_ids: list[str] = []
|
||
if not links and legacy_contract_id and str(legacy_contract_id).strip() == cid:
|
||
legacy_ids = [plan_id]
|
||
scopes = resolve_linked_plan_scopes(links, legacy_ids)
|
||
return scopes.get(plan_id) or {
|
||
"load_ids": set(),
|
||
"delivery_ids": set(),
|
||
"use_all_stops": False,
|
||
"has_explicit": False,
|
||
}
|
||
|
||
|
||
def _load_plan_stop_product_maps(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
plan_id: str,
|
||
) -> tuple[Set[str], Set[str], Dict[str, list], Dict[str, list]]:
|
||
loads = (
|
||
db.query(PlPlanLoad)
|
||
.filter(
|
||
PlPlanLoad.tenant_no == tenant_no,
|
||
PlPlanLoad.del_flag == 0,
|
||
PlPlanLoad.plan_id == plan_id,
|
||
)
|
||
.all()
|
||
)
|
||
deliveries = (
|
||
db.query(PlPlanDelivery)
|
||
.filter(
|
||
PlPlanDelivery.tenant_no == tenant_no,
|
||
PlPlanDelivery.del_flag == 0,
|
||
PlPlanDelivery.plan_id == plan_id,
|
||
)
|
||
.all()
|
||
)
|
||
load_ids = {row.id for row in loads}
|
||
delivery_ids = {row.id for row in deliveries}
|
||
load_products_by_load: DefaultDict[str, list] = defaultdict(list)
|
||
if load_ids:
|
||
for prod in (
|
||
db.query(PlPlanLoadProduct)
|
||
.filter(
|
||
PlPlanLoadProduct.tenant_no == tenant_no,
|
||
PlPlanLoadProduct.del_flag == 0,
|
||
PlPlanLoadProduct.load_id.in_(load_ids),
|
||
)
|
||
.all()
|
||
):
|
||
load_products_by_load[prod.load_id].append(prod)
|
||
delivery_products_by_delivery: DefaultDict[str, list] = defaultdict(list)
|
||
if delivery_ids:
|
||
for prod in (
|
||
db.query(PlPlanDeliveryProduct)
|
||
.filter(
|
||
PlPlanDeliveryProduct.tenant_no == tenant_no,
|
||
PlPlanDeliveryProduct.del_flag == 0,
|
||
PlPlanDeliveryProduct.delivery_id.in_(delivery_ids),
|
||
)
|
||
.all()
|
||
):
|
||
delivery_products_by_delivery[prod.delivery_id].append(prod)
|
||
return load_ids, delivery_ids, dict(load_products_by_load), dict(delivery_products_by_delivery)
|
||
|
||
|
||
def collect_plan_contract_qty_checks(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
plan_id: str,
|
||
) -> tuple[list[Dict[str, Any]], list[Dict[str, Any]]]:
|
||
"""
|
||
返回 (violations, reminders)。
|
||
在写运输前调用；reminders 可缓存到成功响应，勿在状态变更后再算。
|
||
"""
|
||
from app.models.crm_models import CrmContract, CrmContractProduct
|
||
|
||
plan = (
|
||
db.query(PlPlan)
|
||
.filter(
|
||
PlPlan.id == plan_id,
|
||
PlPlan.tenant_no == tenant_no,
|
||
PlPlan.del_flag == 0,
|
||
)
|
||
.first()
|
||
)
|
||
if not plan:
|
||
return [], []
|
||
|
||
link_rows = (
|
||
db.query(PlPlanContract)
|
||
.filter(
|
||
PlPlanContract.tenant_no == tenant_no,
|
||
PlPlanContract.del_flag == 0,
|
||
PlPlanContract.plan_id == plan_id,
|
||
)
|
||
.all()
|
||
)
|
||
contract_ids: Set[str] = set()
|
||
for row in link_rows:
|
||
cid = str(row.contract_id or "").strip()
|
||
if cid:
|
||
contract_ids.add(cid)
|
||
legacy_cid = str(plan.contract_id or "").strip()
|
||
if legacy_cid:
|
||
contract_ids.add(legacy_cid)
|
||
if not contract_ids:
|
||
return [], []
|
||
|
||
load_ids, delivery_ids, load_products_by_load, delivery_products_by_delivery = (
|
||
_load_plan_stop_product_maps(db, tenant_no=tenant_no, plan_id=plan_id)
|
||
)
|
||
in_bucket = (
|
||
_bucket_for_plan(
|
||
archive_status=plan.archive_status,
|
||
follow_status=plan.follow_status,
|
||
transport_status=plan.transport_status,
|
||
)
|
||
is not None
|
||
)
|
||
|
||
violations: list[Dict[str, Any]] = []
|
||
reminders: list[Dict[str, Any]] = []
|
||
|
||
for cid in sorted(contract_ids):
|
||
contract = (
|
||
db.query(CrmContract)
|
||
.filter(
|
||
CrmContract.id == cid,
|
||
CrmContract.tenant_no == tenant_no,
|
||
CrmContract.del_flag == 0,
|
||
)
|
||
.first()
|
||
)
|
||
if not contract:
|
||
continue
|
||
|
||
scope = _plan_scope_for_contract(
|
||
plan_id=plan_id,
|
||
contract_id=cid,
|
||
link_rows=link_rows,
|
||
legacy_contract_id=legacy_cid,
|
||
)
|
||
|
||
plan_qtys = sum_qtys_for_scope(
|
||
load_products_by_load=load_products_by_load,
|
||
delivery_products_by_delivery=delivery_products_by_delivery,
|
||
all_load_ids=load_ids,
|
||
all_delivery_ids=delivery_ids,
|
||
scope=scope,
|
||
)
|
||
if not plan_qtys:
|
||
continue
|
||
|
||
qty_map = aggregate_contract_product_plan_qtys(
|
||
db, tenant_no=tenant_no, contract_id=cid
|
||
)
|
||
products = (
|
||
db.query(CrmContractProduct)
|
||
.filter(
|
||
CrmContractProduct.contract_id == cid,
|
||
CrmContractProduct.tenant_no == tenant_no,
|
||
CrmContractProduct.del_flag == 0,
|
||
)
|
||
.all()
|
||
)
|
||
products_by_pid = {
|
||
str(p.product_id or "").strip(): p for p in products if p.product_id
|
||
}
|
||
stop_kind = _stop_kind_from_scope(scope)
|
||
contract_type = (contract.contract_type or "").strip()
|
||
contract_no = contract.contract_no or ""
|
||
customer_name = (contract.customer_name or "").strip()
|
||
|
||
for pid, new_qty_raw in plan_qtys.items():
|
||
new_qty = _to_float(new_qty_raw)
|
||
if new_qty <= 0:
|
||
continue
|
||
cp = products_by_pid.get(pid)
|
||
# 计划产品不在合同明细：不校验数量（归属另有规则）
|
||
if not cp:
|
||
continue
|
||
|
||
arranged_total = _to_float((qty_map.get(pid) or {}).get("arrangedTotalQty"))
|
||
if in_bucket:
|
||
old_qty = new_qty
|
||
# arranged 已含当前库内量；effective = arranged - old + new = arranged
|
||
else:
|
||
old_qty = 0.0
|
||
|
||
has_fixed = contract_product_has_fixed_qty(
|
||
contract_type=contract_type,
|
||
quantity=cp.quantity,
|
||
quantity_mode=cp.quantity_mode,
|
||
)
|
||
evaluated = evaluate_product_qty_for_arrange(
|
||
has_fixed_qty=has_fixed,
|
||
contract_qty=_to_float(cp.quantity) if has_fixed else None,
|
||
arranged_total=arranged_total,
|
||
old_plan_qty=old_qty,
|
||
new_plan_qty=new_qty,
|
||
)
|
||
if not evaluated:
|
||
continue
|
||
|
||
item = {
|
||
"contractId": cid,
|
||
"contractNo": contract_no,
|
||
"contractType": contract_type,
|
||
"customerName": customer_name,
|
||
"stopKind": stop_kind,
|
||
"productId": pid,
|
||
"productName": cp.product_name or "",
|
||
"model": (cp.model or "").strip() or None,
|
||
"unit": (cp.unit or "").strip() or None,
|
||
"thisPlanQty": evaluated.get("thisPlanQty", round(new_qty, 3)),
|
||
"arrangedTotalQty": evaluated["arrangedTotalQty"],
|
||
}
|
||
kind = evaluated["kind"]
|
||
if kind == "violation":
|
||
violations.append(
|
||
{
|
||
**item,
|
||
"contractQty": evaluated["contractQty"],
|
||
"oldPlanQty": evaluated["oldPlanQty"],
|
||
"newPlanQty": evaluated["newPlanQty"],
|
||
"effectiveTotal": evaluated["effectiveTotal"],
|
||
"overflowQty": evaluated["overflowQty"],
|
||
# 与提醒表同列：剩余可为负，表示超量
|
||
"remainingQty": round(
|
||
evaluated["contractQty"] - evaluated["effectiveTotal"], 3
|
||
),
|
||
}
|
||
)
|
||
elif kind == "remaining":
|
||
reminders.append(
|
||
{
|
||
**item,
|
||
"mode": "remaining",
|
||
"contractQty": evaluated["contractQty"],
|
||
"remainingQty": evaluated["remainingQty"],
|
||
}
|
||
)
|
||
elif kind == "framework_cumulative":
|
||
reminders.append(
|
||
{
|
||
**item,
|
||
"mode": "framework_cumulative",
|
||
}
|
||
)
|
||
|
||
return violations, reminders
|
||
|
||
|
||
def assert_contract_qty_for_arrange(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
plan_id: str,
|
||
) -> list[Dict[str, Any]]:
|
||
"""超限则 422；通过则返回 reminders（供成功响应复用）。"""
|
||
from fastapi import HTTPException
|
||
|
||
violations, reminders = collect_plan_contract_qty_checks(
|
||
db, tenant_no=tenant_no, plan_id=plan_id
|
||
)
|
||
if violations:
|
||
raise HTTPException(
|
||
status_code=422,
|
||
detail={
|
||
"code": CONTRACT_QTY_VIOLATION_CODE,
|
||
"message": "合同产品安排数量超过合同数量",
|
||
"items": violations,
|
||
},
|
||
)
|
||
return reminders
|
||
|
||
|
||
def build_contract_qty_reminder_payload(
|
||
reminders: list[Dict[str, Any]],
|
||
) -> Optional[Dict[str, Any]]:
|
||
if not reminders:
|
||
return None
|
||
return {"items": reminders}
|
||
|
||
|
||
def get_contract_product_arrange_qty_detail(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
contract_id: str,
|
||
product_id: str,
|
||
exclude_plan_id: Optional[str] = None,
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
表单「可安排数量」与问号弹窗数据。
|
||
arrangedQty / availableQty 均不含 exclude_plan_id（若该计划已在已安排桶中）。
|
||
"""
|
||
from app.models.crm_models import CrmContract, CrmContractProduct
|
||
|
||
cid = (contract_id or "").strip()
|
||
pid = (product_id or "").strip()
|
||
exclude_id = (exclude_plan_id or "").strip() or None
|
||
empty = {
|
||
"contractId": cid,
|
||
"contractNo": None,
|
||
"customerName": None,
|
||
"productId": pid,
|
||
"productName": None,
|
||
"hasFixedQty": False,
|
||
"contractQty": None,
|
||
"arrangedQty": 0.0,
|
||
"availableQty": None,
|
||
"plans": [],
|
||
}
|
||
if not cid or not pid:
|
||
return empty
|
||
|
||
contract = (
|
||
db.query(CrmContract)
|
||
.filter(
|
||
CrmContract.id == cid,
|
||
CrmContract.tenant_no == tenant_no,
|
||
CrmContract.del_flag == 0,
|
||
)
|
||
.first()
|
||
)
|
||
if not contract:
|
||
return empty
|
||
|
||
cp = (
|
||
db.query(CrmContractProduct)
|
||
.filter(
|
||
CrmContractProduct.contract_id == cid,
|
||
CrmContractProduct.product_id == pid,
|
||
CrmContractProduct.tenant_no == tenant_no,
|
||
CrmContractProduct.del_flag == 0,
|
||
)
|
||
.first()
|
||
)
|
||
contract_type = (contract.contract_type or "").strip()
|
||
has_fixed = False
|
||
contract_qty: Optional[float] = None
|
||
if cp:
|
||
has_fixed = contract_product_has_fixed_qty(
|
||
contract_type=contract_type,
|
||
quantity=cp.quantity,
|
||
quantity_mode=cp.quantity_mode,
|
||
)
|
||
if has_fixed:
|
||
contract_qty = round(_to_float(cp.quantity), 3)
|
||
|
||
qty_map = aggregate_contract_product_plan_qtys(
|
||
db, tenant_no=tenant_no, contract_id=cid
|
||
)
|
||
arranged_including = _to_float((qty_map.get(pid) or {}).get("arrangedTotalQty"))
|
||
|
||
link_rows = (
|
||
db.query(PlPlanContract)
|
||
.filter(
|
||
PlPlanContract.tenant_no == tenant_no,
|
||
PlPlanContract.del_flag == 0,
|
||
PlPlanContract.contract_id == cid,
|
||
)
|
||
.all()
|
||
)
|
||
link_plan_ids = {str(r.plan_id) for r in link_rows if r.plan_id}
|
||
legacy_plans = (
|
||
db.query(PlPlan)
|
||
.filter(
|
||
PlPlan.tenant_no == tenant_no,
|
||
PlPlan.del_flag == 0,
|
||
PlPlan.contract_id == cid,
|
||
)
|
||
.all()
|
||
)
|
||
legacy_only_ids = {p.id for p in legacy_plans if p.id not in link_plan_ids}
|
||
scopes = resolve_linked_plan_scopes(link_rows, legacy_only_ids)
|
||
plan_ids = list(scopes.keys())
|
||
|
||
exclude_qty = 0.0
|
||
plans_out: list[Dict[str, Any]] = []
|
||
|
||
if plan_ids:
|
||
plans = (
|
||
db.query(PlPlan)
|
||
.filter(
|
||
PlPlan.tenant_no == tenant_no,
|
||
PlPlan.del_flag == 0,
|
||
PlPlan.id.in_(plan_ids),
|
||
PlPlan.follow_status != "cancelled",
|
||
)
|
||
.all()
|
||
)
|
||
plan_by_id = {p.id: p for p in plans}
|
||
|
||
loads = (
|
||
db.query(PlPlanLoad)
|
||
.filter(
|
||
PlPlanLoad.tenant_no == tenant_no,
|
||
PlPlanLoad.del_flag == 0,
|
||
PlPlanLoad.plan_id.in_(plan_ids),
|
||
)
|
||
.all()
|
||
)
|
||
deliveries = (
|
||
db.query(PlPlanDelivery)
|
||
.filter(
|
||
PlPlanDelivery.tenant_no == tenant_no,
|
||
PlPlanDelivery.del_flag == 0,
|
||
PlPlanDelivery.plan_id.in_(plan_ids),
|
||
)
|
||
.all()
|
||
)
|
||
load_ids_by_plan: DefaultDict[str, Set[str]] = defaultdict(set)
|
||
delivery_ids_by_plan: DefaultDict[str, Set[str]] = defaultdict(set)
|
||
for row in loads:
|
||
load_ids_by_plan[row.plan_id].add(row.id)
|
||
for row in deliveries:
|
||
delivery_ids_by_plan[row.plan_id].add(row.id)
|
||
|
||
all_load_ids = {row.id for row in loads}
|
||
all_delivery_ids = {row.id for row in deliveries}
|
||
load_products_by_load: DefaultDict[str, list] = defaultdict(list)
|
||
if all_load_ids:
|
||
for prod in (
|
||
db.query(PlPlanLoadProduct)
|
||
.filter(
|
||
PlPlanLoadProduct.tenant_no == tenant_no,
|
||
PlPlanLoadProduct.del_flag == 0,
|
||
PlPlanLoadProduct.load_id.in_(all_load_ids),
|
||
)
|
||
.all()
|
||
):
|
||
load_products_by_load[prod.load_id].append(prod)
|
||
delivery_products_by_delivery: DefaultDict[str, list] = defaultdict(list)
|
||
if all_delivery_ids:
|
||
for prod in (
|
||
db.query(PlPlanDeliveryProduct)
|
||
.filter(
|
||
PlPlanDeliveryProduct.tenant_no == tenant_no,
|
||
PlPlanDeliveryProduct.del_flag == 0,
|
||
PlPlanDeliveryProduct.delivery_id.in_(all_delivery_ids),
|
||
)
|
||
.all()
|
||
):
|
||
delivery_products_by_delivery[prod.delivery_id].append(prod)
|
||
|
||
for plan_id, scope in scopes.items():
|
||
plan = plan_by_id.get(plan_id)
|
||
if not plan:
|
||
continue
|
||
bucket = _bucket_for_plan(
|
||
archive_status=plan.archive_status,
|
||
follow_status=plan.follow_status,
|
||
transport_status=plan.transport_status,
|
||
)
|
||
if not bucket:
|
||
continue
|
||
qty_mode = (
|
||
QTY_MODE_ARCHIVE_ACTUAL if bucket == "archived" else QTY_MODE_PLAN
|
||
)
|
||
qtys = sum_qtys_for_scope(
|
||
load_products_by_load=load_products_by_load,
|
||
delivery_products_by_delivery=delivery_products_by_delivery,
|
||
all_load_ids=load_ids_by_plan.get(plan_id, set()),
|
||
all_delivery_ids=delivery_ids_by_plan.get(plan_id, set()),
|
||
scope=scope,
|
||
qty_mode=qty_mode,
|
||
)
|
||
pq = _to_float(qtys.get(pid))
|
||
if pq <= 0:
|
||
continue
|
||
if exclude_id and str(plan_id) == exclude_id:
|
||
exclude_qty = pq
|
||
continue
|
||
plan_date = getattr(plan, "plan_date", None)
|
||
plans_out.append(
|
||
{
|
||
"planId": plan_id,
|
||
"planNo": plan.plan_no or "",
|
||
"planDate": plan_date.isoformat() if plan_date else None,
|
||
"planQty": round(pq, 3),
|
||
"planType": getattr(plan, "plan_type", None),
|
||
"belongsCompany": getattr(plan, "belongs_company", None),
|
||
"supplierName": getattr(plan, "supplier_name", None),
|
||
"customerName": getattr(plan, "customer_name", None),
|
||
"licensePlate": getattr(plan, "license_plate", None),
|
||
"driverName": getattr(plan, "driver_name", None),
|
||
"followUserName": getattr(plan, "follow_user_name", None),
|
||
"followStatus": plan.follow_status,
|
||
"transportStatus": plan.transport_status,
|
||
"archiveStatus": plan.archive_status,
|
||
"createTime": plan.create_time.isoformat(sep=" ", timespec="seconds")
|
||
if getattr(plan, "create_time", None)
|
||
else None,
|
||
}
|
||
)
|
||
|
||
plans_out.sort(key=lambda x: (x.get("planDate") or "", x.get("planNo") or ""), reverse=True)
|
||
|
||
arranged_qty = round(max(0.0, arranged_including - exclude_qty), 3)
|
||
available_qty = None
|
||
if has_fixed and contract_qty is not None:
|
||
available_qty = round(contract_qty - arranged_qty, 3)
|
||
|
||
return {
|
||
"contractId": cid,
|
||
"contractNo": contract.contract_no or "",
|
||
"customerName": (contract.customer_name or "").strip() or None,
|
||
"productId": pid,
|
||
"productName": (cp.product_name if cp else None) or None,
|
||
"hasFixedQty": has_fixed,
|
||
"contractQty": contract_qty,
|
||
"arrangedQty": arranged_qty,
|
||
"availableQty": available_qty,
|
||
"plans": plans_out,
|
||
}
|