shanxia2.0/backend/scripts/fix_pl_transport_db_link_stop_ids.py

"""
修复 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()