sqxp/backend/app/services/settlement_reports.py

"""结算报表导出行数据组装（01–06）。

无系统数据的字段返回 None，禁止用 0 冒充已付/欠款/费率等空列。
"""
from __future__ import annotations

from datetime import datetime
from decimal import Decimal
from typing import Any, Dict, List, Optional, Sequence, Tuple

from sqlalchemy import func, or_
from sqlalchemy.orm import Session

from app.models.models import (
    BasProduct,
    BasProductUnit,
    BizDeliveryTask,
    BizDeliveryTaskItem,
    BizOrder,
    BizOrderItem,
    BizPriceList,
    BizPriceListItem,
    SysUser,
    SysUserHis,
)

EXPORT_ROW_LIMIT = 20000

_ORDER_STATUS_TEXT = {
    "pending_payment": "待支付",
    "pending_confirm": "待确认收货",
    "completed": "已完结",
    "cancelled": "已取消",
}

_SUPPLIER_SETTLEMENT_TEXT = {
    "pending": "待供应商结算",
    "settled": "已与供应商结算",
}

_DELIVERY_STATUS_TEXT = {
    "pending": "待取货",
    "assigned": "已分配",
    "supplier_stocking": "供应商备货中",
    "supplier_ready": "备货完成（待取货）",
    "picking": "取货中",
    "delivering": "配送中",
    "completed": "已完成",
    "cancelled": "已取消",
}


def _user_code_name(user: Optional[SysUser]) -> Tuple[Optional[str], Optional[str]]:
    if not user:
        return None, None
    code = (user.username or "").strip() or user.id
    name = (user.real_name or user.username or "").strip() or None
    return code, name


def _person_display_name(user: Optional[Any]) -> Optional[str]:
    """人员展示名：真实姓名优先，其次用户名。"""
    if not user:
        return None
    return (getattr(user, "real_name", None) or getattr(user, "username", None) or "").strip() or None


def _supplier_display_name(user: Optional[Any]) -> Optional[str]:
    """供应商名称：用户名（商户名）优先，其次真实姓名。"""
    if not user:
        return None
    return (getattr(user, "username", None) or getattr(user, "real_name", None) or "").strip() or None


def _client_site_display_name(
    user: Optional[Any],
    *,
    contact_person: Optional[str] = None,
) -> Optional[str]:
    """客户名称/站点：用户名（站点）优先，其次真实姓名，再回退联系人。"""
    if user:
        un = (getattr(user, "username", None) or "").strip() or None
        rn = (getattr(user, "real_name", None) or "").strip() or None
        if un and un != "客户":
            return un
        if rn and rn != "客户":
            return rn
        if un or rn:
            return un or rn
    return (contact_person or "").strip() or None


def _user_map_by_ids(db: Session, user_ids: Sequence[Optional[str]]) -> Dict[str, Any]:
    """按 ID 取用户；含已软删；缺失时回退 sys_user_his 最新快照。"""
    ids = list({str(x).strip() for x in user_ids if x and str(x).strip()})
    if not ids:
        return {}
    out: Dict[str, Any] = {
        u.id: u for u in db.query(SysUser).filter(SysUser.id.in_(ids)).all()
    }
    missing = [i for i in ids if i not in out]
    if not missing:
        return out
    his_rows = (
        db.query(SysUserHis)
        .filter(SysUserHis.id.in_(missing))
        .order_by(
            SysUserHis.update_at.desc(),
            SysUserHis.his_create_at.desc(),
        )
        .all()
    )
    for h in his_rows:
        if h.id not in out:
            out[h.id] = h
    return out


def _supplier_unit_yuan_from_pli(pli: Optional[BizPriceListItem]) -> Optional[float]:
    """与 admin 结算列表一致：供货单价 = price / (1 + 调价系数/100)。"""
    if pli is None or getattr(pli, "price", None) is None:
        return None
    final_price = Decimal(str(pli.price))
    raw_coeff = getattr(pli, "pricing_coefficient", None)
    if raw_coeff is None:
        coeff_pct = Decimal("0")
    else:
        try:
            coeff_pct = Decimal(str(raw_coeff))
        except Exception:
            coeff_pct = Decimal("0")
    factor = Decimal("1") + coeff_pct / Decimal("100")
    if factor <= Decimal("0"):
        factor = Decimal("1")
    return float((final_price / factor).quantize(Decimal("0.01")))


def _price_list_item_map(db: Session, price_list_ids: List[str]) -> dict:
    if not price_list_ids:
        return {}
    rows = (
        db.query(BizPriceListItem)
        .filter(BizPriceListItem.price_list_id.in_(price_list_ids), BizPriceListItem.is_del == 0)
        .all()
    )
    return {(r.price_list_id, r.product_id, r.unit_id): r for r in rows}


def _task_quantity_total(items: List[BizDeliveryTaskItem]) -> Optional[float]:
    if not items:
        return None
    use_delivered = any(float(it.delivered_quantity or 0) > 0 for it in items)
    if use_delivered:
        total = sum(float(it.delivered_quantity or 0) for it in items)
    else:
        total = sum(float(it.quantity or 0) for it in items)
    return total


def _task_effective_dt(task: BizDeliveryTask) -> Optional[datetime]:
    return task.delivery_time or task.create_at


def _order_supply_dt(order: BizOrder) -> Optional[datetime]:
    return order.completed_time or order.create_at


def parse_year_month(year_month: str) -> Tuple[datetime, datetime]:
    """year_month: 'YYYY-MM' -> [month_start, next_month_start)"""
    parts = (year_month or "").strip().split("-")
    if len(parts) != 2:
        raise ValueError("yearMonth 格式应为 YYYY-MM")
    try:
        y, m = int(parts[0]), int(parts[1])
    except ValueError as e:
        raise ValueError("yearMonth 格式应为 YYYY-MM") from e
    if m < 1 or m > 12:
        raise ValueError("yearMonth 月份无效")
    start = datetime(y, m, 1, 0, 0, 0)
    if m == 12:
        end = datetime(y + 1, 1, 1, 0, 0, 0)
    else:
        end = datetime(y, m + 1, 1, 0, 0, 0)
    return start, end


def resolve_year_month_range(
    *,
    year_month: Optional[str] = None,
    year_month_start: Optional[str] = None,
    year_month_end: Optional[str] = None,
) -> Tuple[datetime, datetime, str]:
    """解析月份或月份区间，返回 [range_start, range_end_exclusive) 与展示标签。

    - 仅 year_month：等同起止同月
    - year_month_start / year_month_end：闭区间月份（结束月整月含入）
    - 起止都有时优先区间；起止缺一时用另一端补齐；再缺则回退 year_month
    标签：同月为 YYYY-MM，跨月为 YYYY-MM~YYYY-MM
    """
    ym = (year_month or "").strip() or None
    ys = (year_month_start or "").strip() or None
    ye = (year_month_end or "").strip() or None
    if ys or ye:
        if not ys:
            ys = ye
        if not ye:
            ye = ys
        assert ys and ye
        start, _ = parse_year_month(ys)
        _, end = parse_year_month(ye)
        if start >= end:
            raise ValueError("起始月份不能晚于结束月份")
        label = ys if ys == ye else f"{ys}~{ye}"
        return start, end, label
    if ym:
        start, end = parse_year_month(ym)
        return start, end, ym
    raise ValueError("请提供 yearMonth 或 yearMonthStart/yearMonthEnd")


def _parse_optional_dt(value: Optional[Any]) -> Optional[datetime]:
    if value is None or value == "":
        return None
    if isinstance(value, datetime):
        return value
    s = str(value).strip().replace("Z", "+00:00")
    try:
        return datetime.fromisoformat(s)
    except ValueError:
        for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
            try:
                return datetime.strptime(s[:19] if len(s) >= 19 else s, fmt)
            except ValueError:
                continue
    raise ValueError(f"无法解析日期时间: {value}")


def _raise_if_over_limit(n: int) -> None:
    if n > EXPORT_ROW_LIMIT:
        raise ValueError("导出数据超过 20000 行，请缩小筛选范围")


def _client_display(user: Optional[SysUser]) -> Tuple[Optional[str], Optional[str]]:
    """返回 (客户编号, 客户名称)。编号优先 username。兼容旧调用。"""
    return _user_code_name(user)


def _client_id_and_name(
    client_id: Optional[str],
    user: Optional[SysUser],
    *,
    contact_person: Optional[str] = None,
) -> Tuple[Optional[str], Optional[str]]:
    """客户编号=客户id；客户名称优先用户真实姓名/用户名，缺省回退订单联系人。"""
    code = (str(client_id).strip() if client_id is not None else "") or None
    name = None
    if user:
        rn = (user.real_name or "").strip() or None
        un = (user.username or "").strip() or None
        # real_name 仅为占位「客户」时改用 username
        if rn and rn != "客户":
            name = rn
        else:
            name = un or rn
    if not name:
        name = (contact_person or "").strip() or None
    return code, name


def _filter_client_orders(
    db: Session,
    *,
    region_id: Optional[str],
    client_id: Optional[str] = None,
    keyword: Optional[str] = None,
    start: Optional[Any] = None,
    end: Optional[Any] = None,
    order_ids: Optional[Sequence[str]] = None,
) -> List[BizOrder]:
    q = db.query(BizOrder).filter(
        BizOrder.is_del == 0,
        BizOrder.order_status != "cancelled",
    )
    if region_id:
        q = q.filter(BizOrder.region_id == region_id)
    ids = [x.strip() for x in (order_ids or []) if x and str(x).strip()]
    if ids:
        q = q.filter(BizOrder.id.in_(ids))
    else:
        if client_id:
            q = q.filter(BizOrder.client_id == client_id)
        if keyword:
            q = q.filter(BizOrder.order_no.like(f"%{keyword.strip()}%"))
        dt_start = _parse_optional_dt(start)
        dt_end = _parse_optional_dt(end)
        if dt_start:
            q = q.filter(BizOrder.create_at >= dt_start)
        if dt_end:
            q = q.filter(BizOrder.create_at <= dt_end)
    return q.order_by(BizOrder.create_at.desc(), BizOrder.id.desc()).all()


def build_client_order_lines(
    db: Session,
    *,
    region_id: Optional[str],
    client_id: Optional[str] = None,
    keyword: Optional[str] = None,
    start: Optional[Any] = None,
    end: Optional[Any] = None,
    order_ids: Optional[Sequence[str]] = None,
    year_month: Optional[str] = None,
    year_month_start: Optional[str] = None,
    year_month_end: Optional[str] = None,
) -> List[Dict[str, Any]]:
    """01 客户下单清单：一行 = 一个订单商品行。

    提供 year_month 或 year_month_start/end 时：按订单 create_at 落在区间，忽略筛选/勾选。
    """
    use_month = bool(
        (year_month or "").strip()
        or (year_month_start or "").strip()
        or (year_month_end or "").strip()
    )
    if use_month:
        month_start, month_end, _ = resolve_year_month_range(
            year_month=year_month,
            year_month_start=year_month_start,
            year_month_end=year_month_end,
        )
        q = db.query(BizOrder).filter(
            BizOrder.is_del == 0,
            BizOrder.order_status != "cancelled",
            BizOrder.create_at >= month_start,
            BizOrder.create_at < month_end,
        )
        if region_id:
            q = q.filter(BizOrder.region_id == region_id)
        orders = q.order_by(BizOrder.create_at.desc(), BizOrder.id.desc()).all()
    else:
        orders = _filter_client_orders(
            db,
            region_id=region_id,
            client_id=client_id,
            keyword=keyword,
            start=start,
            end=end,
            order_ids=order_ids,
        )
    if not orders:
        return []

    order_ids_list = [o.id for o in orders]
    lines = (
        db.query(BizOrderItem)
        .filter(BizOrderItem.order_id.in_(order_ids_list), BizOrderItem.is_del == 0)
        .order_by(BizOrderItem.order_id.asc(), BizOrderItem.create_at.asc())
        .all()
    )
    _raise_if_over_limit(len(lines))

    by_oid: Dict[str, List[BizOrderItem]] = {}
    for ln in lines:
        by_oid.setdefault(ln.order_id, []).append(ln)

    product_ids = list({ln.product_id for ln in lines if ln.product_id})
    unit_ids = list({ln.unit_id for ln in lines if ln.unit_id})
    client_ids = list({o.client_id for o in orders if o.client_id})

    product_map = {
        p.id: p
        for p in (
            db.query(BasProduct).filter(BasProduct.id.in_(product_ids), BasProduct.is_del == 0).all()
            if product_ids
            else []
        )
    }
    unit_map = {
        u.id: u
        for u in (
            db.query(BasProductUnit)
            .filter(BasProductUnit.id.in_(unit_ids), BasProductUnit.is_del == 0)
            .all()
            if unit_ids
            else []
        )
    }
    client_map = {
        u.id: u
        for u in (
            # 与结算订单列表一致：按 id 批量取，不过滤 is_del，避免名称缺失
            db.query(SysUser).filter(SysUser.id.in_(client_ids)).all()
            if client_ids
            else []
        )
    }

    out: List[Dict[str, Any]] = []
    for o in orders:
        c_code, c_name = _client_id_and_name(
            o.client_id,
            client_map.get(o.client_id),
            contact_person=getattr(o, "contact_person", None),
        )
        status_text = _ORDER_STATUS_TEXT.get(o.order_status) or o.order_status
        order_date = o.create_at.strftime("%Y-%m-%d %H:%M:%S") if o.create_at else None
        for ln in by_oid.get(o.id, []):
            p = product_map.get(ln.product_id)
            u = unit_map.get(ln.unit_id)
            qty = float(ln.quantity) if ln.quantity is not None else None
            sell_u = float(ln.unit_price) if ln.unit_price is not None else None
            if ln.amount is not None:
                sell_amt: Optional[float] = float(ln.amount)
            elif qty is not None and sell_u is not None:
                sell_amt = round(qty * sell_u, 2)
            else:
                sell_amt = None
            out.append(
                {
                    "orderNo": o.order_no,
                    "orderDate": order_date,
                    "clientId": c_code,
                    "clientName": c_name,
                    "productName": p.name if p else None,
                    "spec": getattr(p, "spec", None) if p else None,
                    "unitName": u.name if u else None,
                    "quantity": qty,
                    "unitPrice": sell_u,
                    "amount": sell_amt,
                    "orderStatusText": status_text,
                    "remark": (o.remark or None) if getattr(o, "remark", None) else None,
                }
            )
    _raise_if_over_limit(len(out))
    return out


def build_client_monthly_summary(
    db: Session,
    *,
    region_id: Optional[str],
    year_month: Optional[str] = None,
    year_month_start: Optional[str] = None,
    year_month_end: Optional[str] = None,
) -> List[Dict[str, Any]]:
    """04 月度客户结算汇总：一行 = 区间内一个客户（跨月合计）。"""
    range_start, range_end, period_label = resolve_year_month_range(
        year_month=year_month,
        year_month_start=year_month_start,
        year_month_end=year_month_end,
    )
    q = db.query(BizOrder).filter(
        BizOrder.is_del == 0,
        BizOrder.order_status != "cancelled",
        BizOrder.create_at >= range_start,
        BizOrder.create_at < range_end,
    )
    if region_id:
        q = q.filter(BizOrder.region_id == region_id)
    orders = q.order_by(BizOrder.create_at.asc()).all()
    if not orders:
        return []

    order_ids = [o.id for o in orders]
    lines = (
        db.query(BizOrderItem)
        .filter(BizOrderItem.order_id.in_(order_ids), BizOrderItem.is_del == 0)
        .all()
    )
    qty_by_oid: Dict[str, float] = {}
    for ln in lines:
        qty_by_oid[ln.order_id] = qty_by_oid.get(ln.order_id, 0.0) + float(ln.quantity or 0)

    # client_id -> agg
    agg: Dict[str, Dict[str, Any]] = {}
    for o in orders:
        cid = o.client_id or ""
        bucket = agg.get(cid)
        if not bucket:
            bucket = {
                "clientIdRaw": o.client_id,
                "contactPerson": (getattr(o, "contact_person", None) or "").strip() or None,
                "orderIds": set(),
                "quantitySum": 0.0,
                "receivableAmount": 0.0,
            }
            agg[cid] = bucket
        elif not bucket.get("contactPerson"):
            cp = (getattr(o, "contact_person", None) or "").strip() or None
            if cp:
                bucket["contactPerson"] = cp
        bucket["orderIds"].add(o.id)
        bucket["quantitySum"] += qty_by_oid.get(o.id, 0.0)
        if o.total_amount is not None:
            bucket["receivableAmount"] += float(o.total_amount)

    client_ids = [b["clientIdRaw"] for b in agg.values() if b["clientIdRaw"]]
    client_map = {
        u.id: u
        for u in (
            db.query(SysUser).filter(SysUser.id.in_(client_ids)).all()
            if client_ids
            else []
        )
    }

    rows: List[Dict[str, Any]] = []
    for cid, bucket in sorted(agg.items(), key=lambda x: x[0]):
        c_code, c_name = _client_id_and_name(
            bucket["clientIdRaw"],
            client_map.get(bucket["clientIdRaw"]),
            contact_person=bucket.get("contactPerson"),
        )
        order_count = len(bucket["orderIds"])
        qty_sum = bucket["quantitySum"]
        recv = bucket["receivableAmount"]
        rows.append(
            {
                "settlementMonth": period_label,
                "clientId": c_code,
                "clientName": c_name,
                "orderCount": order_count,
                "quantitySum": qty_sum if qty_sum else None,
                "receivableAmount": round(recv, 2) if recv else None,
                # 以下无系统对账数据：一律 None
                "discountAdjust": None,
                "actualReceived": None,
                "paidAmount": None,
                "debtAmount": None,
                "settlementStatus": None,
                "reconcileDate": None,
                "remark": None,
            }
        )
    _raise_if_over_limit(len(rows))
    return rows


def build_supplier_supply_lines(
    db: Session,
    *,
    region_id: Optional[str],
    supplier_id: Optional[str] = None,
    keyword: Optional[str] = None,
    start: Optional[Any] = None,
    end: Optional[Any] = None,
    order_ids: Optional[Sequence[str]] = None,
    year_month: Optional[str] = None,
    year_month_start: Optional[str] = None,
    year_month_end: Optional[str] = None,
) -> List[Dict[str, Any]]:
    """02 供货商供货结算：一行 = 订单商品行（供货价）。

    提供 year_month 或 year_month_start/end 时：按订单 create_at 落在区间，忽略筛选/勾选。
    """
    use_month = bool(
        (year_month or "").strip()
        or (year_month_start or "").strip()
        or (year_month_end or "").strip()
    )
    if use_month:
        month_start, month_end, _ = resolve_year_month_range(
            year_month=year_month,
            year_month_start=year_month_start,
            year_month_end=year_month_end,
        )
        q = (
            db.query(BizOrder, BizPriceList.supplier_id)
            .join(BizPriceList, BizPriceList.id == BizOrder.price_list_id)
            .filter(
                BizOrder.is_del == 0,
                BizOrder.order_status != "cancelled",
                BizOrder.create_at >= month_start,
                BizOrder.create_at < month_end,
            )
        )
        if region_id:
            q = q.filter(BizOrder.region_id == region_id)
    else:
        supply_dt = func.coalesce(BizOrder.completed_time, BizOrder.create_at)
        q = (
            db.query(BizOrder, BizPriceList.supplier_id)
            .join(BizPriceList, BizPriceList.id == BizOrder.price_list_id)
            .filter(BizOrder.is_del == 0, BizOrder.order_status != "cancelled")
        )
        if region_id:
            q = q.filter(BizOrder.region_id == region_id)

        ids = [x.strip() for x in (order_ids or []) if x and str(x).strip()]
        if ids:
            q = q.filter(BizOrder.id.in_(ids))
        else:
            if supplier_id:
                q = q.filter(BizPriceList.supplier_id == supplier_id)
            if keyword:
                q = q.filter(BizOrder.order_no.like(f"%{keyword.strip()}%"))
            dt_start = _parse_optional_dt(start)
            dt_end = _parse_optional_dt(end)
            if dt_start:
                q = q.filter(supply_dt >= dt_start)
            if dt_end:
                q = q.filter(supply_dt <= dt_end)

    pairs = q.order_by(BizOrder.create_at.desc(), BizOrder.id.desc()).all()
    if not pairs:
        return []

    orders = [o for o, _ in pairs]
    supplier_by_oid = {o.id: sid for o, sid in pairs}
    order_id_list = [o.id for o in orders]

    lines = (
        db.query(BizOrderItem)
        .filter(BizOrderItem.order_id.in_(order_id_list), BizOrderItem.is_del == 0)
        .order_by(BizOrderItem.order_id.asc(), BizOrderItem.create_at.asc())
        .all()
    )
    _raise_if_over_limit(len(lines))

    by_oid: Dict[str, List[BizOrderItem]] = {}
    for ln in lines:
        by_oid.setdefault(ln.order_id, []).append(ln)

    product_ids = list({ln.product_id for ln in lines if ln.product_id})
    unit_ids = list({ln.unit_id for ln in lines if ln.unit_id})
    supplier_ids = list({sid for sid in supplier_by_oid.values() if sid})
    pl_ids = list({o.price_list_id for o in orders if o.price_list_id})

    product_map = {
        p.id: p
        for p in (
            db.query(BasProduct).filter(BasProduct.id.in_(product_ids), BasProduct.is_del == 0).all()
            if product_ids
            else []
        )
    }
    unit_map = {
        u.id: u
        for u in (
            db.query(BasProductUnit)
            .filter(BasProductUnit.id.in_(unit_ids), BasProductUnit.is_del == 0)
            .all()
            if unit_ids
            else []
        )
    }
    supplier_map = {
        u.id: u
        for u in (
            db.query(SysUser).filter(SysUser.id.in_(supplier_ids), SysUser.is_del == 0).all()
            if supplier_ids
            else []
        )
    }
    pli_map = _price_list_item_map(db, pl_ids)

    out: List[Dict[str, Any]] = []
    for o in orders:
        sid = supplier_by_oid.get(o.id)
        s_code, s_name = _user_code_name(supplier_map.get(sid) if sid else None)
        settle_text = _SUPPLIER_SETTLEMENT_TEXT.get(
            getattr(o, "supplier_settlement_status", None) or "pending"
        )
        supply_date = _order_supply_dt(o)
        supply_date_s = supply_date.strftime("%Y-%m-%d %H:%M:%S") if supply_date else None
        for ln in by_oid.get(o.id, []):
            p = product_map.get(ln.product_id)
            u = unit_map.get(ln.unit_id)
            qty = float(ln.quantity) if ln.quantity is not None else None
            pli = pli_map.get((o.price_list_id, ln.product_id, ln.unit_id))
            sup_u = _supplier_unit_yuan_from_pli(pli)
            if sup_u is None and ln.unit_price is not None:
                # 无清单价时回退订单行单价（与结算列表一致）
                try:
                    sup_u = float(ln.unit_price)
                except (TypeError, ValueError):
                    sup_u = None
            if qty is not None and sup_u is not None:
                sup_amt: Optional[float] = round(qty * float(sup_u), 2)
            else:
                sup_amt = None
            out.append(
                {
                    "supplyNo": o.order_no,
                    "orderNo": o.order_no,
                    "supplyDate": supply_date_s,
                    "supplierId": s_code,
                    "supplierName": s_name,
                    "skuCode": p.sku_code if p else None,
                    "productName": p.name if p else None,
                    "spec": getattr(p, "spec", None) if p else None,
                    "unitName": u.name if u else None,
                    "quantity": qty,
                    "unitPrice": float(sup_u) if sup_u is not None else None,
                    "amount": sup_amt,
                    "acceptStatus": None,
                    "deduction": None,
                    "actualAmount": None,
                    "settlementStatus": settle_text,
                    "remark": (ln.note or None) if getattr(ln, "note", None) else None,
                }
            )
    _raise_if_over_limit(len(out))
    return out


def build_supplier_monthly_summary(
    db: Session,
    *,
    region_id: Optional[str],
    year_month: Optional[str] = None,
    year_month_start: Optional[str] = None,
    year_month_end: Optional[str] = None,
) -> List[Dict[str, Any]]:
    """05 月度供货商结算汇总：一行 = 区间内一个供应商；笔数=订单笔数（跨月合计）。"""
    range_start, range_end, period_label = resolve_year_month_range(
        year_month=year_month,
        year_month_start=year_month_start,
        year_month_end=year_month_end,
    )
    pairs = (
        db.query(BizOrder, BizPriceList.supplier_id)
        .join(BizPriceList, BizPriceList.id == BizOrder.price_list_id)
        .filter(
            BizOrder.is_del == 0,
            BizOrder.order_status != "cancelled",
            BizOrder.create_at >= range_start,
            BizOrder.create_at < range_end,
        )
    )
    if region_id:
        pairs = pairs.filter(BizOrder.region_id == region_id)
    pairs = pairs.all()
    if not pairs:
        return []

    orders = [o for o, _ in pairs]
    supplier_by_oid = {o.id: sid for o, sid in pairs}
    order_ids = [o.id for o in orders]
    lines = (
        db.query(BizOrderItem)
        .filter(BizOrderItem.order_id.in_(order_ids), BizOrderItem.is_del == 0)
        .all()
    )
    by_oid: Dict[str, List[BizOrderItem]] = {}
    for ln in lines:
        by_oid.setdefault(ln.order_id, []).append(ln)

    pl_ids = list({o.price_list_id for o in orders if o.price_list_id})
    pli_map = _price_list_item_map(db, pl_ids)

    agg: Dict[str, Dict[str, Any]] = {}
    for o in orders:
        sid = supplier_by_oid.get(o.id) or ""
        bucket = agg.get(sid)
        if not bucket:
            bucket = {
                "supplierIdRaw": supplier_by_oid.get(o.id),
                "orderIds": set(),
                "quantitySum": 0.0,
                "payableAmount": 0.0,
            }
            agg[sid] = bucket
        bucket["orderIds"].add(o.id)
        for ln in by_oid.get(o.id, []):
            qty = float(ln.quantity or 0)
            bucket["quantitySum"] += qty
            pli = pli_map.get((o.price_list_id, ln.product_id, ln.unit_id))
            sup_u = _supplier_unit_yuan_from_pli(pli)
            if sup_u is None and ln.unit_price is not None:
                try:
                    sup_u = float(ln.unit_price)
                except (TypeError, ValueError):
                    sup_u = None
            if sup_u is not None:
                bucket["payableAmount"] += qty * float(sup_u)

    supplier_ids = [b["supplierIdRaw"] for b in agg.values() if b["supplierIdRaw"]]
    supplier_map = {
        u.id: u
        for u in (
            db.query(SysUser).filter(SysUser.id.in_(supplier_ids), SysUser.is_del == 0).all()
            if supplier_ids
            else []
        )
    }

    rows: List[Dict[str, Any]] = []
    for sid, bucket in sorted(agg.items(), key=lambda x: x[0]):
        s_code, s_name = _user_code_name(supplier_map.get(bucket["supplierIdRaw"]))
        pay = bucket["payableAmount"]
        qty_sum = bucket["quantitySum"]
        rows.append(
            {
                "settlementMonth": period_label,
                "supplierId": s_code,
                "supplierName": s_name,
                "orderCount": len(bucket["orderIds"]),
                "quantitySum": qty_sum if qty_sum else None,
                "payableAmount": round(pay, 2) if pay else None,
                "deduction": None,
                "actualPaid": None,
                "paidAmount": None,
                "debtAmount": None,
                "settlementStatus": None,
                "reconcileDate": None,
                "remark": None,
            }
        )
    _raise_if_over_limit(len(rows))
    return rows


def build_delivery_lines(
    db: Session,
    *,
    region_id: Optional[str],
    keyword: Optional[str] = None,
    status: Optional[str] = None,
    deliver_id: Optional[str] = None,
    start: Optional[Any] = None,
    end: Optional[Any] = None,
    year_month: Optional[str] = None,
    year_month_start: Optional[str] = None,
    year_month_end: Optional[str] = None,
) -> List[Dict[str, Any]]:
    """03 配送商配送结算：一行 = 一个配送任务。

    提供 year_month 或 year_month_start/end 时：按 eff_dt 落在区间，忽略筛选参数。
    """
    eff_dt = func.coalesce(BizDeliveryTask.delivery_time, BizDeliveryTask.create_at)
    use_month = bool(
        (year_month or "").strip()
        or (year_month_start or "").strip()
        or (year_month_end or "").strip()
    )
    q = db.query(BizDeliveryTask).filter(
        BizDeliveryTask.is_del == 0,
        BizDeliveryTask.status != "cancelled",
    )
    if region_id:
        q = q.filter(BizDeliveryTask.region_id == region_id)

    if use_month:
        month_start, month_end, _ = resolve_year_month_range(
            year_month=year_month,
            year_month_start=year_month_start,
            year_month_end=year_month_end,
        )
        q = q.filter(eff_dt >= month_start, eff_dt < month_end)
    else:
        if status:
            q = q.filter(BizDeliveryTask.status == status)
        if deliver_id:
            q = q.filter(BizDeliveryTask.deliver_id == deliver_id)

        kw = (keyword or "").strip()
        if kw:
            order_ids_kw = [
                r[0]
                for r in db.query(BizOrder.id)
                .filter(BizOrder.is_del == 0, BizOrder.order_no.like(f"%{kw}%"))
                .all()
            ]
            conds = [BizDeliveryTask.task_no.like(f"%{kw}%")]
            if order_ids_kw:
                conds.append(BizDeliveryTask.order_id.in_(order_ids_kw))
            q = q.filter(or_(*conds))

        dt_start = _parse_optional_dt(start)
        dt_end = _parse_optional_dt(end)
        if dt_start:
            q = q.filter(eff_dt >= dt_start)
        if dt_end:
            q = q.filter(eff_dt <= dt_end)

    tasks = q.order_by(BizDeliveryTask.create_at.desc(), BizDeliveryTask.id.desc()).all()
    _raise_if_over_limit(len(tasks))
    if not tasks:
        return []

    task_ids = [t.id for t in tasks]
    order_ids = list({t.order_id for t in tasks if t.order_id})
    deliver_ids = list({t.deliver_id for t in tasks if t.deliver_id})
    supplier_ids = list({t.supplier_id for t in tasks if t.supplier_id})

    items = (
        db.query(BizDeliveryTaskItem)
        .filter(BizDeliveryTaskItem.task_id.in_(task_ids), BizDeliveryTaskItem.is_del == 0)
        .all()
    )
    items_by_tid: Dict[str, List[BizDeliveryTaskItem]] = {}
    for it in items:
        items_by_tid.setdefault(it.task_id, []).append(it)

    orders = (
        db.query(BizOrder).filter(BizOrder.id.in_(order_ids), BizOrder.is_del == 0).all()
        if order_ids
        else []
    )
    order_map = {o.id: o for o in orders}
    # 订单删了也能从配送明细拿客户
    client_ids = list(
        {
            *(o.client_id for o in orders if o.client_id),
            *(it.client_id for it in items if it.client_id),
        }
    )

    user_ids = list({*deliver_ids, *supplier_ids, *client_ids})
    user_map = _user_map_by_ids(db, user_ids)

    out: List[Dict[str, Any]] = []
    for t in tasks:
        o = order_map.get(t.order_id)
        deliver_user = user_map.get(t.deliver_id) if t.deliver_id else None
        client_id = None
        contact_person = None
        if o and o.client_id:
            client_id = o.client_id
            contact_person = (o.contact_person or "").strip() or None
        elif items_by_tid.get(t.id):
            first_item = items_by_tid[t.id][0]
            client_id = first_item.client_id or None
            contact_person = (first_item.contact_person or "").strip() or None
        client_user = user_map.get(client_id) if client_id else None
        supplier_user = user_map.get(t.supplier_id) if t.supplier_id else None

        # 配送商编号=用户id；配送商名称=真实姓名/用户名
        deliver_id_out = t.deliver_id or None
        deliver_name_out = _person_display_name(deliver_user)
        # 客户编号=客户id；客户名称/站点=站点用户名优先，缺省联系人
        client_name_out = _client_site_display_name(client_user, contact_person=contact_person)
        # 起点（供货商/仓）= 供应商名称（用户名/商户名优先）
        origin_name_out = _supplier_display_name(supplier_user)

        dest = None
        if o:
            dest = (o.delivery_address or "").strip() or client_name_out
        elif items_by_tid.get(t.id):
            dest = (items_by_tid[t.id][0].delivery_address or "").strip() or client_name_out
        eff = _task_effective_dt(t)
        qty = _task_quantity_total(items_by_tid.get(t.id, []))
        out.append(
            {
                "taskNo": t.task_no,
                "orderNo": o.order_no if o else None,
                "deliveryDate": eff.strftime("%Y-%m-%d %H:%M:%S") if eff else None,
                "deliverId": deliver_id_out,
                "deliverName": deliver_name_out,
                "clientId": client_id,
                "clientName": client_name_out,
                "originName": origin_name_out,
                "destName": dest,
                "quantityTotal": qty,
                "rate": None,
                "payableFee": None,
                "deduction": None,
                "actualFee": None,
                "settlementStatus": _DELIVERY_STATUS_TEXT.get(t.status) or t.status,
                "remark": (t.delivery_remark or t.dispatch_remark or None),
            }
        )
    return out


def build_delivery_monthly_summary(
    db: Session,
    *,
    region_id: Optional[str],
    year_month: Optional[str] = None,
    year_month_start: Optional[str] = None,
    year_month_end: Optional[str] = None,
) -> List[Dict[str, Any]]:
    """06 月度配送商结算汇总：一行 = 区间内一个配送员；金额列全空。"""
    range_start, range_end, period_label = resolve_year_month_range(
        year_month=year_month,
        year_month_start=year_month_start,
        year_month_end=year_month_end,
    )
    eff_dt = func.coalesce(BizDeliveryTask.delivery_time, BizDeliveryTask.create_at)
    q = db.query(BizDeliveryTask).filter(
        BizDeliveryTask.is_del == 0,
        BizDeliveryTask.status != "cancelled",
        eff_dt >= range_start,
        eff_dt < range_end,
    )
    if region_id:
        q = q.filter(BizDeliveryTask.region_id == region_id)
    tasks = q.all()
    if not tasks:
        return []

    task_ids = [t.id for t in tasks]
    items = (
        db.query(BizDeliveryTaskItem)
        .filter(BizDeliveryTaskItem.task_id.in_(task_ids), BizDeliveryTaskItem.is_del == 0)
        .all()
    )
    items_by_tid: Dict[str, List[BizDeliveryTaskItem]] = {}
    for it in items:
        items_by_tid.setdefault(it.task_id, []).append(it)

    agg: Dict[str, Dict[str, Any]] = {}
    for t in tasks:
        key = t.deliver_id or ""
        bucket = agg.get(key)
        if not bucket:
            bucket = {
                "deliverIdRaw": t.deliver_id,
                "taskCount": 0,
                "quantitySum": 0.0,
            }
            agg[key] = bucket
        bucket["taskCount"] += 1
        qtot = _task_quantity_total(items_by_tid.get(t.id, []))
        if qtot is not None:
            bucket["quantitySum"] += qtot

    deliver_ids = [b["deliverIdRaw"] for b in agg.values() if b["deliverIdRaw"]]
    deliver_map = _user_map_by_ids(db, deliver_ids)

    rows: List[Dict[str, Any]] = []
    for key, bucket in sorted(agg.items(), key=lambda x: x[0]):
        if bucket["deliverIdRaw"]:
            deliver_user = deliver_map.get(bucket["deliverIdRaw"])
            d_code = bucket["deliverIdRaw"]
            d_name = _person_display_name(deliver_user)
        else:
            d_code, d_name = None, "未分配"
        qty_sum = bucket["quantitySum"]
        rows.append(
            {
                "settlementMonth": period_label,
                "deliverId": d_code,
                "deliverName": d_name,
                "taskCount": bucket["taskCount"],
                "quantitySum": qty_sum if qty_sum else None,
                "payableFee": None,
                "deduction": None,
                "actualFee": None,
                "paidAmount": None,
                "debtAmount": None,
                "settlementStatus": None,
                "reconcileDate": None,
                "remark": None,
            }
        )
    _raise_if_over_limit(len(rows))
    return rows