chore(pl): add script to remount orphaned transport dibang stop links.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/scripts/fix_pl_transport_db_link_stop_ids.py

@@ -0,0 +1,337 @@
+"""
+修复 pl_transport_db_link 悬空的 stop_id / plan_product_id。
+
+段主键 upsert 上线前，计划保存会软删重建送货段，导致 link 仍指向旧 stop/product，
+详情「送货信息段次」过滤不到地磅单。
+
+用法（在 backend 目录、已配置 DATABASE_URL / .env）：
+  python scripts/fix_pl_transport_db_link_stop_ids.py --plan-no PO20260904024
+  python scripts/fix_pl_transport_db_link_stop_ids.py --plan-no PO20260904024 --apply
+  python scripts/fix_pl_transport_db_link_stop_ids.py --apply   # 全租户扫描
+"""
+from __future__ import annotations
+
+import argparse
+import os
+import sys
+from dataclasses import dataclass
+from datetime import datetime
+from typing import Optional
+from urllib.parse import urlparse
+
+from sqlalchemy import create_engine
+from sqlalchemy.engine import Engine
+from sqlalchemy.orm import Session, sessionmaker
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from app.models.pl_models import (  # noqa: E402
+    PlPlan,
+    PlPlanDelivery,
+    PlPlanLoad,
+    PlPlanProduct,
+    PlTransportDbLink,
+)
+
+DEFAULT_TENANT = "shanxia2.0"
+
+
+@dataclass
+class FixProposal:
+    link_id: str
+    plan_id: str
+    plan_no: str
+    old_stop_id: Optional[str]
+    new_stop_id: Optional[str]
+    old_plan_product_id: str
+    new_plan_product_id: Optional[str]
+    reason: str
+    can_apply: bool
+
+
+def _engine_from_env() -> Engine:
+    url = os.environ.get("DATABASE_URL")
+    if not url:
+        from app.core.config import settings
+
+        url = settings.DATABASE_URL
+    return create_engine(url, pool_pre_ping=True)
+
+
+def _stop_alive(db: Session, *, stop_type: Optional[str], stop_id: Optional[str], tenant_no: str) -> bool:
+    if not stop_id:
+        return True
+    if stop_type == "load":
+        return (
+            db.query(PlPlanLoad.id)
+            .filter(PlPlanLoad.id == stop_id, PlPlanLoad.tenant_no == tenant_no, PlPlanLoad.del_flag == 0)
+            .first()
+            is not None
+        )
+    if stop_type == "delivery":
+        return (
+            db.query(PlPlanDelivery.id)
+            .filter(
+                PlPlanDelivery.id == stop_id,
+                PlPlanDelivery.tenant_no == tenant_no,
+                PlPlanDelivery.del_flag == 0,
+            )
+            .first()
+            is not None
+        )
+    return True
+
+
+def _product_alive(db: Session, plan_product_id: str) -> bool:
+    if not plan_product_id:
+        return False
+    return (
+        db.query(PlPlanProduct.id)
+        .filter(PlPlanProduct.id == plan_product_id, PlPlanProduct.del_flag == 0)
+        .first()
+        is not None
+    )
+
+
+def _resolve_delivery_stop_id(
+    db: Session,
+    *,
+    plan_id: str,
+    tenant_no: str,
+    old_stop_id: Optional[str],
+) -> tuple[Optional[str], str]:
+    active = (
+        db.query(PlPlanDelivery)
+        .filter(
+            PlPlanDelivery.plan_id == plan_id,
+            PlPlanDelivery.tenant_no == tenant_no,
+            PlPlanDelivery.del_flag == 0,
+        )
+        .order_by(PlPlanDelivery.seq.asc(), PlPlanDelivery.create_time.asc())
+        .all()
+    )
+    if not active:
+        return None, "无有效送货段可改挂"
+    if len(active) == 1:
+        return active[0].id, "仅一条有效送货段，唯一匹配"
+
+    old_name = None
+    if old_stop_id:
+        old = db.query(PlPlanDelivery).filter(PlPlanDelivery.id == old_stop_id).first()
+        if old:
+            old_name = (old.customer_name or "").strip()
+    if old_name:
+        matched = [d for d in active if (d.customer_name or "").strip() == old_name]
+        if len(matched) == 1:
+            return matched[0].id, f"按客户名匹配: {old_name}"
+        if len(matched) > 1:
+            return None, f"同客户名送货段不唯一: {old_name}"
+    return None, "多条送货段且无法唯一匹配客户名"
+
+
+def _resolve_load_stop_id(
+    db: Session,
+    *,
+    plan_id: str,
+    tenant_no: str,
+) -> tuple[Optional[str], str]:
+    active = (
+        db.query(PlPlanLoad)
+        .filter(
+            PlPlanLoad.plan_id == plan_id,
+            PlPlanLoad.tenant_no == tenant_no,
+            PlPlanLoad.del_flag == 0,
+        )
+        .all()
+    )
+    if not active:
+        return None, "无有效装货段可改挂"
+    if len(active) == 1:
+        return active[0].id, "仅一条有效装货段，唯一匹配"
+    return None, "多条装货段无法自动改挂"
+
+
+def _resolve_plan_product_id(
+    db: Session,
+    *,
+    plan_id: str,
+    tenant_no: str,
+    old_plan_product_id: str,
+) -> tuple[Optional[str], str]:
+    if _product_alive(db, old_plan_product_id):
+        return old_plan_product_id, "计划产品仍有效"
+    old = (
+        db.query(PlPlanProduct)
+        .filter(PlPlanProduct.id == old_plan_product_id, PlPlanProduct.plan_id == plan_id)
+        .first()
+    )
+    crm_product_id = (old.product_id if old else "") or ""
+    if not crm_product_id:
+        return None, "旧计划产品无 product_id，无法改挂"
+    active = (
+        db.query(PlPlanProduct)
+        .filter(
+            PlPlanProduct.plan_id == plan_id,
+            PlPlanProduct.tenant_no == tenant_no,
+            PlPlanProduct.del_flag == 0,
+            PlPlanProduct.product_id == crm_product_id,
+        )
+        .order_by(PlPlanProduct.order_num.asc(), PlPlanProduct.create_time.asc())
+        .all()
+    )
+    if len(active) == 1:
+        return active[0].id, f"按 CRM product_id 匹配: {crm_product_id}"
+    if not active:
+        return None, f"无有效计划产品匹配 product_id={crm_product_id}"
+    return None, f"多条计划产品匹配 product_id={crm_product_id}"
+
+
+def collect_proposals(
+    db: Session,
+    *,
+    tenant_no: str,
+    plan_no: Optional[str],
+) -> list[FixProposal]:
+    q = (
+        db.query(PlTransportDbLink, PlPlan)
+        .join(PlPlan, PlPlan.id == PlTransportDbLink.plan_id)
+        .filter(
+            PlTransportDbLink.tenant_no == tenant_no,
+            PlTransportDbLink.del_flag == 0,
+            PlPlan.tenant_no == tenant_no,
+            PlPlan.del_flag == 0,
+        )
+    )
+    if plan_no:
+        q = q.filter(PlPlan.plan_no == plan_no)
+
+    proposals: list[FixProposal] = []
+    for link, plan in q.all():
+        stop_type = (link.stop_type or "").strip() or None
+        old_stop_id = (link.stop_id or "").strip() or None
+        old_pp = (link.plan_product_id or "").strip()
+
+        reasons: list[str] = []
+        new_stop_id = old_stop_id
+        stop_ok = True
+        if not _stop_alive(db, stop_type=stop_type, stop_id=old_stop_id, tenant_no=tenant_no):
+            if stop_type == "delivery":
+                new_stop_id, stop_reason = _resolve_delivery_stop_id(
+                    db, plan_id=plan.id, tenant_no=tenant_no, old_stop_id=old_stop_id
+                )
+            elif stop_type == "load":
+                new_stop_id, stop_reason = _resolve_load_stop_id(
+                    db, plan_id=plan.id, tenant_no=tenant_no
+                )
+            else:
+                new_stop_id, stop_reason = None, f"未知 stop_type={stop_type}"
+            reasons.append(f"stop: {stop_reason}")
+            stop_ok = bool(new_stop_id)
+
+        new_pp = old_pp
+        product_ok = True
+        if not _product_alive(db, old_pp):
+            new_pp, pp_reason = _resolve_plan_product_id(
+                db, plan_id=plan.id, tenant_no=tenant_no, old_plan_product_id=old_pp
+            )
+            reasons.append(f"product: {pp_reason}")
+            product_ok = bool(new_pp)
+
+        if not reasons:
+            continue
+
+        proposals.append(
+            FixProposal(
+                link_id=link.id,
+                plan_id=plan.id,
+                plan_no=plan.plan_no or "",
+                old_stop_id=old_stop_id,
+                new_stop_id=new_stop_id,
+                old_plan_product_id=old_pp,
+                new_plan_product_id=new_pp,
+                reason="; ".join(reasons),
+                can_apply=stop_ok and product_ok,
+            )
+        )
+    return proposals
+
+
+def apply_proposals(db: Session, proposals: list[FixProposal]) -> int:
+    applied = 0
+    now = datetime.now()
+    for p in proposals:
+        if not p.can_apply:
+            print(f"  SKIP link={p.link_id} plan={p.plan_no}: {p.reason}")
+            continue
+        link = (
+            db.query(PlTransportDbLink)
+            .filter(PlTransportDbLink.id == p.link_id, PlTransportDbLink.del_flag == 0)
+            .first()
+        )
+        if not link:
+            continue
+        changed = False
+        if p.new_stop_id and p.new_stop_id != (link.stop_id or ""):
+            link.stop_id = p.new_stop_id
+            changed = True
+        if p.new_plan_product_id and p.new_plan_product_id != (link.plan_product_id or ""):
+            link.plan_product_id = p.new_plan_product_id
+            changed = True
+        if changed:
+            link.update_time = now
+            applied += 1
+            print(
+                f"  APPLY link={p.link_id} plan={p.plan_no}: "
+                f"stop_id {p.old_stop_id} → {p.new_stop_id}; "
+                f"plan_product_id {p.old_plan_product_id} → {p.new_plan_product_id}"
+            )
+    return applied
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="修复悬空的运输地磅 link stop/product id")
+    parser.add_argument("--apply", action="store_true", help="实际写库（默认 dry-run）")
+    parser.add_argument("--plan-no", default=None, help="仅处理指定计划编号")
+    parser.add_argument("--tenant", default=DEFAULT_TENANT, help="租户编号")
+    args = parser.parse_args()
+
+    engine = _engine_from_env()
+    parsed = urlparse(str(engine.url))
+    print(f"目标库: {parsed.path.lstrip('/')} @ {parsed.hostname}:{parsed.port or 3306}")
+    print(f"租户: {args.tenant}")
+    print(f"模式: {'APPLY' if args.apply else 'DRY-RUN'}")
+    if args.plan_no:
+        print(f"计划: {args.plan_no}")
+    print("---")
+
+    SessionLocal = sessionmaker(bind=engine)
+    db = SessionLocal()
+    try:
+        proposals = collect_proposals(db, tenant_no=args.tenant, plan_no=args.plan_no)
+        if not proposals:
+            print("无可修复的悬空 link。")
+            return
+        for p in proposals:
+            flag = "OK" if p.can_apply else "NEED-MANUAL"
+            print(
+                f"[{flag}] link={p.link_id} plan={p.plan_no} ({p.plan_id})\n"
+                f"  stop_id {p.old_stop_id} → {p.new_stop_id}\n"
+                f"  plan_product_id {p.old_plan_product_id} → {p.new_plan_product_id}\n"
+                f"  reason: {p.reason}"
+            )
+        print("---")
+        if not args.apply:
+            print(f"共 {len(proposals)} 条候选。加 --apply 写库。")
+            return
+        n = apply_proposals(db, proposals)
+        db.commit()
+        print(f"已更新 {n} 条 link。")
+    except Exception:
+        db.rollback()
+        raise
+    finally:
+        db.close()
+
+
+if __name__ == "__main__":
+    main()

docs/sql/migrations/2026-09-05-fix-pl-db-link-orphaned-stop.sql

@@ -0,0 +1,17 @@
+-- 备忘：修复 PO20260904024 悬空地磅 link（stop/product 软删重建后未改挂）
+-- 生产确认后执行；亦可使用 scripts/fix_pl_transport_db_link_stop_ids.py --plan-no PO20260904024 --apply
+--
+-- 背景：
+--   link 4AAE759627054CAAB6B5F36D4888CBA4 仍指向软删送货段/产品，
+--   详情按当前段 id 过滤 →「暂无关联地磅单」。
+-- 目标：
+--   stop_id 1D4CB037… → 2036E1B2CEC54509ABC2B22DAA5786F3
+--   plan_product_id DA6B3548… → 537B42A6B7E54D7CBD4C90DC73B185E7
+
+UPDATE pl_transport_db_link
+SET stop_id = '2036E1B2CEC54509ABC2B22DAA5786F3',
+    plan_product_id = '537B42A6B7E54D7CBD4C90DC73B185E7',
+    update_time = NOW()
+WHERE id = '4AAE759627054CAAB6B5F36D4888CBA4'
+  AND plan_id = '551307353C99489786756A3F26C49020'
+  AND del_flag = 0;