zkxw/backend/app/crud/collection_protocol.py

"""采集协议单 / 审核 / 缴费明细 CRUD（细胞存储 2.2.6～2.2.10）"""
from __future__ import annotations

import json
import math
import calendar
from datetime import datetime, timedelta
from decimal import Decimal
from typing import Any, Dict, List, Optional, Tuple

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

from app.core.utils import generate_uuid, parse_datetime, format_date, optional_date
from app.crud.base import camel_to_snake
from app.crud.biz_no_generator import allocate_next_biz_no
from app.crud.protocol_donation_convert import (
    assert_biz_no_available,
    get_partner_sample_codes,
)
from app.services.protocol_payment_installment import (
    calc_fq_installment_due_date,
    calc_installment_due_amounts_by_ratios,
    parse_installment_ratios,
    resolve_installment_due_amounts,
    resolve_locked_remaining_due_amounts,
)
from app.services.protocol_pay_mode_switch import validate_payment_details_for_save
from app.utils.sample_special import (
    apply_sample_special_fields,
    sample_special_to_dict,
    validate_sample_special_raw,
)
from app.models.research_donation import BizResearchDonation, BizResearchDonationSample
from app.models.models import (
    BizAttachment,
    BizCollectionProtocol,
    BizProtocolAudit,
    BizProtocolStatusChange,
    BizProtocolCollectionInfo,
    BizProtocolFullmoonTodo,
    BizProtocolInvoiceRequest,
    BizProtocolInvoiceRequestItem,
    BizProtocolPartyCustomer,
    BizProtocolPayment,
    BizProtocolPaymentDetail,
    BizProtocolSample,
    BizProtocolSampleTransport,
    BizProtocolStorageCert,
    BizQcWorkOrder,
    BizSampleReception,
    BizStorageReleaseReview,
    BizCollectionRegion,
    BizCollectionHospital,
    BizCellSampleType,
    BizExpressCompany,
    BizItemUnit,
    SysSite,
    SysUser,
)

_NEWBORN_SEX_LABEL = {"1": "男", "0": "女"}
_SAMPLE_STATUS_LABEL = {
    "ycj": "已采集",
    "jcz": "检测中",
    "yrk": "已入库",
    "yzf": "已作废",
}


def get_protocol_nos_used_in_protocols(
    db: Session,
    *,
    tenant_no: str,
    protocol_nos: List[str],
) -> set:
    """返回已在采集协议单或科研捐赠单中占用的协议编号集合（同租户、未删除）。"""
    nos = list({(n or "").strip() for n in protocol_nos if (n or "").strip()})
    if not nos:
        return set()
    used: set = set()
    proto_rows = (
        db.query(BizCollectionProtocol.protocol_no)
        .filter(
            BizCollectionProtocol.tenant_no == tenant_no,
            BizCollectionProtocol.del_flag == 0,
            BizCollectionProtocol.protocol_no.in_(nos),
        )
        .all()
    )
    used.update(r[0] for r in proto_rows if r[0])
    donation_rows = (
        db.query(BizResearchDonation.donation_no)
        .filter(
            BizResearchDonation.tenant_no == tenant_no,
            BizResearchDonation.del_flag == 0,
            BizResearchDonation.donation_no.in_(nos),
        )
        .all()
    )
    used.update(r[0] for r in donation_rows if r[0])
    return used


def get_sample_codes_used_in_protocol_samples(
    db: Session,
    *,
    tenant_no: str,
    sample_codes: List[str],
) -> set:
    """返回已在采集协议单或科研捐赠单样本表中占用的样本编码集合（同租户、未删除）。"""
    codes = list({(c or "").strip() for c in sample_codes if (c or "").strip()})
    if not codes:
        return set()
    used: set = set()
    proto_rows = (
        db.query(BizProtocolSample.sample_code)
        .filter(
            BizProtocolSample.tenant_no == tenant_no,
            BizProtocolSample.del_flag == 0,
            BizProtocolSample.sample_code.in_(codes),
        )
        .all()
    )
    used.update(r[0] for r in proto_rows if r[0])
    donation_rows = (
        db.query(BizResearchDonationSample.sample_code)
        .filter(
            BizResearchDonationSample.tenant_no == tenant_no,
            BizResearchDonationSample.del_flag == 0,
            BizResearchDonationSample.sample_code.in_(codes),
        )
        .all()
    )
    used.update(r[0] for r in donation_rows if r[0])
    return used


def validate_sample_codes_not_used(
    db: Session,
    *,
    tenant_no: str,
    samples: list,
    exclude_protocol_id: Optional[str] = None,
    exclude_donation_id: Optional[str] = None,
) -> None:
    """创建/更新样本行前校验：样本编码不得已被任一业务单占用（转移配对单豁免）。"""
    codes: List[str] = []
    for raw_item in samples or []:
        raw = _normalize_fields(raw_item)
        c = (raw.get("sample_code") or "").strip()
        if c:
            codes.append(c)
    if not codes:
        return
    used = get_sample_codes_used_in_protocol_samples(
        db, tenant_no=tenant_no, sample_codes=codes
    )
    exempt = get_partner_sample_codes(
        db,
        tenant_no,
        protocol_id=exclude_protocol_id,
        donation_id=exclude_donation_id,
    )
    used -= exempt
    conflicts = set(codes) & used
    if conflicts:
        first = sorted(conflicts)[0]
        raise ValueError(f"样本编码「{first}」已被占用")


def total_pages(total: int, page_size: int) -> int:
    if total <= 0:
        return 0
    return int(math.ceil(total / float(page_size)))


def _fmt_dt(t: Optional[datetime]) -> str:
    return t.strftime("%Y-%m-%d %H:%M:%S") if t else ""


def _dec(v: Any) -> Optional[float]:
    if v is None:
        return None
    if isinstance(v, Decimal):
        return float(v)
    return float(v)


def _money_string(v: Any) -> Optional[str]:
    if v is None:
        return None
    return format(Decimal(str(v)).quantize(Decimal("0.01")), "f")


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


def _today_yyyymmdd() -> str:
    return datetime.now().strftime("%Y%m%d")


def _normalize_yyyymmdd_date(val: Any) -> Optional[str]:
    """将 YYYYMMDD / YYYY-MM-DD 等格式规范为 8 位日期字符串。"""
    if val is None:
        return None
    s = str(val).strip()
    if not s:
        return None
    compact = s.replace("-", "").replace("/", "")[:8]
    if len(compact) == 8 and compact.isdigit():
        return compact
    return None


def _validate_yyyymmdd_date(val: Any, field_label: str) -> Optional[str]:
    """Validate supplied business dates instead of silently clearing bad input."""
    if val is None or str(val).strip() == "":
        return None
    compact = str(val).strip().replace("-", "").replace("/", "")
    if len(compact) != 8 or not compact.isdigit():
        raise ValueError(f"{field_label}日期格式无效")
    try:
        datetime.strptime(compact, "%Y%m%d")
    except ValueError as exc:
        raise ValueError(f"{field_label}日期无效") from exc
    return compact


_SNAKE_ALIASES = {
    "apgar1min": "apgar_1min",
    "apgar5min": "apgar_5min",
}


def _normalize_fields(fields: Optional[dict]) -> dict:
    """将前端 camelCase 字段名转为数据库 snake_case。"""
    if not fields:
        return {}
    out: Dict[str, Any] = {}
    for k, v in fields.items():
        sk = _SNAKE_ALIASES.get(k) or camel_to_snake(k)
        out[sk] = v
    return out


def _pop_attachment_urls(raw: Dict[str, Any], *keys: str) -> Optional[str]:
    """从保存载荷中取出附件字段（支持 snake/camel、URL 串或附件对象列表）。"""
    for key in keys:
        if key not in raw:
            continue
        urls = attachments_to_urls(raw.pop(key))
        if urls:
            return urls
    return None


def attachments_to_urls(attachments: Any) -> Optional[str]:
    """附件列表或字符串 → 逗号分隔 URL（MinIO）。"""
    if attachments is None:
        return None
    if isinstance(attachments, str):
        return attachments.strip() or None
    if isinstance(attachments, list):
        urls: List[str] = []
        for item in attachments:
            if isinstance(item, dict):
                u = (item.get("fileUrl") or item.get("file_url") or "").strip()
                if u:
                    urls.append(u)
            elif isinstance(item, str) and item.strip():
                urls.append(item.strip())
        return ",".join(urls) if urls else None
    return None


def _normalize_attachment_url_key(url: str) -> str:
    u = (url or "").strip()
    if not u:
        return ""
    if "?" in u:
        u = u.split("?", 1)[0]
    return u.lstrip("/")


def _format_attachment_file_size(size_bytes: int) -> str:
    if size_bytes < 1024:
        return f"{size_bytes} B"
    if size_bytes < 1024 * 1024:
        return f"{size_bytes / 1024:.1f} KB"
    return f"{size_bytes / (1024 * 1024):.1f} MB"


def _lookup_attachment_by_url(
    db: Session,
    url: str,
    tenant_no: Optional[str],
) -> Optional[BizAttachment]:
    key = _normalize_attachment_url_key(url)
    if not key:
        return None
    q = db.query(BizAttachment).filter(BizAttachment.del_flag == 0)
    if tenant_no:
        q = q.filter(BizAttachment.tenant_no == tenant_no)
    row = q.filter(BizAttachment.file_url == key).first()
    if row:
        return row
    tail = key.rsplit("/", 1)[-1]
    if tail:
        row = (
            q.filter(BizAttachment.file_url.like(f"%/{tail}"))
            .order_by(BizAttachment.create_time.desc())
            .first()
        )
    return row


def lookup_attachments_by_urls(
    db: Session,
    urls: List[str],
    tenant_no: Optional[str],
) -> Dict[str, BizAttachment]:
    """批量按 file_url 精确匹配附件，避免列表场景逐 URL 查库。

    返回 normalize(url) -> BizAttachment。未命中的 URL 不回退 LIKE，
    由调用方按无元数据展示；详情单条查询仍可用 `_lookup_attachment_by_url`。
    """
    keys: List[str] = []
    seen = set()
    for url in urls:
        key = _normalize_attachment_url_key(url)
        if not key or key in seen:
            continue
        seen.add(key)
        keys.append(key)
    if not keys:
        return {}
    q = db.query(BizAttachment).filter(
        BizAttachment.del_flag == 0,
        BizAttachment.file_url.in_(keys),
    )
    if tenant_no:
        q = q.filter(BizAttachment.tenant_no == tenant_no)
    out: Dict[str, BizAttachment] = {}
    for row in q.all():
        k = _normalize_attachment_url_key(row.file_url or "")
        if k and k not in out:
            out[k] = row
    return out


def _attachment_item_from_row(url: str, att: Optional[BizAttachment]) -> Dict[str, Any]:
    if att:
        size = att.file_size or 0
        return {
            "id": att.id,
            "fileUrl": url,
            "fileName": att.original_name or url.rsplit("/", 1)[-1],
            "fileSize": size,
            "fileSizeDisplay": _format_attachment_file_size(size),
            "fileType": att.file_type or "",
        }
    name = url.rsplit("/", 1)[-1] if "/" in url else url
    return {"fileUrl": url, "fileName": name, "fileSize": 0}


def urls_to_attachment_list(
    urls_str: Optional[str],
    db: Optional[Session] = None,
    tenant_no: Optional[str] = None,
    *,
    attachment_by_url: Optional[Dict[str, BizAttachment]] = None,
) -> List[Dict[str, Any]]:
    """逗号分隔 URL → 前端附件卡片结构；有 db 时从 biz_attachment 取原始文件名。

    传入 `attachment_by_url`（normalize key → row）时只做内存匹配，不再逐条查库。
    """
    if not urls_str:
        return []
    result: List[Dict[str, Any]] = []
    for url in urls_str.split(","):
        u = url.strip()
        if not u:
            continue
        att: Optional[BizAttachment] = None
        if attachment_by_url is not None:
            att = attachment_by_url.get(_normalize_attachment_url_key(u))
        elif db:
            att = _lookup_attachment_by_url(db, u, tenant_no)
        result.append(_attachment_item_from_row(u, att))
    return result


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


def _protocol_to_dict(p: BizCollectionProtocol) -> Dict[str, Any]:
    return {
        "id": p.id,
        "protocolNoId": p.protocol_no_id or "",
        "protocolNo": p.protocol_no or "",
        "protocolStatus": p.protocol_status or "",
        "protocolAttachments": p.protocol_attachments or "",
        "consentFormUrls": p.consent_form_urls or "",
        "healthSurveyFormUrls": p.health_survey_form_urls or "",
        "registerDate": (p.register_date or "").strip() if p.register_date else "",
        "effectiveTime": _fmt_dt(p.effective_time),
        "effectiveDesc": p.effective_desc or "",
        "expireTime": _fmt_dt(p.expire_time),
        "expireDesc": p.expire_desc or "",
        "voidTime": _fmt_dt(p.void_time),
        "voidDesc": p.void_desc or "",
        "voidType": p.void_type or "",
        "toResearchTime": _fmt_dt(p.to_research_time),
        "toResearchDesc": p.to_research_desc or "",
        "collectorUserId": p.collector_user_id or "",
        "protocolTotalAmount": _dec(p.protocol_total_amount)
        if getattr(p, "protocol_total_amount", None) is not None
        else None,
        "sourcePresignOrderId": getattr(p, "source_presign_order_id", None) or "",
        "paymentModelVersion": int(
            getattr(p, "payment_model_version", None) or 1
        ),
        "tenantNo": p.tenant_no,
        "orderNum": p.order_num if p.order_num is not None else 0,
        "remark": p.remark or "",
        "createTime": _fmt_dt(p.create_time),
        "updateTime": _fmt_dt(p.update_time),
    }


def _payment_to_dict(p: BizProtocolPayment) -> Dict[str, Any]:
    ratios_list = [
        format(value, "f")
        for value in parse_installment_ratios(p.installment_ratios)
    ]
    return {
        "id": p.id,
        "collectionProtocolId": p.collection_protocol_id,
        "payMode": p.pay_mode or "",
        "contractAmount": _money_string(p.contract_amount),
        "installmentCount": p.installment_count,
        "installmentIntervalMonths": _dec(p.installment_interval_months),
        "lumpSumDeadline": (p.lump_sum_deadline or "").strip() if p.lump_sum_deadline else "",
        "installmentStartDate": (p.installment_start_date or "").strip() if p.installment_start_date else "",
        "installmentAdvanceDays": p.installment_advance_days if p.installment_advance_days is not None else 30,
        "installmentRatios": ratios_list,
        "tenantNo": p.tenant_no,
        "orderNum": p.order_num if p.order_num is not None else 0,
        "remark": p.remark or "",
        "createTime": _fmt_dt(p.create_time),
        "updateTime": _fmt_dt(p.update_time),
    }


def _detail_to_dict(
    d: BizProtocolPaymentDetail,
    *,
    protocol_no: str = "",
    collection_protocol_id: str = "",
    protocol_status: str = "",
    pay_mode: str = "",
    mother_name: str = "",
    mother_phone: str = "",
    mother_id_card: str = "",
    newborn_name: str = "",
    newborn_id_card: str = "",
) -> Dict[str, Any]:
    return {
        "id": d.id,
        "paymentId": d.payment_id,
        "collectionProtocolId": collection_protocol_id,
        "protocolNo": protocol_no,
        "protocolStatus": protocol_status,
        "payMode": pay_mode,
        "motherName": mother_name,
        "motherPhone": mother_phone,
        "motherIdCard": mother_id_card,
        "newbornName": newborn_name,
        "newbornIdCard": newborn_id_card,
        "transferType": d.transfer_type or "",
        "dueAmount": _money_string(d.due_amount),
        "transferAmount": _money_string(d.transfer_amount),
        "transferTime": _fmt_dt(d.transfer_time),
        "transferAttachments": d.transfer_attachments or "",
        "financeConfirmAttachments": d.finance_confirm_attachments or "",
        "financeRemark": d.finance_remark or "",
        "refundFinanceConfirmAttachments": d.refund_finance_confirm_attachments or "",
        "refundFinanceRemark": d.refund_finance_remark or "",
        "payStatus": d.pay_status or "",
        "installmentNo": d.installment_no,
        "installmentDueDate": (d.installment_due_date or "").strip() if d.installment_due_date else "",
        "refundAmount": _money_string(d.refund_amount),
        "refundRemark": d.refund_remark or "",
        "refundAttachments": d.refund_attachments or "",
        "tenantNo": d.tenant_no,
        "orderNum": d.order_num if d.order_num is not None else 0,
        "remark": d.remark or "",
        "createTime": _fmt_dt(d.create_time),
        "updateTime": _fmt_dt(d.update_time),
    }


def _audit_to_dict(
    a: BizProtocolAudit,
    *,
    protocol_no: str = "",
    auditor_user_name: str = "",
) -> Dict[str, Any]:
    return {
        "id": a.id,
        "collectionProtocolId": a.collection_protocol_id,
        "protocolNo": protocol_no,
        "auditorUserId": a.auditor_user_id or "",
        "auditorUserName": auditor_user_name or "",
        "auditRemark": a.audit_remark or "",
        "auditStatus": a.audit_status or "",
        "tenantNo": a.tenant_no,
        "orderNum": a.order_num if a.order_num is not None else 0,
        "remark": a.remark or "",
        "createTime": _fmt_dt(a.create_time),
        "updateTime": _fmt_dt(a.update_time),
    }


_PROTOCOL_STATUS_CHANGE_TYPE = {
    "sxz": "activate",
    "ygq": "expire",
    "yzf": "void",
    "zky": "to_research",
}


def _status_change_to_dict(
    row: BizProtocolStatusChange,
    *,
    operator_user_name: str = "",
) -> Dict[str, Any]:
    return {
        "id": row.id,
        "collectionProtocolId": row.collection_protocol_id,
        "changeType": row.change_type or "",
        "fromStatus": row.from_status or "",
        "toStatus": row.to_status or "",
        "changeTime": _fmt_dt(row.change_time),
        "changeDesc": row.change_desc or "",
        "voidType": row.void_type or "",
        "operatorUserId": row.operator_user_id or "",
        "operatorUserName": operator_user_name or "",
        "tenantNo": row.tenant_no,
        "createTime": _fmt_dt(row.create_time),
        "updateTime": _fmt_dt(row.update_time),
    }


def append_protocol_status_change(
    db: Session,
    *,
    collection_protocol_id: str,
    change_type: str,
    from_status: Optional[str],
    to_status: str,
    change_time: datetime,
    change_desc: Optional[str] = None,
    void_type: Optional[str] = None,
    tenant_no: str,
    user_id: Optional[str] = None,
) -> BizProtocolStatusChange:
    """追加一条状态变更历史（不 commit）。"""
    now = datetime.now()
    obj = BizProtocolStatusChange(
        id=generate_uuid(),
        collection_protocol_id=collection_protocol_id,
        change_type=(change_type or "").strip(),
        from_status=(from_status or "").strip() or None,
        to_status=(to_status or "").strip(),
        change_time=change_time,
        change_desc=(change_desc or "").strip() or None,
        void_type=(void_type or "").strip() or None,
        operator_user_id=user_id,
        tenant_no=tenant_no,
        order_num=0,
        remark=None,
        create_by=user_id,
        create_dept_id="",
        create_time=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(obj)
    return obj


def list_protocol_status_changes(
    db: Session,
    *,
    collection_protocol_id: str,
    tenant_no: str,
) -> List[Dict[str, Any]]:
    pid = (collection_protocol_id or "").strip()
    if not pid:
        return []
    rows = (
        db.query(BizProtocolStatusChange)
        .filter(
            BizProtocolStatusChange.collection_protocol_id == pid,
            BizProtocolStatusChange.tenant_no == tenant_no,
            BizProtocolStatusChange.del_flag == 0,
        )
        .order_by(
            BizProtocolStatusChange.change_time.asc(),
            BizProtocolStatusChange.create_time.asc(),
        )
        .all()
    )
    op_ids = list({r.operator_user_id for r in rows if r.operator_user_id})
    name_map: Dict[str, str] = {}
    if op_ids:
        for u in db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(op_ids)).all():
            name_map[u.id] = u.nick_name or ""
    return [
        _status_change_to_dict(r, operator_user_name=name_map.get(r.operator_user_id or "", ""))
        for r in rows
    ]


def _party_customer_to_dict(p: BizProtocolPartyCustomer) -> Dict[str, Any]:
    return {
        "id": p.id,
        "collectionProtocolId": p.collection_protocol_id,
        "customerId": p.customer_id or "",
        "fatherCustomerId": p.father_customer_id or "",
        "newbornCustomerId": p.newborn_customer_id or "",
        "motherName": p.mother_name or "",
        "motherPhone": p.mother_phone or "",
        "motherIdCard": p.mother_id_card or "",
        "motherBirthYm": (p.mother_birth_ym or "").strip() if p.mother_birth_ym else "",
        "motherEthnic": p.mother_ethnic or "",
        "motherEmail": p.mother_email or "",
        "motherAddress": p.mother_address or "",
        "fatherName": p.father_name or "",
        "fatherPhone": p.father_phone or "",
        "fatherIdCard": p.father_id_card or "",
        "newbornName": p.newborn_name or "",
        "newbornIdCard": p.newborn_id_card or "",
        "expectedDeliveryDate": format_date(p.expected_delivery_date),
        "newbornDeliveryTime": _fmt_dt(p.newborn_delivery_time),
        "newbornSex": p.newborn_sex or "",
        "deliveryMode": p.delivery_mode or "",
        "birthOrder": p.birth_order,
        "newbornWeightGram": p.newborn_weight_gram,
        "apgar1min": _dec(p.apgar_1min),
        "apgar5min": _dec(p.apgar_5min),
        "collectorUserId": p.collector_user_id or "",
        "tenantNo": p.tenant_no,
        "orderNum": p.order_num if p.order_num is not None else 0,
        "remark": p.remark or "",
        "createTime": _fmt_dt(p.create_time),
        "updateTime": _fmt_dt(p.update_time),
    }


def _collection_info_to_dict(c: BizProtocolCollectionInfo) -> Dict[str, Any]:
    return {
        "id": c.id,
        "collectionProtocolId": c.collection_protocol_id,
        "regionId": c.region_id or "",
        "hospitalId": c.hospital_id or "",
        "medicalStaff": c.medical_staff or "",
        "salesUserId": c.sales_user_id or "",
        "regionManagerUserId": c.region_manager_user_id or "",
        "sampleCollectTime": _fmt_dt(c.sample_collect_time),
        "maternalBloodCollectTime": _fmt_dt(c.maternal_blood_collect_time),
        "tenantNo": c.tenant_no,
        "orderNum": c.order_num if c.order_num is not None else 0,
        "remark": c.remark or "",
        "createTime": _fmt_dt(c.create_time),
        "updateTime": _fmt_dt(c.update_time),
    }


def _sample_to_dict(s: BizProtocolSample) -> Dict[str, Any]:
    return {
        "id": s.id,
        "collectionProtocolId": s.collection_protocol_id,
        "sampleTypeId": s.sample_type_id or "",
        "sampleCodeId": s.sample_code_id or "",
        "sampleCode": s.sample_code or "",
        "collectAmount": _dec(s.collect_amount),
        "unitId": s.unit_id or "",
        "price": _dec(s.price),
        "originalPrice": _dec(s.original_price),
        "storageYears": s.storage_years,
        "sampleStatus": (s.sample_status or "").strip() if (s.sample_status or "").strip() else "",
        "storageInDate": (s.storage_in_date or "").strip() if s.storage_in_date else "",
        "detectStartDate": (s.detect_start_date or "").strip() if s.detect_start_date else "",
        "voidDate": (s.void_date or "").strip() if s.void_date else "",
        "storageExpireDate": (s.storage_expire_date or "").strip() if s.storage_expire_date else "",
        **sample_special_to_dict(s),
        "tenantNo": s.tenant_no,
        "orderNum": s.order_num if s.order_num is not None else 0,
        "remark": s.remark or "",
        "createTime": _fmt_dt(s.create_time),
        "updateTime": _fmt_dt(s.update_time),
    }


def _transport_to_dict(t: BizProtocolSampleTransport) -> Dict[str, Any]:
    return {
        "id": t.id,
        "collectionProtocolId": t.collection_protocol_id,
        "sampleId": t.sample_id,
        "transportType": t.transport_type or "",
        "expressCompanyId": t.express_company_id or "",
        "waybillNo": t.waybill_no or "",
        "destSiteId": t.dest_site_id or "",
        "expressFee": _dec(t.express_fee),
        "etaTime": _fmt_dt(t.eta_time),
        "carrierName": t.carrier_name or "",
        "carrierPhone": t.carrier_phone or "",
        "remark": t.remark or "",
        "orderNum": t.order_num if t.order_num is not None else 0,
        "createTime": _fmt_dt(t.create_time),
        "updateTime": _fmt_dt(t.update_time),
    }


def _storage_cert_to_dict(s: BizProtocolStorageCert) -> Dict[str, Any]:
    from app.services.sample_info_progress import storage_cert_handled

    return {
        "id": s.id,
        "collectionProtocolId": s.collection_protocol_id,
        "makeStatus": s.make_status or "0",
        "certImageUrls": s.cert_image_urls or "",
        "makeDate": (s.make_date or "").strip() if s.make_date else "",
        "handleTime": _fmt_dt(s.handle_time) if getattr(s, "handle_time", None) else "",
        "handlerUserId": getattr(s, "handler_user_id", None) or "",
        "handled": storage_cert_handled(s),
        "tenantNo": s.tenant_no,
        "orderNum": s.order_num if s.order_num is not None else 0,
        "remark": s.remark or "",
        "createTime": _fmt_dt(s.create_time),
        "updateTime": _fmt_dt(s.update_time),
    }


def _format_newborn_info(pc: Optional[BizProtocolPartyCustomer]) -> str:
    if not pc:
        return ""
    parts: List[str] = []
    name = (pc.newborn_name or "").strip()
    if name:
        parts.append(name)
    sex = _NEWBORN_SEX_LABEL.get((pc.newborn_sex or "").strip(), "")
    if sex:
        parts.append(sex)
    if pc.newborn_weight_gram is not None:
        parts.append(f"{pc.newborn_weight_gram}g")
    if pc.newborn_delivery_time:
        parts.append(_fmt_dt(pc.newborn_delivery_time))
    return " / ".join(parts)


def _format_sample_collect_summary(samples: List[BizProtocolSample]) -> str:
    """汇总协议单下各样本采集情况，如「1个已采集，1个未采集」。"""
    collected = 0
    not_collected = 0
    for s in samples:
        status = (s.sample_status or "").strip()
        if status:
            collected += 1
        else:
            not_collected += 1
    parts: List[str] = []
    if collected:
        parts.append(f"{collected}个已采集")
    if not_collected:
        parts.append(f"{not_collected}个未采集")
    return "，".join(parts)


def _format_storage_cert_sample_info(
    samples: List[Dict[str, Any]], type_map: Dict[str, str]
) -> str:
    parts: List[str] = []
    for s in samples:
        type_name = type_map.get(s.get("sampleTypeId") or "", "") or "样本"
        code = (s.get("sampleCode") or "").strip()
        status = (s.get("sampleStatus") or "").strip()
        seg = type_name
        if code:
            seg += f"({code})"
        if status:
            seg += f"[{_SAMPLE_STATUS_LABEL.get(status, status)}]"
        parts.append(seg)
    return "；".join(parts)


def _pick_storage_in_date(samples: List[Dict[str, Any]]) -> str:
    for s in samples:
        d = (s.get("storageInDate") or "").strip()
        if d:
            return d
    return ""


def _load_protocol_storage_in_date_map(
    db: Session,
    protocol_ids: List[str],
    tenant_no: str,
) -> Dict[str, str]:
    """从入库放行审核记录补全入库日期（历史数据可能未写入样本 storage_in_date）。"""
    if not protocol_ids:
        return {}
    rows = (
        db.query(BizStorageReleaseReview)
        .filter(
            BizStorageReleaseReview.protocol_id.in_(protocol_ids),
            BizStorageReleaseReview.tenant_no == tenant_no,
            BizStorageReleaseReview.del_flag == 0,
            BizStorageReleaseReview.storage_status == "yrk",
            BizStorageReleaseReview.storage_time.isnot(None),
        )
        .all()
    )
    out: Dict[str, str] = {}
    for rev in rows:
        pid = (rev.protocol_id or "").strip()
        if not pid or not rev.storage_time:
            continue
        d = rev.storage_time.strftime("%Y%m%d")
        prev = out.get(pid)
        if not prev or d < prev:
            out[pid] = d
    return out


def _load_sample_type_name_map(db: Session, samples: List[BizProtocolSample]) -> Dict[str, str]:
    type_ids = list({s.sample_type_id for s in samples if s.sample_type_id})
    if not type_ids:
        return {}
    type_map: Dict[str, str] = {}
    for t in (
        db.query(BizCellSampleType)
        .filter(BizCellSampleType.id.in_(type_ids), BizCellSampleType.del_flag == 0)
        .all()
    ):
        type_map[t.id] = t.type_name or ""
    return type_map


def _load_unit_name_map(db: Session, samples: List[BizProtocolSample]) -> Dict[str, str]:
    unit_ids = list({s.unit_id for s in samples if s.unit_id})
    if not unit_ids:
        return {}
    unit_map: Dict[str, str] = {}
    for u in (
        db.query(BizItemUnit)
        .filter(BizItemUnit.id.in_(unit_ids), BizItemUnit.del_flag == 0)
        .all()
    ):
        unit_map[u.id] = u.unit_name or ""
    return unit_map


def _load_site_name_map(db: Session, site_ids: List[str]) -> Dict[str, str]:
    ids = list({sid for sid in site_ids if sid})
    if not ids:
        return {}
    site_map: Dict[str, str] = {}
    for site in (
        db.query(SysSite)
        .filter(SysSite.id.in_(ids), SysSite.del_flag == 0)
        .all()
    ):
        site_map[site.id] = site.site_name or ""
    return site_map


def _load_express_name_map(db: Session, express_ids: List[str]) -> Dict[str, str]:
    ids = list({eid for eid in express_ids if eid})
    if not ids:
        return {}
    express_map: Dict[str, str] = {}
    for row in (
        db.query(BizExpressCompany)
        .filter(BizExpressCompany.id.in_(ids), BizExpressCompany.del_flag == 0)
        .all()
    ):
        express_map[row.id] = row.express_name or ""
    return express_map


def list_storage_certs(
    db: Session,
    *,
    tenant_no: str,
    keyword: Optional[str] = None,
    make_status: Optional[str] = None,
    storage_in_date_start: Optional[str] = None,
    storage_in_date_end: Optional[str] = None,
    page: int = 1,
    page_size: int = 10,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    q = db.query(BizCollectionProtocol).filter(
        BizCollectionProtocol.del_flag == 0,
        BizCollectionProtocol.tenant_no == tenant_no,
        BizCollectionProtocol.protocol_status == "sxz",
    )
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        party_ids = (
            db.query(BizProtocolPartyCustomer.collection_protocol_id)
            .filter(
                BizProtocolPartyCustomer.del_flag == 0,
                BizProtocolPartyCustomer.tenant_no == tenant_no,
                BizProtocolPartyCustomer.mother_name.like(like),
            )
        )
        q = q.filter(BizCollectionProtocol.id.in_(party_ids))
    if user_info:
        from app.core.data_scope import apply_data_scope
        q = apply_data_scope(q, BizCollectionProtocol, user_info, db, tenant_no)

    rows_all = q.order_by(BizCollectionProtocol.create_time.desc()).all()
    protocol_ids = [r.id for r in rows_all]
    pc_map: Dict[str, BizProtocolPartyCustomer] = {}
    sc_map: Dict[str, BizProtocolStorageCert] = {}
    sample_map: Dict[str, List[Dict[str, Any]]] = {}
    type_map: Dict[str, str] = {}
    storage_in_date_map: Dict[str, str] = {}
    if protocol_ids:
        for pc in (
            db.query(BizProtocolPartyCustomer)
            .filter(
                BizProtocolPartyCustomer.collection_protocol_id.in_(protocol_ids),
                BizProtocolPartyCustomer.del_flag == 0,
            )
            .all()
        ):
            pc_map[pc.collection_protocol_id] = pc

        for sc in (
            db.query(BizProtocolStorageCert)
            .filter(
                BizProtocolStorageCert.collection_protocol_id.in_(protocol_ids),
                BizProtocolStorageCert.del_flag == 0,
            )
            .all()
        ):
            sc_map[sc.collection_protocol_id] = sc

        s_rows = (
            db.query(BizProtocolSample)
            .filter(
                BizProtocolSample.collection_protocol_id.in_(protocol_ids),
                BizProtocolSample.del_flag == 0,
            )
            .all()
        )
        type_map = _load_sample_type_name_map(db, s_rows)
        for s in s_rows:
            sample_map.setdefault(s.collection_protocol_id, []).append(_sample_to_dict(s))

        storage_in_date_map = _load_protocol_storage_in_date_map(db, protocol_ids, tenant_no)

    items: List[Dict[str, Any]] = []
    ms_filter = (make_status or "").strip()
    start_in = _yyyymmdd_from_date_param(storage_in_date_start) if storage_in_date_start else ""
    end_in = _yyyymmdd_from_date_param(storage_in_date_end) if storage_in_date_end else ""
    for r in rows_all:
        sc = sc_map.get(r.id)
        cert_ms = (sc.make_status or "0") if sc else "0"
        if ms_filter and cert_ms != ms_filter:
            continue
        pc = pc_map.get(r.id)
        samples = sample_map.get(r.id, [])
        storage_in = _pick_storage_in_date(samples) or storage_in_date_map.get(r.id, "")
        if start_in or end_in:
            sid = _storage_in_date_to_yyyymmdd(storage_in)
            if not sid:
                continue
            if start_in and sid < start_in:
                continue
            if end_in and sid > end_in:
                continue
        cert_dict = _storage_cert_to_dict(sc) if sc else {"makeStatus": "0", "certImageUrls": "", "makeDate": "", "remark": ""}
        items.append(
            {
                "id": r.id,
                "protocolId": r.id,
                "protocolNo": r.protocol_no or "",
                "protocolStatus": r.protocol_status or "",
                "motherName": (pc.mother_name or "") if pc else "",
                "motherPhone": (pc.mother_phone or "") if pc else "",
                "motherIdCard": (pc.mother_id_card or "") if pc else "",
                "newbornName": (pc.newborn_name or "") if pc else "",
                "newbornIdCard": (pc.newborn_id_card or "") if pc else "",
                "newbornDeliveryTime": _fmt_dt(pc.newborn_delivery_time) if pc and pc.newborn_delivery_time else "",
                "newbornSex": (pc.newborn_sex or "") if pc else "",
                "newbornWeightGram": pc.newborn_weight_gram if pc else None,
                "newbornInfo": _format_newborn_info(pc),
                "sampleInfo": _format_storage_cert_sample_info(samples, type_map),
                "storageInDate": storage_in,
                "samples": samples,
                "makeStatus": cert_dict.get("makeStatus", "0"),
                "certImageUrls": cert_dict.get("certImageUrls", ""),
                "makeDate": cert_dict.get("makeDate", ""),
                "remark": cert_dict.get("remark", ""),
            }
        )

    total = len(items)
    page_safe = max(1, page)
    page_size_safe = min(max(1, page_size), 100)
    start = (page_safe - 1) * page_size_safe
    return items[start : start + page_size_safe], total


def _normalize_date_bound(value: str, *, end_of_day: bool = False) -> str:
    val = (value or "").strip()
    if len(val) == 10:
        return f"{val} 23:59:59" if end_of_day else f"{val} 00:00:00"
    return val


def _yyyymmdd_from_date_param(value: str) -> str:
    val = (value or "").strip()
    if len(val) == 10 and val[4] == "-" and val[7] == "-":
        return val.replace("-", "")
    digits = "".join(c for c in val if c.isdigit())
    return digits[:8] if len(digits) >= 8 else digits


def _storage_in_date_to_yyyymmdd(value: str) -> str:
    digits = "".join(c for c in (value or "").strip() if c.isdigit())
    return digits[:8] if len(digits) >= 8 else ""


def list_collection_protocols(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    protocol_status: Optional[str] = None,
    create_time_start: Optional[str] = None,
    create_time_end: Optional[str] = None,
    sample_collect_time_start: Optional[str] = None,
    sample_collect_time_end: Optional[str] = None,
    tenant_no: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    def _filtered():
        q0 = db.query(BizCollectionProtocol).filter(BizCollectionProtocol.del_flag == 0)
        if tenant_no:
            q0 = q0.filter(BizCollectionProtocol.tenant_no == tenant_no)
        if protocol_status:
            q0 = q0.filter(BizCollectionProtocol.protocol_status == protocol_status.strip())
        if create_time_start:
            q0 = q0.filter(
                BizCollectionProtocol.create_time >= _normalize_date_bound(create_time_start)
            )
        if create_time_end:
            q0 = q0.filter(
                BizCollectionProtocol.create_time
                <= _normalize_date_bound(create_time_end, end_of_day=True)
            )
        if sample_collect_time_start or sample_collect_time_end:
            ci_q = db.query(BizProtocolCollectionInfo.collection_protocol_id).filter(
                BizProtocolCollectionInfo.del_flag == 0,
            )
            if tenant_no:
                ci_q = ci_q.filter(BizProtocolCollectionInfo.tenant_no == tenant_no)
            if sample_collect_time_start:
                ci_q = ci_q.filter(
                    BizProtocolCollectionInfo.sample_collect_time
                    >= _normalize_date_bound(sample_collect_time_start)
                )
            if sample_collect_time_end:
                ci_q = ci_q.filter(
                    BizProtocolCollectionInfo.sample_collect_time
                    <= _normalize_date_bound(sample_collect_time_end, end_of_day=True)
                )
            q0 = q0.filter(BizCollectionProtocol.id.in_(ci_q))
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            party_q = db.query(BizProtocolPartyCustomer.collection_protocol_id).filter(
                BizProtocolPartyCustomer.del_flag == 0,
                or_(
                    BizProtocolPartyCustomer.mother_name.like(like),
                    BizProtocolPartyCustomer.mother_id_card.like(like),
                    BizProtocolPartyCustomer.mother_phone.like(like),
                ),
            )
            if tenant_no:
                party_q = party_q.filter(BizProtocolPartyCustomer.tenant_no == tenant_no)
            kw_conds = [
                BizCollectionProtocol.id.in_(party_q),
            ]
            if tenant_no:
                from app.crud import sample_reception as reception_svc

                proto_ids = reception_svc.protocol_ids_matching_lab_storage_keyword(
                    db, tenant_no, kw
                )
                if proto_ids:
                    kw_conds.append(BizCollectionProtocol.id.in_(proto_ids))
            q0 = q0.filter(or_(*kw_conds))
        if user_info:
            from app.core.data_scope import apply_data_scope
            q0 = apply_data_scope(q0, BizCollectionProtocol, user_info, db, tenant_no)
        return q0

    total = _filtered().order_by(None).with_entities(func.count(BizCollectionProtocol.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        _filtered()
        .order_by(BizCollectionProtocol.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )

    # 批量加载子表数据，避免 N+1 查询
    protocol_ids = [r.id for r in rows]
    pc_map: Dict[str, Dict[str, Any]] = {}
    ci_map: Dict[str, Dict[str, Any]] = {}
    if protocol_ids:
        pc_rows = (
            db.query(BizProtocolPartyCustomer)
            .filter(
                BizProtocolPartyCustomer.collection_protocol_id.in_(protocol_ids),
                BizProtocolPartyCustomer.del_flag == 0,
            )
            .all()
        )
        for pc in pc_rows:
            pc_map[pc.collection_protocol_id] = _party_customer_to_dict(pc)

        ci_rows = (
            db.query(BizProtocolCollectionInfo)
            .filter(
                BizProtocolCollectionInfo.collection_protocol_id.in_(protocol_ids),
                BizProtocolCollectionInfo.del_flag == 0,
            )
            .all()
        )
        for ci in ci_rows:
            ci_map[ci.collection_protocol_id] = _collection_info_to_dict(ci)

    # 批量解析地区/医院名称
    region_ids = list({ci_map.get(pid, {}).get("regionId", "") for pid in protocol_ids} - {""})
    hospital_ids = list({ci_map.get(pid, {}).get("hospitalId", "") for pid in protocol_ids} - {""})
    region_name_map: Dict[str, str] = {}
    hospital_name_map: Dict[str, str] = {}
    if region_ids:
        for r in db.query(BizCollectionRegion.id, BizCollectionRegion.region_name).filter(BizCollectionRegion.id.in_(region_ids)).all():
            region_name_map[r.id] = r.region_name or ""
    if hospital_ids:
        for h in db.query(BizCollectionHospital.id, BizCollectionHospital.hospital_name).filter(BizCollectionHospital.id.in_(hospital_ids)).all():
            hospital_name_map[h.id] = h.hospital_name or ""

    # 批量解析市场专员 / 地区经理名称
    sales_user_ids = list({ci_map.get(pid, {}).get("salesUserId", "") for pid in protocol_ids} - {""})
    region_manager_user_ids = list(
        {ci_map.get(pid, {}).get("regionManagerUserId", "") for pid in protocol_ids} - {""}
    )
    collector_user_ids = list({_protocol_to_dict(r).get("collectorUserId", "") for r in rows} - {""})
    all_user_ids = list(set(sales_user_ids + region_manager_user_ids + collector_user_ids))
    user_name_map: Dict[str, str] = {}
    if all_user_ids:
        for u in db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(all_user_ids)).all():
            user_name_map[u.id] = u.nick_name or ""

    # 批量加载协议样本（列表展示样本类型、样本编码）
    sample_map: Dict[str, List[BizProtocolSample]] = {}
    sample_type_name_map: Dict[str, str] = {}
    if protocol_ids:
        s_rows = (
            db.query(BizProtocolSample)
            .filter(
                BizProtocolSample.collection_protocol_id.in_(protocol_ids),
                BizProtocolSample.del_flag == 0,
            )
            .order_by(BizProtocolSample.order_num.asc(), BizProtocolSample.create_time.asc())
            .all()
        )
        sample_type_name_map = _load_sample_type_name_map(db, s_rows)
        for s in s_rows:
            sample_map.setdefault(s.collection_protocol_id, []).append(s)

    all_sample_codes: List[str] = []
    for samples in sample_map.values():
        for s in samples:
            c = (s.sample_code or "").strip()
            if c:
                all_sample_codes.append(c)
    lab_code_map: Dict[str, str] = {}
    reception_summary_map: Dict[str, Dict[str, str]] = {}
    qc_status_map: Dict[str, Dict[str, str]] = {}
    from app.services.protocol_sample_list_summary import (
        build_sample_item_reception_qc_fields,
        get_list_qc_status_map_by_sample_codes,
        get_list_reception_summary_map_by_sample_codes,
    )

    if tenant_no and all_sample_codes:
        from app.crud import sample_reception as reception_svc

        lab_code_map = reception_svc.get_lab_storage_code_map_by_sample_codes(
            db, tenant_no, all_sample_codes, reception_source="cell_storage"
        )
        reception_summary_map = get_list_reception_summary_map_by_sample_codes(
            db, tenant_no, all_sample_codes, reception_source="cell_storage"
        )
        qc_status_map = get_list_qc_status_map_by_sample_codes(
            db, tenant_no, all_sample_codes, reception_source="cell_storage"
        )

    transport_by_sample_id: Dict[str, BizProtocolSampleTransport] = {}
    if protocol_ids:
        for t in (
            db.query(BizProtocolSampleTransport)
            .filter(
                BizProtocolSampleTransport.collection_protocol_id.in_(protocol_ids),
                BizProtocolSampleTransport.tenant_no == tenant_no,
                BizProtocolSampleTransport.del_flag == 0,
            )
            .all()
        ):
            transport_by_sample_id[t.sample_id] = t
    site_name_map = _load_site_name_map(
        db, [t.dest_site_id for t in transport_by_sample_id.values() if t.dest_site_id]
    )
    express_name_map = _load_express_name_map(
        db,
        [t.express_company_id for t in transport_by_sample_id.values() if t.express_company_id],
    )

    from app.services.qc_work_order_fee_summary import build_protocol_payment_summaries

    payment_summary_map = build_protocol_payment_summaries(
        db, protocol_ids, tenant_no, include_sample_payment_modes=True
    )

    # 合并主表 + 子表数据
    result = []
    for r in rows:
        d = _protocol_to_dict(r)
        pc = pc_map.get(r.id)
        ci = ci_map.get(r.id)
        d["collectorUserName"] = user_name_map.get(d.get("collectorUserId", ""), "")
        if pc:
            d["motherName"] = pc.get("motherName", "")
            d["motherPhone"] = pc.get("motherPhone", "")
            d["motherIdCard"] = pc.get("motherIdCard", "")
            d["motherBirthYm"] = pc.get("motherBirthYm", "")
            d["motherEthnic"] = pc.get("motherEthnic", "")
            d["fatherName"] = pc.get("fatherName", "")
            d["fatherPhone"] = pc.get("fatherPhone", "")
            d["newbornName"] = pc.get("newbornName", "")
            d["expectedDeliveryDate"] = pc.get("expectedDeliveryDate", "")
            d["newbornDeliveryTime"] = pc.get("newbornDeliveryTime", "")
            d["newbornSex"] = pc.get("newbornSex", "")
            d["deliveryMode"] = pc.get("deliveryMode", "")
            d["birthOrder"] = pc.get("birthOrder")
            d["newbornWeightGram"] = pc.get("newbornWeightGram")
        else:
            d["motherName"] = ""
            d["motherPhone"] = ""
            d["motherIdCard"] = ""
            d["motherBirthYm"] = ""
            d["motherEthnic"] = ""
            d["fatherName"] = ""
            d["fatherPhone"] = ""
            d["newbornName"] = ""
            d["expectedDeliveryDate"] = ""
            d["newbornDeliveryTime"] = ""
            d["newbornSex"] = ""
            d["deliveryMode"] = ""
            d["birthOrder"] = None
            d["newbornWeightGram"] = None
        if ci:
            d["regionName"] = region_name_map.get(ci.get("regionId", ""), "")
            d["hospitalName"] = hospital_name_map.get(ci.get("hospitalId", ""), "")
            d["salesUserName"] = user_name_map.get(ci.get("salesUserId", ""), "")
            d["regionManagerUserName"] = user_name_map.get(ci.get("regionManagerUserId", ""), "")
            d["sampleCollectTime"] = ci.get("sampleCollectTime", "")
        else:
            d["regionName"] = ""
            d["hospitalName"] = ""
            d["salesUserName"] = ""
            d["regionManagerUserName"] = ""
            d["sampleCollectTime"] = ""
        protocol_samples = sample_map.get(r.id, [])
        type_names: List[str] = []
        codes: List[str] = []
        sample_items: List[Dict[str, Any]] = []
        for s in protocol_samples:
            type_name = sample_type_name_map.get(s.sample_type_id or "", "")
            code = (s.sample_code or "").strip()
            if type_name:
                type_names.append(type_name)
            if code:
                codes.append(code)
            transport_obj = transport_by_sample_id.get(s.id)
            transport_dict: Optional[Dict[str, Any]] = None
            if transport_obj and (
                (transport_obj.transport_type or "").strip() or transport_obj.eta_time
            ):
                dest_site_id = transport_obj.dest_site_id or ""
                express_company_id = transport_obj.express_company_id or ""
                transport_dict = {
                    "transportType": transport_obj.transport_type or "",
                    "destSiteId": dest_site_id,
                    "destSiteName": site_name_map.get(dest_site_id, ""),
                    "expressCompanyId": express_company_id,
                    "expressCompanyName": express_name_map.get(express_company_id, ""),
                    "waybillNo": transport_obj.waybill_no or "",
                    "carrierName": transport_obj.carrier_name or "",
                    "carrierPhone": transport_obj.carrier_phone or "",
                    "etaTime": _fmt_dt(transport_obj.eta_time),
                }
            sample_items.append(
                {
                    "sampleTypeName": type_name,
                    "sampleCode": code,
                    "sampleStatus": (s.sample_status or "").strip(),
                    "labStorageCode": lab_code_map.get(code, "") if code else "",
                    "transport": transport_dict,
                    **sample_special_to_dict(s),
                    **build_sample_item_reception_qc_fields(
                        reception_summary_map.get(code) if code else None,
                        qc_status_map.get(code) if code else None,
                    ),
                }
            )
        d["sampleItems"] = sample_items
        d["sampleTypeNames"] = ",".join(type_names)
        d["sampleCodes"] = ",".join(codes)
        d["labStorageCodes"] = ",".join((lab_code_map.get(c) or "") for c in codes)
        d["sampleCollectSummary"] = _format_sample_collect_summary(protocol_samples)
        d["paymentSummary"] = payment_summary_map.get(r.id)
        result.append(d)
    return result, int(total)


def get_collection_protocol(
    db: Session,
    protocol_id: str,
    tenant_no: str,
    *,
    scope_collector_user_id: Optional[str] = None,
) -> Optional[BizCollectionProtocol]:
    filters = [
        BizCollectionProtocol.id == protocol_id,
        BizCollectionProtocol.del_flag == 0,
        BizCollectionProtocol.tenant_no == tenant_no,
    ]
    if scope_collector_user_id:
        filters.append(
            BizCollectionProtocol.collector_user_id == scope_collector_user_id
        )
    return db.query(BizCollectionProtocol).filter(and_(*filters)).first()


def get_payment_by_protocol_id(db: Session, collection_protocol_id: str, tenant_no: str) -> Optional[BizProtocolPayment]:
    return (
        db.query(BizProtocolPayment)
        .filter(
            and_(
                BizProtocolPayment.collection_protocol_id == collection_protocol_id,
                BizProtocolPayment.tenant_no == tenant_no,
                BizProtocolPayment.del_flag == 0,
            )
        )
        .first()
    )


def get_protocol_detail_bundle(
    db: Session,
    protocol_id: str,
    tenant_no: str,
    *,
    scope_collector_user_id: Optional[str] = None,
) -> Optional[Dict[str, Any]]:
    p = get_collection_protocol(
        db, protocol_id, tenant_no, scope_collector_user_id=scope_collector_user_id
    )
    if not p:
        return None
    pay = get_payment_by_protocol_id(db, protocol_id, tenant_no)
    pc = get_party_customer(db, protocol_id, tenant_no)
    ci = get_collection_info(db, protocol_id, tenant_no)
    s_rows = list_samples(db, protocol_id, tenant_no)
    sc = get_storage_cert(db, protocol_id, tenant_no)
    protocol_dict = _protocol_to_dict(p)
    ci_dict = _collection_info_to_dict(ci) if ci else None
    user_ids: List[str] = []
    if protocol_dict.get("collectorUserId"):
        user_ids.append(str(protocol_dict["collectorUserId"]))
    if ci_dict and ci_dict.get("salesUserId"):
        user_ids.append(str(ci_dict["salesUserId"]))
    if ci_dict and ci_dict.get("regionManagerUserId"):
        user_ids.append(str(ci_dict["regionManagerUserId"]))
    user_name_map = _user_nick_name_map(db, user_ids)
    protocol_dict["collectorUserName"] = user_name_map.get(
        str(protocol_dict.get("collectorUserId") or ""), ""
    )
    if ci_dict:
        ci_dict["salesUserName"] = user_name_map.get(str(ci_dict.get("salesUserId") or ""), "")
        ci_dict["regionManagerUserName"] = user_name_map.get(
            str(ci_dict.get("regionManagerUserId") or ""), ""
        )
        region_id = str(ci_dict.get("regionId") or "")
        if region_id:
            r = (
                db.query(BizCollectionRegion.region_name)
                .filter(BizCollectionRegion.id == region_id)
                .first()
            )
            ci_dict["regionName"] = r.region_name if r else ""
        hospital_id = str(ci_dict.get("hospitalId") or "")
        if hospital_id:
            h = (
                db.query(BizCollectionHospital.hospital_name)
                .filter(BizCollectionHospital.id == hospital_id)
                .first()
            )
            ci_dict["hospitalName"] = h.hospital_name if h else ""
    out: Dict[str, Any] = {"protocol": protocol_dict}
    out["payment"] = _payment_to_dict(pay) if pay else None
    out["partyCustomer"] = _party_customer_to_dict(pc) if pc else None
    out["collectionInfo"] = ci_dict
    transport_rows = list_sample_transports(db, protocol_id, tenant_no)
    site_name_map = _load_site_name_map(
        db, [t.dest_site_id for t in transport_rows if t.dest_site_id]
    )
    transport_by_sample: Dict[str, Dict[str, Any]] = {}
    for t in transport_rows:
        td = _transport_to_dict(t)
        td["destSiteName"] = site_name_map.get(t.dest_site_id or "", "")
        transport_by_sample[t.sample_id] = td
    from app.crud import sample_reception as sample_reception_svc

    sample_code_by_id = {s.id: s.sample_code or "" for s in s_rows}
    reception_by_sample = sample_reception_svc.get_receptions_map_for_samples(
        db,
        tenant_no,
        sample_code_by_id=sample_code_by_id,
    )
    from app.crud import qc_work_order as qc_work_order_svc
    from app.crud import production_work_order as production_work_order_svc
    from app.crud import storage_release_review as storage_release_review_svc

    qc_by_sample = qc_work_order_svc.get_qc_info_map_for_samples(
        db,
        [s.id for s in s_rows],
        tenant_no,
        sample_code_by_id=sample_code_by_id,
    )
    process_qc_by_sample = qc_work_order_svc.get_process_qc_info_map_for_samples(
        db,
        [s.id for s in s_rows],
        tenant_no,
        sample_code_by_id=sample_code_by_id,
    )
    production_by_sample = production_work_order_svc.get_production_info_map_for_samples(
        db,
        [s.id for s in s_rows],
        tenant_no,
        sample_code_by_id=sample_code_by_id,
    )
    storage_release_by_sample = storage_release_review_svc.get_storage_release_map_for_samples(
        db,
        [s.id for s in s_rows],
        tenant_no,
        sample_code_by_id=sample_code_by_id,
    )
    unit_name_map = _load_unit_name_map(db, s_rows)
    type_name_map = _load_sample_type_name_map(db, s_rows)
    from app.crud import sample_freight as sample_freight_svc

    freight_by_sample = sample_freight_svc.freight_map_for_samples(
        db,
        tenant_no=tenant_no,
        sample_source=sample_freight_svc.SAMPLE_SOURCE_CELL,
        sample_ids=[s.id for s in s_rows],
    )
    samples_out: List[Dict[str, Any]] = []
    for s in s_rows:
        item = _sample_to_dict(s)
        item["sampleTypeName"] = type_name_map.get(s.sample_type_id or "", "")
        item["unitName"] = unit_name_map.get(s.unit_id or "", "")
        item["transport"] = transport_by_sample.get(s.id)
        item["freight"] = freight_by_sample.get(s.id)
        item["sampleReception"] = reception_by_sample.get(s.id)
        item["productionReception"] = (reception_by_sample.get(s.id) or {}).get(
            "productionReception"
        )
        item["qcInfo"] = qc_by_sample.get(s.id)
        item["processQcInfoList"] = process_qc_by_sample.get(s.id, [])
        item["productionInfo"] = production_by_sample.get(s.id)
        item["storageReleaseInfo"] = storage_release_by_sample.get(s.id)
        samples_out.append(item)
    out["samples"] = samples_out
    out["storageCert"] = _storage_cert_to_dict(sc) if sc else None
    if out["storageCert"]:
        hid = (out["storageCert"].get("handlerUserId") or "").strip()
        if hid:
            hu = db.query(SysUser).filter(SysUser.id == hid).first()
            out["storageCert"]["handlerUserName"] = (hu.nick_name or "") if hu else ""
        else:
            out["storageCert"]["handlerUserName"] = ""
    ft = get_fullmoon_todo(db, protocol_id, tenant_no)
    delivery_dt = pc.newborn_delivery_time if pc else None
    out["fullmoonTodo"] = _fullmoon_todo_detail_dict(ft, newborn_delivery_time=delivery_dt)
    hid2 = (out["fullmoonTodo"].get("handlerUserId") or "").strip()
    if hid2:
        hu2 = db.query(SysUser).filter(SysUser.id == hid2).first()
        out["fullmoonTodo"]["handlerUserName"] = (hu2.nick_name or "") if hu2 else ""
    else:
        out["fullmoonTodo"]["handlerUserName"] = ""
    out["attachments"] = urls_to_attachment_list(p.protocol_attachments, db, tenant_no)
    out["consentAttachmentFiles"] = urls_to_attachment_list(p.consent_form_urls, db, tenant_no)
    out["healthSurveyAttachmentFiles"] = urls_to_attachment_list(
        p.health_survey_form_urls, db, tenant_no
    )
    if pay:
        detail_rows = list_payment_details(db, pay.id, tenant_no)
        details_out: List[Dict[str, Any]] = []
        for d in detail_rows:
            item = _detail_to_dict(
                d,
                protocol_no=p.protocol_no or "",
                collection_protocol_id=protocol_id,
            )
            item["attachments"] = urls_to_attachment_list(d.transfer_attachments, db, tenant_no)
            item["financeConfirmAttachmentFiles"] = urls_to_attachment_list(
                d.finance_confirm_attachments, db, tenant_no
            )
            item["refundFinanceConfirmAttachmentFiles"] = urls_to_attachment_list(
                d.refund_finance_confirm_attachments, db, tenant_no
            )
            item["refundAttachmentFiles"] = urls_to_attachment_list(
                d.refund_attachments, db, tenant_no
            )
            details_out.append(item)
        out["paymentDetails"] = details_out
    else:
        out["paymentDetails"] = []

    from app.crud import protocol_sample_payment as sp_pay_svc
    from app.crud import protocol_refund_record as rr_svc

    version = int(getattr(p, "payment_model_version", None) or 1)
    if version >= 2:
        out["samplePayments"] = sp_pay_svc.get_sample_payments_bundle(
            db, protocol_id, tenant_no
        )
    else:
        out["samplePayments"] = []
    out["refundRecords"] = rr_svc.list_refund_records(
        db, collection_protocol_id=protocol_id, tenant_no=tenant_no
    )
    source_presign_id = getattr(p, "source_presign_order_id", None)
    # 意向金与预签共用 biz_presign_transfer，仅通过 source_presign_order_id 关联
    out["depositTransfers"] = sp_pay_svc.list_deposit_transfers_for_protocol(
        db, source_presign_order_id=source_presign_id, tenant_no=tenant_no
    )
    out["presignDepositHint"] = sp_pay_svc.get_presign_deposit_hint(
        db,
        source_presign_order_id=source_presign_id,
        tenant_no=tenant_no,
    )
    return out


def list_audits(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    audit_status: Optional[str] = None,
    collection_protocol_id: Optional[str] = None,
    tenant_no: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    q_base = (
        db.query(BizProtocolAudit, BizCollectionProtocol.protocol_no)
        .join(
            BizCollectionProtocol,
            and_(
                BizProtocolAudit.collection_protocol_id == BizCollectionProtocol.id,
                BizCollectionProtocol.del_flag == 0,
            ),
        )
        .filter(BizProtocolAudit.del_flag == 0)
    )
    if tenant_no:
        q_base = q_base.filter(BizProtocolAudit.tenant_no == tenant_no)
    if audit_status:
        q_base = q_base.filter(BizProtocolAudit.audit_status == audit_status.strip())
    pid = (collection_protocol_id or "").strip()
    if pid:
        q_base = q_base.filter(BizProtocolAudit.collection_protocol_id == pid)
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        q_base = q_base.filter(
            or_(
                BizCollectionProtocol.protocol_no.like(like),
                BizProtocolAudit.audit_remark.like(like),
            )
        )
    if user_info:
        from app.core.data_scope import apply_data_scope
        q_base = apply_data_scope(q_base, BizProtocolAudit, user_info, db, tenant_no)

    total = q_base.order_by(None).with_entities(func.count(BizProtocolAudit.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        q_base.order_by(
            BizProtocolAudit.order_num.asc(),
            BizProtocolAudit.create_time.desc(),
        )
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    auditor_ids = list({a.auditor_user_id for a, _ in rows if a.auditor_user_id})
    auditor_name_map: Dict[str, str] = {}
    if auditor_ids:
        for u in db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(auditor_ids)).all():
            auditor_name_map[u.id] = u.nick_name or ""
    out = [
        _audit_to_dict(
            a,
            protocol_no=pn or "",
            auditor_user_name=auditor_name_map.get(a.auditor_user_id or "", ""),
        )
        for a, pn in rows
    ]
    return out, int(total)


def create_audit(
    db: Session,
    *,
    collection_protocol_id: str,
    audit_status: str,
    audit_remark: Optional[str],
    tenant_no: str,
    user_id: Optional[str],
) -> BizProtocolAudit:
    p = get_collection_protocol(db, collection_protocol_id, tenant_no)
    if not p:
        raise ValueError("协议单不存在")
    st = (audit_status or "").strip()
    if st not in ("shtg", "shth"):
        raise ValueError("审核状态无效")
    obj = BizProtocolAudit(
        id=generate_uuid(),
        collection_protocol_id=collection_protocol_id,
        auditor_user_id=user_id,
        audit_remark=(audit_remark or "").strip() or None,
        audit_status=st,
        tenant_no=tenant_no,
        order_num=0,
        remark=None,
        create_by=user_id,
        create_dept_id="",
        create_time=datetime.now(),
        update_by=user_id,
        update_time=datetime.now(),
        del_flag=0,
    )
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def _detail_query_with_protocol(
    db: Session,
    tenant_no: str,
    pay_status_filter: str,
):
    return (
        db.query(BizProtocolPaymentDetail, BizProtocolPayment, BizCollectionProtocol)
        .join(BizProtocolPayment, BizProtocolPaymentDetail.payment_id == BizProtocolPayment.id)
        .join(BizCollectionProtocol, BizProtocolPayment.collection_protocol_id == BizCollectionProtocol.id)
        .filter(
            BizProtocolPaymentDetail.del_flag == 0,
            BizProtocolPayment.del_flag == 0,
            BizCollectionProtocol.del_flag == 0,
            BizProtocolPaymentDetail.tenant_no == tenant_no,
            BizProtocolPaymentDetail.pay_status == pay_status_filter,
        )
    )


def _apply_payment_detail_list_filters(
    q,
    db: Session,
    *,
    tenant_no: str,
    keyword: Optional[str] = None,
    installment_due_date_start: Optional[str] = None,
    installment_due_date_end: Optional[str] = None,
):
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        party_q = db.query(BizProtocolPartyCustomer.collection_protocol_id).filter(
            BizProtocolPartyCustomer.del_flag == 0,
            or_(
                BizProtocolPartyCustomer.mother_name.like(like),
                BizProtocolPartyCustomer.mother_id_card.like(like),
                BizProtocolPartyCustomer.newborn_name.like(like),
                BizProtocolPartyCustomer.newborn_id_card.like(like),
            ),
        )
        if tenant_no:
            party_q = party_q.filter(BizProtocolPartyCustomer.tenant_no == tenant_no)
        q = q.filter(BizCollectionProtocol.id.in_(party_q))
    if installment_due_date_start:
        start_due = _yyyymmdd_from_date_param(installment_due_date_start)
        if start_due:
            q = q.filter(BizProtocolPaymentDetail.installment_due_date >= start_due)
    if installment_due_date_end:
        end_due = _yyyymmdd_from_date_param(installment_due_date_end)
        if end_due:
            q = q.filter(BizProtocolPaymentDetail.installment_due_date <= end_due)
    return q


def _rows_to_payment_detail_dicts(
    db: Session,
    rows: List[Any],
    tenant_no: str,
    *,
    post_process: Optional[Any] = None,
) -> List[Dict[str, Any]]:
    """缴费明细行 → 前端字典（含协议、客户、转账附件）。"""
    protocol_ids = [
        pay.collection_protocol_id for _, pay, _ in rows if pay and pay.collection_protocol_id
    ]
    pc_map = _party_customer_map_by_protocol_ids(db, protocol_ids)
    out: List[Dict[str, Any]] = []
    for d, pay, prot in rows:
        cpid = pay.collection_protocol_id if pay else ""
        pc = pc_map.get(cpid)
        item = _detail_to_dict(
            d,
            protocol_no=(prot.protocol_no if prot else "") or "",
            collection_protocol_id=cpid,
            protocol_status=(prot.protocol_status if prot else "") or "",
            pay_mode=(pay.pay_mode if pay else "") or "",
            mother_name=(pc.mother_name if pc else "") or "",
            mother_id_card=(pc.mother_id_card if pc else "") or "",
            newborn_name=(pc.newborn_name if pc else "") or "",
            newborn_id_card=(pc.newborn_id_card if pc else "") or "",
        )
        item["attachments"] = urls_to_attachment_list(d.transfer_attachments, db, tenant_no)
        if post_process:
            post_process(item, d)
        out.append(item)
    return out


def list_receipt_confirm_details(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    installment_due_date_start: Optional[str] = None,
    installment_due_date_end: Optional[str] = None,
    tenant_no: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    if not tenant_no:
        return [], 0
    q_base = _detail_query_with_protocol(db, tenant_no, "jfqrz")
    q_base = _apply_payment_detail_list_filters(
        q_base,
        db,
        tenant_no=tenant_no,
        keyword=keyword,
        installment_due_date_start=installment_due_date_start,
        installment_due_date_end=installment_due_date_end,
    )
    if user_info:
        from app.core.data_scope import apply_data_scope
        q_base = apply_data_scope(q_base, BizProtocolPaymentDetail, user_info, db, tenant_no)
    total = q_base.order_by(None).with_entities(func.count(BizProtocolPaymentDetail.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        q_base.order_by(
            BizProtocolPaymentDetail.order_num.asc(),
            BizProtocolPaymentDetail.create_time.desc(),
        )
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    return _rows_to_payment_detail_dicts(db, rows, tenant_no), int(total)


def list_receipt_confirmed_details(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    installment_due_date_start: Optional[str] = None,
    installment_due_date_end: Optional[str] = None,
    tenant_no: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    if not tenant_no:
        return [], 0
    q_base = _detail_query_with_protocol(db, tenant_no, "yjf")
    q_base = _apply_payment_detail_list_filters(
        q_base,
        db,
        tenant_no=tenant_no,
        keyword=keyword,
        installment_due_date_start=installment_due_date_start,
        installment_due_date_end=installment_due_date_end,
    )
    if user_info:
        from app.core.data_scope import apply_data_scope
        q_base = apply_data_scope(q_base, BizProtocolPaymentDetail, user_info, db, tenant_no)
    total = q_base.order_by(None).with_entities(func.count(BizProtocolPaymentDetail.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        q_base.order_by(BizProtocolPaymentDetail.update_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )

    def _add_finance_files(item: Dict[str, Any], d: BizProtocolPaymentDetail) -> None:
        item["financeConfirmAttachmentFiles"] = urls_to_attachment_list(
            d.finance_confirm_attachments, db, tenant_no
        )

    return (
        _rows_to_payment_detail_dicts(db, rows, tenant_no, post_process=_add_finance_files),
        int(total),
    )


def confirm_receipt(
    db: Session,
    *,
    detail_id: str,
    tenant_no: str,
    user_id: Optional[str],
    finance_confirm_attachments: Optional[str] = None,
    finance_remark: Optional[str] = None,
) -> BizProtocolPaymentDetail:
    d = (
        db.query(BizProtocolPaymentDetail)
        .filter(
            and_(
                BizProtocolPaymentDetail.id == detail_id,
                BizProtocolPaymentDetail.tenant_no == tenant_no,
                BizProtocolPaymentDetail.del_flag == 0,
            )
        )
        .first()
    )
    if not d:
        raise ValueError("缴费明细不存在")
    if d.pay_status != "jfqrz":
        if d.pay_status == "yjf":
            raise ValueError("该明细已确认收款")
        raise ValueError("当前状态不可确认收款")
    d.pay_status = "yjf"
    if finance_confirm_attachments is not None:
        d.finance_confirm_attachments = finance_confirm_attachments.strip() or None
    if finance_remark is not None:
        d.finance_remark = finance_remark.strip() or None
    d.update_by = user_id
    d.update_time = datetime.now()
    db.add(d)
    db.commit()
    db.refresh(d)
    return d


def reject_receipt(
    db: Session,
    *,
    detail_id: str,
    tenant_no: str,
    user_id: Optional[str],
) -> BizProtocolPaymentDetail:
    d = (
        db.query(BizProtocolPaymentDetail)
        .filter(
            and_(
                BizProtocolPaymentDetail.id == detail_id,
                BizProtocolPaymentDetail.tenant_no == tenant_no,
                BizProtocolPaymentDetail.del_flag == 0,
            )
        )
        .first()
    )
    if not d:
        raise ValueError("缴费明细不存在")
    if d.pay_status != "jfqrz":
        raise ValueError("当前状态不可退回")
    d.pay_status = "dfj"
    d.update_by = user_id
    d.update_time = datetime.now()
    db.add(d)
    db.commit()
    db.refresh(d)
    return d


def list_refund_confirm_details(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    installment_due_date_start: Optional[str] = None,
    installment_due_date_end: Optional[str] = None,
    tenant_no: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    if not tenant_no:
        return [], 0
    q_base = _detail_query_with_protocol(db, tenant_no, "tfqrz")
    q_base = _apply_payment_detail_list_filters(
        q_base,
        db,
        tenant_no=tenant_no,
        keyword=keyword,
        installment_due_date_start=installment_due_date_start,
        installment_due_date_end=installment_due_date_end,
    )
    if user_info:
        from app.core.data_scope import apply_data_scope
        q_base = apply_data_scope(q_base, BizProtocolPaymentDetail, user_info, db, tenant_no)
    total = q_base.order_by(None).with_entities(func.count(BizProtocolPaymentDetail.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        q_base.order_by(
            BizProtocolPaymentDetail.order_num.asc(),
            BizProtocolPaymentDetail.create_time.desc(),
        )
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    def _add_refund_apply_files(item: Dict[str, Any], d: BizProtocolPaymentDetail) -> None:
        item["refundAttachmentFiles"] = urls_to_attachment_list(
            d.refund_attachments, db, tenant_no
        )

    return (
        _rows_to_payment_detail_dicts(db, rows, tenant_no, post_process=_add_refund_apply_files),
        int(total),
    )


def list_refund_confirmed_details(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    installment_due_date_start: Optional[str] = None,
    installment_due_date_end: Optional[str] = None,
    tenant_no: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    if not tenant_no:
        return [], 0
    q_base = _detail_query_with_protocol(db, tenant_no, "ytf")
    q_base = _apply_payment_detail_list_filters(
        q_base,
        db,
        tenant_no=tenant_no,
        keyword=keyword,
        installment_due_date_start=installment_due_date_start,
        installment_due_date_end=installment_due_date_end,
    )
    if user_info:
        from app.core.data_scope import apply_data_scope
        q_base = apply_data_scope(q_base, BizProtocolPaymentDetail, user_info, db, tenant_no)
    total = q_base.order_by(None).with_entities(func.count(BizProtocolPaymentDetail.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        q_base.order_by(BizProtocolPaymentDetail.update_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    def _add_refund_confirmed_files(item: Dict[str, Any], d: BizProtocolPaymentDetail) -> None:
        item["refundAttachmentFiles"] = urls_to_attachment_list(
            d.refund_attachments, db, tenant_no
        )
        item["refundFinanceConfirmAttachmentFiles"] = urls_to_attachment_list(
            d.refund_finance_confirm_attachments, db, tenant_no
        )

    return (
        _rows_to_payment_detail_dicts(db, rows, tenant_no, post_process=_add_refund_confirmed_files),
        int(total),
    )


def confirm_refund(
    db: Session,
    *,
    detail_id: str,
    tenant_no: str,
    user_id: Optional[str],
    refund_finance_confirm_attachments: Optional[str] = None,
    refund_finance_remark: Optional[str] = None,
) -> BizProtocolPaymentDetail:
    d = (
        db.query(BizProtocolPaymentDetail)
        .filter(
            and_(
                BizProtocolPaymentDetail.id == detail_id,
                BizProtocolPaymentDetail.tenant_no == tenant_no,
                BizProtocolPaymentDetail.del_flag == 0,
            )
        )
        .first()
    )
    if not d:
        raise ValueError("缴费明细不存在")
    if d.pay_status != "tfqrz":
        if d.pay_status == "ytf":
            raise ValueError("该明细已确认退费")
        raise ValueError("当前状态不可确认退费")
    d.pay_status = "ytf"
    if refund_finance_confirm_attachments is not None:
        d.refund_finance_confirm_attachments = refund_finance_confirm_attachments.strip() or None
    if refund_finance_remark is not None:
        d.refund_finance_remark = refund_finance_remark.strip() or None
    d.update_by = user_id
    d.update_time = datetime.now()
    db.add(d)
    db.commit()
    db.refresh(d)
    return d


def reject_refund(
    db: Session,
    *,
    detail_id: str,
    tenant_no: str,
    user_id: Optional[str],
) -> BizProtocolPaymentDetail:
    d = (
        db.query(BizProtocolPaymentDetail)
        .filter(
            and_(
                BizProtocolPaymentDetail.id == detail_id,
                BizProtocolPaymentDetail.tenant_no == tenant_no,
                BizProtocolPaymentDetail.del_flag == 0,
            )
        )
        .first()
    )
    if not d:
        raise ValueError("缴费明细不存在")
    if d.pay_status != "tfqrz":
        raise ValueError("当前状态不可退回")
    d.pay_status = "yjf"
    d.update_by = user_id
    d.update_time = datetime.now()
    db.add(d)
    db.commit()
    db.refresh(d)
    return d


def _get_payment_detail_or_raise(
    db: Session,
    *,
    detail_id: str,
    tenant_no: str,
) -> BizProtocolPaymentDetail:
    d = (
        db.query(BizProtocolPaymentDetail)
        .filter(
            and_(
                BizProtocolPaymentDetail.id == detail_id,
                BizProtocolPaymentDetail.tenant_no == tenant_no,
                BizProtocolPaymentDetail.del_flag == 0,
            )
        )
        .first()
    )
    if not d:
        raise ValueError("缴费明细不存在")
    return d


def _parse_transfer_time_value(value: Any) -> Optional[datetime]:
    if value is None or value == "":
        return None
    if isinstance(value, datetime):
        return value
    if isinstance(value, str):
        return parse_datetime(value)
    return None


def _apply_payment_detail_draft_fields(
    db: Session,
    *,
    tenant_no: str,
    d: BizProtocolPaymentDetail,
    user_id: Optional[str],
    fields: Dict[str, Any],
) -> None:
    from app.crud.dict_data import normalize_transfer_type

    if d.pay_status != "dfj":
        raise ValueError("仅已付款状态可编辑缴费明细")
    if "transfer_type" in fields:
        d.transfer_type = normalize_transfer_type(
            db, value=str(fields.get("transfer_type") or ""), tenant_no=tenant_no
        )
    if "transfer_amount" in fields:
        amount = fields.get("transfer_amount")
        if amount is None or amount == "":
            d.transfer_amount = None
        else:
            dec_amount = Decimal(str(amount))
            if dec_amount < 0:
                raise ValueError("转账金额无效")
            d.transfer_amount = dec_amount
    if "transfer_time" in fields:
        d.transfer_time = _parse_transfer_time_value(fields.get("transfer_time"))
    if "transfer_attachments" in fields:
        d.transfer_attachments = (fields.get("transfer_attachments") or "").strip() or None
    if "remark" in fields:
        d.remark = (fields.get("remark") or "").strip() or None
    d.update_by = user_id
    d.update_time = datetime.now()


def _validate_receipt_finance_submit_ready(d: BizProtocolPaymentDetail) -> None:
    if not (d.transfer_type or "").strip():
        raise ValueError("请选择转账方式")
    if d.transfer_amount is None or Decimal(str(d.transfer_amount)) <= 0:
        raise ValueError("请填写转账金额")


def update_payment_detail_draft(
    db: Session,
    *,
    detail_id: str,
    tenant_no: str,
    user_id: Optional[str],
    draft_fields: Dict[str, Any],
) -> BizProtocolPaymentDetail:
    d = _get_payment_detail_or_raise(db, detail_id=detail_id, tenant_no=tenant_no)
    _apply_payment_detail_draft_fields(
        db, tenant_no=tenant_no, d=d, user_id=user_id, fields=draft_fields
    )
    db.add(d)
    db.commit()
    db.refresh(d)
    return d


def _apply_submit_receipt_finance(
    d: BizProtocolPaymentDetail,
    *,
    user_id: Optional[str],
) -> None:
    if d.pay_status != "dfj":
        raise ValueError("仅已付款状态可提交财务确认")
    d.pay_status = "jfqrz"
    d.update_by = user_id
    d.update_time = datetime.now()


def submit_receipt_finance(
    db: Session,
    *,
    detail_id: str,
    tenant_no: str,
    user_id: Optional[str],
    draft_fields: Optional[Dict[str, Any]] = None,
) -> BizProtocolPaymentDetail:
    d = _get_payment_detail_or_raise(db, detail_id=detail_id, tenant_no=tenant_no)
    if draft_fields:
        _apply_payment_detail_draft_fields(
            db, tenant_no=tenant_no, d=d, user_id=user_id, fields=draft_fields
        )
    _validate_receipt_finance_submit_ready(d)
    _apply_submit_receipt_finance(d, user_id=user_id)
    db.add(d)
    db.commit()
    db.refresh(d)
    return d


def batch_submit_receipt_finance(
    db: Session,
    *,
    detail_ids: List[str],
    tenant_no: str,
    user_id: Optional[str],
) -> int:
    ids = [str(i).strip() for i in (detail_ids or []) if str(i).strip()]
    if not ids:
        raise ValueError("请选择缴费明细")
    for detail_id in ids:
        d = _get_payment_detail_or_raise(db, detail_id=detail_id, tenant_no=tenant_no)
        _apply_submit_receipt_finance(d, user_id=user_id)
        db.add(d)
    db.commit()
    return len(ids)


def _apply_submit_refund_finance(
    d: BizProtocolPaymentDetail,
    *,
    user_id: Optional[str],
    refund_amount: Any,
    refund_remark: Optional[str] = None,
    refund_attachments: Optional[str] = None,
) -> None:
    if d.pay_status != "yjf":
        raise ValueError("仅已付款状态可提交退费财务确认")
    if refund_amount is None or Decimal(str(refund_amount)) <= 0:
        raise ValueError("请填写有效的退费金额")
    d.pay_status = "tfqrz"
    d.refund_amount = Decimal(str(refund_amount))
    d.refund_remark = (refund_remark or "").strip() or None
    d.refund_attachments = (refund_attachments or "").strip() or None
    d.update_by = user_id
    d.update_time = datetime.now()


def submit_refund_finance(
    db: Session,
    *,
    detail_id: str,
    tenant_no: str,
    user_id: Optional[str],
    refund_amount: Any,
    refund_remark: Optional[str] = None,
    refund_attachments: Optional[str] = None,
) -> BizProtocolPaymentDetail:
    d = _get_payment_detail_or_raise(db, detail_id=detail_id, tenant_no=tenant_no)
    _apply_submit_refund_finance(
        d,
        user_id=user_id,
        refund_amount=refund_amount,
        refund_remark=refund_remark,
        refund_attachments=refund_attachments,
    )
    db.add(d)
    db.commit()
    db.refresh(d)
    return d


def batch_submit_refund_finance(
    db: Session,
    *,
    detail_ids: List[str],
    tenant_no: str,
    user_id: Optional[str],
    refund_amount: Any,
    refund_remark: Optional[str] = None,
    refund_attachments: Optional[str] = None,
) -> int:
    ids = [str(i).strip() for i in (detail_ids or []) if str(i).strip()]
    if not ids:
        raise ValueError("请选择缴费明细")
    for detail_id in ids:
        d = _get_payment_detail_or_raise(db, detail_id=detail_id, tenant_no=tenant_no)
        _apply_submit_refund_finance(
            d,
            user_id=user_id,
            refund_amount=refund_amount,
            refund_remark=refund_remark,
            refund_attachments=refund_attachments,
        )
        db.add(d)
    db.commit()
    return len(ids)


def _party_customer_map_by_protocol_ids(
    db: Session,
    protocol_ids: List[str],
) -> Dict[str, BizProtocolPartyCustomer]:
    if not protocol_ids:
        return {}
    rows = (
        db.query(BizProtocolPartyCustomer)
        .filter(
            BizProtocolPartyCustomer.collection_protocol_id.in_(protocol_ids),
            BizProtocolPartyCustomer.del_flag == 0,
        )
        .all()
    )
    return {r.collection_protocol_id: r for r in rows}


def _yyyymmdd_from_date_param(value: str) -> str:
    val = (value or "").strip()
    if len(val) == 10 and val[4] == "-" and val[7] == "-":
        return val.replace("-", "")
    return val


def list_payment_details_page(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    pay_status: Optional[str] = None,
    keyword: Optional[str] = None,
    transfer_type: Optional[str] = None,
    installment_due_date_start: Optional[str] = None,
    installment_due_date_end: Optional[str] = None,
    tenant_no: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    if not tenant_no:
        return [], 0
    q = (
        db.query(BizProtocolPaymentDetail, BizProtocolPayment, BizCollectionProtocol)
        .join(BizProtocolPayment, BizProtocolPaymentDetail.payment_id == BizProtocolPayment.id)
        .join(BizCollectionProtocol, BizProtocolPayment.collection_protocol_id == BizCollectionProtocol.id)
        .filter(
            BizProtocolPaymentDetail.del_flag == 0,
            BizProtocolPayment.del_flag == 0,
            BizCollectionProtocol.del_flag == 0,
            BizProtocolPaymentDetail.tenant_no == tenant_no,
        )
    )
    if pay_status:
        q = q.filter(BizProtocolPaymentDetail.pay_status == pay_status)
    if keyword and keyword.strip():
        like = f"%{keyword.strip()}%"
        party_q = db.query(BizProtocolPartyCustomer.collection_protocol_id).filter(
            BizProtocolPartyCustomer.del_flag == 0,
            or_(
                BizProtocolPartyCustomer.mother_name.like(like),
                BizProtocolPartyCustomer.mother_id_card.like(like),
                BizProtocolPartyCustomer.newborn_name.like(like),
                BizProtocolPartyCustomer.newborn_id_card.like(like),
            ),
        )
        if tenant_no:
            party_q = party_q.filter(BizProtocolPartyCustomer.tenant_no == tenant_no)
        q = q.filter(BizCollectionProtocol.id.in_(party_q))
    if transfer_type and transfer_type.strip():
        q = q.filter(BizProtocolPaymentDetail.transfer_type == transfer_type.strip())
    if installment_due_date_start:
        start_due = _yyyymmdd_from_date_param(installment_due_date_start)
        if start_due:
            q = q.filter(BizProtocolPaymentDetail.installment_due_date >= start_due)
    if installment_due_date_end:
        end_due = _yyyymmdd_from_date_param(installment_due_date_end)
        if end_due:
            q = q.filter(BizProtocolPaymentDetail.installment_due_date <= end_due)
    if user_info:
        from app.core.data_scope import apply_data_scope
        q = apply_data_scope(q, BizProtocolPaymentDetail, user_info, db, tenant_no)
    total = q.order_by(None).with_entities(func.count(BizProtocolPaymentDetail.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    due_col = BizProtocolPaymentDetail.installment_due_date
    # 缴费信息 Tab：已付款/付款确认中/退费确认中 → 应缴日期正序；付款完成/已退费 → 倒序
    due_order = due_col.desc() if pay_status in ("yjf", "ytf") else due_col.asc()
    rows = (
        q.order_by(due_order)
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    protocol_ids = [
        pay.collection_protocol_id for _, pay, _ in rows if pay and pay.collection_protocol_id
    ]
    pc_map = _party_customer_map_by_protocol_ids(db, protocol_ids)
    out = []
    for d, pay, prot in rows:
        cpid = pay.collection_protocol_id if pay else ""
        pc = pc_map.get(cpid)
        item = _detail_to_dict(
            d,
            protocol_no=(prot.protocol_no if prot else "") or "",
            collection_protocol_id=cpid,
            protocol_status=(prot.protocol_status if prot else "") or "",
            pay_mode=(pay.pay_mode if pay else "") or "",
            mother_name=(pc.mother_name if pc else "") or "",
            mother_phone=(pc.mother_phone if pc else "") or "",
            mother_id_card=(pc.mother_id_card if pc else "") or "",
            newborn_name=(pc.newborn_name if pc else "") or "",
            newborn_id_card=(pc.newborn_id_card if pc else "") or "",
        )
        item["attachments"] = urls_to_attachment_list(d.transfer_attachments, db, tenant_no)
        out.append(item)
    return out, int(total)


def _classify_installment_warning(due: str, today: str) -> str:
    """overdue | warn | upcoming — due 为 YYYYMMDD"""
    if not due or len(due) != 8 or not due.isdigit():
        return "unknown"
    if due < today:
        return "overdue"
    last_warn = (datetime.strptime(today, "%Y%m%d") + timedelta(days=15)).strftime("%Y%m%d")
    if today <= due <= last_warn:
        return "warn"
    return "upcoming"


def list_installment_warning(
    db: Session,
    *,
    page: int = 1,
    page_size: int = 10,
    warning_type: str = "all",
    tenant_no: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    if not tenant_no:
        return [], 0
    today = _today_yyyymmdd()
    q0 = _detail_query_with_protocol(db, tenant_no, "dfj").filter(
        BizProtocolPaymentDetail.installment_no.isnot(None),
    )
    if user_info:
        from app.core.data_scope import apply_data_scope
        q0 = apply_data_scope(q0, BizProtocolPaymentDetail, user_info, db, tenant_no)

    # 拉取当前租户下候选（分期号非空）；日期过滤在内存中完成以保证与 classify 一致
    rows_all = q0.order_by(BizProtocolPaymentDetail.installment_due_date.asc()).all()
    flattened: List[Dict[str, Any]] = []
    for d, pay, prot in rows_all:
        due_raw = (d.installment_due_date or "").strip() if d.installment_due_date else ""
        wt = _classify_installment_warning(due_raw, today)
        cpid = pay.collection_protocol_id if pay else ""
        item = _detail_to_dict(
            d,
            protocol_no=(prot.protocol_no if prot else "") or "",
            collection_protocol_id=cpid,
            protocol_status=(prot.protocol_status if prot else "") or "",
            pay_mode=(pay.pay_mode if pay else "") or "",
        )
        item["warningType"] = wt
        wt_filter = (warning_type or "all").strip().lower()
        if wt_filter == "overdue" and wt != "overdue":
            continue
        if wt_filter == "warn" and wt != "warn":
            continue
        if wt_filter == "all":
            pass
        elif wt_filter not in ("all", "warn", "overdue"):
            pass
        flattened.append(item)

    total = len(flattened)
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    start = (page - 1) * page_size
    page_rows = flattened[start : start + page_size]
    return page_rows, total


def create_collection_protocol(
    db: Session,
    *,
    protocol_no_id: Optional[str] = None,
    protocol_no: str = "",
    protocol_status: str = "dsx",
    protocol_attachments: Optional[str] = None,
    consent_form_urls: Optional[str] = None,
    health_survey_form_urls: Optional[str] = None,
    register_date: Optional[str] = None,
    collector_user_id: Optional[str] = None,
    protocol_total_amount: Any = None,
    source_presign_order_id: Optional[str] = None,
    payment_model_version: Optional[int] = None,
    remark: Optional[str] = None,
    tenant_no: str = "zkxw",
    user_id: Optional[str] = None,
    create_dept_id: str = "",
    commit: bool = True,
    exclude_presign_order_id: Optional[str] = None,
) -> BizCollectionProtocol:
    pn = (protocol_no or "").strip()
    if not pn:
        pn = allocate_next_biz_no(
            db, tenant_no=tenant_no, kind="collection", user_id=user_id
        )
    # 预签转正式时排除来源预签订单自身占用的编号
    assert_biz_no_available(
        db,
        tenant_no=tenant_no,
        biz_no=pn,
        exclude_presign_order_id=exclude_presign_order_id or source_presign_order_id,
    )
    version = int(payment_model_version) if payment_model_version is not None else 2
    if version not in (1, 2):
        version = 2
    total_amt = None
    if protocol_total_amount not in (None, ""):
        total_amt = Decimal(str(protocol_total_amount))
    obj = BizCollectionProtocol(
        id=generate_uuid(),
        protocol_no_id=(protocol_no_id or "").strip() or None,
        protocol_no=pn,
        protocol_status=protocol_status.strip() or "dsx",
        protocol_attachments=(protocol_attachments or "").strip() or None,
        consent_form_urls=(consent_form_urls or "").strip() or None,
        health_survey_form_urls=(health_survey_form_urls or "").strip() or None,
        register_date=(register_date or "").strip() or None,
        collector_user_id=(collector_user_id or "").strip() or None,
        protocol_total_amount=total_amt,
        source_presign_order_id=(source_presign_order_id or "").strip() or None,
        payment_model_version=version,
        tenant_no=tenant_no,
        order_num=0,
        remark=(remark or "").strip() or None,
        create_by=user_id,
        create_dept_id=create_dept_id,
        create_time=datetime.now(),
        update_by=user_id,
        update_time=datetime.now(),
        del_flag=0,
    )
    db.add(obj)
    if commit:
        db.commit()
        db.refresh(obj)
    else:
        db.flush()
    return obj


def update_collection_protocol(
    db: Session,
    protocol_id: str,
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
    scope_collector_user_id: Optional[str] = None,
    protocol_no: Optional[str] = None,
    protocol_status: Optional[str] = None,
    protocol_attachments: Optional[str] = None,
    consent_form_urls: Optional[str] = None,
    health_survey_form_urls: Optional[str] = None,
    register_date: Optional[str] = None,
    effective_time: Optional[datetime] = None,
    effective_desc: Optional[str] = None,
    expire_time: Optional[datetime] = None,
    expire_desc: Optional[str] = None,
    void_time: Optional[datetime] = None,
    void_desc: Optional[str] = None,
    void_type: Optional[str] = None,
    to_research_time: Optional[datetime] = None,
    to_research_desc: Optional[str] = None,
    collector_user_id: Optional[str] = None,
    protocol_total_amount: Any = None,
    source_presign_order_id: Optional[str] = None,
    remark: Optional[str] = None,
    commit: bool = True,
) -> BizCollectionProtocol:
    obj = get_collection_protocol(
        db, protocol_id, tenant_no, scope_collector_user_id=scope_collector_user_id
    )
    if not obj:
        raise ValueError("协议单不存在或无权操作")
    from_status = (obj.protocol_status or "").strip()
    status_log_payload: Optional[Dict[str, Any]] = None
    if protocol_no is not None:
        pn = protocol_no.strip()
        if pn != (obj.protocol_no or ""):
            raise ValueError("协议号不可修改")
    if protocol_status is not None:
        from app.core.protocol_status_display import assert_protocol_status_transition

        next_status = protocol_status.strip()
        assert_protocol_status_transition(obj.protocol_status or "", next_status, kind="protocol")
        obj.protocol_status = next_status
        if next_status == "yzf" and void_time is not None:
            if void_type is None or not str(void_type).strip():
                raise ValueError("请选择作废类型")
        change_type = _PROTOCOL_STATUS_CHANGE_TYPE.get(next_status)
        change_time_val: Optional[datetime] = None
        change_desc_val: Optional[str] = None
        void_type_val: Optional[str] = None
        if change_type == "activate" and effective_time is not None:
            change_time_val = effective_time
            change_desc_val = effective_desc
        elif change_type == "expire" and expire_time is not None:
            change_time_val = expire_time
            change_desc_val = expire_desc
        elif change_type == "void" and void_time is not None:
            change_time_val = void_time
            change_desc_val = void_desc
            void_type_val = void_type
        elif change_type == "to_research" and to_research_time is not None:
            change_time_val = to_research_time
            change_desc_val = to_research_desc
        if change_type and change_time_val is not None:
            status_log_payload = {
                "change_type": change_type,
                "to_status": next_status,
                "change_time": change_time_val,
                "change_desc": change_desc_val,
                "void_type": void_type_val,
            }
    if protocol_attachments is not None:
        obj.protocol_attachments = protocol_attachments.strip() or None
    if consent_form_urls is not None:
        obj.consent_form_urls = consent_form_urls.strip() or None
    if health_survey_form_urls is not None:
        obj.health_survey_form_urls = health_survey_form_urls.strip() or None
    if register_date is not None:
        obj.register_date = register_date.strip() or None
    if effective_time is not None:
        obj.effective_time = effective_time
    if effective_desc is not None:
        obj.effective_desc = effective_desc.strip() or None
    if expire_time is not None:
        obj.expire_time = expire_time
    if expire_desc is not None:
        obj.expire_desc = expire_desc.strip() or None
    if void_time is not None:
        obj.void_time = void_time
    if void_desc is not None:
        obj.void_desc = void_desc.strip() or None
    if void_type is not None:
        from app.crud.dict_data import normalize_void_type

        obj.void_type = (
            normalize_void_type(db, value=void_type, tenant_no=tenant_no)
            if (void_type or "").strip()
            else None
        )
        if status_log_payload and status_log_payload.get("change_type") == "void":
            status_log_payload["void_type"] = obj.void_type
    if to_research_time is not None:
        obj.to_research_time = to_research_time
    if to_research_desc is not None:
        obj.to_research_desc = to_research_desc.strip() or None
    if collector_user_id is not None:
        obj.collector_user_id = collector_user_id.strip() or None
    if protocol_total_amount is not None:
        if protocol_total_amount == "":
            obj.protocol_total_amount = None
        else:
            obj.protocol_total_amount = Decimal(str(protocol_total_amount))
    if source_presign_order_id is not None:
        obj.source_presign_order_id = (source_presign_order_id or "").strip() or None
    if remark is not None:
        obj.remark = remark.strip() or None
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    if status_log_payload:
        append_protocol_status_change(
            db,
            collection_protocol_id=obj.id,
            change_type=status_log_payload["change_type"],
            from_status=from_status,
            to_status=status_log_payload["to_status"],
            change_time=status_log_payload["change_time"],
            change_desc=status_log_payload.get("change_desc"),
            void_type=status_log_payload.get("void_type"),
            tenant_no=tenant_no,
            user_id=user_id,
        )
    if commit:
        db.commit()
        db.refresh(obj)
    else:
        db.flush()
    return obj


def soft_delete_collection_protocol(
    db: Session,
    protocol_id: str,
    tenant_no: str,
    user_id: Optional[str] = None,
    scope_collector_user_id: Optional[str] = None,
) -> None:
    obj = get_collection_protocol(
        db, protocol_id, tenant_no, scope_collector_user_id=scope_collector_user_id
    )
    if not obj:
        raise ValueError("协议单不存在或无权操作")
    now = datetime.now()
    # 同步软删样本与运输信息，释放样本编码占用（占用校验只看样本 del_flag）
    for s in list_samples(db, protocol_id, tenant_no):
        s.del_flag = 1
        s.update_by = user_id
        s.update_time = now
        db.add(s)
    _soft_delete_transports_for_protocol(db, protocol_id, tenant_no, user_id)
    obj.del_flag = 1
    obj.update_by = user_id
    obj.update_time = now
    db.add(obj)
    db.commit()


# ---------------------------------------------------------------------------
# 子表：当事人客户 / 采集信息 / 样本 / 储存证书
# ---------------------------------------------------------------------------


def get_party_customer(db: Session, collection_protocol_id: str, tenant_no: str) -> Optional[BizProtocolPartyCustomer]:
    return (
        db.query(BizProtocolPartyCustomer)
        .filter(
            and_(
                BizProtocolPartyCustomer.collection_protocol_id == collection_protocol_id,
                BizProtocolPartyCustomer.tenant_no == tenant_no,
                BizProtocolPartyCustomer.del_flag == 0,
            )
        )
        .first()
    )


def upsert_party_customer(
    db: Session,
    collection_protocol_id: str,
    tenant_no: str,
    user_id: Optional[str],
    commit: bool = True,
    create_dept_id: str = "",
    **fields,
) -> BizProtocolPartyCustomer:
    fields = _normalize_fields(fields)
    obj = get_party_customer(db, collection_protocol_id, tenant_no)
    _STR_FIELDS = [
        "customer_id", "father_customer_id", "newborn_customer_id",
        "mother_name", "mother_phone", "mother_id_card",
        "mother_birth_ym", "mother_ethnic", "mother_email", "mother_address",
        "father_name", "father_phone", "father_id_card",
        "newborn_name", "newborn_id_card",
        "newborn_sex", "delivery_mode", "collector_user_id",
    ]
    _INT_FIELDS = ["birth_order", "newborn_weight_gram"]
    _DEC_FIELDS = ["apgar_1min", "apgar_5min"]
    _DATE_FIELDS = ["expected_delivery_date"]
    _DT_FIELDS = ["newborn_delivery_time"]
    if obj:
        for k in _STR_FIELDS:
            if k in fields:
                setattr(obj, k, (fields[k] or "").strip() or None)
        for k in _INT_FIELDS:
            if k in fields:
                setattr(obj, k, fields[k])
        for k in _DEC_FIELDS:
            if k in fields:
                setattr(obj, k, _optional_decimal(fields[k]))
        for k in _DATE_FIELDS:
            if k in fields:
                setattr(obj, k, optional_date(fields[k]))
        for k in _DT_FIELDS:
            if k in fields:
                setattr(obj, k, fields[k])
        if "remark" in fields:
            obj.remark = (fields["remark"] or "").strip() or None
        obj.update_by = user_id
        obj.update_time = datetime.now()
        db.add(obj)
        if commit:
            db.commit()
            db.refresh(obj)
        else:
            db.flush()
        return obj
    obj = BizProtocolPartyCustomer(
        id=generate_uuid(),
        collection_protocol_id=collection_protocol_id,
        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=datetime.now(),
        update_by=user_id,
        update_time=datetime.now(),
        del_flag=0,
    )
    for k in _STR_FIELDS:
        if k in fields:
            setattr(obj, k, (fields[k] or "").strip() or None)
    for k in _INT_FIELDS:
        if k in fields:
            setattr(obj, k, fields[k])
    for k in _DEC_FIELDS:
        if k in fields:
            setattr(obj, k, _optional_decimal(fields[k]))
    for k in _DATE_FIELDS:
        if k in fields:
            setattr(obj, k, optional_date(fields[k]))
    for k in _DT_FIELDS:
        if k in fields:
            setattr(obj, k, fields[k])
    db.add(obj)
    if commit:
        db.commit()
        db.refresh(obj)
    else:
        db.flush()
    return obj


def get_collection_info(db: Session, collection_protocol_id: str, tenant_no: str) -> Optional[BizProtocolCollectionInfo]:
    return (
        db.query(BizProtocolCollectionInfo)
        .filter(
            and_(
                BizProtocolCollectionInfo.collection_protocol_id == collection_protocol_id,
                BizProtocolCollectionInfo.tenant_no == tenant_no,
                BizProtocolCollectionInfo.del_flag == 0,
            )
        )
        .first()
    )


def upsert_collection_info(
    db: Session,
    collection_protocol_id: str,
    tenant_no: str,
    user_id: Optional[str],
    commit: bool = True,
    create_dept_id: str = "",
    **fields,
) -> BizProtocolCollectionInfo:
    fields = _normalize_fields(fields)
    obj = get_collection_info(db, collection_protocol_id, tenant_no)
    _STR_FIELDS = [
        "region_id",
        "hospital_id",
        "medical_staff",
        "sales_user_id",
        "region_manager_user_id",
    ]
    _DT_FIELDS = ["sample_collect_time", "maternal_blood_collect_time"]
    if obj:
        for k in _STR_FIELDS:
            if k in fields:
                setattr(obj, k, (fields[k] or "").strip() or None)
        for k in _DT_FIELDS:
            if k in fields:
                setattr(obj, k, fields[k])
        if "remark" in fields:
            obj.remark = (fields["remark"] or "").strip() or None
        obj.update_by = user_id
        obj.update_time = datetime.now()
        db.add(obj)
        if commit:
            db.commit()
            db.refresh(obj)
        else:
            db.flush()
        return obj
    obj = BizProtocolCollectionInfo(
        id=generate_uuid(),
        collection_protocol_id=collection_protocol_id,
        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=datetime.now(),
        update_by=user_id,
        update_time=datetime.now(),
        del_flag=0,
    )
    for k in _STR_FIELDS:
        if k in fields:
            setattr(obj, k, (fields[k] or "").strip() or None)
    for k in _DT_FIELDS:
        if k in fields:
            setattr(obj, k, fields[k])
    db.add(obj)
    if commit:
        db.commit()
        db.refresh(obj)
    else:
        db.flush()
    return obj


def list_samples(db: Session, collection_protocol_id: str, tenant_no: str) -> List[BizProtocolSample]:
    return (
        db.query(BizProtocolSample)
        .filter(
            and_(
                BizProtocolSample.collection_protocol_id == collection_protocol_id,
                BizProtocolSample.tenant_no == tenant_no,
                BizProtocolSample.del_flag == 0,
            )
        )
        .order_by(BizProtocolSample.order_num.asc(), BizProtocolSample.create_time.asc())
        .all()
    )


def _migrate_sample_child_refs_on_replace(
    db: Session,
    *,
    tenant_no: str,
    old_id_to_code: Dict[str, str],
    new_code_to_id: Dict[str, str],
    user_id: Optional[str],
) -> None:
    """协议样本行重建后，将接收/质检工单的 protocol_sample_id 迁移到新样本 ID。"""
    if not old_id_to_code or not new_code_to_id:
        return
    old_ids = list(old_id_to_code.keys())
    now = datetime.now()
    for model in (BizSampleReception, BizQcWorkOrder):
        rows = (
            db.query(model)
            .filter(
                model.protocol_sample_id.in_(old_ids),
                model.tenant_no == tenant_no,
                model.del_flag == 0,
            )
            .all()
        )
        for row in rows:
            old_sid = (row.protocol_sample_id or "").strip()
            code = old_id_to_code.get(old_sid) or (row.sample_code or "").strip()
            new_sid = new_code_to_id.get(code) if code else None
            if not new_sid:
                continue
            row.protocol_sample_id = new_sid
            row.update_by = user_id
            row.update_time = now
            db.add(row)


def save_samples(
    db: Session,
    collection_protocol_id: str,
    tenant_no: str,
    user_id: Optional[str],
    samples: list,
    create_dept_id: str = "",
    commit: bool = True,
) -> List[BizProtocolSample]:
    existing = list_samples(db, collection_protocol_id, tenant_no)
    old_id_to_code = {
        s.id: (s.sample_code or "").strip()
        for s in existing
        if (s.sample_code or "").strip()
    }
    for s in existing:
        s.del_flag = 1
        s.update_by = user_id
        s.update_time = datetime.now()
        db.add(s)
    if existing:
        db.flush()
    validate_sample_codes_not_used(
        db,
        tenant_no=tenant_no,
        samples=samples,
        exclude_protocol_id=collection_protocol_id,
    )
    _STR_FIELDS = [
        "sample_type_id", "sample_code_id", "sample_code", "unit_id",
        "sample_status", "storage_in_date", "detect_start_date",
        "void_date", "storage_expire_date",
    ]
    _DEC_FIELDS = ["collect_amount", "price", "original_price"]
    _INT_FIELDS = ["storage_years"]
    result = []
    for idx, raw in enumerate(samples):
        raw = _normalize_fields(raw)
        validate_sample_special_raw(raw, index=idx)
        obj = BizProtocolSample(
            id=generate_uuid(),
            collection_protocol_id=collection_protocol_id,
            tenant_no=tenant_no,
            order_num=idx,
            remark=(raw.get("remark") or "").strip() or None,
            create_by=user_id,
            create_dept_id=create_dept_id,
            create_time=datetime.now(),
            update_by=user_id,
            update_time=datetime.now(),
            del_flag=0,
        )
        for k in _STR_FIELDS:
            if k == "sample_status":
                continue
            if k in raw:
                setattr(obj, k, (raw[k] or "").strip() or None)
        # 样本状态：显式支持 null/空字符串，避免 ORM/库默认值 ycj 回填
        if "sample_status" in raw:
            status_val = raw.get("sample_status")
            obj.sample_status = (
                str(status_val).strip() if status_val is not None and str(status_val).strip() else None
            )
        for k in _DEC_FIELDS:
            if k in raw and raw[k] is not None:
                setattr(obj, k, Decimal(str(raw[k])))
        for k in _INT_FIELDS:
            if k in raw:
                setattr(obj, k, raw[k])
        apply_sample_special_fields(obj, raw)
        db.add(obj)
        result.append(obj)
    new_code_to_id = {
        (o.sample_code or "").strip(): o.id
        for o in result
        if (o.sample_code or "").strip()
    }
    _migrate_sample_child_refs_on_replace(
        db,
        tenant_no=tenant_no,
        old_id_to_code=old_id_to_code,
        new_code_to_id=new_code_to_id,
        user_id=user_id,
    )
    db.flush()
    for o in result:
        db.refresh(o)
    # 运费批次 + 预计到达（旧运输明细仅保留 eta 等仍在用的字段）
    from app.crud import sample_freight as sample_freight_svc

    sample_freight_svc.apply_freight_on_protocol_save(
        db,
        tenant_no=tenant_no,
        user_id=user_id,
        sample_source=sample_freight_svc.SAMPLE_SOURCE_CELL,
        protocol_id=collection_protocol_id,
        saved_samples=result,
        input_samples=samples,
        replaced_sample_ids=[sample.id for sample in existing],
    )
    save_sample_transports(
        db,
        collection_protocol_id=collection_protocol_id,
        tenant_no=tenant_no,
        user_id=user_id,
        saved_samples=result,
        input_samples=samples,
        commit=False,
    )
    if commit:
        db.commit()
        for o in result:
            db.refresh(o)
    else:
        db.flush()
    return result


def list_sample_transports(
    db: Session,
    collection_protocol_id: str,
    tenant_no: str,
) -> List[BizProtocolSampleTransport]:
    return (
        db.query(BizProtocolSampleTransport)
        .filter(
            and_(
                BizProtocolSampleTransport.collection_protocol_id == collection_protocol_id,
                BizProtocolSampleTransport.tenant_no == tenant_no,
                BizProtocolSampleTransport.del_flag == 0,
            )
        )
        .order_by(BizProtocolSampleTransport.order_num.asc(), BizProtocolSampleTransport.create_time.asc())
        .all()
    )


def _soft_delete_transports_for_protocol(
    db: Session,
    collection_protocol_id: str,
    tenant_no: str,
    user_id: Optional[str],
) -> None:
    rows = list_sample_transports(db, collection_protocol_id, tenant_no)
    for r in rows:
        r.del_flag = 1
        r.update_by = user_id
        r.update_time = datetime.now()
        db.add(r)
    if rows:
        db.flush()


def save_sample_transports(
    db: Session,
    collection_protocol_id: str,
    tenant_no: str,
    user_id: Optional[str],
    saved_samples: List[BizProtocolSample],
    input_samples: list,
    commit: bool = True,
) -> None:
    """写回样本运输明细。运费批次改造后仍保留预计到达等字段（可仅有 eta_time）。"""
    _soft_delete_transports_for_protocol(db, collection_protocol_id, tenant_no, user_id)
    for idx, sample_obj in enumerate(saved_samples):
        if idx >= len(input_samples):
            break
        raw = _normalize_fields(input_samples[idx] if isinstance(input_samples[idx], dict) else {})
        transport_raw = raw.get("transport")
        if not isinstance(transport_raw, dict):
            transport_raw = {}
        tnorm = _normalize_fields(transport_raw)
        # 移动端可能把 etaTime 放在样本顶层
        if not tnorm.get("eta_time") and raw.get("eta_time"):
            tnorm["eta_time"] = raw.get("eta_time")

        transport_type = (tnorm.get("transport_type") or "").strip()
        eta_time = tnorm.get("eta_time")
        if isinstance(eta_time, str):
            eta_time = parse_datetime(eta_time)
        if not transport_type and not eta_time:
            continue

        express_fee = tnorm.get("express_fee")
        if express_fee is not None and express_fee != "":
            express_fee = Decimal(str(express_fee))
        else:
            express_fee = None

        obj = BizProtocolSampleTransport(
            id=generate_uuid(),
            collection_protocol_id=collection_protocol_id,
            sample_id=sample_obj.id,
            sample_code=sample_obj.sample_code or None,
            transport_type=transport_type,
            express_company_id=(tnorm.get("express_company_id") or "").strip() or None,
            waybill_no=(tnorm.get("waybill_no") or "").strip() or None,
            dest_site_id=(tnorm.get("dest_site_id") or "").strip() or None,
            express_fee=express_fee,
            eta_time=eta_time,
            carrier_name=(tnorm.get("carrier_name") or "").strip() or None,
            carrier_phone=(tnorm.get("carrier_phone") or "").strip() or None,
            tenant_no=tenant_no,
            order_num=idx,
            remark=(tnorm.get("remark") or "").strip() or None,
            create_by=user_id,
            create_dept_id="",
            create_time=datetime.now(),
            update_by=user_id,
            update_time=datetime.now(),
            del_flag=0,
        )
        db.add(obj)
    if commit:
        db.commit()
    else:
        db.flush()


def get_storage_cert(db: Session, collection_protocol_id: str, tenant_no: str) -> Optional[BizProtocolStorageCert]:
    return (
        db.query(BizProtocolStorageCert)
        .filter(
            and_(
                BizProtocolStorageCert.collection_protocol_id == collection_protocol_id,
                BizProtocolStorageCert.tenant_no == tenant_no,
                BizProtocolStorageCert.del_flag == 0,
            )
        )
        .first()
    )


def _add_months_yyyymmdd(date_str: str, months: int) -> str:
    raw = (date_str or "").strip().replace("-", "")
    if len(raw) != 8:
        return ""
    y = int(raw[0:4])
    m = int(raw[4:6])
    d = int(raw[6:8])
    month_index = m - 1 + months
    ny = y + month_index // 12
    nm = month_index % 12 + 1
    last_day = calendar.monthrange(ny, nm)[1]
    nd = min(d, last_day)
    return f"{ny:04d}{nm:02d}{nd:02d}"


def _calc_installment_due_amounts(
    contract_amount: Optional[Decimal],
    count: int,
) -> List[Decimal]:
    if contract_amount is None or count < 1:
        return []
    total_cents = int((Decimal(str(contract_amount)) * Decimal("100")).to_integral_value())
    base_cents = total_cents // count
    amounts: List[Decimal] = []
    sum_cents = 0
    for i in range(count):
        if i == count - 1:
            cents = total_cents - sum_cents
        else:
            cents = base_cents
            sum_cents += base_cents
        amounts.append(Decimal(cents) / Decimal(100))
    return amounts


def align_fq_payment_details_payload(
    pay: BizProtocolPayment,
    details: Optional[list],
) -> list:
    if pay.pay_mode not in ("fq", "djfq") or not pay.installment_count or int(pay.installment_count) < 1:
        return list(details or [])
    count = int(pay.installment_count)
    rows = [_normalize_fields(dict(d)) for d in (details or [])]
    explicit_nos = [
        int(row["installment_no"])
        for row in rows
        if row.get("installment_no") not in (None, "")
    ]
    if (
        any(number < 1 or number > count for number in explicit_nos)
        or len(explicit_nos) != len(set(explicit_nos))
    ):
        return rows
    if len(rows) >= count:
        needed = set(range(1, count + 1))
        have = {
            int(r["installment_no"])
            for r in rows
            if r.get("installment_no") is not None
        }
        if needed - have == set():
            return rows[:count]

    by_no: Dict[int, Dict[str, Any]] = {}
    unnumbered: List[Dict[str, Any]] = []
    for r in rows:
        no = r.get("installment_no")
        if no is not None:
            by_no[int(no)] = r
        else:
            unnumbered.append(r)

    ratios = parse_installment_ratios(pay.installment_ratios)
    amounts = resolve_installment_due_amounts(pay.contract_amount, ratios, count)
    start = (pay.installment_start_date or "").strip()
    interval = pay.installment_interval_months
    interval_val = Decimal(str(interval)) if interval is not None else Decimal("0")
    advance_days = (
        30
        if pay.installment_advance_days is None
        else int(pay.installment_advance_days)
    )
    result: List[Dict[str, Any]] = []
    for i in range(count):
        no = i + 1
        if no in by_no:
            result.append(by_no[no])
            continue
        if i < len(unnumbered):
            raw = dict(unnumbered[i])
            raw["installment_no"] = no
            if not raw.get("installment_due_date") and start and interval_val:
                raw["installment_due_date"] = calc_fq_installment_due_date(
                    start, i, interval_val, advance_days
                )
            result.append(raw)
            continue
        due_date = (
            calc_fq_installment_due_date(start, i, interval_val, advance_days)
            if start and interval_val
            else ""
        )
        due = amounts[i] if i < len(amounts) else None
        result.append(
            {
                "installment_no": no,
                "installment_due_date": due_date or None,
                "due_amount": format(due, "f") if due is not None else None,
                "pay_status": "dfj",
                "transfer_type": "",
                "transfer_amount": None,
                "attachments": [],
            }
        )
    return result


def list_payment_details(
    db: Session,
    payment_id: str,
    tenant_no: str,
) -> List[BizProtocolPaymentDetail]:
    return (
        db.query(BizProtocolPaymentDetail)
        .filter(
            and_(
                BizProtocolPaymentDetail.payment_id == payment_id,
                BizProtocolPaymentDetail.tenant_no == tenant_no,
                BizProtocolPaymentDetail.del_flag == 0,
            )
        )
        .order_by(
            BizProtocolPaymentDetail.installment_no.asc(),
            BizProtocolPaymentDetail.order_num.asc(),
            BizProtocolPaymentDetail.create_time.asc(),
        )
        .all()
    )


def upsert_payment(
    db: Session,
    collection_protocol_id: str,
    tenant_no: str,
    user_id: Optional[str],
    commit: bool = True,
    **fields,
) -> BizProtocolPayment:
    fields = _normalize_fields(fields)
    if "pay_mode" in fields and fields["pay_mode"] not in {
        "dj", "fq", "djfq", "fgd"
    }:
        raise ValueError("不支持的缴费模式")
    if commit:
        obj = get_payment_by_protocol_id(db, collection_protocol_id, tenant_no)
    else:
        obj = (
            db.query(BizProtocolPayment)
            .filter(
                and_(
                    BizProtocolPayment.collection_protocol_id
                    == collection_protocol_id,
                    BizProtocolPayment.tenant_no == tenant_no,
                    BizProtocolPayment.del_flag == 0,
                )
            )
            .populate_existing()
            .with_for_update()
            .first()
        )
    _STR_FIELDS = ["pay_mode"]
    _STR_DATE_FIELDS = ["lump_sum_deadline", "installment_start_date"]
    _DEC_FIELDS = ["contract_amount", "installment_interval_months"]
    _INT_FIELDS = ["installment_count", "installment_advance_days"]

    def _apply(target: BizProtocolPayment) -> None:
        for k in _STR_FIELDS:
            if k in fields:
                setattr(target, k, (fields[k] or "").strip() or None)
        for k in _STR_DATE_FIELDS:
            if k in fields:
                setattr(target, k, _validate_yyyymmdd_date(fields[k], "缴费方案"))
        for k in _DEC_FIELDS:
            if k in fields and fields[k] is not None:
                setattr(target, k, Decimal(str(fields[k])))
        for k in _INT_FIELDS:
            if k in fields:
                setattr(target, k, fields[k])
        if "installment_ratios" in fields:
            val = fields["installment_ratios"]
            if val is None:
                target.installment_ratios = None
            elif isinstance(val, str):
                target.installment_ratios = val.strip() or None
            else:
                target.installment_ratios = json.dumps(list(val), ensure_ascii=False)
        if "remark" in fields:
            target.remark = (fields["remark"] or "").strip() or None

    if obj:
        _apply(obj)
        obj.update_by = user_id
        obj.update_time = datetime.now()
        db.add(obj)
        if commit:
            db.commit()
            db.refresh(obj)
        else:
            db.flush()
        return obj

    pay_mode = (fields.get("pay_mode") or "").strip() or "dj"
    obj = BizProtocolPayment(
        id=generate_uuid(),
        collection_protocol_id=collection_protocol_id,
        pay_mode=pay_mode,
        tenant_no=tenant_no,
        order_num=0,
        remark=(fields.get("remark") or "").strip() or None,
        create_by=user_id,
        create_dept_id="",
        create_time=datetime.now(),
        update_by=user_id,
        update_time=datetime.now(),
        del_flag=0,
    )
    _apply(obj)
    db.add(obj)
    if commit:
        db.commit()
        db.refresh(obj)
    else:
        db.flush()
    return obj


_PAYMENT_PLAN_FIELDS = frozenset(
    {
        "pay_mode",
        "contract_amount",
        "lump_sum_deadline",
        "installment_count",
        "installment_interval_months",
        "installment_start_date",
        "installment_advance_days",
        "installment_ratios",
    }
)


def _payment_plan_changed_fields(
    current: BizProtocolPayment,
    incoming: Dict[str, Any],
) -> set:
    fields = _normalize_fields(dict(incoming))
    changed = set()
    for key in _PAYMENT_PLAN_FIELDS.intersection(fields):
        value = fields[key]
        existing = getattr(current, key, None)
        if key in {"contract_amount", "installment_interval_months"}:
            normalized = Decimal(str(value)) if value is not None else None
            if existing != normalized:
                changed.add(key)
            continue
        if key == "installment_ratios":
            normalized = (
                [Decimal(str(item)) for item in value]
                if value is not None and not isinstance(value, str)
                else parse_installment_ratios(value)
            )
            if parse_installment_ratios(existing) != normalized:
                changed.add(key)
            continue
        if key in {"pay_mode", "lump_sum_deadline", "installment_start_date"}:
            if str(existing or "").strip() != str(value or "").strip():
                changed.add(key)
            continue
        if existing != value:
            changed.add(key)
    return changed


def _assert_locked_payment_plan_transition_allowed(
    current: BizProtocolPayment,
    incoming: Dict[str, Any],
    locked_details: List[BizProtocolPaymentDetail],
    draft_details: List[BizProtocolPaymentDetail],
    changed_fields: set,
) -> None:
    if not changed_fields or not locked_details:
        return
    statuses = {str(item.pay_status or "") for item in locked_details}
    if statuses != {"yjf"}:
        raise ValueError("存在确认中或异常锁定明细，无法变更缴费计划")

    fields = _normalize_fields(dict(incoming))
    from_mode = str(current.pay_mode or "").strip()
    to_mode = str(fields.get("pay_mode", from_mode) or "").strip()
    installment_modes = {"fq", "djfq"}
    if "pay_mode" in changed_fields:
        if from_mode == "dj" and to_mode in installment_modes:
            raise ValueError("趸交已有缴费记录，无法改为分期")
        if from_mode in installment_modes and to_mode in installment_modes:
            raise ValueError("分期已有缴费记录，无法切换分期类型")
        if from_mode in installment_modes and to_mode == "dj":
            if not draft_details:
                raise ValueError("分期已全部缴清，无需改为趸交")
            contract_amount = Decimal(
                str(fields.get("contract_amount", current.contract_amount) or 0)
            )
            paid_due = sum(
                (item.due_amount or Decimal("0")) for item in locked_details
            )
            if contract_amount <= paid_due:
                raise ValueError("已缴清或合同金额不足，无法改为趸交")


def _compare_money(val: Any) -> Optional[str]:
    if val is None or val == "":
        return None
    try:
        return f"{Decimal(str(val)).quantize(Decimal('0.01'))}"
    except Exception:
        return str(val).strip() or None


def _compare_datetime(val: Any) -> Optional[str]:
    if val is None or val == "":
        return None
    if isinstance(val, datetime):
        return _fmt_dt(val)
    if isinstance(val, str):
        dt = parse_datetime(val)
        return _fmt_dt(dt) if dt else val.strip()
    return str(val)


def _payment_detail_compare_row(raw: Dict[str, Any]) -> Dict[str, Any]:
    """Normalize one payment detail row for equality compare."""
    raw = _normalize_fields(dict(raw))
    transfer_att = _pop_attachment_urls(
        raw, "attachments", "transfer_attachments", "transferAttachments"
    )
    finance_att = _pop_attachment_urls(
        raw,
        "finance_confirm_attachments",
        "financeConfirmAttachments",
        "finance_confirm_attachment_files",
        "financeConfirmAttachmentFiles",
    )
    refund_att = _pop_attachment_urls(
        raw,
        "refund_attachments",
        "refundAttachments",
        "refund_attachment_files",
        "refundAttachmentFiles",
    )
    refund_finance_att = _pop_attachment_urls(
        raw,
        "refund_finance_confirm_attachments",
        "refundFinanceConfirmAttachments",
        "refund_finance_confirm_attachment_files",
        "refundFinanceConfirmAttachmentFiles",
    )
    transfer_time = raw.get("transfer_time")
    if isinstance(transfer_time, str):
        transfer_time = parse_datetime(transfer_time)

    pay_status = (raw.get("pay_status") or "").strip() or "dfj"

    return {
        "installment_no": raw.get("installment_no"),
        "installment_due_date": _normalize_yyyymmdd_date(raw.get("installment_due_date")),
        "due_amount": _compare_money(raw.get("due_amount")),
        "transfer_type": (raw.get("transfer_type") or "").strip() or None,
        "transfer_amount": _compare_money(raw.get("transfer_amount")),
        "transfer_time": _compare_datetime(transfer_time),
        "pay_status": pay_status,
        "remark": (raw.get("remark") or "").strip() or None,
        "finance_remark": (raw.get("finance_remark") or "").strip() or None,
        "refund_remark": (raw.get("refund_remark") or "").strip() or None,
        "refund_finance_remark": (raw.get("refund_finance_remark") or "").strip() or None,
        "refund_amount": _compare_money(raw.get("refund_amount")),
        "transfer_attachments": transfer_att or None,
        "finance_confirm_attachments": finance_att or None,
        "refund_attachments": refund_att or None,
        "refund_finance_confirm_attachments": refund_finance_att or None,
    }


def _payment_detail_compare_from_model(d: BizProtocolPaymentDetail) -> Dict[str, Any]:
    pay_status = (d.pay_status or "").strip() or "dfj"
    return {
        "installment_no": d.installment_no,
        "installment_due_date": _normalize_yyyymmdd_date(d.installment_due_date),
        "due_amount": _compare_money(d.due_amount),
        "transfer_type": (d.transfer_type or "").strip() or None,
        "transfer_amount": _compare_money(d.transfer_amount),
        "transfer_time": _compare_datetime(d.transfer_time),
        "pay_status": pay_status,
        "remark": (d.remark or "").strip() or None,
        "finance_remark": (d.finance_remark or "").strip() or None,
        "refund_remark": (d.refund_remark or "").strip() or None,
        "refund_finance_remark": (d.refund_finance_remark or "").strip() or None,
        "refund_amount": _compare_money(d.refund_amount),
        "transfer_attachments": (d.transfer_attachments or "").strip() or None,
        "finance_confirm_attachments": (d.finance_confirm_attachments or "").strip() or None,
        "refund_attachments": (d.refund_attachments or "").strip() or None,
        "refund_finance_confirm_attachments": (
            (d.refund_finance_confirm_attachments or "").strip() or None
        ),
    }


def _payment_detail_sort_key(row: Dict[str, Any]) -> tuple:
    no = row.get("installment_no")
    if no is not None:
        return (0, int(no))
    return (1, 0)


def _payment_details_payload_unchanged(
    existing: List[BizProtocolPaymentDetail],
    incoming: list,
) -> bool:
    incoming_rows = [_payment_detail_compare_row(dict(d)) for d in (incoming or [])]
    existing_rows = [_payment_detail_compare_from_model(d) for d in existing]
    if len(existing_rows) != len(incoming_rows):
        return False
    existing_sorted = sorted(existing_rows, key=_payment_detail_sort_key)
    incoming_sorted = sorted(incoming_rows, key=_payment_detail_sort_key)
    return existing_sorted == incoming_sorted


def _is_draft_payment_detail(detail: BizProtocolPaymentDetail) -> bool:
    return detail.pay_status == "dfj"


def save_payment_details(
    db: Session,
    payment_id: str,
    tenant_no: str,
    user_id: Optional[str],
    details: list,
    commit: bool = True,
    enforce_locked_plan_consistency: bool = False,
) -> List[BizProtocolPaymentDetail]:
    pay = (
        db.query(BizProtocolPayment)
        .filter(
            and_(
                BizProtocolPayment.id == payment_id,
                BizProtocolPayment.tenant_no == tenant_no,
                BizProtocolPayment.del_flag == 0,
            )
        )
        .populate_existing()
        .with_for_update()
        .first()
    )
    existing = (
        db.query(BizProtocolPaymentDetail)
        .filter(
            and_(
                BizProtocolPaymentDetail.payment_id == payment_id,
                BizProtocolPaymentDetail.tenant_no == tenant_no,
                BizProtocolPaymentDetail.del_flag == 0,
            )
        )
        .order_by(
            BizProtocolPaymentDetail.installment_no.asc(),
            BizProtocolPaymentDetail.order_num.asc(),
            BizProtocolPaymentDetail.create_time.asc(),
        )
        .populate_existing()
        .with_for_update()
        .all()
    )
    existing_by_id = {row.id: row for row in existing}
    normalized_details = [_normalize_fields(dict(detail)) for detail in (details or [])]
    for raw in normalized_details:
        if "installment_due_date" in raw:
            raw["installment_due_date"] = _validate_yyyymmdd_date(
                raw["installment_due_date"], "缴费明细"
            )
    submitted_ids: set = set()
    for raw in normalized_details:
        detail_id = str(raw.get("id") or "").strip()
        if not detail_id:
            continue
        if detail_id not in existing_by_id:
            raise ValueError("缴费明细不存在或不属于当前缴费方案")
        if detail_id in submitted_ids:
            raise ValueError("缴费明细ID重复")
        submitted_ids.add(detail_id)

    effective_details: List[Dict[str, Any]] = []
    for raw in normalized_details:
        detail_id = str(raw.get("id") or "").strip()
        current = existing_by_id.get(detail_id)
        if current is not None and not _is_draft_payment_detail(current):
            authoritative = _payment_detail_compare_from_model(current)
            authoritative["id"] = current.id
            authoritative["pay_status"] = current.pay_status
            effective_details.append(authoritative)
        else:
            effective_details.append(dict(raw))
    for current in existing:
        if current.id in submitted_ids or _is_draft_payment_detail(current):
            continue
        authoritative = _payment_detail_compare_from_model(current)
        authoritative["id"] = current.id
        authoritative["pay_status"] = current.pay_status
        effective_details.append(authoritative)

    if pay and (pay.pay_mode or "") in ("fq", "djfq", "fgd"):
        installment_nos = [
            int(raw["installment_no"])
            for raw in effective_details
            if raw.get("installment_no") not in (None, "")
        ]
        if len(installment_nos) != len(set(installment_nos)):
            raise ValueError("分期期号不能重复")

    if pay:
        installment_ratios = parse_installment_ratios(pay.installment_ratios)
        validate_payment_details_for_save(
            pay.pay_mode or "",
            pay.contract_amount,
            pay.installment_count,
            effective_details,
            installment_ratios=installment_ratios or None,
            installment_advance_days=pay.installment_advance_days,
            locked_rows_are_read_only=True,
        )
        installment_plan = (pay.pay_mode or "") in ("fq", "djfq")
        if enforce_locked_plan_consistency or installment_plan:
            count = 0
            ratios: List[Decimal] = []
            if installment_plan:
                count = int(pay.installment_count or 0)
                ratios = parse_installment_ratios(pay.installment_ratios)
                effective_nos = sorted(
                    int(item.get("installment_no") or 0)
                    for item in effective_details
                )
                if effective_nos != list(range(1, count + 1)):
                    raise ValueError("分期明细期号必须完整且为 1 至分期期数")
            locked_due = sum(
                (item.due_amount or Decimal("0"))
                for item in existing
                if not _is_draft_payment_detail(item)
            )
            contract_amount = pay.contract_amount
            if contract_amount is None or contract_amount < locked_due:
                raise ValueError("合同金额不能小于已缴或锁定明细金额")
            incoming_due = []
            for item in effective_details:
                value = item.get("due_amount", item.get("dueAmount"))
                if value in (None, ""):
                    raise ValueError("变更缴费计划时明细应缴金额不能为空")
                incoming_due.append(Decimal(str(value)))
            if sum(incoming_due, Decimal("0")) != contract_amount:
                raise ValueError("变更后的缴费计划明细金额须与合同金额一致")
            if (pay.pay_mode or "") == "dj":
                locked_ids = {
                    item.id for item in existing
                    if not _is_draft_payment_detail(item)
                }
                editable_plan = [
                    item for item in effective_details
                    if str(item.get("id") or "") not in locked_ids
                ]
                if len(editable_plan) != 1:
                    raise ValueError("切换趸交后须仅保留一条待缴计划")
                draft = editable_plan[0]
                if draft.get("installment_no") not in (None, ""):
                    raise ValueError("切换趸交后待缴计划不能保留分期期号")
                expected_amount = contract_amount - locked_due
                if Decimal(str(draft.get("due_amount"))) != expected_amount:
                    raise ValueError("切换趸交后待缴金额应为剩余合同金额")
                expected_date = _normalize_yyyymmdd_date(pay.lump_sum_deadline)
                actual_date = _normalize_yyyymmdd_date(
                    draft.get("installment_due_date")
                )
                if actual_date != expected_date:
                    raise ValueError("切换趸交后待缴应缴时间须与趸交截止时间一致")
            if installment_plan:
                locked_ids = {
                    item.id for item in existing
                    if not _is_draft_payment_detail(item)
                }
                editable_installments = []
                for item in effective_details:
                    if str(item.get("id") or "") in locked_ids:
                        continue
                    installment_no = int(item.get("installment_no") or 0)
                    if installment_no < 1 or installment_no > count:
                        raise ValueError("变更缴费计划时待缴明细期号无效")
                    editable_installments.append((item, installment_no))
                editable_installments.sort(key=lambda entry: entry[1])
                planned_amounts = resolve_locked_remaining_due_amounts(
                    contract_amount,
                    locked_due,
                    ratios,
                    [number - 1 for _, number in editable_installments],
                )
                for (item, installment_no), expected_amount in zip(
                    editable_installments, planned_amounts
                ):
                    actual_amount = Decimal(str(item.get("due_amount")))
                    if actual_amount != expected_amount:
                        raise ValueError("变更缴费计划后待缴明细应缴金额与新方案不一致")
                    expected_date = calc_fq_installment_due_date(
                        pay.installment_start_date or "",
                        installment_no - 1,
                        pay.installment_interval_months or Decimal("0"),
                        (
                            30
                            if pay.installment_advance_days is None
                            else int(pay.installment_advance_days)
                        ),
                    )
                    actual_date = _normalize_yyyymmdd_date(
                        item.get("installment_due_date")
                    )
                    if expected_date and actual_date != expected_date:
                        raise ValueError("变更缴费计划后待缴明细应缴时间与新方案不一致")

    if _payment_details_payload_unchanged(existing, effective_details):
        return existing

    retained_locked: List[BizProtocolPaymentDetail] = []
    for d in existing:
        if d.id not in submitted_ids:
            if _is_draft_payment_detail(d):
                d.del_flag = 1
                d.update_by = user_id
                d.update_time = datetime.now()
                db.add(d)
            else:
                retained_locked.append(d)

    _STR_FIELDS = ["transfer_type"]
    _STR_DATE_FIELDS = ["installment_due_date"]
    _DEC_FIELDS = ["due_amount", "transfer_amount"]
    _INT_FIELDS = ["installment_no"]
    result: List[BizProtocolPaymentDetail] = []
    for idx, raw in enumerate(normalized_details):
        detail_id = str(raw.get("id") or "").strip()
        obj = existing_by_id.get(detail_id)
        if obj is not None and not _is_draft_payment_detail(obj):
            result.append(obj)
            continue

        has_transfer_attachments = any(
            key in raw for key in ("attachments", "transfer_attachments")
        )
        transfer_att = _pop_attachment_urls(
            raw, "attachments", "transfer_attachments", "transferAttachments"
        )

        transfer_time = raw.get("transfer_time")
        if isinstance(transfer_time, str):
            transfer_time = parse_datetime(transfer_time)

        installment_no = raw.get("installment_no")
        if (
            installment_no is None
            and raw.get("installment_due_date")
            and pay is not None
            and (pay.pay_mode or "") in ("fq", "djfq", "fgd")
        ):
            installment_no = idx + 1

        if obj is None:
            obj = BizProtocolPaymentDetail(
                id=generate_uuid(),
                payment_id=payment_id,
                tenant_no=tenant_no,
                create_by=user_id,
                create_dept_id="",
                create_time=datetime.now(),
                del_flag=0,
                pay_status="dfj",
            )
        obj.order_num = idx
        obj.remark = (raw.get("remark") or "").strip() or None
        obj.update_by = user_id
        obj.update_time = datetime.now()
        obj.transfer_time = transfer_time
        obj.installment_no = installment_no
        if has_transfer_attachments:
            obj.transfer_attachments = transfer_att
        for k in _STR_FIELDS:
            if k in raw:
                setattr(obj, k, (raw[k] or "").strip() or None)
        for k in _STR_DATE_FIELDS:
            if k in raw:
                setattr(obj, k, _validate_yyyymmdd_date(raw[k], "缴费明细"))
        for k in _DEC_FIELDS:
            if k in raw and raw[k] is not None:
                setattr(obj, k, Decimal(str(raw[k])))
        for k in _INT_FIELDS:
            if k in raw and raw[k] is not None:
                setattr(obj, k, raw[k])
        db.add(obj)
        result.append(obj)
    result.extend(retained_locked)
    if commit:
        db.commit()
        for o in result:
            db.refresh(o)
    else:
        db.flush()
    return result


def save_payment_with_details(
    db: Session,
    collection_protocol_id: str,
    tenant_no: str,
    user_id: Optional[str],
    *,
    payment: Optional[dict],
    payment_details: Optional[list],
    commit: bool = True,
) -> Optional[BizProtocolPayment]:
    """原子保存 v1 父缴费方案与明细，供 Web/移动端聚合表单复用。"""
    try:
        current_pay = (
            db.query(BizProtocolPayment)
            .filter(
                BizProtocolPayment.collection_protocol_id
                == collection_protocol_id,
                BizProtocolPayment.tenant_no == tenant_no,
                BizProtocolPayment.del_flag == 0,
            )
            .populate_existing()
            .with_for_update()
            .first()
        )
        current_details = (
            db.query(BizProtocolPaymentDetail)
            .filter(
                BizProtocolPaymentDetail.payment_id == current_pay.id,
                BizProtocolPaymentDetail.tenant_no == tenant_no,
                BizProtocolPaymentDetail.del_flag == 0,
            )
            .populate_existing()
            .with_for_update()
            .all()
            if current_pay is not None
            else []
        )
        if payment_details is not None and current_pay is None and not payment:
            raise ValueError("提交缴费明细前必须先提交缴费主表方案")
        changed_plan_fields = (
            _payment_plan_changed_fields(current_pay, payment)
            if current_pay is not None and payment
            else set()
        )
        plan_changed = bool(changed_plan_fields)
        has_locked_details = any(
            not _is_draft_payment_detail(item) for item in current_details
        )
        if current_pay is not None and payment and has_locked_details:
            _assert_locked_payment_plan_transition_allowed(
                current_pay,
                payment,
                [item for item in current_details if not _is_draft_payment_detail(item)],
                [item for item in current_details if _is_draft_payment_detail(item)],
                changed_plan_fields,
            )
        if plan_changed and has_locked_details and payment_details is None:
            raise ValueError("存在已缴或锁定明细，变更缴费计划须同步提交明细")
        pay: Optional[BizProtocolPayment] = None
        if payment:
            pay = upsert_payment(
                db,
                collection_protocol_id,
                tenant_no,
                user_id,
                commit=False,
                **payment,
            )
        if payment_details is not None:
            if not pay:
                pay = get_payment_by_protocol_id(
                    db, collection_protocol_id, tenant_no
                )
            if pay:
                aligned_details = align_fq_payment_details_payload(
                    pay, payment_details
                )
                save_payment_details(
                    db,
                    pay.id,
                    tenant_no,
                    user_id,
                    aligned_details,
                    commit=False,
                    enforce_locked_plan_consistency=(
                        plan_changed and has_locked_details
                    ),
                )
        if commit:
            db.commit()
            if pay:
                db.refresh(pay)
        else:
            db.flush()
        return pay
    except Exception:
        if commit:
            db.rollback()
        raise


def upsert_storage_cert(
    db: Session,
    collection_protocol_id: str,
    tenant_no: str,
    user_id: Optional[str],
    **fields,
) -> BizProtocolStorageCert:
    fields = _normalize_fields(fields)
    obj = get_storage_cert(db, collection_protocol_id, tenant_no)
    _STR_FIELDS = ["make_status", "cert_image_urls", "make_date", "handler_user_id"]
    if obj:
        for k in _STR_FIELDS:
            if k in fields:
                setattr(obj, k, (fields[k] or "").strip() or None if isinstance(fields[k], str) or fields[k] is None else fields[k])
        if "handle_time" in fields:
            obj.handle_time = fields.get("handle_time")
        if "remark" in fields:
            obj.remark = (fields["remark"] or "").strip() or None
        obj.update_by = user_id
        obj.update_time = datetime.now()
        db.add(obj)
        db.commit()
        db.refresh(obj)
        return obj
    obj = BizProtocolStorageCert(
        id=generate_uuid(),
        collection_protocol_id=collection_protocol_id,
        tenant_no=tenant_no,
        order_num=0,
        remark=(fields.get("remark") or "").strip() or None,
        create_by=user_id,
        create_dept_id="",
        create_time=datetime.now(),
        update_by=user_id,
        update_time=datetime.now(),
        del_flag=0,
    )
    for k in _STR_FIELDS:
        if k in fields:
            val = fields[k]
            setattr(obj, k, (val or "").strip() or None if isinstance(val, str) or val is None else val)
    if "handle_time" in fields:
        obj.handle_time = fields.get("handle_time")
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def upsert_storage_cert_handle(
    db: Session,
    collection_protocol_id: str,
    tenant_no: str,
    user_id: Optional[str],
    *,
    handle_time: Optional[datetime] = None,
    handler_user_id: Optional[str] = None,
    cert_image_urls: Optional[str] = None,
    remark: Optional[str] = None,
) -> BizProtocolStorageCert:
    from app.services.sample_info_progress import apply_handle_payload

    payload = apply_handle_payload(
        handle_time=handle_time,
        handler_user_id=handler_user_id,
        attachments=cert_image_urls or "",
        remark=remark or "",
    )
    return upsert_storage_cert(
        db,
        collection_protocol_id,
        tenant_no,
        user_id,
        handle_time=payload["handle_time"],
        handler_user_id=payload["handler_user_id"],
        make_status=payload["make_or_todo_status"],
        make_date=payload["legacy_date"] or None,
        cert_image_urls=payload["attachments"] or None,
        remark=payload["remark"] or None,
    )


# ---------------------------------------------------------------------------
# 子表：满月待办
# ---------------------------------------------------------------------------


def _fullmoon_todo_empty_dict() -> Dict[str, Any]:
    return {
        "todoStatus": "0",
        "handleDate": "",
        "handleTime": "",
        "handlerUserId": "",
        "handled": False,
        "handleRemark": "",
        "insuranceAttachments": "",
        "insuranceCompanyId": "",
        "insuranceCompanyName": "",
        "insuranceType": "",
        "insurancePolicyNo": "",
        "insuranceEffectiveDate": "",
        "insuranceExpiryDate": "",
        "insuranceSubscribed": "0",
        "insuranceFee": None,
        "fullmoonDate": "",
        "isFullmoonPassed": False,
    }


def _fullmoon_todo_detail_dict(
    ft: Optional[BizProtocolFullmoonTodo],
    *,
    newborn_delivery_time: Optional[datetime] = None,
) -> Dict[str, Any]:
    """协议详情 bundle 用：满月待办字段 + 由分娩日推算的满月日期。"""
    base = _fullmoon_todo_to_dict(ft) if ft else dict(_fullmoon_todo_empty_dict())
    fullmoon_date_str = ""
    is_fullmoon_passed = False
    if newborn_delivery_time:
        fullmoon_date = newborn_delivery_time + timedelta(days=30)
        fullmoon_date_str = fullmoon_date.strftime("%Y-%m-%d")
        is_fullmoon_passed = fullmoon_date <= datetime.now()
    base["fullmoonDate"] = fullmoon_date_str
    base["isFullmoonPassed"] = is_fullmoon_passed
    return base


def _fullmoon_todo_to_dict(ft: BizProtocolFullmoonTodo) -> Dict[str, Any]:
    from app.services.sample_info_progress import insurance_handled

    return {
        "id": ft.id,
        "collectionProtocolId": ft.collection_protocol_id,
        "todoStatus": ft.todo_status or "0",
        "handleDate": (ft.handle_date or "").strip() if ft.handle_date else "",
        "handleTime": _fmt_dt(ft.handle_time) if getattr(ft, "handle_time", None) else "",
        "handlerUserId": getattr(ft, "handler_user_id", None) or "",
        "handled": insurance_handled(ft),
        "handleRemark": ft.handle_remark or "",
        "insuranceAttachments": ft.insurance_attachments or "",
        "insuranceCompanyId": ft.insurance_company_id or "",
        "insuranceCompanyName": ft.insurance_company_name or "",
        "insuranceType": ft.insurance_type or "",
        "insurancePolicyNo": ft.insurance_policy_no or "",
        "insuranceEffectiveDate": (ft.insurance_effective_date or "").strip() if ft.insurance_effective_date else "",
        "insuranceExpiryDate": (ft.insurance_expiry_date or "").strip() if ft.insurance_expiry_date else "",
        "insuranceSubscribed": ft.insurance_subscribed or "0",
        "insuranceFee": _dec(ft.insurance_fee),
        "tenantNo": ft.tenant_no,
        "orderNum": ft.order_num if ft.order_num is not None else 0,
        "remark": ft.remark or "",
        "createTime": _fmt_dt(ft.create_time),
        "updateTime": _fmt_dt(ft.update_time),
    }


def get_fullmoon_todo(
    db: Session,
    collection_protocol_id: str,
    tenant_no: str,
) -> Optional[BizProtocolFullmoonTodo]:
    return (
        db.query(BizProtocolFullmoonTodo)
        .filter(
            and_(
                BizProtocolFullmoonTodo.collection_protocol_id == collection_protocol_id,
                BizProtocolFullmoonTodo.tenant_no == tenant_no,
                BizProtocolFullmoonTodo.del_flag == 0,
            )
        )
        .first()
    )


def upsert_fullmoon_todo(
    db: Session,
    collection_protocol_id: str,
    tenant_no: str,
    user_id: Optional[str],
    **fields,
) -> BizProtocolFullmoonTodo:
    fields = _normalize_fields(fields)
    obj = get_fullmoon_todo(db, collection_protocol_id, tenant_no)
    _STR_FIELDS = [
        "todo_status", "handle_date", "handle_remark", "handler_user_id",
        "insurance_attachments", "insurance_company_id",
        "insurance_company_name", "insurance_type",
        "insurance_policy_no", "insurance_effective_date",
        "insurance_expiry_date", "insurance_subscribed",
    ]
    _DEC_FIELDS = ["insurance_fee"]
    if obj:
        for k in _STR_FIELDS:
            if k in fields:
                setattr(obj, k, (fields[k] or "").strip() or None)
        if "handle_time" in fields:
            obj.handle_time = fields.get("handle_time")
        for k in _DEC_FIELDS:
            if k in fields and fields[k] is not None:
                setattr(obj, k, Decimal(str(fields[k])))
        if "remark" in fields:
            obj.remark = (fields["remark"] or "").strip() or None
        obj.update_by = user_id
        obj.update_time = datetime.now()
        db.add(obj)
        db.commit()
        db.refresh(obj)
        return obj
    obj = BizProtocolFullmoonTodo(
        id=generate_uuid(),
        collection_protocol_id=collection_protocol_id,
        tenant_no=tenant_no,
        order_num=0,
        remark=(fields.get("remark") or "").strip() or None,
        create_by=user_id,
        create_dept_id="",
        create_time=datetime.now(),
        update_by=user_id,
        update_time=datetime.now(),
        del_flag=0,
    )
    for k in _STR_FIELDS:
        if k in fields:
            setattr(obj, k, (fields[k] or "").strip() or None)
    if "handle_time" in fields:
        obj.handle_time = fields.get("handle_time")
    for k in _DEC_FIELDS:
        if k in fields and fields[k] is not None:
            setattr(obj, k, Decimal(str(fields[k])))
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def upsert_fullmoon_todo_handle(
    db: Session,
    collection_protocol_id: str,
    tenant_no: str,
    user_id: Optional[str],
    *,
    handle_time: Optional[datetime] = None,
    handler_user_id: Optional[str] = None,
    insurance_attachments: Optional[str] = None,
    handle_remark: Optional[str] = None,
) -> BizProtocolFullmoonTodo:
    from app.services.sample_info_progress import apply_handle_payload

    payload = apply_handle_payload(
        handle_time=handle_time,
        handler_user_id=handler_user_id,
        attachments=insurance_attachments or "",
        remark=handle_remark or "",
    )
    return upsert_fullmoon_todo(
        db,
        collection_protocol_id,
        tenant_no,
        user_id,
        handle_time=payload["handle_time"],
        handler_user_id=payload["handler_user_id"],
        todo_status=payload["make_or_todo_status"],
        handle_date=payload["legacy_date"] or None,
        insurance_attachments=payload["attachments"] or None,
        handle_remark=payload["remark"] or None,
    )


# ---------------------------------------------------------------------------
# 开票申请（主表 + 明细子表）
# ---------------------------------------------------------------------------


def build_installment_summary(installment_nos: List[Optional[int]]) -> str:
    nums = sorted({n for n in installment_nos if n is not None})
    if not nums:
        return "趸交"
    if len(nums) == 1:
        return f"第{nums[0]}期"
    return "第" + "、".join(str(n) for n in nums) + "期"


def _invoice_request_item_to_dict(
    item: BizProtocolInvoiceRequestItem,
    *,
    payment_detail: Optional[BizProtocolPaymentDetail] = None,
) -> Dict[str, Any]:
    out: Dict[str, Any] = {
        "id": item.id,
        "invoiceRequestId": item.invoice_request_id,
        "paymentId": item.payment_id,
        "paymentDetailId": item.payment_detail_id,
        "installmentNo": item.installment_no,
        "lineAmount": float(item.line_amount) if item.line_amount is not None else None,
        "orderNum": item.order_num,
        "remark": item.remark or "",
        "createTime": item.create_time.strftime("%Y-%m-%d %H:%M:%S") if item.create_time else "",
    }
    if payment_detail is not None:
        out["transferTime"] = (
            payment_detail.transfer_time.strftime("%Y-%m-%d %H:%M:%S")
            if payment_detail.transfer_time
            else ""
        )
        out["transferType"] = payment_detail.transfer_type or ""
        out["installmentDueDate"] = payment_detail.installment_due_date or ""
    return out


def _invoice_request_to_dict(
    req: BizProtocolInvoiceRequest,
    *,
    items: Optional[List[Dict[str, Any]]] = None,
    installment_summary: Optional[str] = None,
    protocol_ctx: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
    out: Dict[str, Any] = {
        "id": req.id,
        "collectionProtocolId": req.collection_protocol_id,
        "invoiceAmount": float(req.invoice_amount) if req.invoice_amount is not None else None,
        "invoiceTitle": req.invoice_title or "",
        "taxNo": req.tax_no or "",
        "companyAddress": req.company_address or "",
        "companyPhone": req.company_phone or "",
        "bankName": req.bank_name or "",
        "bankAccount": req.bank_account or "",
        "applyRemark": req.apply_remark or "",
        "invoiceAttachments": req.invoice_attachments or "",
        "financeRemark": req.finance_remark or "",
        "status": req.status,
        "financeHandleTime": (
            req.finance_handle_time.strftime("%Y-%m-%d %H:%M:%S") if req.finance_handle_time else ""
        ),
        "tenantNo": req.tenant_no,
        "orderNum": req.order_num,
        "remark": req.remark or "",
        "createTime": req.create_time.strftime("%Y-%m-%d %H:%M:%S") if req.create_time else "",
        "updateTime": req.update_time.strftime("%Y-%m-%d %H:%M:%S") if req.update_time else "",
    }
    if items is not None:
        out["items"] = items
        out["itemCount"] = len(items)
    if installment_summary is not None:
        out["installmentSummary"] = installment_summary
    if protocol_ctx:
        out.update(protocol_ctx)
    return out


def _active_invoiced_payment_detail_ids(db: Session, tenant_no: str) -> set:
    rows = (
        db.query(BizProtocolInvoiceRequestItem.payment_detail_id)
        .join(
            BizProtocolInvoiceRequest,
            BizProtocolInvoiceRequest.id == BizProtocolInvoiceRequestItem.invoice_request_id,
        )
        .filter(
            BizProtocolInvoiceRequest.tenant_no == tenant_no,
            BizProtocolInvoiceRequest.del_flag == 0,
            BizProtocolInvoiceRequest.status.in_(("dkp", "ykp")),
            BizProtocolInvoiceRequestItem.del_flag == 0,
        )
        .all()
    )
    return {r[0] for r in rows}


def _pack_invoice_request_with_items(
    db: Session,
    req: BizProtocolInvoiceRequest,
    *,
    tenant_no: str,
    with_protocol_ctx: bool = False,
) -> Dict[str, Any]:
    items_orm = (
        db.query(BizProtocolInvoiceRequestItem)
        .filter(
            BizProtocolInvoiceRequestItem.invoice_request_id == req.id,
            BizProtocolInvoiceRequestItem.del_flag == 0,
        )
        .order_by(BizProtocolInvoiceRequestItem.order_num.asc())
        .all()
    )
    detail_ids = [i.payment_detail_id for i in items_orm]
    detail_map: Dict[str, BizProtocolPaymentDetail] = {}
    if detail_ids:
        details = (
            db.query(BizProtocolPaymentDetail)
            .filter(
                BizProtocolPaymentDetail.id.in_(detail_ids),
                BizProtocolPaymentDetail.del_flag == 0,
            )
            .all()
        )
        detail_map = {d.id: d for d in details}
    item_dicts = [
        _invoice_request_item_to_dict(i, payment_detail=detail_map.get(i.payment_detail_id))
        for i in items_orm
    ]
    summary = build_installment_summary([i.installment_no for i in items_orm])
    protocol_ctx: Optional[Dict[str, Any]] = None
    if with_protocol_ctx:
        prot = (
            db.query(BizCollectionProtocol)
            .filter(
                BizCollectionProtocol.id == req.collection_protocol_id,
                BizCollectionProtocol.del_flag == 0,
            )
            .first()
        )
        pc_map = _party_customer_map_by_protocol_ids(db, [req.collection_protocol_id])
        pc = pc_map.get(req.collection_protocol_id)
        protocol_ctx = {
            "protocolNo": (prot.protocol_no if prot else "") or "",
            "protocolStatus": (prot.protocol_status if prot else "") or "",
            "motherName": (pc.mother_name if pc else "") or "",
            "motherIdCard": (pc.mother_id_card if pc else "") or "",
            "newbornName": (pc.newborn_name if pc else "") or "",
            "newbornIdCard": (pc.newborn_id_card if pc else "") or "",
        }
    out = _invoice_request_to_dict(
        req, items=item_dicts, installment_summary=summary, protocol_ctx=protocol_ctx
    )
    if req.status == "ykp" and req.invoice_attachments:
        out["invoiceAttachmentFiles"] = urls_to_attachment_list(
            req.invoice_attachments, db, tenant_no
        )
    else:
        out["invoiceAttachmentFiles"] = []
    return out


def list_invoice_eligible_payment_details(
    db: Session,
    *,
    collection_protocol_id: str,
    tenant_no: str,
) -> List[Dict[str, Any]]:
    if not tenant_no or not collection_protocol_id:
        return []
    invoiced_ids = _active_invoiced_payment_detail_ids(db, tenant_no)
    payment = (
        db.query(BizProtocolPayment)
        .filter(
            BizProtocolPayment.collection_protocol_id == collection_protocol_id,
            BizProtocolPayment.tenant_no == tenant_no,
            BizProtocolPayment.del_flag == 0,
        )
        .first()
    )
    if not payment:
        return []
    details = (
        db.query(BizProtocolPaymentDetail)
        .filter(
            BizProtocolPaymentDetail.payment_id == payment.id,
            BizProtocolPaymentDetail.tenant_no == tenant_no,
            BizProtocolPaymentDetail.del_flag == 0,
            BizProtocolPaymentDetail.pay_status == "yjf",
        )
        .order_by(
            BizProtocolPaymentDetail.order_num.asc(),
            BizProtocolPaymentDetail.create_time.desc(),
        )
        .all()
    )
    out: List[Dict[str, Any]] = []
    for d in details:
        if d.id in invoiced_ids:
            continue
        out.append(
            {
                "id": d.id,
                "paymentId": payment.id,
                "installmentNo": d.installment_no,
                "dueAmount": float(d.due_amount) if d.due_amount is not None else None,
                "transferAmount": float(d.transfer_amount) if d.transfer_amount is not None else None,
                "transferTime": (
                    d.transfer_time.strftime("%Y-%m-%d %H:%M:%S") if d.transfer_time else ""
                ),
                "transferType": d.transfer_type or "",
                "installmentDueDate": d.installment_due_date or "",
            }
        )
    return out


def list_invoice_requests_by_protocol(
    db: Session,
    *,
    collection_protocol_id: str,
    tenant_no: str,
) -> List[Dict[str, Any]]:
    rows = (
        db.query(BizProtocolInvoiceRequest)
        .filter(
            BizProtocolInvoiceRequest.collection_protocol_id == collection_protocol_id,
            BizProtocolInvoiceRequest.tenant_no == tenant_no,
            BizProtocolInvoiceRequest.del_flag == 0,
        )
        .order_by(BizProtocolInvoiceRequest.create_time.desc())
        .all()
    )
    return [_pack_invoice_request_with_items(db, r, tenant_no=tenant_no) for r in rows]


def create_invoice_request(
    db: Session,
    *,
    collection_protocol_id: str,
    payment_detail_ids: List[str],
    tenant_no: str,
    user_info: Optional[dict] = None,
    invoice_title: Optional[str] = None,
    tax_no: Optional[str] = None,
    company_address: Optional[str] = None,
    company_phone: Optional[str] = None,
    bank_name: Optional[str] = None,
    bank_account: Optional[str] = None,
    apply_remark: Optional[str] = None,
) -> Dict[str, Any]:
    ids = list(dict.fromkeys([str(i).strip() for i in (payment_detail_ids or []) if str(i).strip()]))
    if not ids:
        raise ValueError("请至少选择一条缴费明细")

    invoiced_ids = _active_invoiced_payment_detail_ids(db, tenant_no)
    overlap = set(ids) & invoiced_ids
    if overlap:
        raise ValueError("所选缴费明细中存在已申请开票的记录")

    payment = (
        db.query(BizProtocolPayment)
        .filter(
            BizProtocolPayment.collection_protocol_id == collection_protocol_id,
            BizProtocolPayment.tenant_no == tenant_no,
            BizProtocolPayment.del_flag == 0,
        )
        .first()
    )
    if not payment:
        raise ValueError("未找到缴费信息")

    details = (
        db.query(BizProtocolPaymentDetail)
        .filter(
            BizProtocolPaymentDetail.payment_id == payment.id,
            BizProtocolPaymentDetail.id.in_(ids),
            BizProtocolPaymentDetail.tenant_no == tenant_no,
            BizProtocolPaymentDetail.del_flag == 0,
            BizProtocolPaymentDetail.pay_status == "yjf",
        )
        .all()
    )
    if len(details) != len(ids):
        raise ValueError("部分缴费明细不可开票")

    detail_map = {d.id: d for d in details}
    ordered_details = [detail_map[i] for i in ids]

    uid = (user_info or {}).get("user_id") or ""
    dept_id = (user_info or {}).get("dept_id") or (user_info or {}).get("create_dept_id") or ""

    req = BizProtocolInvoiceRequest(
        id=generate_uuid(),
        collection_protocol_id=collection_protocol_id,
        invoice_amount=Decimal("0"),
        invoice_title=(invoice_title or "").strip() or None,
        tax_no=(tax_no or "").strip() or None,
        company_address=(company_address or "").strip() or None,
        company_phone=(company_phone or "").strip() or None,
        bank_name=(bank_name or "").strip() or None,
        bank_account=(bank_account or "").strip() or None,
        apply_remark=(apply_remark or "").strip() or None,
        status="dkp",
        tenant_no=tenant_no,
        create_by=uid,
        create_dept_id=dept_id,
        update_by=uid,
    )
    db.add(req)
    db.flush()

    total = Decimal("0")
    for idx, d in enumerate(ordered_details, start=1):
        line_amt = Decimal(str(d.transfer_amount or 0))
        total += line_amt
        db.add(
            BizProtocolInvoiceRequestItem(
                id=generate_uuid(),
                invoice_request_id=req.id,
                payment_id=payment.id,
                payment_detail_id=d.id,
                installment_no=d.installment_no,
                line_amount=line_amt,
                tenant_no=tenant_no,
                order_num=idx,
                create_by=uid,
                create_dept_id=dept_id,
                update_by=uid,
            )
        )

    req.invoice_amount = total
    db.commit()
    db.refresh(req)
    return _pack_invoice_request_with_items(db, req, tenant_no=tenant_no)


def _invoice_request_list_query(
    db: Session,
    *,
    tenant_no: str,
    status: str,
    keyword: Optional[str] = None,
):
    q = db.query(BizProtocolInvoiceRequest).filter(
        BizProtocolInvoiceRequest.tenant_no == tenant_no,
        BizProtocolInvoiceRequest.del_flag == 0,
        BizProtocolInvoiceRequest.status == status,
    )
    kw = (keyword or "").strip()
    if kw:
        like = f"%{kw}%"
        party_q = db.query(BizProtocolPartyCustomer.collection_protocol_id).filter(
            BizProtocolPartyCustomer.del_flag == 0,
            BizProtocolPartyCustomer.tenant_no == tenant_no,
            or_(
                BizProtocolPartyCustomer.mother_name.like(like),
                BizProtocolPartyCustomer.mother_id_card.like(like),
                BizProtocolPartyCustomer.newborn_name.like(like),
                BizProtocolPartyCustomer.newborn_id_card.like(like),
            ),
        )
        q = q.filter(BizProtocolInvoiceRequest.collection_protocol_id.in_(party_q))
    return q


def list_protocol_invoice_requests(
    db: Session,
    *,
    status: str,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    tenant_no: Optional[str] = None,
    user_info=None,
) -> Tuple[List[Dict[str, Any]], int]:
    if not tenant_no:
        return [], 0
    q_base = _invoice_request_list_query(db, tenant_no=tenant_no, status=status, keyword=keyword)
    if user_info:
        from app.core.data_scope import apply_data_scope

        q_base = apply_data_scope(q_base, BizProtocolInvoiceRequest, user_info, db, tenant_no)
    total = q_base.order_by(None).with_entities(func.count(BizProtocolInvoiceRequest.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    order_col = (
        BizProtocolInvoiceRequest.finance_handle_time.desc()
        if status == "ykp"
        else BizProtocolInvoiceRequest.create_time.desc()
    )
    rows = (
        q_base.order_by(order_col)
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    items = [
        _pack_invoice_request_with_items(db, r, tenant_no=tenant_no, with_protocol_ctx=True)
        for r in rows
    ]
    return items, int(total)


def count_pending_protocol_invoice_requests(
    db: Session,
    *,
    tenant_no: str,
    user_info=None,
) -> int:
    q = _invoice_request_list_query(db, tenant_no=tenant_no, status="dkp")
    if user_info:
        from app.core.data_scope import apply_data_scope

        q = apply_data_scope(q, BizProtocolInvoiceRequest, user_info, db, tenant_no)
    return int(q.with_entities(func.count(BizProtocolInvoiceRequest.id)).scalar() or 0)


def complete_invoice_request(
    db: Session,
    *,
    request_id: str,
    tenant_no: str,
    invoice_attachments: str,
    finance_remark: Optional[str] = None,
    user_info: Optional[dict] = None,
) -> Dict[str, Any]:
    if not (invoice_attachments or "").strip():
        raise ValueError("请上传发票附件")
    req = (
        db.query(BizProtocolInvoiceRequest)
        .filter(
            BizProtocolInvoiceRequest.id == request_id,
            BizProtocolInvoiceRequest.tenant_no == tenant_no,
            BizProtocolInvoiceRequest.del_flag == 0,
        )
        .first()
    )
    if not req:
        raise ValueError("开票申请不存在")
    if req.status != "dkp":
        raise ValueError("当前状态不可完成开票")
    uid = (user_info or {}).get("user_id") or ""
    req.status = "ykp"
    req.invoice_attachments = invoice_attachments.strip()
    req.finance_remark = (finance_remark or "").strip() or None
    req.finance_handle_time = datetime.now()
    req.update_by = uid
    req.update_time = datetime.now()
    db.commit()
    db.refresh(req)
    return _pack_invoice_request_with_items(
        db, req, tenant_no=tenant_no, with_protocol_ctx=True
    )