"""预签约订单 / 意向金转账 / 转正式协议 CRUD"""

from __future__ import annotations

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

from sqlalchemy import and_, func, or_
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from app.services.project_assignment import assignment_brief, filter_assignment, guard_project_edit
from app.core.utils import generate_uuid, parse_datetime
from app.crud.child_row_sync import sync_child_rows
from app.crud.biz_no_generator import allocate_next_biz_no
from app.crud.collection_protocol import urls_to_attachment_list
from app.models.models import (
    BizCellSampleType,
    BizCollectionHospital,
    BizCollectionProtocol,
    BizCollectionRegion,
    BizPresignOrder,
    BizPresignOrderSampleType,
    BizPresignTransfer,
    BizProtocolPartyCustomer,
    SysUser,
)
from app.services.protocol_transfer_status import (
    can_confirm_receipt,
    can_submit_receipt,
    can_withdraw_receipt,
)


def _now() -> datetime:
    return datetime.now()


def assert_presign_data_access(
    db: Session, record_id: str, tenant_no: str, user_info: dict,
    *, transfer: bool = False,
) -> None:
    """Web 单条操作沿用对应订单/财务列表的租户与多角色数据范围。"""
    from app.core.data_scope import apply_data_scope

    model = BizPresignTransfer if transfer else BizPresignOrder
    query = db.query(model.id).filter(
        model.id == record_id,
        model.tenant_no == tenant_no,
        model.del_flag == 0,
    )
    if apply_data_scope(query, model, user_info, db, tenant_no).first() is None:
        raise PermissionError("记录不存在或无权访问")


def _dec(v: Any) -> Optional[Decimal]:
    if v is None or v == "":
        return None
    return Decimal(str(v))


def _user_nick_name_map(
    db: Session, user_ids: List[str], tenant_no: str
) -> Dict[str, str]:
    ids = [str(x).strip() for x in user_ids if str(x or "").strip()]
    if not ids:
        return {}
    rows = (
        db.query(SysUser.id, SysUser.nick_name)
        .filter(
            SysUser.id.in_(ids),
            SysUser.tenant_no == tenant_no,
        )
        .all()
    )
    out: Dict[str, str] = {}
    for u in rows:
        out[str(u.id)] = (u.nick_name or "").strip()
    return out


def assert_presign_convertible(*, order_status: Optional[str]) -> None:
    st = (order_status or "").strip() or "draft"
    if st == "converted":
        raise ValueError("该预签约订单已转正式协议")
    if st == "void":
        raise ValueError("已作废的预签约订单不能转正式")


def resolve_convert_staff_fields(
    *,
    collector_user_id: Optional[str],
    sales_user_id: Optional[str],
    region_manager_user_id: Optional[str],
    fallback_user_id: Optional[str] = None,
) -> Dict[str, Optional[str]]:
    """转正式人员映射：采集专员→采集专员+市场专员；采集经理→地区经理。"""
    _ = sales_user_id  # 正式单市场专员以采集专员为准，不再沿用预签市场专员
    collector = (collector_user_id or "").strip() or (fallback_user_id or "").strip() or None
    region_mgr = (region_manager_user_id or "").strip() or None
    return {
        "collector_user_id": collector,
        "sales_user_id": collector,
        "region_manager_user_id": region_mgr,
    }


def _get_order_or_raise(
    db: Session, *, order_id: str, tenant_no: str, for_update: bool = False
) -> BizPresignOrder:
    query = (
        db.query(BizPresignOrder)
        .filter(
            BizPresignOrder.id == order_id,
            BizPresignOrder.tenant_no == tenant_no,
            BizPresignOrder.del_flag == 0,
        )
    )
    if for_update:
        query = query.populate_existing().with_for_update()
    obj = query.first()
    if not obj:
        raise ValueError("预签约订单不存在")
    return obj


def validate_presign_person_ids(
    db: Session,
    *,
    tenant_no: str,
    collector_user_id: Optional[str],
    sales_user_id: Optional[str],
    region_manager_user_id: Optional[str],
) -> None:
    """所有预签人员必须是当前租户有效且启用的系统用户。"""
    requested = {
        str(user_id).strip()
        for user_id in (
            collector_user_id,
            sales_user_id,
            region_manager_user_id,
        )
        if str(user_id or "").strip()
    }
    if not requested:
        return
    existing = {
        str(row[0])
        for row in db.query(SysUser.id)
        .filter(
            SysUser.id.in_(requested),
            SysUser.tenant_no == tenant_no,
            SysUser.del_flag == 0,
            SysUser.status == "1",
        )
        .all()
    }
    if existing != requested:
        raise ValueError("人员用户不存在、不属于当前租户、已删除或已停用")


def _get_transfer_or_raise(
    db: Session, *, transfer_id: str, tenant_no: str
) -> BizPresignTransfer:
    obj = (
        db.query(BizPresignTransfer)
        .filter(
            BizPresignTransfer.id == transfer_id,
            BizPresignTransfer.tenant_no == tenant_no,
            BizPresignTransfer.del_flag == 0,
        )
        .first()
    )
    if not obj:
        raise ValueError("意向金转账笔不存在")
    return obj


def _update_transfer_expected_status(
    db: Session,
    *,
    transfer_id: str,
    tenant_no: str,
    expected_status: str,
    values: Dict[Any, Any],
    extra_conditions: Tuple[Any, ...] = (),
    commit: bool = True,
) -> BizPresignTransfer:
    """以旧状态作为 CAS 条件更新，并返回事务内刷新后的稳定快照。"""
    changed = (
        db.query(BizPresignTransfer)
        .filter(
            BizPresignTransfer.id == transfer_id,
            BizPresignTransfer.tenant_no == tenant_no,
            BizPresignTransfer.del_flag == 0,
            BizPresignTransfer.pay_status == expected_status,
            *extra_conditions,
        )
        .update(values, synchronize_session=False)
    )
    if changed != 1:
        raise ValueError("转账状态已变化，请刷新后重试")
    obj = (
        db.query(BizPresignTransfer)
        .filter(
            BizPresignTransfer.id == transfer_id,
            BizPresignTransfer.tenant_no == tenant_no,
            BizPresignTransfer.del_flag == 0,
        )
        .populate_existing()
        .first()
    )
    if obj is None:
        raise ValueError("意向金转账笔不存在")
    db.flush()
    if commit:
        # 提交会 expire 会话内 ORM 对象；先分离，调用方序列化时只读取
        # 本次事务内已刷新的标量快照，不会在提交后重新读取并发新状态。
        db.expunge(obj)
        db.commit()
    return obj


def _transfer_to_dict(
    db: Session, t: BizPresignTransfer, tenant_no: str
) -> Dict[str, Any]:
    return {
        "id": t.id,
        "presignOrderId": t.presign_order_id,
        "transferAmount": float(t.transfer_amount) if t.transfer_amount is not None else None,
        "transferTime": t.transfer_time.strftime("%Y-%m-%d %H:%M:%S") if t.transfer_time else "",
        "transferType": t.transfer_type or "",
        "transferAttachments": t.transfer_attachments or "",
        "attachments": urls_to_attachment_list(t.transfer_attachments, db, tenant_no),
        "payStatus": t.pay_status or "",
        "financeConfirmAttachments": t.finance_confirm_attachments or "",
        "financeConfirmAttachmentFiles": urls_to_attachment_list(
            t.finance_confirm_attachments, db, tenant_no
        ),
        "financeRemark": t.finance_remark or "",
        "remark": t.remark or "",
        "source": "presign",
        "createTime": t.create_time.strftime("%Y-%m-%d %H:%M:%S") if t.create_time else "",
        "updateTime": t.update_time.strftime("%Y-%m-%d %H:%M:%S") if t.update_time else "",
    }


def confirmed_deposit_amount(
    db: Session, *, presign_order_id: str, tenant_no: str
) -> Decimal:
    rows = (
        db.query(BizPresignTransfer.transfer_amount)
        .filter(
            BizPresignTransfer.presign_order_id == presign_order_id,
            BizPresignTransfer.tenant_no == tenant_no,
            BizPresignTransfer.del_flag == 0,
            BizPresignTransfer.pay_status == "yjf",
        )
        .all()
    )
    total = Decimal("0")
    for (amt,) in rows:
        if amt is not None:
            total += Decimal(str(amt))
    return total


def confirmed_deposit_amount_map(
    db: Session, *, presign_order_ids: List[str], tenant_no: str
) -> Dict[str, Decimal]:
    ids = [str(order_id).strip() for order_id in presign_order_ids if str(order_id or "").strip()]
    if not ids:
        return {}
    rows = (
        db.query(
            BizPresignTransfer.presign_order_id,
            func.sum(BizPresignTransfer.transfer_amount),
        )
        .filter(
            BizPresignTransfer.presign_order_id.in_(ids),
            BizPresignTransfer.tenant_no == tenant_no,
            BizPresignTransfer.del_flag == 0,
            BizPresignTransfer.pay_status == "yjf",
        )
        .group_by(BizPresignTransfer.presign_order_id)
        .all()
    )
    return {
        str(order_id): Decimal(str(amount or 0)) for order_id, amount in rows
    }


def _sample_type_names(
    db: Session, *, presign_order_id: str, tenant_no: str
) -> str:
    ids = list_sample_type_ids(
        db, presign_order_id=presign_order_id, tenant_no=tenant_no
    )
    if not ids:
        return ""
    rows = (
        db.query(BizCellSampleType.id, BizCellSampleType.type_name)
        .filter(
            BizCellSampleType.id.in_(ids),
            BizCellSampleType.del_flag == 0,
        )
        .all()
    )
    name_map = {str(r.id): (r.type_name or "").strip() for r in rows}
    names = [name_map.get(i, "") for i in ids if name_map.get(i)]
    return "、".join(names)


def list_sample_type_ids(
    db: Session, *, presign_order_id: str, tenant_no: str
) -> List[str]:
    rows = (
        db.query(BizPresignOrderSampleType)
        .filter(
            BizPresignOrderSampleType.presign_order_id == presign_order_id,
            BizPresignOrderSampleType.tenant_no == tenant_no,
            BizPresignOrderSampleType.del_flag == 0,
        )
        .order_by(
            BizPresignOrderSampleType.order_num.asc(),
            BizPresignOrderSampleType.create_time.asc(),
        )
        .all()
    )
    return [r.sample_type_id for r in rows if r.sample_type_id]


def replace_sample_types(
    db: Session,
    *,
    presign_order_id: str,
    tenant_no: str,
    user_id: Optional[str],
    sample_type_ids: List[str],
    create_dept_id: str = "",
) -> None:
    now = _now()
    payload: List[Dict[str, Any]] = []
    seen = set()
    for tid in sample_type_ids or []:
        sid = (tid or "").strip()
        if not sid or sid in seen:
            continue
        seen.add(sid)
        payload.append({"sample_type_id": sid})

    def build_row(row_id: str) -> BizPresignOrderSampleType:
        return BizPresignOrderSampleType(
            id=row_id,
            presign_order_id=presign_order_id,
            tenant_no=tenant_no,
            create_by=user_id,
            create_dept_id=create_dept_id,
            create_time=now,
        )

    def apply_row(row: BizPresignOrderSampleType, item: Dict[str, Any], idx: int, _now: datetime) -> None:
        row.sample_type_id = item["sample_type_id"]
        row.order_num = idx

    sync_child_rows(
        db,
        BizPresignOrderSampleType,
        parent_attr="presign_order_id",
        parent_id=presign_order_id,
        tenant_no=tenant_no,
        items=payload,
        user_id=user_id,
        now=now,
        build_row=build_row,
        apply_fields=apply_row,
        natural_key=lambda item: (item.get("sample_type_id") or "").strip(),
        row_natural_key=lambda row: (row.sample_type_id or "").strip(),
        duplicate_message="样本类型编号重复",
        foreign_message="样本类型不属于当前预签约订单",
    )
    db.flush()


def list_transfers(
    db: Session, *, presign_order_id: str, tenant_no: str
) -> List[Dict[str, Any]]:
    rows = (
        db.query(BizPresignTransfer)
        .filter(
            BizPresignTransfer.presign_order_id == presign_order_id,
            BizPresignTransfer.tenant_no == tenant_no,
            BizPresignTransfer.del_flag == 0,
        )
        .order_by(BizPresignTransfer.create_time.asc())
        .all()
    )
    return [_transfer_to_dict(db, t, tenant_no) for t in rows]


def order_to_dict(
    db: Session,
    o: BizPresignOrder,
    tenant_no: str,
    *,
    with_children: bool = True,
    deposit_amount: Optional[Decimal] = None,
    user_name_map: Optional[Dict[str, str]] = None,
) -> Dict[str, Any]:
    deposit = (
        deposit_amount
        if deposit_amount is not None
        else confirmed_deposit_amount(
            db, presign_order_id=o.id, tenant_no=tenant_no
        )
    )
    data: Dict[str, Any] = {
        "id": o.id,
        **assignment_brief(db, tenant_no, o.project_id),
        "orderDate": o.order_date or "",
        "protocolNo": o.protocol_no or "",
        "motherName": o.mother_name or "",
        "expectedDeliveryDate": o.expected_delivery_date or "",
        "regionId": o.region_id or "",
        "hospitalId": o.hospital_id or "",
        "sampleCollectTime": o.sample_collect_time.strftime("%Y-%m-%d %H:%M:%S")
        if o.sample_collect_time
        else "",
        "payMode": o.pay_mode or "",
        "contractAmount": float(o.contract_amount) if o.contract_amount is not None else None,
        "collectorUserId": o.collector_user_id or "",
        "salesUserId": o.sales_user_id or "",
        "regionManagerUserId": o.region_manager_user_id or "",
        "orderStatus": o.order_status or "",
        "convertedProtocolId": o.converted_protocol_id or "",
        "confirmedDepositAmount": float(deposit),
        "remark": o.remark or "",
        "createTime": o.create_time.strftime("%Y-%m-%d %H:%M:%S") if o.create_time else "",
        "updateTime": o.update_time.strftime("%Y-%m-%d %H:%M:%S") if o.update_time else "",
    }
    name_map = user_name_map
    if name_map is None:
        name_map = _user_nick_name_map(
            db,
            [
                data["collectorUserId"],
                data["salesUserId"],
                data["regionManagerUserId"],
            ],
            tenant_no,
        )
    data["collectorUserName"] = name_map.get(data["collectorUserId"], "")
    data["salesUserName"] = name_map.get(data["salesUserId"], "")
    data["regionManagerUserName"] = name_map.get(data["regionManagerUserId"], "")
    if with_children:
        data["sampleTypeIds"] = list_sample_type_ids(
            db, presign_order_id=o.id, tenant_no=tenant_no
        )
        data["transfers"] = list_transfers(
            db, presign_order_id=o.id, tenant_no=tenant_no
        )
    return data


def list_presign_orders(
    db: Session,
    *,
    tenant_no: str,
    project_id: Optional[str] = None,
    team_id: Optional[str] = None,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    order_status: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    q = db.query(BizPresignOrder).filter(
        BizPresignOrder.tenant_no == tenant_no,
        BizPresignOrder.del_flag == 0,
    )
    q = filter_assignment(q, BizPresignOrder, db, tenant_no, project_id, team_id)
    st = (order_status or "").strip()
    if st:
        q = q.filter(BizPresignOrder.order_status == st)
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q = q.filter(
            or_(
                BizPresignOrder.protocol_no.like(like),
                BizPresignOrder.mother_name.like(like),
            )
        )
    if user_info:
        from app.core.data_scope import apply_data_scope

        q = apply_data_scope(q, BizPresignOrder, user_info, db, tenant_no)
    total = q.order_by(None).with_entities(func.count(BizPresignOrder.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        q.order_by(BizPresignOrder.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    items = [order_to_dict(db, o, tenant_no, with_children=False) for o in rows]
    attach_presign_ledger_fields(db, items, tenant_no)
    return items, int(total)


def attach_presign_ledger_fields(
    db: Session, items: List[Dict[str, Any]], tenant_no: str
) -> None:
    """补列表导出需要的地区、医院、样本类型和转账时间。含已停用主数据名称。"""
    if not items:
        return
    order_ids = [str(item.get("id") or "") for item in items if item.get("id")]
    region_ids = {str(item.get("regionId") or "") for item in items if item.get("regionId")}
    hospital_ids = {
        str(item.get("hospitalId") or "") for item in items if item.get("hospitalId")
    }
    region_names: Dict[str, str] = {}
    if region_ids:
        region_names = {
            str(row.id): str(row.region_name or "").strip()
            for row in db.query(BizCollectionRegion).filter(
                BizCollectionRegion.id.in_(region_ids),
                BizCollectionRegion.tenant_no == tenant_no,
            )
        }
    hospital_names: Dict[str, str] = {}
    if hospital_ids:
        hospital_names = {
            str(row.id): str(row.hospital_name or "").strip()
            for row in db.query(BizCollectionHospital).filter(
                BizCollectionHospital.id.in_(hospital_ids),
                BizCollectionHospital.tenant_no == tenant_no,
            )
        }
    type_ids_by_order: Dict[str, List[str]] = {order_id: [] for order_id in order_ids}
    if order_ids:
        links = (
            db.query(BizPresignOrderSampleType)
            .filter(
                BizPresignOrderSampleType.presign_order_id.in_(order_ids),
                BizPresignOrderSampleType.tenant_no == tenant_no,
                BizPresignOrderSampleType.del_flag == 0,
            )
            .order_by(
                BizPresignOrderSampleType.order_num.asc(),
                BizPresignOrderSampleType.create_time.asc(),
                BizPresignOrderSampleType.id.asc(),
            )
            .all()
        )
        for link in links:
            if link.sample_type_id:
                type_ids_by_order.setdefault(link.presign_order_id, []).append(
                    str(link.sample_type_id)
                )
    type_ids = {
        type_id
        for order_type_ids in type_ids_by_order.values()
        for type_id in order_type_ids
    }
    type_names: Dict[str, str] = {}
    if type_ids:
        type_names = {
            str(row.id): str(row.type_name or "").strip()
            for row in db.query(BizCellSampleType).filter(
                BizCellSampleType.id.in_(type_ids),
                BizCellSampleType.tenant_no == tenant_no,
            )
        }
    times_by_order: Dict[str, List[str]] = {}
    if order_ids:
        transfers = (
            db.query(BizPresignTransfer)
            .filter(
                BizPresignTransfer.presign_order_id.in_(order_ids),
                BizPresignTransfer.tenant_no == tenant_no,
                BizPresignTransfer.del_flag == 0,
            )
            .order_by(
                BizPresignTransfer.transfer_time.asc(),
                BizPresignTransfer.create_time.asc(),
                BizPresignTransfer.id.asc(),
            )
            .all()
        )
        for transfer in transfers:
            if not transfer.transfer_time:
                continue
            times_by_order.setdefault(transfer.presign_order_id, []).append(
                transfer.transfer_time.strftime("%Y-%m-%d %H:%M:%S")
            )
    for item in items:
        order_id = str(item.get("id") or "")
        item["regionName"] = region_names.get(str(item.get("regionId") or ""), "")
        item["hospitalName"] = hospital_names.get(str(item.get("hospitalId") or ""), "")
        names = [
            type_names.get(type_id, "")
            for type_id in type_ids_by_order.get(order_id, [])
        ]
        item["sampleTypeNames"] = [name for name in names if name]
        item["transferTimes"] = times_by_order.get(order_id, [])


def get_presign_order(
    db: Session, *, order_id: str, tenant_no: str
) -> Optional[Dict[str, Any]]:
    obj = (
        db.query(BizPresignOrder)
        .filter(
            BizPresignOrder.id == order_id,
            BizPresignOrder.tenant_no == tenant_no,
            BizPresignOrder.del_flag == 0,
        )
        .first()
    )
    if not obj:
        return None
    return order_to_dict(db, obj, tenant_no, with_children=True)


def create_presign_order(
    db: Session,
    *,
    tenant_no: str,
    user_id: Optional[str],
    fields: Dict[str, Any],
    sample_type_ids: Optional[List[str]] = None,
    transfers: Optional[List[Dict[str, Any]]] = None,
    create_dept_id: str = "",
    commit: bool = True,
) -> BizPresignOrder:
    now = _now()
    sample_collect = fields.get("sample_collect_time") or fields.get("sampleCollectTime")
    sct = None
    if sample_collect not in (None, ""):
        sct = (
            sample_collect
            if isinstance(sample_collect, datetime)
            else parse_datetime(str(sample_collect))
        )
    pay_mode = (fields.get("pay_mode") or fields.get("payMode") or "").strip() or None
    if pay_mode and pay_mode not in ("dj", "fq"):
        raise ValueError("缴费方式无效")
    # 协议编号由服务端发号，忽略客户端传入值
    protocol_no = allocate_next_biz_no(
        db, tenant_no=tenant_no, kind="collection", user_id=user_id
    )
    obj = BizPresignOrder(
        project_id=fields.get("project_id", fields.get("projectId")) or None,
        id=generate_uuid(),
        order_date=(fields.get("order_date") or fields.get("orderDate") or "").strip()
        or None,
        protocol_no=protocol_no,
        mother_name=(fields.get("mother_name") or fields.get("motherName") or "").strip()
        or None,
        expected_delivery_date=(
            fields.get("expected_delivery_date")
            or fields.get("expectedDeliveryDate")
            or ""
        ).strip()
        or None,
        region_id=(fields.get("region_id") or fields.get("regionId") or "").strip() or None,
        hospital_id=(fields.get("hospital_id") or fields.get("hospitalId") or "").strip()
        or None,
        sample_collect_time=sct,
        pay_mode=pay_mode,
        contract_amount=_dec(fields.get("contract_amount", fields.get("contractAmount"))),
        collector_user_id=(
            fields.get("collector_user_id") or fields.get("collectorUserId") or ""
        ).strip()
        or user_id
        or None,
        sales_user_id=(fields.get("sales_user_id") or fields.get("salesUserId") or "").strip()
        or None,
        region_manager_user_id=(
            fields.get("region_manager_user_id")
            or fields.get("regionManagerUserId")
            or ""
        ).strip()
        or None,
        order_status=(fields.get("order_status") or fields.get("orderStatus") or "active")
        .strip()
        or "active",
        tenant_no=tenant_no,
        order_num=0,
        remark=(fields.get("remark") or "").strip() or None,
        create_by=user_id,
        create_dept_id=create_dept_id,
        create_time=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(obj)
    db.flush()
    replace_sample_types(
        db,
        presign_order_id=obj.id,
        tenant_no=tenant_no,
        user_id=user_id,
        sample_type_ids=sample_type_ids or [],
        create_dept_id=create_dept_id,
    )
    for raw in transfers or []:
        if isinstance(raw, dict):
            add_transfer(
                db,
                presign_order_id=obj.id,
                tenant_no=tenant_no,
                user_id=user_id,
                fields=raw,
                create_dept_id=create_dept_id,
                commit=False,
            )
    if commit:
        db.commit()
        db.refresh(obj)
    else:
        db.flush()
    return obj


def update_presign_order(
    db: Session,
    *,
    order_id: str,
    tenant_no: str,
    user_id: Optional[str],
    fields: Dict[str, Any],
    sample_type_ids: Optional[List[str]] = None,
    create_dept_id: str = "",
    commit: bool = True,
) -> BizPresignOrder:
    obj = _get_order_or_raise(db, order_id=order_id, tenant_no=tenant_no)
    guard_project_edit(obj, fields)
    if (obj.order_status or "") == "converted":
        raise ValueError("已转正式的预签约订单不可修改")
    mapping = {
        "order_date": "order_date",
        "orderDate": "order_date",
        "mother_name": "mother_name",
        "motherName": "mother_name",
        "expected_delivery_date": "expected_delivery_date",
        "expectedDeliveryDate": "expected_delivery_date",
        "region_id": "region_id",
        "regionId": "region_id",
        "hospital_id": "hospital_id",
        "hospitalId": "hospital_id",
        "pay_mode": "pay_mode",
        "payMode": "pay_mode",
        "collector_user_id": "collector_user_id",
        "collectorUserId": "collector_user_id",
        "sales_user_id": "sales_user_id",
        "salesUserId": "sales_user_id",
        "region_manager_user_id": "region_manager_user_id",
        "regionManagerUserId": "region_manager_user_id",
        "order_status": "order_status",
        "orderStatus": "order_status",
        "remark": "remark",
    }
    for k, attr in mapping.items():
        if k in fields:
            val = fields[k]
            if attr in (
                "order_date",
                "mother_name",
                "expected_delivery_date",
                "region_id",
                "hospital_id",
                "pay_mode",
                "collector_user_id",
                "sales_user_id",
                "region_manager_user_id",
                "order_status",
                "remark",
            ):
                setattr(obj, attr, (str(val).strip() if val is not None else "") or None)
    if "contract_amount" in fields or "contractAmount" in fields:
        obj.contract_amount = _dec(
            fields.get("contract_amount", fields.get("contractAmount"))
        )
    if "sample_collect_time" in fields or "sampleCollectTime" in fields:
        v = fields.get("sample_collect_time", fields.get("sampleCollectTime"))
        if v in (None, ""):
            obj.sample_collect_time = None
        elif isinstance(v, datetime):
            obj.sample_collect_time = v
        else:
            obj.sample_collect_time = parse_datetime(str(v))
    if obj.pay_mode and obj.pay_mode not in ("dj", "fq"):
        raise ValueError("缴费方式无效")
    # 协议编号不可改；历史空号在编辑保存时补发
    if not (obj.protocol_no or "").strip():
        obj.protocol_no = allocate_next_biz_no(
            db, tenant_no=tenant_no, kind="collection", user_id=user_id
        )
    obj.update_by = user_id
    obj.update_time = _now()
    db.add(obj)
    if sample_type_ids is not None:
        replace_sample_types(
            db,
            presign_order_id=obj.id,
            tenant_no=tenant_no,
            user_id=user_id,
            sample_type_ids=sample_type_ids,
            create_dept_id=create_dept_id,
        )
    if commit:
        db.commit()
        db.refresh(obj)
    else:
        db.flush()
    return obj


def soft_delete_presign_order(
    db: Session, *, order_id: str, tenant_no: str, user_id: Optional[str]
) -> None:
    obj = _get_order_or_raise(db, order_id=order_id, tenant_no=tenant_no)
    if (obj.order_status or "") == "converted":
        raise ValueError("已转正式的预签约订单不可删除")
    now = _now()
    obj.del_flag = 1
    obj.update_by = user_id
    obj.update_time = now
    db.add(obj)
    db.commit()


def add_transfer(
    db: Session,
    *,
    presign_order_id: str,
    tenant_no: str,
    user_id: Optional[str],
    fields: Dict[str, Any],
    create_dept_id: str = "",
    commit: bool = True,
) -> BizPresignTransfer:
    _get_order_or_raise(db, order_id=presign_order_id, tenant_no=tenant_no)
    now = _now()
    tt = fields.get("transfer_time", fields.get("transferTime"))
    transfer_time = None
    if tt not in (None, ""):
        transfer_time = tt if isinstance(tt, datetime) else parse_datetime(str(tt))
    obj = BizPresignTransfer(
        id=generate_uuid(),
        presign_order_id=presign_order_id,
        transfer_amount=_dec(fields.get("transfer_amount", fields.get("transferAmount"))),
        transfer_time=transfer_time,
        transfer_type=(
            fields.get("transfer_type") or fields.get("transferType") or ""
        ).strip()
        or None,
        transfer_attachments=(
            fields.get("transfer_attachments")
            or fields.get("transferAttachments")
            or ""
        ).strip()
        or None,
        pay_status="dfj",
        tenant_no=tenant_no,
        order_num=0,
        remark=(fields.get("remark") or "").strip() or None,
        create_by=user_id,
        create_dept_id=create_dept_id,
        create_time=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(obj)
    if commit:
        db.commit()
        db.refresh(obj)
    else:
        db.flush()
    return obj


def update_transfer_draft(
    db: Session,
    *,
    transfer_id: str,
    tenant_no: str,
    user_id: Optional[str],
    fields: Dict[str, Any],
    commit: bool = True,
) -> BizPresignTransfer:
    t = _get_transfer_or_raise(db, transfer_id=transfer_id, tenant_no=tenant_no)
    if t.pay_status != "dfj":
        raise ValueError("仅已付款状态可编辑意向金转账笔")
    values: Dict[Any, Any] = {
        BizPresignTransfer.update_by: user_id,
        BizPresignTransfer.update_time: _now(),
    }
    if "transfer_amount" in fields or "transferAmount" in fields:
        values[BizPresignTransfer.transfer_amount] = _dec(
            fields.get("transfer_amount", fields.get("transferAmount"))
        )
    if "transfer_time" in fields or "transferTime" in fields:
        v = fields.get("transfer_time", fields.get("transferTime"))
        if v in (None, ""):
            values[BizPresignTransfer.transfer_time] = None
        elif isinstance(v, datetime):
            values[BizPresignTransfer.transfer_time] = v
        else:
            parsed = parse_datetime(str(v))
            if parsed is None:
                raise ValueError("转账时间格式无效")
            values[BizPresignTransfer.transfer_time] = parsed
    if "transfer_type" in fields or "transferType" in fields:
        values[BizPresignTransfer.transfer_type] = (
            str(fields.get("transfer_type", fields.get("transferType")) or "")
        ).strip() or None
    if "transfer_attachments" in fields or "transferAttachments" in fields:
        values[BizPresignTransfer.transfer_attachments] = (
            str(
                fields.get("transfer_attachments", fields.get("transferAttachments"))
                or ""
            )
        ).strip() or None
    if "remark" in fields:
        values[BizPresignTransfer.remark] = (
            fields.get("remark") or ""
        ).strip() or None
    return _update_transfer_expected_status(
        db,
        transfer_id=transfer_id,
        tenant_no=tenant_no,
        expected_status="dfj",
        values=values,
        commit=commit,
    )


def submit_receipt_finance(
    db: Session,
    *,
    transfer_id: str,
    tenant_no: str,
    user_id: Optional[str],
    commit: bool = True,
) -> BizPresignTransfer:
    t = _get_transfer_or_raise(db, transfer_id=transfer_id, tenant_no=tenant_no)
    if not can_submit_receipt(t.pay_status):
        raise ValueError("仅已付款状态可提交财务确认")
    if not (t.transfer_type or "").strip():
        raise ValueError("请选择转账方式")
    if t.transfer_amount is None or Decimal(str(t.transfer_amount)) <= 0:
        raise ValueError("请填写转账金额")
    return _update_transfer_expected_status(
        db,
        transfer_id=transfer_id,
        tenant_no=tenant_no,
        expected_status="dfj",
        values={
            BizPresignTransfer.pay_status: "jfqrz",
            BizPresignTransfer.update_by: user_id,
            BizPresignTransfer.update_time: _now(),
        },
        extra_conditions=(
            BizPresignTransfer.transfer_amount.isnot(None),
            BizPresignTransfer.transfer_amount > 0,
            BizPresignTransfer.transfer_type.isnot(None),
            func.length(func.trim(BizPresignTransfer.transfer_type)) > 0,
        ),
        commit=commit,
    )


def withdraw_receipt(
    db: Session,
    *,
    transfer_id: str,
    tenant_no: str,
    user_id: Optional[str],
    commit: bool = True,
) -> BizPresignTransfer:
    t = _get_transfer_or_raise(db, transfer_id=transfer_id, tenant_no=tenant_no)
    if not can_withdraw_receipt(t.pay_status):
        raise ValueError("仅付款确认中可撤回")
    # 故意不改 finance_remark / finance_confirm_attachments
    return _update_transfer_expected_status(
        db,
        transfer_id=transfer_id,
        tenant_no=tenant_no,
        expected_status="jfqrz",
        values={
            BizPresignTransfer.pay_status: "dfj",
            BizPresignTransfer.update_by: user_id,
            BizPresignTransfer.update_time: _now(),
        },
        commit=commit,
    )


def confirm_receipt(
    db: Session,
    *,
    transfer_id: str,
    tenant_no: str,
    user_id: Optional[str],
    finance_confirm_attachments: Optional[str] = None,
    finance_remark: Optional[str] = None,
    commit: bool = True,
) -> BizPresignTransfer:
    t = _get_transfer_or_raise(db, transfer_id=transfer_id, tenant_no=tenant_no)
    if not can_confirm_receipt(t.pay_status):
        raise ValueError("仅付款确认中状态可确认收款")
    values: Dict[Any, Any] = {
        BizPresignTransfer.pay_status: "yjf",
        BizPresignTransfer.update_by: user_id,
        BizPresignTransfer.update_time: _now(),
    }
    if finance_confirm_attachments is not None:
        values[BizPresignTransfer.finance_confirm_attachments] = (
            finance_confirm_attachments or ""
        ).strip() or None
    if finance_remark is not None:
        values[BizPresignTransfer.finance_remark] = (
            finance_remark or ""
        ).strip() or None
    return _update_transfer_expected_status(
        db,
        transfer_id=transfer_id,
        tenant_no=tenant_no,
        expected_status="jfqrz",
        values=values,
        commit=commit,
    )


def reject_receipt(
    db: Session,
    *,
    transfer_id: str,
    tenant_no: str,
    user_id: Optional[str],
    commit: bool = True,
) -> BizPresignTransfer:
    t = _get_transfer_or_raise(db, transfer_id=transfer_id, tenant_no=tenant_no)
    if not can_confirm_receipt(t.pay_status):
        raise ValueError("仅付款确认中状态可退回")
    return _update_transfer_expected_status(
        db,
        transfer_id=transfer_id,
        tenant_no=tenant_no,
        expected_status="jfqrz",
        values={
            BizPresignTransfer.pay_status: "dfj",
            BizPresignTransfer.update_by: user_id,
            BizPresignTransfer.update_time: _now(),
        },
        commit=commit,
    )


def list_presign_transfers_for_finance(
    db: Session,
    *,
    pay_status: str,
    tenant_no: str,
    keyword: Optional[str] = None,
    page: int = 1,
    page_size: int = 10,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    if not tenant_no:
        return [], 0
    st = (pay_status or "").strip()
    q = (
        db.query(BizPresignTransfer, BizPresignOrder)
        .join(
            BizPresignOrder,
            and_(
                BizPresignOrder.id == BizPresignTransfer.presign_order_id,
                BizPresignOrder.del_flag == 0,
            ),
        )
        .filter(
            BizPresignTransfer.tenant_no == tenant_no,
            BizPresignTransfer.del_flag == 0,
            BizPresignTransfer.pay_status == st,
        )
    )
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q = q.filter(
            or_(
                BizPresignOrder.protocol_no.like(like),
                BizPresignOrder.mother_name.like(like),
            )
        )
    if user_info:
        from app.core.data_scope import apply_data_scope

        q = apply_data_scope(q, BizPresignTransfer, user_info, db, tenant_no)
    total = q.order_by(None).with_entities(func.count(BizPresignTransfer.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        q.order_by(BizPresignTransfer.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    out: List[Dict[str, Any]] = []
    for t, order in rows:
        item = _transfer_to_dict(db, t, tenant_no)
        item["protocolNo"] = (order.protocol_no if order else "") or ""
        item["motherName"] = (order.mother_name if order else "") or ""
        item["sampleCode"] = ""
        item["presignOrderId"] = order.id if order else ""
        item["orderDate"] = (order.order_date if order else "") or ""
        item["expectedDeliveryDate"] = (
            (order.expected_delivery_date if order else "") or ""
        )
        item["contractAmount"] = (
            float(order.contract_amount)
            if order and order.contract_amount is not None
            else None
        )
        item["contractPayMode"] = (order.pay_mode if order else "") or ""
        item["orderStatus"] = (order.order_status if order else "") or ""
        item["orderRemark"] = (order.remark if order else "") or ""
        name_map = _user_nick_name_map(
            db,
            [
                (order.collector_user_id if order else "") or "",
                (order.sales_user_id if order else "") or "",
                (order.region_manager_user_id if order else "") or "",
            ],
            tenant_no,
        )
        item["collectorUserName"] = name_map.get(
            (order.collector_user_id if order else "") or "", ""
        )
        item["salesUserName"] = name_map.get(
            (order.sales_user_id if order else "") or "", ""
        )
        item["regionManagerUserName"] = name_map.get(
            (order.region_manager_user_id if order else "") or "", ""
        )
        item["sampleTypeNames"] = (
            _sample_type_names(
                db, presign_order_id=order.id, tenant_no=tenant_no
            )
            if order
            else ""
        )
        out.append(item)
    return out, int(total)


def convert_to_protocol(
    db: Session,
    *,
    order_id: str,
    tenant_no: str,
    user_id: Optional[str],
    create_dept_id: str = "",
    commit: bool = True,
) -> Dict[str, Any]:
    """转正式协议：创建协议主单 + 当事人/采集信息 + 样本类型行；意向金仅关联预签订单 ID，不拷贝。"""
    from app.crud import collection_protocol as cp_svc

    order = _get_order_or_raise(
        db, order_id=order_id, tenant_no=tenant_no, for_update=True
    )
    assert_presign_convertible(order_status=order.order_status)
    validate_presign_person_ids(
        db,
        tenant_no=tenant_no,
        collector_user_id=order.collector_user_id or user_id,
        sales_user_id=order.sales_user_id,
        region_manager_user_id=order.region_manager_user_id,
    )

    staff = resolve_convert_staff_fields(
        collector_user_id=order.collector_user_id,
        sales_user_id=order.sales_user_id,
        region_manager_user_id=order.region_manager_user_id,
        fallback_user_id=user_id,
    )

    type_ids = list_sample_type_ids(db, presign_order_id=order.id, tenant_no=tenant_no)
    try:
        protocol = cp_svc.create_collection_protocol(
            db,
            protocol_no=order.protocol_no or "",
            protocol_status="sxz",
            register_date=order.order_date,
            collector_user_id=staff["collector_user_id"],
            source_presign_order_id=order.id,
            project_id=order.project_id,
            payment_model_version=2,
            protocol_total_amount=None,
            remark=order.remark,
            tenant_no=tenant_no,
            user_id=user_id,
            create_dept_id=create_dept_id,
            commit=False,
            exclude_presign_order_id=order.id,
        )
        party_fields: Dict[str, Any] = {
            "mother_name": order.mother_name,
            "expected_delivery_date": order.expected_delivery_date,
        }
        cp_svc.upsert_party_customer(
            db,
            protocol.id,
            tenant_no,
            user_id,
            create_dept_id=create_dept_id,
            commit=False,
            **party_fields,
        )
        collection_fields: Dict[str, Any] = {
            "region_id": order.region_id,
            "hospital_id": order.hospital_id,
            "sales_user_id": staff["sales_user_id"],
            "region_manager_user_id": staff["region_manager_user_id"],
            "sample_collect_time": order.sample_collect_time,
        }
        cp_svc.upsert_collection_info(
            db,
            protocol.id,
            tenant_no,
            user_id,
            create_dept_id=create_dept_id,
            commit=False,
            **collection_fields,
        )
        if type_ids:
            samples = [
                {
                    "sampleTypeId": tid,
                    "sampleStatus": None,
                }
                for tid in type_ids
            ]
            cp_svc.save_samples(
                db,
                protocol.id,
                tenant_no,
                user_id,
                samples,
                create_dept_id=create_dept_id,
                commit=False,
            )

        order.order_status = "converted"
        order.converted_protocol_id = protocol.id
        if not (order.protocol_no or "").strip():
            order.protocol_no = protocol.protocol_no
        order.update_by = user_id
        order.update_time = _now()
        db.add(order)
        if commit:
            db.commit()
            db.refresh(order)
        else:
            db.flush()
    except IntegrityError as exc:
        db.rollback()
        message = str(exc).lower()
        if (
            "uk_collection_protocol_presign_source" in message
            or (
                "unique constraint failed" in message
                and "source_presign_order_id" in message
            )
        ):
            raise ValueError("该预签约订单已转正式协议") from exc
        raise
    except Exception:
        db.rollback()
        raise
    return {
        "presignOrder": order_to_dict(db, order, tenant_no, with_children=True),
        "protocolId": protocol.id,
        "protocolNo": protocol.protocol_no,
    }