zkxw/backend/app/crud/pilot_service.py

"""中试服务管理 CRUD（细胞产品类型/规格/价格体系/项目价格、渠道信息、体检机构）"""
from __future__ import annotations

import json
import math
from datetime import datetime
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
from app.crud.collection_protocol import urls_to_attachment_list
from app.models.models import (
    BizCellProductType,
    BizCellProductSpec,
    BizCellProductPrice,
    BizCellProductProjectSystem,
    BizCellProductProjectSystemItem,
    BizChannel,
    BizChannelAccount,
    BizChannelAgreement,
    BizChannelAgreementProductPrice,
    BizChannelInvoice,
    BizChannelMailAddress,
    BizChannelProductQuota,
    BizChannelServiceOrder,
    BizChannelServiceOrderProduct,
    BizChannelServiceOrderTxn,
    BizCheckupInstitution,
    BizCheckupInstitutionInvoice,
    BizCheckupInstitutionProject,
    BizCheckupInstitutionPackage,
    BizCheckupInstitutionPackageItem,
    SysUser,
)


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 _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 "").strip() for u in rows}


# ────────────────────────────────────────────
# Entity 1: BizCellProductType  细胞产品类型
# ────────────────────────────────────────────

def _row_to_dict_cell_product_type(row) -> Dict[str, Any]:
    return {
        "id": row.id,
        "productTypeCode": row.product_type_code or "",
        "productTypeName": row.product_type_name or "",
        "tenantNo": row.tenant_no or "",
        "orderNum": row.order_num if row.order_num is not None else 0,
        "remark": row.remark or "",
        "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
        "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
    }


def list_cell_product_types(db, *, page=1, page_size=10, keyword=None, tenant_no=None, user_info=None):
    def _filtered():
        q0 = db.query(BizCellProductType).filter(BizCellProductType.del_flag == 0)
        if tenant_no:
            q0 = q0.filter(BizCellProductType.tenant_no == tenant_no)
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            q0 = q0.filter(
                or_(
                    BizCellProductType.product_type_code.like(like),
                    BizCellProductType.product_type_name.like(like),
                )
            )
        if user_info:
            from app.core.data_scope import apply_data_scope
            q0 = apply_data_scope(q0, BizCellProductType, user_info, db, tenant_no)
        return q0

    total = _filtered().order_by(None).with_entities(func.count(BizCellProductType.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = _filtered().order_by(BizCellProductType.create_time.desc()).offset((page - 1) * page_size).limit(page_size).all()
    return [_row_to_dict_cell_product_type(x) for x in rows], int(total)


def get_cell_product_type(db, type_id, tenant_no=None):
    q = db.query(BizCellProductType).filter(and_(BizCellProductType.id == type_id, BizCellProductType.del_flag == 0))
    if tenant_no:
        q = q.filter(BizCellProductType.tenant_no == tenant_no)
    return q.first()


def cell_product_type_code_exists(db, code, tenant_no, exclude_id=None):
    q = db.query(BizCellProductType).filter(
        and_(BizCellProductType.product_type_code == code, BizCellProductType.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizCellProductType.tenant_no == tenant_no)
    if exclude_id:
        q = q.filter(BizCellProductType.id != exclude_id)
    return db.query(q.exists()).scalar()


def create_cell_product_type(db, *, product_type_code, product_type_name, tenant_no="zkxw", order_num=None, remark=None, user_id=None):
    if cell_product_type_code_exists(db, product_type_code, tenant_no):
        raise ValueError(f"产品类型编码 {product_type_code} 已存在")
    obj = BizCellProductType(
        id=generate_uuid(),
        product_type_code=product_type_code,
        product_type_name=product_type_name,
        tenant_no=tenant_no,
        order_num=order_num or 0,
        remark=remark or "",
        create_by=user_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 update_cell_product_type(db, type_id, *, product_type_code=None, product_type_name=None, order_num=None, remark=None, user_id=None, tenant_no=None):
    obj = get_cell_product_type(db, type_id, tenant_no)
    if not obj:
        raise ValueError("记录不存在")
    if product_type_code is not None and product_type_code != obj.product_type_code:
        if cell_product_type_code_exists(db, product_type_code, tenant_no, exclude_id=type_id):
            raise ValueError(f"产品类型编码 {product_type_code} 已存在")
        obj.product_type_code = product_type_code
    if product_type_name is not None:
        obj.product_type_name = product_type_name
    if order_num is not None:
        obj.order_num = order_num
    if remark is not None:
        obj.remark = remark
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def soft_delete_cell_product_type(db, type_id, tenant_no=None):
    obj = get_cell_product_type(db, type_id, tenant_no)
    if not obj:
        raise ValueError("记录不存在")
    obj.del_flag = 1
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


# ────────────────────────────────────────────
# Entity 2: BizCellProductSpec  细胞产品规格
# ────────────────────────────────────────────

def _row_to_dict_cell_product_spec(row) -> Dict[str, Any]:
    return {
        "id": row.id,
        "specCode": row.spec_code or "",
        "specName": row.spec_name or "",
        "tenantNo": row.tenant_no or "",
        "orderNum": row.order_num if row.order_num is not None else 0,
        "remark": row.remark or "",
        "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
        "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
    }


def list_cell_product_specs(db, *, page=1, page_size=10, keyword=None, tenant_no=None, user_info=None):
    def _filtered():
        q0 = db.query(BizCellProductSpec).filter(BizCellProductSpec.del_flag == 0)
        if tenant_no:
            q0 = q0.filter(BizCellProductSpec.tenant_no == tenant_no)
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            q0 = q0.filter(
                or_(
                    BizCellProductSpec.spec_code.like(like),
                    BizCellProductSpec.spec_name.like(like),
                )
            )
        if user_info:
            from app.core.data_scope import apply_data_scope
            q0 = apply_data_scope(q0, BizCellProductSpec, user_info, db, tenant_no)
        return q0

    total = _filtered().order_by(None).with_entities(func.count(BizCellProductSpec.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = _filtered().order_by(BizCellProductSpec.create_time.desc()).offset((page - 1) * page_size).limit(page_size).all()
    return [_row_to_dict_cell_product_spec(x) for x in rows], int(total)


def get_cell_product_spec(db, spec_id, tenant_no=None):
    q = db.query(BizCellProductSpec).filter(and_(BizCellProductSpec.id == spec_id, BizCellProductSpec.del_flag == 0))
    if tenant_no:
        q = q.filter(BizCellProductSpec.tenant_no == tenant_no)
    return q.first()


def cell_product_spec_code_exists(db, code, tenant_no, exclude_id=None):
    q = db.query(BizCellProductSpec).filter(
        and_(BizCellProductSpec.spec_code == code, BizCellProductSpec.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizCellProductSpec.tenant_no == tenant_no)
    if exclude_id:
        q = q.filter(BizCellProductSpec.id != exclude_id)
    return db.query(q.exists()).scalar()


def create_cell_product_spec(db, *, spec_code, spec_name, tenant_no="zkxw", order_num=None, remark=None, user_id=None):
    if cell_product_spec_code_exists(db, spec_code, tenant_no):
        raise ValueError(f"规格编码 {spec_code} 已存在")
    obj = BizCellProductSpec(
        id=generate_uuid(),
        spec_code=spec_code,
        spec_name=spec_name,
        tenant_no=tenant_no,
        order_num=order_num or 0,
        remark=remark or "",
        create_by=user_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 update_cell_product_spec(db, spec_id, *, spec_code=None, spec_name=None, order_num=None, remark=None, user_id=None, tenant_no=None):
    obj = get_cell_product_spec(db, spec_id, tenant_no)
    if not obj:
        raise ValueError("记录不存在")
    if spec_code is not None and spec_code != obj.spec_code:
        if cell_product_spec_code_exists(db, spec_code, tenant_no, exclude_id=spec_id):
            raise ValueError(f"规格编码 {spec_code} 已存在")
        obj.spec_code = spec_code
    if spec_name is not None:
        obj.spec_name = spec_name
    if order_num is not None:
        obj.order_num = order_num
    if remark is not None:
        obj.remark = remark
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def soft_delete_cell_product_spec(db, spec_id, tenant_no=None):
    obj = get_cell_product_spec(db, spec_id, tenant_no)
    if not obj:
        raise ValueError("记录不存在")
    obj.del_flag = 1
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


# ────────────────────────────────────────────
# Entity 3: BizCellProductPrice  细胞产品价格
# ────────────────────────────────────────────

def _row_to_dict_cell_product_price(row) -> Dict[str, Any]:
    return {
        "id": row.id,
        "projectName": row.project_name or "",
        "projectCode": row.project_code or "",
        "cellTypeCode": row.cell_type_code or "",
        "cellTypeName": row.cell_type_name or "",
        "specText": row.spec_text or "",
        "specId": row.spec_id or "",
        "cellProductTypeId": row.cell_product_type_id or "",
        "unitPrice": float(row.unit_price) if row.unit_price is not None else None,
        "tenantNo": row.tenant_no or "",
        "orderNum": row.order_num if row.order_num is not None else 0,
        "remark": row.remark or "",
        "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
        "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
    }


def list_cell_product_prices(db, *, page=1, page_size=10, keyword=None, tenant_no=None, cell_product_type_id=None, user_info=None):
    def _filtered():
        q0 = db.query(BizCellProductPrice).filter(BizCellProductPrice.del_flag == 0)
        if tenant_no:
            q0 = q0.filter(BizCellProductPrice.tenant_no == tenant_no)
        if cell_product_type_id:
            q0 = q0.filter(BizCellProductPrice.cell_product_type_id == cell_product_type_id)
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            q0 = q0.filter(
                or_(
                    BizCellProductPrice.project_name.like(like),
                    BizCellProductPrice.project_code.like(like),
                    BizCellProductPrice.cell_type_name.like(like),
                )
            )
        if user_info:
            from app.core.data_scope import apply_data_scope
            q0 = apply_data_scope(q0, BizCellProductPrice, user_info, db, tenant_no)
        return q0

    total = _filtered().order_by(None).with_entities(func.count(BizCellProductPrice.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = _filtered().order_by(BizCellProductPrice.create_time.desc()).offset((page - 1) * page_size).limit(page_size).all()
    return [_row_to_dict_cell_product_price(x) for x in rows], int(total)


def get_cell_product_price(db, price_id, tenant_no=None):
    q = db.query(BizCellProductPrice).filter(and_(BizCellProductPrice.id == price_id, BizCellProductPrice.del_flag == 0))
    if tenant_no:
        q = q.filter(BizCellProductPrice.tenant_no == tenant_no)
    return q.first()


def cell_product_price_code_exists(db, code, tenant_no, exclude_id=None):
    q = db.query(BizCellProductPrice).filter(
        and_(BizCellProductPrice.project_code == code, BizCellProductPrice.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizCellProductPrice.tenant_no == tenant_no)
    if exclude_id:
        q = q.filter(BizCellProductPrice.id != exclude_id)
    return db.query(q.exists()).scalar()


def create_cell_product_price(
    db,
    *,
    project_name,
    project_code,
    cell_type_code,
    cell_type_name,
    spec_text,
    spec_id,
    cell_product_type_id,
    unit_price,
    tenant_no="zkxw",
    order_num=None,
    remark=None,
    user_id=None,
):
    if cell_product_price_code_exists(db, project_code, tenant_no):
        raise ValueError(f"项目编码 {project_code} 已存在")
    obj = BizCellProductPrice(
        id=generate_uuid(),
        project_name=project_name,
        project_code=project_code,
        cell_type_code=cell_type_code,
        cell_type_name=cell_type_name,
        spec_text=spec_text or "",
        spec_id=spec_id or "",
        cell_product_type_id=cell_product_type_id or "",
        unit_price=unit_price,
        tenant_no=tenant_no,
        order_num=order_num or 0,
        remark=remark or "",
        create_by=user_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 update_cell_product_price(
    db,
    price_id,
    *,
    project_name=None,
    project_code=None,
    cell_type_code=None,
    cell_type_name=None,
    spec_text=None,
    spec_id=None,
    cell_product_type_id=None,
    unit_price=None,
    order_num=None,
    remark=None,
    user_id=None,
    tenant_no=None,
):
    obj = get_cell_product_price(db, price_id, tenant_no)
    if not obj:
        raise ValueError("记录不存在")
    if project_code is not None and project_code != obj.project_code:
        if cell_product_price_code_exists(db, project_code, tenant_no, exclude_id=price_id):
            raise ValueError(f"项目编码 {project_code} 已存在")
        obj.project_code = project_code
    if project_name is not None:
        obj.project_name = project_name
    if cell_type_code is not None:
        obj.cell_type_code = cell_type_code
    if cell_type_name is not None:
        obj.cell_type_name = cell_type_name
    if spec_text is not None:
        obj.spec_text = spec_text
    if spec_id is not None:
        obj.spec_id = spec_id
    if cell_product_type_id is not None:
        obj.cell_product_type_id = cell_product_type_id
    if unit_price is not None:
        obj.unit_price = unit_price
    if order_num is not None:
        obj.order_num = order_num
    if remark is not None:
        obj.remark = remark
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def soft_delete_cell_product_price(db, price_id, tenant_no=None):
    obj = get_cell_product_price(db, price_id, tenant_no)
    if not obj:
        raise ValueError("记录不存在")
    obj.del_flag = 1
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


# ────────────────────────────────────────────
# Entity 4: BizCellProductProjectSystem  项目价格体系（主）
# ────────────────────────────────────────────

def _row_to_dict_project_system(row) -> Dict[str, Any]:
    return {
        "id": row.id,
        "systemName": row.system_name or "",
        "systemCode": row.system_code or "",
        "status": row.status or "1",
        "tenantNo": row.tenant_no or "",
        "orderNum": row.order_num if row.order_num is not None else 0,
        "remark": row.remark or "",
        "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
        "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
    }


def list_project_systems(db, *, page=1, page_size=10, keyword=None, tenant_no=None, user_info=None):
    def _filtered():
        q0 = db.query(BizCellProductProjectSystem).filter(BizCellProductProjectSystem.del_flag == 0)
        if tenant_no:
            q0 = q0.filter(BizCellProductProjectSystem.tenant_no == tenant_no)
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            q0 = q0.filter(
                or_(
                    BizCellProductProjectSystem.system_code.like(like),
                    BizCellProductProjectSystem.system_name.like(like),
                )
            )
        if user_info:
            from app.core.data_scope import apply_data_scope
            q0 = apply_data_scope(q0, BizCellProductProjectSystem, user_info, db, tenant_no)
        return q0

    total = _filtered().order_by(None).with_entities(func.count(BizCellProductProjectSystem.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = _filtered().order_by(BizCellProductProjectSystem.create_time.desc()).offset((page - 1) * page_size).limit(page_size).all()
    return [_row_to_dict_project_system(x) for x in rows], int(total)


def get_project_system(db, system_id, tenant_no=None):
    q = db.query(BizCellProductProjectSystem).filter(
        and_(BizCellProductProjectSystem.id == system_id, BizCellProductProjectSystem.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizCellProductProjectSystem.tenant_no == tenant_no)
    return q.first()


def project_system_code_exists(db, code, tenant_no, exclude_id=None):
    q = db.query(BizCellProductProjectSystem).filter(
        and_(BizCellProductProjectSystem.system_code == code, BizCellProductProjectSystem.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizCellProductProjectSystem.tenant_no == tenant_no)
    if exclude_id:
        q = q.filter(BizCellProductProjectSystem.id != exclude_id)
    return db.query(q.exists()).scalar()


def create_project_system(db, *, system_name, system_code, status="1", tenant_no="zkxw", order_num=None, remark=None, user_id=None):
    if project_system_code_exists(db, system_code, tenant_no):
        raise ValueError(f"体系编码 {system_code} 已存在")
    obj = BizCellProductProjectSystem(
        id=generate_uuid(),
        system_name=system_name,
        system_code=system_code,
        status=status or "1",
        tenant_no=tenant_no,
        order_num=order_num or 0,
        remark=remark or "",
        create_by=user_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 update_project_system(db, system_id, *, system_name=None, system_code=None, status=None, order_num=None, remark=None, user_id=None, tenant_no=None):
    obj = get_project_system(db, system_id, tenant_no)
    if not obj:
        raise ValueError("记录不存在")
    if system_code is not None and system_code != obj.system_code:
        if project_system_code_exists(db, system_code, tenant_no, exclude_id=system_id):
            raise ValueError(f"体系编码 {system_code} 已存在")
        obj.system_code = system_code
    if system_name is not None:
        obj.system_name = system_name
    if status is not None:
        obj.status = status
    if order_num is not None:
        obj.order_num = order_num
    if remark is not None:
        obj.remark = remark
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def soft_delete_project_system(db, system_id, tenant_no=None):
    obj = get_project_system(db, system_id, tenant_no)
    if not obj:
        raise ValueError("记录不存在")
    obj.del_flag = 1
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


# ────────────────────────────────────────────
# Entity 5: BizCellProductProjectSystemItem  项目价格体系明细
# ────────────────────────────────────────────

def _row_to_dict_project_system_item(row) -> Dict[str, Any]:
    return {
        "id": row.id,
        "projectSystemId": row.project_system_id or "",
        "sourcePriceId": row.source_price_id or "",
        "projectName": row.project_name or "",
        "projectCode": row.project_code or "",
        "cellTypeCode": row.cell_type_code or "",
        "cellTypeName": row.cell_type_name or "",
        "specText": row.spec_text or "",
        "specId": row.spec_id or "",
        "cellProductTypeId": row.cell_product_type_id or "",
        "unitPrice": float(row.unit_price) if row.unit_price is not None else None,
        "tenantNo": row.tenant_no or "",
        "orderNum": row.order_num if row.order_num is not None else 0,
        "remark": row.remark or "",
        "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
        "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
    }


def list_project_system_items(db, project_system_id, tenant_no=None, user_info=None):
    q = db.query(BizCellProductProjectSystemItem).filter(
        and_(
            BizCellProductProjectSystemItem.project_system_id == project_system_id,
            BizCellProductProjectSystemItem.del_flag == 0,
        )
    )
    if tenant_no:
        q = q.filter(BizCellProductProjectSystemItem.tenant_no == tenant_no)
    if user_info:
        from app.core.data_scope import apply_data_scope
        q = apply_data_scope(q, BizCellProductProjectSystemItem, user_info, db, tenant_no)
    rows = q.order_by(
        BizCellProductProjectSystemItem.order_num.asc(),
        BizCellProductProjectSystemItem.create_time.asc(),
    ).all()
    return [_row_to_dict_project_system_item(x) for x in rows]


def create_project_system_item(
    db,
    *,
    project_system_id,
    source_price_id="",
    project_name,
    project_code,
    cell_type_code,
    cell_type_name,
    spec_text="",
    spec_id="",
    cell_product_type_id="",
    unit_price,
    tenant_no="zkxw",
    order_num=None,
    remark=None,
    user_id=None,
):
    obj = BizCellProductProjectSystemItem(
        id=generate_uuid(),
        project_system_id=project_system_id,
        source_price_id=source_price_id,
        project_name=project_name,
        project_code=project_code,
        cell_type_code=cell_type_code,
        cell_type_name=cell_type_name,
        spec_text=spec_text or "",
        spec_id=spec_id or "",
        cell_product_type_id=cell_product_type_id or "",
        unit_price=unit_price,
        tenant_no=tenant_no,
        order_num=order_num or 0,
        remark=remark or "",
        create_by=user_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 update_project_system_item(
    db,
    item_id,
    *,
    source_price_id=None,
    project_name=None,
    project_code=None,
    cell_type_code=None,
    cell_type_name=None,
    spec_text=None,
    spec_id=None,
    cell_product_type_id=None,
    unit_price=None,
    order_num=None,
    remark=None,
    user_id=None,
    tenant_no=None,
):
    q = db.query(BizCellProductProjectSystemItem).filter(
        and_(BizCellProductProjectSystemItem.id == item_id, BizCellProductProjectSystemItem.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizCellProductProjectSystemItem.tenant_no == tenant_no)
    obj = q.first()
    if not obj:
        raise ValueError("记录不存在")
    if source_price_id is not None:
        obj.source_price_id = source_price_id
    if project_name is not None:
        obj.project_name = project_name
    if project_code is not None:
        obj.project_code = project_code
    if cell_type_code is not None:
        obj.cell_type_code = cell_type_code
    if cell_type_name is not None:
        obj.cell_type_name = cell_type_name
    if spec_text is not None:
        obj.spec_text = spec_text
    if spec_id is not None:
        obj.spec_id = spec_id
    if cell_product_type_id is not None:
        obj.cell_product_type_id = cell_product_type_id
    if unit_price is not None:
        obj.unit_price = unit_price
    if order_num is not None:
        obj.order_num = order_num
    if remark is not None:
        obj.remark = remark
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def soft_delete_project_system_item(db, item_id, tenant_no=None):
    q = db.query(BizCellProductProjectSystemItem).filter(
        and_(BizCellProductProjectSystemItem.id == item_id, BizCellProductProjectSystemItem.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizCellProductProjectSystemItem.tenant_no == tenant_no)
    obj = q.first()
    if not obj:
        raise ValueError("记录不存在")
    obj.del_flag = 1
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


# ────────────────────────────────────────────
# Entity 6: BizChannel  渠道信息
# ────────────────────────────────────────────

def _row_to_dict_channel(
    row,
    db: Optional[Session] = None,
    tenant_no: Optional[str] = None,
) -> Dict[str, Any]:
    tn = tenant_no or row.tenant_no
    return {
        "id": row.id,
        "channelName": row.channel_name or "",
        "creditCode": row.credit_code or "",
        "channelType": row.channel_type or "",
        "contactName": row.contact_name or "",
        "contactPhone": row.contact_phone or "",
        "contactEmail": row.contact_email or "",
        "protocolAttachments": row.protocol_attachments or "",
        "attachmentFiles": urls_to_attachment_list(row.protocol_attachments, db, tn) if db else [],
        "tenantNo": row.tenant_no or "",
        "orderNum": row.order_num if row.order_num is not None else 0,
        "remark": row.remark or "",
        "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
        "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
    }


def list_channels(db, *, page=1, page_size=10, keyword=None, tenant_no=None, user_info=None):
    def _filtered():
        q0 = db.query(BizChannel).filter(BizChannel.del_flag == 0)
        if tenant_no:
            q0 = q0.filter(BizChannel.tenant_no == tenant_no)
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            q0 = q0.filter(
                or_(
                    BizChannel.channel_name.like(like),
                    BizChannel.credit_code.like(like),
                    BizChannel.contact_name.like(like),
                )
            )
        if user_info:
            from app.core.data_scope import apply_data_scope
            q0 = apply_data_scope(q0, BizChannel, user_info, db, tenant_no)
        return q0

    total = _filtered().order_by(None).with_entities(func.count(BizChannel.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = _filtered().order_by(BizChannel.create_time.desc()).offset((page - 1) * page_size).limit(page_size).all()
    return [_row_to_dict_channel(x, db, tenant_no) for x in rows], int(total)


def get_channel(db, channel_id, tenant_no=None):
    q = db.query(BizChannel).filter(and_(BizChannel.id == channel_id, BizChannel.del_flag == 0))
    if tenant_no:
        q = q.filter(BizChannel.tenant_no == tenant_no)
    return q.first()


def create_channel(
    db,
    *,
    channel_name,
    credit_code=None,
    channel_type=None,
    contact_name=None,
    contact_phone=None,
    contact_email=None,
    protocol_attachments=None,
    tenant_no="zkxw",
    order_num=None,
    remark=None,
    user_id=None,
):
    obj = BizChannel(
        id=generate_uuid(),
        channel_name=channel_name,
        credit_code=credit_code or "",
        channel_type=channel_type or "",
        contact_name=contact_name or "",
        contact_phone=contact_phone or "",
        contact_email=contact_email or "",
        protocol_attachments=protocol_attachments or "",
        tenant_no=tenant_no,
        order_num=order_num or 0,
        remark=remark or "",
        create_by=user_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 update_channel(
    db,
    channel_id,
    *,
    channel_name=None,
    credit_code=None,
    channel_type=None,
    contact_name=None,
    contact_phone=None,
    contact_email=None,
    protocol_attachments=None,
    order_num=None,
    remark=None,
    user_id=None,
    tenant_no=None,
):
    obj = get_channel(db, channel_id, tenant_no)
    if not obj:
        raise ValueError("记录不存在")
    if channel_name is not None:
        obj.channel_name = channel_name
    if credit_code is not None:
        obj.credit_code = credit_code
    if channel_type is not None:
        obj.channel_type = channel_type
    if contact_name is not None:
        obj.contact_name = contact_name
    if contact_phone is not None:
        obj.contact_phone = contact_phone
    if contact_email is not None:
        obj.contact_email = contact_email
    if protocol_attachments is not None:
        obj.protocol_attachments = protocol_attachments
    if order_num is not None:
        obj.order_num = order_num
    if remark is not None:
        obj.remark = remark
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def soft_delete_channel(db, channel_id, tenant_no=None):
    obj = get_channel(db, channel_id, tenant_no)
    if not obj:
        raise ValueError("记录不存在")
    obj.del_flag = 1
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


# ────────────────────────────────────────────
# Entity 7: BizChannelInvoice  渠道开票信息
# ────────────────────────────────────────────

def _row_to_dict_channel_invoice(row) -> Dict[str, Any]:
    return {
        "id": row.id,
        "channelId": row.channel_id or "",
        "companyName": row.company_name or "",
        "taxNo": row.tax_no or "",
        "companyAddress": row.company_address or "",
        "companyPhone": row.company_phone or "",
        "bankName": row.bank_name or "",
        "bankAccount": row.bank_account or "",
        "tenantNo": row.tenant_no or "",
        "orderNum": row.order_num if row.order_num is not None else 0,
        "remark": row.remark or "",
        "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
        "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
    }


def list_channel_invoices(db, channel_id, tenant_no=None, user_info=None):
    q = db.query(BizChannelInvoice).filter(
        and_(
            BizChannelInvoice.channel_id == channel_id,
            BizChannelInvoice.del_flag == 0,
        )
    )
    if tenant_no:
        q = q.filter(BizChannelInvoice.tenant_no == tenant_no)
    if user_info:
        from app.core.data_scope import apply_data_scope
        q = apply_data_scope(q, BizChannelInvoice, user_info, db, tenant_no)
    rows = q.order_by(BizChannelInvoice.create_time.desc()).all()
    return [_row_to_dict_channel_invoice(x) for x in rows]


def create_channel_invoice(
    db,
    *,
    channel_id,
    company_name,
    tax_no=None,
    company_address=None,
    company_phone=None,
    bank_name=None,
    bank_account=None,
    tenant_no="zkxw",
    order_num=None,
    remark=None,
    user_id=None,
):
    obj = BizChannelInvoice(
        id=generate_uuid(),
        channel_id=channel_id,
        company_name=company_name,
        tax_no=tax_no or "",
        company_address=company_address or "",
        company_phone=company_phone or "",
        bank_name=bank_name or "",
        bank_account=bank_account or "",
        tenant_no=tenant_no,
        order_num=order_num or 0,
        remark=remark or "",
        create_by=user_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 update_channel_invoice(
    db,
    invoice_id,
    *,
    company_name=None,
    tax_no=None,
    company_address=None,
    company_phone=None,
    bank_name=None,
    bank_account=None,
    order_num=None,
    remark=None,
    user_id=None,
    tenant_no=None,
):
    q = db.query(BizChannelInvoice).filter(
        and_(BizChannelInvoice.id == invoice_id, BizChannelInvoice.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizChannelInvoice.tenant_no == tenant_no)
    obj = q.first()
    if not obj:
        raise ValueError("记录不存在")
    if company_name is not None:
        obj.company_name = company_name
    if tax_no is not None:
        obj.tax_no = tax_no
    if company_address is not None:
        obj.company_address = company_address
    if company_phone is not None:
        obj.company_phone = company_phone
    if bank_name is not None:
        obj.bank_name = bank_name
    if bank_account is not None:
        obj.bank_account = bank_account
    if order_num is not None:
        obj.order_num = order_num
    if remark is not None:
        obj.remark = remark
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def soft_delete_channel_invoice(db, invoice_id, tenant_no=None):
    q = db.query(BizChannelInvoice).filter(
        and_(BizChannelInvoice.id == invoice_id, BizChannelInvoice.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizChannelInvoice.tenant_no == tenant_no)
    obj = q.first()
    if not obj:
        raise ValueError("记录不存在")
    obj.del_flag = 1
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


# ────────────────────────────────────────────
# Entity 8: BizChannelMailAddress  渠道邮寄地址
# ────────────────────────────────────────────

def _row_to_dict_channel_mail_address(row) -> Dict[str, Any]:
    return {
        "id": row.id,
        "channelId": row.channel_id or "",
        "mailAddress": row.mail_address or "",
        "mailContact": row.mail_contact or "",
        "mailPhone": row.mail_phone or "",
        "tenantNo": row.tenant_no or "",
        "orderNum": row.order_num if row.order_num is not None else 0,
        "remark": row.remark or "",
        "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
        "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
    }


def list_channel_mail_addresses(db, channel_id, tenant_no=None, user_info=None):
    q = db.query(BizChannelMailAddress).filter(
        and_(
            BizChannelMailAddress.channel_id == channel_id,
            BizChannelMailAddress.del_flag == 0,
        )
    )
    if tenant_no:
        q = q.filter(BizChannelMailAddress.tenant_no == tenant_no)
    if user_info:
        from app.core.data_scope import apply_data_scope
        q = apply_data_scope(q, BizChannelMailAddress, user_info, db, tenant_no)
    rows = q.order_by(BizChannelMailAddress.create_time.desc()).all()
    return [_row_to_dict_channel_mail_address(x) for x in rows]


def create_channel_mail_address(
    db,
    *,
    channel_id,
    mail_address,
    mail_contact=None,
    mail_phone=None,
    tenant_no="zkxw",
    order_num=None,
    remark=None,
    user_id=None,
):
    obj = BizChannelMailAddress(
        id=generate_uuid(),
        channel_id=channel_id,
        mail_address=mail_address,
        mail_contact=mail_contact or "",
        mail_phone=mail_phone or "",
        tenant_no=tenant_no,
        order_num=order_num or 0,
        remark=remark or "",
        create_by=user_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 update_channel_mail_address(
    db,
    address_id,
    *,
    mail_address=None,
    mail_contact=None,
    mail_phone=None,
    order_num=None,
    remark=None,
    user_id=None,
    tenant_no=None,
):
    q = db.query(BizChannelMailAddress).filter(
        and_(BizChannelMailAddress.id == address_id, BizChannelMailAddress.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizChannelMailAddress.tenant_no == tenant_no)
    obj = q.first()
    if not obj:
        raise ValueError("记录不存在")
    if mail_address is not None:
        obj.mail_address = mail_address
    if mail_contact is not None:
        obj.mail_contact = mail_contact
    if mail_phone is not None:
        obj.mail_phone = mail_phone
    if order_num is not None:
        obj.order_num = order_num
    if remark is not None:
        obj.remark = remark
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def soft_delete_channel_mail_address(db, address_id, tenant_no=None):
    q = db.query(BizChannelMailAddress).filter(
        and_(BizChannelMailAddress.id == address_id, BizChannelMailAddress.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizChannelMailAddress.tenant_no == tenant_no)
    obj = q.first()
    if not obj:
        raise ValueError("记录不存在")
    obj.del_flag = 1
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


# ────────────────────────────────────────────
# Entity 9: BizChannelAgreement  渠道协议
# ────────────────────────────────────────────

def _row_to_dict_channel_agreement(row) -> Dict[str, Any]:
    def _fmt_date(v):
        """CHAR(8) YYYYMMDD → YYYY-MM-DD"""
        if not v or len(v) != 8:
            return v or ""
        return f"{v[:4]}-{v[4:6]}-{v[6:8]}"

    return {
        "id": row.id,
        "channelId": row.channel_id or "",
        "agreementType": row.agreement_type or "",
        "businessType": row.business_type or "",
        "settleMode": row.settle_mode or "",
        "creditLimitAmount": float(row.credit_limit_amount) if row.credit_limit_amount is not None else None,
        "effectiveDate": _fmt_date(row.effective_date),
        "expireDate": _fmt_date(row.expire_date),
        "agreementStatus": row.agreement_status or "",
        "tenantNo": row.tenant_no or "",
        "orderNum": row.order_num if row.order_num is not None else 0,
        "remark": row.remark or "",
        "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
        "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
    }


def list_channel_agreements(db, channel_id, tenant_no=None, user_info=None):
    q = db.query(BizChannelAgreement).filter(
        and_(
            BizChannelAgreement.channel_id == channel_id,
            BizChannelAgreement.del_flag == 0,
        )
    )
    if tenant_no:
        q = q.filter(BizChannelAgreement.tenant_no == tenant_no)
    if user_info:
        from app.core.data_scope import apply_data_scope
        q = apply_data_scope(q, BizChannelAgreement, user_info, db, tenant_no)
    rows = q.order_by(BizChannelAgreement.create_time.desc()).all()
    return [_row_to_dict_channel_agreement(x) for x in rows]


def create_channel_agreement(
    db,
    *,
    channel_id,
    agreement_type,
    business_type=None,
    settle_mode=None,
    credit_limit_amount=None,
    effective_date=None,
    expire_date=None,
    agreement_status="dsx",
    tenant_no="zkxw",
    order_num=None,
    remark=None,
    user_id=None,
):
    _ed = (effective_date or "").replace("-", "") if effective_date else ""
    _xd = (expire_date or "").replace("-", "") if expire_date else ""
    obj = BizChannelAgreement(
        id=generate_uuid(),
        channel_id=channel_id,
        agreement_type=agreement_type,
        business_type=business_type or "",
        settle_mode=settle_mode or "",
        credit_limit_amount=credit_limit_amount,
        effective_date=_ed,
        expire_date=_xd,
        agreement_status=agreement_status or "dsx",
        tenant_no=tenant_no,
        order_num=order_num or 0,
        remark=remark or "",
        create_by=user_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 update_channel_agreement(
    db,
    agreement_id,
    *,
    agreement_type=None,
    business_type=None,
    settle_mode=None,
    credit_limit_amount=None,
    effective_date=None,
    expire_date=None,
    order_num=None,
    remark=None,
    user_id=None,
    tenant_no=None,
):
    q = db.query(BizChannelAgreement).filter(
        and_(BizChannelAgreement.id == agreement_id, BizChannelAgreement.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizChannelAgreement.tenant_no == tenant_no)
    obj = q.first()
    if not obj:
        raise ValueError("记录不存在")
    if agreement_type is not None:
        obj.agreement_type = agreement_type
    if business_type is not None:
        obj.business_type = business_type
    if settle_mode is not None:
        obj.settle_mode = settle_mode
    if credit_limit_amount is not None:
        obj.credit_limit_amount = credit_limit_amount
    if effective_date is not None:
        obj.effective_date = effective_date.replace("-", "") if effective_date else ""
    if expire_date is not None:
        obj.expire_date = expire_date.replace("-", "") if expire_date else ""
    if order_num is not None:
        obj.order_num = order_num
    if remark is not None:
        obj.remark = remark
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def soft_delete_channel_agreement(db, agreement_id, tenant_no=None):
    q = db.query(BizChannelAgreement).filter(
        and_(BizChannelAgreement.id == agreement_id, BizChannelAgreement.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizChannelAgreement.tenant_no == tenant_no)
    obj = q.first()
    if not obj:
        raise ValueError("记录不存在")
    obj.del_flag = 1
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


def activate_channel_agreement(db, agreement_id, tenant_no=None, user_id=None):
    """生效协议：仅待生效(dsx)可变为生效中(sxz)"""
    q = db.query(BizChannelAgreement).filter(
        and_(BizChannelAgreement.id == agreement_id, BizChannelAgreement.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizChannelAgreement.tenant_no == tenant_no)
    obj = q.first()
    if not obj:
        raise ValueError("记录不存在")
    if (obj.agreement_status or "").strip() != "dsx":
        raise ValueError("仅待生效状态的协议可执行生效操作")
    obj.agreement_status = "sxz"
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def void_channel_agreement(db, agreement_id, tenant_no=None, user_id=None):
    """作废协议：设置 agreement_status='yzf'"""
    q = db.query(BizChannelAgreement).filter(
        and_(BizChannelAgreement.id == agreement_id, BizChannelAgreement.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizChannelAgreement.tenant_no == tenant_no)
    obj = q.first()
    if not obj:
        raise ValueError("记录不存在")
    obj.agreement_status = "yzf"
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


# ────────────────────────────────────────────
# Entity 10: BizChannelAgreementProductPrice  渠道协议产品价格
# ────────────────────────────────────────────

def _row_to_dict_channel_agreement_price(row) -> Dict[str, Any]:
    return {
        "id": row.id,
        "channelAgreementId": row.channel_agreement_id or "",
        "projectName": row.project_name or "",
        "projectCode": row.project_code or "",
        "cellTypeCode": row.cell_type_code or "",
        "cellTypeName": row.cell_type_name or "",
        "specText": row.spec_text or "",
        "originalPrice": float(row.original_price) if row.original_price is not None else None,
        "agreementPrice": float(row.agreement_price) if row.agreement_price is not None else None,
        "tenantNo": row.tenant_no or "",
        "orderNum": row.order_num if row.order_num is not None else 0,
        "remark": row.remark or "",
        "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
        "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
    }


def _channel_agreement_price_item_key(
    project_code=None,
    project_name=None,
    cell_type_code=None,
    cell_type_name=None,
    spec_text=None,
) -> str:
    code = (project_code or project_name or "").strip().lower()
    cell = (cell_type_code or cell_type_name or "").strip().lower()
    spec = (spec_text or "").strip().lower()
    return f"{code}|{cell}|{spec}"


def _channel_agreement_price_exists(
    db,
    channel_agreement_id,
    *,
    project_code=None,
    project_name=None,
    cell_type_code=None,
    cell_type_name=None,
    spec_text=None,
    tenant_no=None,
    exclude_price_id=None,
):
    target_key = _channel_agreement_price_item_key(
        project_code=project_code,
        project_name=project_name,
        cell_type_code=cell_type_code,
        cell_type_name=cell_type_name,
        spec_text=spec_text,
    )
    q = db.query(BizChannelAgreementProductPrice).filter(
        and_(
            BizChannelAgreementProductPrice.channel_agreement_id == channel_agreement_id,
            BizChannelAgreementProductPrice.del_flag == 0,
        )
    )
    if tenant_no:
        q = q.filter(BizChannelAgreementProductPrice.tenant_no == tenant_no)
    if exclude_price_id:
        q = q.filter(BizChannelAgreementProductPrice.id != exclude_price_id)
    for row in q.all():
        if _channel_agreement_price_item_key(
            project_code=row.project_code,
            project_name=row.project_name,
            cell_type_code=row.cell_type_code,
            cell_type_name=row.cell_type_name,
            spec_text=row.spec_text,
        ) == target_key:
            return True
    return False


def list_channel_agreement_prices(db, channel_agreement_id, tenant_no=None, user_info=None):
    q = db.query(BizChannelAgreementProductPrice).filter(
        and_(
            BizChannelAgreementProductPrice.channel_agreement_id == channel_agreement_id,
            BizChannelAgreementProductPrice.del_flag == 0,
        )
    )
    if tenant_no:
        q = q.filter(BizChannelAgreementProductPrice.tenant_no == tenant_no)
    if user_info:
        from app.core.data_scope import apply_data_scope
        q = apply_data_scope(q, BizChannelAgreementProductPrice, user_info, db, tenant_no)
    rows = q.order_by(BizChannelAgreementProductPrice.create_time.desc()).all()
    return [_row_to_dict_channel_agreement_price(x) for x in rows]


def create_channel_agreement_price(
    db,
    *,
    channel_agreement_id,
    project_name=None,
    project_code=None,
    cell_type_code=None,
    cell_type_name=None,
    spec_text=None,
    original_price=None,
    agreement_price=None,
    tenant_no="zkxw",
    order_num=None,
    remark=None,
    user_id=None,
):
    if _channel_agreement_price_exists(
        db,
        channel_agreement_id,
        project_code=project_code,
        project_name=project_name,
        cell_type_code=cell_type_code,
        cell_type_name=cell_type_name,
        spec_text=spec_text,
        tenant_no=tenant_no,
    ):
        raise ValueError("该项目信息已存在，不可重复添加")
    obj = BizChannelAgreementProductPrice(
        id=generate_uuid(),
        channel_agreement_id=channel_agreement_id,
        project_name=project_name or "",
        project_code=project_code or "",
        cell_type_code=cell_type_code or "",
        cell_type_name=cell_type_name or "",
        spec_text=spec_text or "",
        original_price=original_price,
        agreement_price=agreement_price,
        tenant_no=tenant_no,
        order_num=order_num or 0,
        remark=remark or "",
        create_by=user_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 update_channel_agreement_price(
    db,
    price_id,
    *,
    project_name=None,
    project_code=None,
    cell_type_code=None,
    cell_type_name=None,
    spec_text=None,
    original_price=None,
    agreement_price=None,
    order_num=None,
    remark=None,
    user_id=None,
    tenant_no=None,
):
    q = db.query(BizChannelAgreementProductPrice).filter(
        and_(BizChannelAgreementProductPrice.id == price_id, BizChannelAgreementProductPrice.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizChannelAgreementProductPrice.tenant_no == tenant_no)
    obj = q.first()
    if not obj:
        raise ValueError("记录不存在")
    next_project_name = project_name if project_name is not None else obj.project_name
    next_project_code = project_code if project_code is not None else obj.project_code
    next_cell_type_code = cell_type_code if cell_type_code is not None else obj.cell_type_code
    next_cell_type_name = cell_type_name if cell_type_name is not None else obj.cell_type_name
    next_spec_text = spec_text if spec_text is not None else obj.spec_text
    if _channel_agreement_price_exists(
        db,
        obj.channel_agreement_id,
        project_code=next_project_code,
        project_name=next_project_name,
        cell_type_code=next_cell_type_code,
        cell_type_name=next_cell_type_name,
        spec_text=next_spec_text,
        tenant_no=tenant_no,
        exclude_price_id=price_id,
    ):
        raise ValueError("该项目信息已存在，不可重复添加")
    if project_name is not None:
        obj.project_name = project_name
    if project_code is not None:
        obj.project_code = project_code
    if cell_type_code is not None:
        obj.cell_type_code = cell_type_code
    if cell_type_name is not None:
        obj.cell_type_name = cell_type_name
    if spec_text is not None:
        obj.spec_text = spec_text
    if original_price is not None:
        obj.original_price = original_price
    if agreement_price is not None:
        obj.agreement_price = agreement_price
    if order_num is not None:
        obj.order_num = order_num
    if remark is not None:
        obj.remark = remark
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def soft_delete_channel_agreement_price(db, price_id, tenant_no=None):
    q = db.query(BizChannelAgreementProductPrice).filter(
        and_(BizChannelAgreementProductPrice.id == price_id, BizChannelAgreementProductPrice.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizChannelAgreementProductPrice.tenant_no == tenant_no)
    obj = q.first()
    if not obj:
        raise ValueError("记录不存在")
    obj.del_flag = 1
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


# ────────────────────────────────────────────
# Entity 11: BizCheckupInstitution  体检机构
# ────────────────────────────────────────────

def _row_to_dict_checkup_institution(row, projects: Optional[List[Dict[str, Any]]] = None) -> Dict[str, Any]:
    data = {
        "id": row.id,
        "institutionName": row.institution_name or "",
        "institutionAddress": row.institution_address or "",
        "contactName": row.contact_name or "",
        "contactPhone": row.contact_phone or "",
        "institutionLogo": row.institution_logo or "",
        "status": row.status or "1",
        "tenantNo": row.tenant_no or "",
        "orderNum": row.order_num if row.order_num is not None else 0,
        "remark": row.remark or "",
        "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
        "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
    }
    if projects is not None:
        data["projects"] = projects
    return data


def list_checkup_institutions(db, *, page=1, page_size=10, keyword=None, status=None, tenant_no=None, user_info=None):
    def _filtered():
        q0 = db.query(BizCheckupInstitution).filter(BizCheckupInstitution.del_flag == 0)
        if tenant_no:
            q0 = q0.filter(BizCheckupInstitution.tenant_no == tenant_no)
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            q0 = q0.filter(
                or_(
                    BizCheckupInstitution.institution_name.like(like),
                    BizCheckupInstitution.contact_name.like(like),
                )
            )
        st = (status or "").strip()
        if st in ("0", "1"):
            q0 = q0.filter(BizCheckupInstitution.status == st)
        if user_info:
            from app.core.data_scope import apply_data_scope
            q0 = apply_data_scope(q0, BizCheckupInstitution, user_info, db, tenant_no)
        return q0

    total = _filtered().order_by(None).with_entities(func.count(BizCheckupInstitution.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = _filtered().order_by(BizCheckupInstitution.create_time.desc()).offset((page - 1) * page_size).limit(page_size).all()
    institution_ids = [x.id for x in rows]
    projects_by_inst: Dict[str, List[Dict[str, Any]]] = {iid: [] for iid in institution_ids}
    if institution_ids:
        pq = db.query(BizCheckupInstitutionProject).filter(
            BizCheckupInstitutionProject.checkup_institution_id.in_(institution_ids),
            BizCheckupInstitutionProject.del_flag == 0,
        )
        if tenant_no:
            pq = pq.filter(BizCheckupInstitutionProject.tenant_no == tenant_no)
        if user_info:
            from app.core.data_scope import apply_data_scope
            pq = apply_data_scope(pq, BizCheckupInstitutionProject, user_info, db, tenant_no)
        for pr in pq.order_by(BizCheckupInstitutionProject.create_time.desc()).all():
            projects_by_inst.setdefault(pr.checkup_institution_id, []).append(
                _row_to_dict_checkup_institution_project(pr)
            )
    return [
        _row_to_dict_checkup_institution(x, projects_by_inst.get(x.id, [])) for x in rows
    ], int(total)


def get_checkup_institution(db, institution_id, tenant_no=None):
    q = db.query(BizCheckupInstitution).filter(
        and_(BizCheckupInstitution.id == institution_id, BizCheckupInstitution.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizCheckupInstitution.tenant_no == tenant_no)
    return q.first()


def create_checkup_institution(
    db,
    *,
    institution_name,
    institution_address=None,
    contact_name=None,
    contact_phone=None,
    institution_logo=None,
    status="1",
    tenant_no="zkxw",
    order_num=None,
    remark=None,
    user_id=None,
):
    obj = BizCheckupInstitution(
        id=generate_uuid(),
        institution_name=institution_name,
        institution_address=institution_address or "",
        contact_name=contact_name or "",
        contact_phone=contact_phone or "",
        institution_logo=(institution_logo or "").strip() or None,
        status=status or "1",
        tenant_no=tenant_no,
        order_num=order_num or 0,
        remark=remark or "",
        create_by=user_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 update_checkup_institution(
    db,
    institution_id,
    *,
    institution_name=None,
    institution_address=None,
    contact_name=None,
    contact_phone=None,
    institution_logo=None,
    status=None,
    order_num=None,
    remark=None,
    user_id=None,
    tenant_no=None,
):
    obj = get_checkup_institution(db, institution_id, tenant_no)
    if not obj:
        raise ValueError("记录不存在")
    if institution_name is not None:
        obj.institution_name = institution_name
    if institution_address is not None:
        obj.institution_address = institution_address
    if contact_name is not None:
        obj.contact_name = contact_name
    if contact_phone is not None:
        obj.contact_phone = contact_phone
    if institution_logo is not None:
        obj.institution_logo = (institution_logo or "").strip() or None
    if status is not None:
        obj.status = status
    if order_num is not None:
        obj.order_num = order_num
    if remark is not None:
        obj.remark = remark
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def soft_delete_checkup_institution(db, institution_id, tenant_no=None):
    obj = get_checkup_institution(db, institution_id, tenant_no)
    if not obj:
        raise ValueError("记录不存在")
    obj.del_flag = 1
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


# ────────────────────────────────────────────
# Entity 12: BizCheckupInstitutionInvoice  体检机构开票信息
# ────────────────────────────────────────────

def _row_to_dict_checkup_institution_invoice(row) -> Dict[str, Any]:
    return {
        "id": row.id,
        "checkupInstitutionId": row.checkup_institution_id or "",
        "companyName": row.company_name or "",
        "taxNo": row.tax_no or "",
        "companyAddress": row.company_address or "",
        "companyPhone": row.company_phone or "",
        "bankName": row.bank_name or "",
        "bankAccount": row.bank_account or "",
        "tenantNo": row.tenant_no or "",
        "orderNum": row.order_num if row.order_num is not None else 0,
        "remark": row.remark or "",
        "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
        "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
    }


def list_checkup_institution_invoices(db, checkup_institution_id, tenant_no=None, user_info=None):
    q = db.query(BizCheckupInstitutionInvoice).filter(
        and_(
            BizCheckupInstitutionInvoice.checkup_institution_id == checkup_institution_id,
            BizCheckupInstitutionInvoice.del_flag == 0,
        )
    )
    if tenant_no:
        q = q.filter(BizCheckupInstitutionInvoice.tenant_no == tenant_no)
    if user_info:
        from app.core.data_scope import apply_data_scope
        q = apply_data_scope(q, BizCheckupInstitutionInvoice, user_info, db, tenant_no)
    rows = q.order_by(BizCheckupInstitutionInvoice.create_time.desc()).all()
    return [_row_to_dict_checkup_institution_invoice(x) for x in rows]


def create_checkup_institution_invoice(
    db,
    *,
    checkup_institution_id,
    company_name,
    tax_no=None,
    company_address=None,
    company_phone=None,
    bank_name=None,
    bank_account=None,
    tenant_no="zkxw",
    order_num=None,
    remark=None,
    user_id=None,
):
    obj = BizCheckupInstitutionInvoice(
        id=generate_uuid(),
        checkup_institution_id=checkup_institution_id,
        company_name=company_name,
        tax_no=tax_no or "",
        company_address=company_address or "",
        company_phone=company_phone or "",
        bank_name=bank_name or "",
        bank_account=bank_account or "",
        tenant_no=tenant_no,
        order_num=order_num or 0,
        remark=remark or "",
        create_by=user_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 update_checkup_institution_invoice(
    db,
    invoice_id,
    *,
    company_name=None,
    tax_no=None,
    company_address=None,
    company_phone=None,
    bank_name=None,
    bank_account=None,
    order_num=None,
    remark=None,
    user_id=None,
    tenant_no=None,
):
    q = db.query(BizCheckupInstitutionInvoice).filter(
        and_(BizCheckupInstitutionInvoice.id == invoice_id, BizCheckupInstitutionInvoice.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizCheckupInstitutionInvoice.tenant_no == tenant_no)
    obj = q.first()
    if not obj:
        raise ValueError("记录不存在")
    if company_name is not None:
        obj.company_name = company_name
    if tax_no is not None:
        obj.tax_no = tax_no
    if company_address is not None:
        obj.company_address = company_address
    if company_phone is not None:
        obj.company_phone = company_phone
    if bank_name is not None:
        obj.bank_name = bank_name
    if bank_account is not None:
        obj.bank_account = bank_account
    if order_num is not None:
        obj.order_num = order_num
    if remark is not None:
        obj.remark = remark
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def soft_delete_checkup_institution_invoice(db, invoice_id, tenant_no=None):
    q = db.query(BizCheckupInstitutionInvoice).filter(
        and_(BizCheckupInstitutionInvoice.id == invoice_id, BizCheckupInstitutionInvoice.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizCheckupInstitutionInvoice.tenant_no == tenant_no)
    obj = q.first()
    if not obj:
        raise ValueError("记录不存在")
    obj.del_flag = 1
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


# ────────────────────────────────────────────
# Entity 13: BizCheckupInstitutionProject  体检机构项目及价格
# ────────────────────────────────────────────

def _row_to_dict_checkup_institution_project(row) -> Dict[str, Any]:
    return {
        "id": row.id,
        "checkupInstitutionId": row.checkup_institution_id or "",
        "projectName": row.project_name or "",
        "projectFee": float(row.project_fee) if row.project_fee is not None else None,
        "status": row.status or "1",
        "tenantNo": row.tenant_no or "",
        "orderNum": row.order_num if row.order_num is not None else 0,
        "remark": row.remark or "",
        "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
        "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
    }


def list_checkup_institution_projects(db, checkup_institution_id, tenant_no=None, user_info=None):
    q = db.query(BizCheckupInstitutionProject).filter(
        and_(
            BizCheckupInstitutionProject.checkup_institution_id == checkup_institution_id,
            BizCheckupInstitutionProject.del_flag == 0,
        )
    )
    if tenant_no:
        q = q.filter(BizCheckupInstitutionProject.tenant_no == tenant_no)
    if user_info:
        from app.core.data_scope import apply_data_scope
        q = apply_data_scope(q, BizCheckupInstitutionProject, user_info, db, tenant_no)
    rows = q.order_by(BizCheckupInstitutionProject.create_time.desc()).all()
    return [_row_to_dict_checkup_institution_project(x) for x in rows]


def create_checkup_institution_project(
    db,
    *,
    checkup_institution_id,
    project_name,
    project_fee=None,
    status="1",
    tenant_no="zkxw",
    order_num=None,
    remark=None,
    user_id=None,
):
    obj = BizCheckupInstitutionProject(
        id=generate_uuid(),
        checkup_institution_id=checkup_institution_id,
        project_name=project_name,
        project_fee=project_fee,
        status=status or "1",
        tenant_no=tenant_no,
        order_num=order_num or 0,
        remark=remark or "",
        create_by=user_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 update_checkup_institution_project(
    db,
    project_id,
    *,
    project_name=None,
    project_fee=None,
    status=None,
    order_num=None,
    remark=None,
    user_id=None,
    tenant_no=None,
):
    q = db.query(BizCheckupInstitutionProject).filter(
        and_(BizCheckupInstitutionProject.id == project_id, BizCheckupInstitutionProject.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizCheckupInstitutionProject.tenant_no == tenant_no)
    obj = q.first()
    if not obj:
        raise ValueError("记录不存在")
    if project_name is not None:
        obj.project_name = project_name
    if project_fee is not None:
        obj.project_fee = project_fee
    if status is not None:
        obj.status = status
    if order_num is not None:
        obj.order_num = order_num
    if remark is not None:
        obj.remark = remark
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def soft_delete_checkup_institution_project(db, project_id, tenant_no=None):
    q = db.query(BizCheckupInstitutionProject).filter(
        and_(BizCheckupInstitutionProject.id == project_id, BizCheckupInstitutionProject.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizCheckupInstitutionProject.tenant_no == tenant_no)
    obj = q.first()
    if not obj:
        raise ValueError("记录不存在")
    obj.del_flag = 1
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()


# ────────────────────────────────────────────
# Entity 14: BizCheckupInstitutionPackage  体检机构项目套餐
# ────────────────────────────────────────────

def _sum_package_item_fees(items: List[Dict[str, Any]]) -> float:
    total = 0.0
    for it in items:
        fee = it.get("project_fee")
        if fee is None and "projectFee" in it:
            fee = it.get("projectFee")
        total += float(fee) if fee is not None else 0.0
    return total


def _row_to_dict_checkup_institution_package(
    row,
    *,
    total_fee: Optional[float] = None,
    item_count: Optional[int] = None,
) -> Dict[str, Any]:
    d = {
        "id": row.id,
        "checkupInstitutionId": row.checkup_institution_id or "",
        "packageName": row.package_name or "",
        "packageCode": row.package_code or "",
        "status": row.status or "1",
        "tenantNo": row.tenant_no or "",
        "orderNum": row.order_num if row.order_num is not None else 0,
        "remark": row.remark or "",
        "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
        "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
    }
    if total_fee is not None:
        d["totalFee"] = total_fee
    if item_count is not None:
        d["itemCount"] = item_count
    return d


def _row_to_dict_checkup_institution_package_item(
    row,
    *,
    source_project_deleted: bool = False,
    source_project_status: Optional[str] = None,
) -> Dict[str, Any]:
    return {
        "id": row.id,
        "packageId": row.package_id or "",
        "checkupInstitutionProjectId": row.checkup_institution_project_id or "",
        "projectName": row.project_name or "",
        "projectFee": float(row.project_fee) if row.project_fee is not None else None,
        "status": source_project_status if source_project_status is not None else "1",
        "sourceProjectDeleted": source_project_deleted,
        "tenantNo": row.tenant_no or "",
        "orderNum": row.order_num if row.order_num is not None else 0,
        "remark": row.remark or "",
        "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
        "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
    }


def checkup_package_code_exists(db, institution_id, code, tenant_no, exclude_id=None):
    q = db.query(BizCheckupInstitutionPackage).filter(
        and_(
            BizCheckupInstitutionPackage.checkup_institution_id == institution_id,
            BizCheckupInstitutionPackage.package_code == code,
            BizCheckupInstitutionPackage.del_flag == 0,
        )
    )
    if tenant_no:
        q = q.filter(BizCheckupInstitutionPackage.tenant_no == tenant_no)
    if exclude_id:
        q = q.filter(BizCheckupInstitutionPackage.id != exclude_id)
    return db.query(q.exists()).scalar()


def _get_checkup_project_for_package(db, project_id, institution_id, tenant_no=None):
    q = db.query(BizCheckupInstitutionProject).filter(
        BizCheckupInstitutionProject.id == project_id,
        BizCheckupInstitutionProject.checkup_institution_id == institution_id,
    )
    if tenant_no:
        q = q.filter(BizCheckupInstitutionProject.tenant_no == tenant_no)
    return q.first()


def _validate_package_items_for_save(
    db,
    institution_id,
    items,
    tenant_no=None,
    existing_item_project_ids=None,
):
    if not items:
        raise ValueError("套餐至少包含一个体检项目")
    existing_ids = set(existing_item_project_ids or [])
    seen = set()
    for it in items:
        pid = it.get("checkup_institution_project_id") or it.get("checkupInstitutionProjectId")
        if not pid:
            raise ValueError("套餐明细缺少机构项目")
        if pid in seen:
            raise ValueError("套餐内不能重复添加同一体检项目")
        seen.add(pid)
        proj = _get_checkup_project_for_package(db, pid, institution_id, tenant_no)
        if not proj:
            raise ValueError("存在无效的体检项目")
        if pid in existing_ids:
            continue
        if proj.del_flag != 0:
            raise ValueError("不能新添加已删除的体检项目")
        if (proj.status or "1") != "1":
            raise ValueError("套餐不能包含已停用的体检项目")


def _list_package_items_raw(db, package_id, tenant_no=None):
    q = db.query(BizCheckupInstitutionPackageItem).filter(
        and_(
            BizCheckupInstitutionPackageItem.package_id == package_id,
            BizCheckupInstitutionPackageItem.del_flag == 0,
        )
    )
    if tenant_no:
        q = q.filter(BizCheckupInstitutionPackageItem.tenant_no == tenant_no)
    return q.order_by(BizCheckupInstitutionPackageItem.order_num.asc(), BizCheckupInstitutionPackageItem.create_time.asc()).all()


def _package_items_to_dicts(db, institution_id, item_rows, tenant_no=None):
    result = []
    for row in item_rows:
        proj = _get_checkup_project_for_package(
            db, row.checkup_institution_project_id, institution_id, tenant_no
        )
        deleted = proj is None or proj.del_flag != 0
        proj_status = (proj.status or "1") if proj else "1"
        result.append(
            _row_to_dict_checkup_institution_package_item(
                row,
                source_project_deleted=deleted,
                source_project_status=proj_status,
            )
        )
    return result


def _insert_package_items(
    db,
    *,
    package_id,
    institution_id,
    items,
    tenant_no,
    user_id=None,
):
    for idx, it in enumerate(items):
        pid = it.get("checkup_institution_project_id") or it.get("checkupInstitutionProjectId")
        proj = _get_checkup_project_for_package(db, pid, institution_id, tenant_no)
        name = (it.get("project_name") or it.get("projectName") or "").strip()
        if not name and proj:
            name = proj.project_name or ""
        fee = it.get("project_fee")
        if fee is None:
            fee = it.get("projectFee")
        obj = BizCheckupInstitutionPackageItem(
            id=generate_uuid(),
            package_id=package_id,
            checkup_institution_project_id=pid,
            project_name=name,
            project_fee=fee,
            tenant_no=tenant_no,
            order_num=it.get("order_num") if it.get("order_num") is not None else it.get("orderNum") or (idx + 1),
            remark=it.get("remark") or "",
            create_by=user_id,
            create_time=datetime.now(),
            update_by=user_id,
            update_time=datetime.now(),
            del_flag=0,
        )
        db.add(obj)


def list_checkup_institution_packages(db, checkup_institution_id, tenant_no=None, user_info=None):
    q = db.query(BizCheckupInstitutionPackage).filter(
        and_(
            BizCheckupInstitutionPackage.checkup_institution_id == checkup_institution_id,
            BizCheckupInstitutionPackage.del_flag == 0,
        )
    )
    if tenant_no:
        q = q.filter(BizCheckupInstitutionPackage.tenant_no == tenant_no)
    if user_info:
        from app.core.data_scope import apply_data_scope
        q = apply_data_scope(q, BizCheckupInstitutionPackage, user_info, db, tenant_no)
    rows = q.order_by(BizCheckupInstitutionPackage.create_time.desc()).all()
    out = []
    for row in rows:
        item_rows = _list_package_items_raw(db, row.id, tenant_no)
        fees = [float(x.project_fee) if x.project_fee is not None else 0.0 for x in item_rows]
        out.append(
            _row_to_dict_checkup_institution_package(
                row,
                total_fee=sum(fees),
                item_count=len(item_rows),
            )
        )
    return out


def get_checkup_institution_package(db, package_id, tenant_no=None):
    q = db.query(BizCheckupInstitutionPackage).filter(
        and_(BizCheckupInstitutionPackage.id == package_id, BizCheckupInstitutionPackage.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizCheckupInstitutionPackage.tenant_no == tenant_no)
    return q.first()


def get_checkup_institution_package_detail(db, package_id, tenant_no=None):
    pkg = get_checkup_institution_package(db, package_id, tenant_no)
    if not pkg:
        return None
    item_rows = _list_package_items_raw(db, package_id, tenant_no)
    items = _package_items_to_dicts(db, pkg.checkup_institution_id, item_rows, tenant_no)
    fees = [float(x.project_fee) if x.project_fee is not None else 0.0 for x in item_rows]
    d = _row_to_dict_checkup_institution_package(pkg, total_fee=sum(fees), item_count=len(item_rows))
    d["items"] = items
    return d


def create_checkup_institution_package(
    db,
    *,
    checkup_institution_id,
    package_name,
    package_code,
    status="1",
    items,
    tenant_no="zkxw",
    order_num=None,
    remark=None,
    user_id=None,
):
    code = (package_code or "").strip()
    if not code:
        raise ValueError("套餐编码不能为空")
    if checkup_package_code_exists(db, checkup_institution_id, code, tenant_no):
        raise ValueError(f"套餐编码 {code} 已存在")
    _validate_package_items_for_save(db, checkup_institution_id, items, tenant_no)
    pkg = BizCheckupInstitutionPackage(
        id=generate_uuid(),
        checkup_institution_id=checkup_institution_id,
        package_name=package_name,
        package_code=code,
        status=status or "1",
        tenant_no=tenant_no,
        order_num=order_num or 0,
        remark=remark or "",
        create_by=user_id,
        create_time=datetime.now(),
        update_by=user_id,
        update_time=datetime.now(),
        del_flag=0,
    )
    db.add(pkg)
    db.flush()
    _insert_package_items(
        db,
        package_id=pkg.id,
        institution_id=checkup_institution_id,
        items=items,
        tenant_no=tenant_no,
        user_id=user_id,
    )
    db.commit()
    db.refresh(pkg)
    return get_checkup_institution_package_detail(db, pkg.id, tenant_no)


def update_checkup_institution_package(
    db,
    package_id,
    *,
    package_name=None,
    package_code=None,
    status=None,
    items=None,
    order_num=None,
    remark=None,
    user_id=None,
    tenant_no=None,
):
    pkg = get_checkup_institution_package(db, package_id, tenant_no)
    if not pkg:
        raise ValueError("记录不存在")
    institution_id = pkg.checkup_institution_id
    if package_code is not None:
        code = package_code.strip()
        if not code:
            raise ValueError("套餐编码不能为空")
        if code != pkg.package_code and checkup_package_code_exists(
            db, institution_id, code, tenant_no, exclude_id=package_id
        ):
            raise ValueError(f"套餐编码 {code} 已存在")
        pkg.package_code = code
    if package_name is not None:
        pkg.package_name = package_name
    if status is not None:
        pkg.status = status
    if order_num is not None:
        pkg.order_num = order_num
    if remark is not None:
        pkg.remark = remark
    if items is not None:
        old_rows = _list_package_items_raw(db, package_id, tenant_no)
        existing_pids = {r.checkup_institution_project_id for r in old_rows}
        _validate_package_items_for_save(
            db,
            institution_id,
            items,
            tenant_no,
            existing_item_project_ids=existing_pids,
        )
        # 全量替换：物理删除旧明细，避免 uk(package_id, project_id) 与软删行冲突
        q_items = db.query(BizCheckupInstitutionPackageItem).filter(
            BizCheckupInstitutionPackageItem.package_id == package_id
        )
        if tenant_no:
            q_items = q_items.filter(BizCheckupInstitutionPackageItem.tenant_no == tenant_no)
        for row in q_items.all():
            db.delete(row)
        db.flush()
        _insert_package_items(
            db,
            package_id=package_id,
            institution_id=institution_id,
            items=items,
            tenant_no=tenant_no or pkg.tenant_no,
            user_id=user_id,
        )
    pkg.update_by = user_id
    pkg.update_time = datetime.now()
    db.add(pkg)
    db.commit()
    return get_checkup_institution_package_detail(db, package_id, tenant_no)


def soft_delete_checkup_institution_package(db, package_id, tenant_no=None):
    pkg = get_checkup_institution_package(db, package_id, tenant_no)
    if not pkg:
        raise ValueError("记录不存在")
    now = datetime.now()
    pkg.del_flag = 1
    pkg.update_time = now
    db.add(pkg)
    for row in _list_package_items_raw(db, package_id, tenant_no):
        row.del_flag = 1
        row.update_time = now
        db.add(row)
    db.commit()


# ────────────────────────────────────────────
# Shared utility
# ────────────────────────────────────────────

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


# ────────────────────────────────────────────
# Entity 15: BizChannelAccount  渠道账户
# ────────────────────────────────────────────

def _row_to_dict_channel_account(row) -> Dict[str, Any]:
    return {
        "id": row.id,
        "channelId": row.channel_id or "",
        "balanceAmount": float(row.balance_amount) if row.balance_amount else 0,
        "tenantNo": row.tenant_no or "",
        "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
        "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
    }


def get_channel_account(db, channel_id, tenant_no=None):
    """只读查询渠道账户，不加行锁"""
    q = db.query(BizChannelAccount).filter(
        and_(BizChannelAccount.channel_id == channel_id, BizChannelAccount.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizChannelAccount.tenant_no == tenant_no)
    return q.first()


def _get_or_create_channel_account(db, channel_id, tenant_no, user_id):
    """获取或创建渠道账户（带行锁 with_for_update）"""
    q = db.query(BizChannelAccount).filter(
        and_(BizChannelAccount.channel_id == channel_id, BizChannelAccount.del_flag == 0)
    ).with_for_update()
    if tenant_no:
        q = q.filter(BizChannelAccount.tenant_no == tenant_no)
    obj = q.first()
    if obj:
        return obj
    obj = BizChannelAccount(
        id=generate_uuid(),
        channel_id=channel_id,
        balance_amount=0,
        tenant_no=tenant_no or "zkxw",
        order_num=0,
        remark="",
        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.flush()
    return obj


# ────────────────────────────────────────────
# Entity 16: BizChannelProductQuota  渠道产品份数
# ────────────────────────────────────────────

def _row_to_dict_channel_product_quota(row) -> Dict[str, Any]:
    return {
        "id": row.id,
        "channelId": row.channel_id or "",
        "projectCode": row.project_code or "",
        "projectName": row.project_name or "",
        "cellTypeCode": row.cell_type_code or "",
        "cellTypeName": row.cell_type_name or "",
        "specText": row.spec_text or "",
        "totalQty": float(row.total_qty) if row.total_qty else 0,
        "usedQty": float(row.used_qty) if row.used_qty else 0,
        "remainingQty": float(row.remaining_qty) if row.remaining_qty else 0,
        "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
        "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
    }


def _upsert_channel_product_quota(
    db,
    channel_id,
    project_code,
    project_name,
    cell_type_code,
    cell_type_name,
    spec_text,
    add_qty,
    tenant_no,
    user_id,
):
    """增加份数：已存在则累加，不存在则新建"""
    q = db.query(BizChannelProductQuota).filter(
        and_(
            BizChannelProductQuota.channel_id == channel_id,
            BizChannelProductQuota.project_code == (project_code or ""),
            BizChannelProductQuota.cell_type_code == (cell_type_code or ""),
            BizChannelProductQuota.spec_text == (spec_text or ""),
            BizChannelProductQuota.del_flag == 0,
        )
    )
    if tenant_no:
        q = q.filter(BizChannelProductQuota.tenant_no == tenant_no)
    obj = q.first()
    if obj:
        obj.total_qty = (float(obj.total_qty) if obj.total_qty else 0) + add_qty
        obj.remaining_qty = (float(obj.remaining_qty) if obj.remaining_qty else 0) + add_qty
        obj.update_by = user_id
        obj.update_time = datetime.now()
        db.add(obj)
        db.flush()
        return obj
    obj = BizChannelProductQuota(
        id=generate_uuid(),
        channel_id=channel_id,
        project_code=project_code or "",
        project_name=project_name or "",
        cell_type_code=cell_type_code or "",
        cell_type_name=cell_type_name or "",
        spec_text=spec_text or "",
        total_qty=add_qty,
        used_qty=0,
        remaining_qty=add_qty,
        tenant_no=tenant_no or "zkxw",
        order_num=0,
        remark="",
        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.flush()
    return obj


def _reduce_channel_product_quota(db, channel_id, project_code, cell_type_code, spec_text, reduce_qty, user_id):
    """扣减份数：验证 remaining_qty >= reduce_qty"""
    q = db.query(BizChannelProductQuota).filter(
        and_(
            BizChannelProductQuota.channel_id == channel_id,
            BizChannelProductQuota.project_code == (project_code or ""),
            BizChannelProductQuota.cell_type_code == (cell_type_code or ""),
            BizChannelProductQuota.spec_text == (spec_text or ""),
            BizChannelProductQuota.del_flag == 0,
        )
    )
    obj = q.first()
    if not obj:
        raise ValueError("份数记录不存在")
    remaining = float(obj.remaining_qty) if obj.remaining_qty else 0
    if remaining < reduce_qty:
        raise ValueError(f"剩余份数不足，当前剩余 {remaining}，需扣减 {reduce_qty}")
    obj.total_qty = (float(obj.total_qty) if obj.total_qty else 0) - reduce_qty
    obj.remaining_qty = remaining - reduce_qty
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.flush()
    return obj


def list_channel_product_quotas(db, channel_id, tenant_no=None):
    """查询渠道下所有产品份数"""
    q = db.query(BizChannelProductQuota).filter(
        and_(
            BizChannelProductQuota.channel_id == channel_id,
            BizChannelProductQuota.del_flag == 0,
        )
    )
    if tenant_no:
        q = q.filter(BizChannelProductQuota.tenant_no == tenant_no)
    rows = q.order_by(BizChannelProductQuota.create_time.desc()).all()
    return [_row_to_dict_channel_product_quota(x) for x in rows]


# ────────────────────────────────────────────
# Entity 17: BizChannelServiceOrder  渠道服务订单
# ────────────────────────────────────────────

def _calc_gmcp_purchase_total_from_db(db, order_id, tenant_no=None) -> float:
    """从订单产品持仓汇总购买总额（协议单价 × 累计购买份数）。"""
    q = db.query(BizChannelServiceOrderProduct).filter(
        and_(
            BizChannelServiceOrderProduct.service_order_id == order_id,
            BizChannelServiceOrderProduct.del_flag == 0,
        )
    )
    if tenant_no:
        q = q.filter(BizChannelServiceOrderProduct.tenant_no == tenant_no)
    total = 0.0
    for prod in q.all():
        price = float(prod.agreement_unit_price or 0)
        qty = float(prod.total_buy_qty or 0)
        total += price * qty
    return total


def _batch_gmcp_purchase_totals(db, order_ids, tenant_no=None) -> Dict[str, float]:
    if not order_ids:
        return {}
    amount_expr = BizChannelServiceOrderProduct.agreement_unit_price * BizChannelServiceOrderProduct.total_buy_qty
    q = db.query(
        BizChannelServiceOrderProduct.service_order_id,
        func.coalesce(func.sum(amount_expr), 0),
    ).filter(
        BizChannelServiceOrderProduct.del_flag == 0,
        BizChannelServiceOrderProduct.service_order_id.in_(order_ids),
    )
    if tenant_no:
        q = q.filter(BizChannelServiceOrderProduct.tenant_no == tenant_no)
    return {row[0]: float(row[1] or 0) for row in q.group_by(BizChannelServiceOrderProduct.service_order_id).all()}


def _calc_gmcp_current_holding_amount_from_db(db, order_id, tenant_no=None) -> float:
    """从订单产品持仓汇总当前余额（协议单价 × 当前持有份数）。"""
    q = db.query(BizChannelServiceOrderProduct).filter(
        and_(
            BizChannelServiceOrderProduct.service_order_id == order_id,
            BizChannelServiceOrderProduct.del_flag == 0,
        )
    )
    if tenant_no:
        q = q.filter(BizChannelServiceOrderProduct.tenant_no == tenant_no)
    total = 0.0
    for prod in q.all():
        price = float(prod.agreement_unit_price or 0)
        qty = float(prod.current_qty or 0)
        total += price * qty
    return total


def _batch_gmcp_current_holding_amounts(db, order_ids, tenant_no=None) -> Dict[str, float]:
    if not order_ids:
        return {}
    amount_expr = BizChannelServiceOrderProduct.agreement_unit_price * BizChannelServiceOrderProduct.current_qty
    q = db.query(
        BizChannelServiceOrderProduct.service_order_id,
        func.coalesce(func.sum(amount_expr), 0),
    ).filter(
        BizChannelServiceOrderProduct.del_flag == 0,
        BizChannelServiceOrderProduct.service_order_id.in_(order_ids),
    )
    if tenant_no:
        q = q.filter(BizChannelServiceOrderProduct.tenant_no == tenant_no)
    return {row[0]: float(row[1] or 0) for row in q.group_by(BizChannelServiceOrderProduct.service_order_id).all()}


def _resolve_service_order_purchase_total(row, db=None, tenant_no=None, purchase_total_override=None) -> float:
    stored = float(row.purchase_total_amount) if row.purchase_total_amount else 0
    if (row.order_type or "") != "gmcp":
        return stored
    if purchase_total_override is not None:
        return float(purchase_total_override)
    if db:
        computed = _calc_gmcp_purchase_total_from_db(db, row.id, tenant_no or row.tenant_no)
        if computed > 0 or not row.purchase_total_amount:
            return computed
    return stored


def _resolve_service_order_current_balance(
    row, db=None, tenant_no=None, current_balance_override=None
) -> float:
    if (row.order_type or "") == "gmcp":
        if current_balance_override is not None:
            return float(current_balance_override)
        if db:
            return _calc_gmcp_current_holding_amount_from_db(db, row.id, tenant_no or row.tenant_no)
        return 0.0
    return float(row.current_balance) if row.current_balance else 0


def _row_to_dict_service_order(
    row,
    db=None,
    tenant_no=None,
    *,
    purchase_total_override=None,
    current_balance_override=None,
) -> Dict[str, Any]:
    tn = tenant_no or row.tenant_no
    purchase_total = _resolve_service_order_purchase_total(
        row, db, tn, purchase_total_override=purchase_total_override
    )
    current_balance = _resolve_service_order_current_balance(
        row, db, tn, current_balance_override=current_balance_override
    )
    d = {
        "id": row.id,
        "channelId": row.channel_id or "",
        "channelName": row.channel_name or "",
        "orderName": row.order_name or "",
        "deptName": row.dept_name or "",
        "orderType": row.order_type or "",
        "prepayAmount": float(row.prepay_amount) if row.prepay_amount else 0,
        "purchaseTotalAmount": purchase_total,
        "orderStatus": row.order_status or "",
        "channelAgreementId": row.channel_agreement_id or "",
        "totalRechargeAmount": float(row.total_recharge_amount) if row.total_recharge_amount else 0,
        "totalRefundAmount": float(row.total_refund_amount) if row.total_refund_amount else 0,
        "currentBalance": current_balance,
        "balanceBefore": float(row.balance_before) if row.balance_before else 0,
        "balanceAfter": float(row.balance_after) if row.balance_after else 0,
        "financeSubmitBy": row.finance_submit_by or "",
        "financeSubmitTime": row.finance_submit_time.strftime("%Y-%m-%d %H:%M:%S") if row.finance_submit_time else "",
        "financeSubmitRemark": row.finance_submit_remark or "",
        "financeSubmitAttachments": row.finance_submit_attachments or "",
        "financeSubmitAttachmentFiles": urls_to_attachment_list(row.finance_submit_attachments, db, tn) if db and row.finance_submit_attachments else [],
        "financeConfirmTime": row.finance_confirm_time.strftime("%Y-%m-%d %H:%M:%S") if row.finance_confirm_time else "",
        "financeConfirmAttachments": row.finance_confirm_attachments or "",
        "financeConfirmAttachmentFiles": urls_to_attachment_list(row.finance_confirm_attachments, db, tn) if db and row.finance_confirm_attachments else [],
        "financeRemark": row.finance_remark or "",
        "refundApplyBy": row.refund_apply_by or "",
        "refundApplyTime": row.refund_apply_time.strftime("%Y-%m-%d %H:%M:%S") if row.refund_apply_time else "",
        "refundAmount": float(row.refund_amount) if row.refund_amount else 0,
        "refundRemark": row.refund_remark or "",
        "refundAttachments": row.refund_attachments or "",
        "refundAttachmentFiles": urls_to_attachment_list(row.refund_attachments, db, tn) if db and row.refund_attachments else [],
        "refundFinanceConfirmTime": row.refund_finance_confirm_time.strftime("%Y-%m-%d %H:%M:%S") if row.refund_finance_confirm_time else "",
        "refundFinanceConfirmAttachments": row.refund_finance_confirm_attachments or "",
        "refundFinanceConfirmAttachmentFiles": urls_to_attachment_list(row.refund_finance_confirm_attachments, db, tn) if db and row.refund_finance_confirm_attachments else [],
        "refundFinanceRemark": row.refund_finance_remark or "",
        "tenantNo": row.tenant_no or "",
        "remark": row.remark or "",
        "createBy": row.create_by or "",
        "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
        "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
    }
    return d


def _row_to_dict_service_order_product(row) -> Dict[str, Any]:
    return {
        "id": row.id,
        "serviceOrderId": row.service_order_id or "",
        "projectCode": row.project_code or "",
        "projectName": row.project_name or "",
        "cellType": row.cell_type or "",
        "specText": row.spec_text or "",
        "agreementUnitPrice": float(row.agreement_unit_price) if row.agreement_unit_price else 0,
        "buyQty": float(row.buy_qty) if row.buy_qty else 0,
        "buyAmount": float(row.buy_amount) if row.buy_amount else 0,
        "totalBuyQty": float(row.total_buy_qty) if row.total_buy_qty else 0,
        "totalUnsubscribeQty": float(row.total_unsubscribe_qty) if row.total_unsubscribe_qty else 0,
        "currentQty": float(row.current_qty) if row.current_qty else 0,
        "totalUnsubscribeAmount": float(row.total_unsubscribe_amount) if row.total_unsubscribe_amount else 0,
    }


def _normalize_txn_product_item(p: Dict[str, Any]) -> Dict[str, Any]:
    """流水 products_json 项统一为 camelCase（入库时为 snake_case）。"""
    qty = p.get("qty")
    if qty is None:
        qty = p.get("buyQty") or p.get("buy_qty")
    unit_price = p.get("agreementUnitPrice") or p.get("agreement_unit_price") or 0
    qty_val = float(qty or 0)
    amount = p.get("amount")
    if amount is None and qty_val:
        amount = float(unit_price or 0) * qty_val
    return {
        "projectCode": p.get("projectCode") or p.get("project_code") or "",
        "projectName": p.get("projectName") or p.get("project_name") or "",
        "cellType": p.get("cellType") or p.get("cell_type") or "",
        "specText": p.get("specText") or p.get("spec_text") or "",
        "agreementUnitPrice": float(unit_price or 0),
        "qty": qty_val,
        "amount": float(amount or 0),
    }


def _enrich_txn_products_from_order(
    db,
    service_order_id: str,
    products: List[Dict[str, Any]],
    tenant_no=None,
) -> List[Dict[str, Any]]:
    """退订等流水可能仅存 projectCode，从订单持仓补全项目名与规格。"""
    if not db or not service_order_id or not products:
        return products
    holdings = get_service_order_products(db, service_order_id, tenant_no)
    if not holdings:
        return products
    by_code_spec = {
        (h.get("projectCode") or "", h.get("specText") or ""): h for h in holdings
    }
    by_code: Dict[str, List[Dict[str, Any]]] = {}
    for h in holdings:
        code = h.get("projectCode") or ""
        by_code.setdefault(code, []).append(h)
    enriched: List[Dict[str, Any]] = []
    for p in products:
        item = dict(p)
        code = item.get("projectCode") or ""
        spec = item.get("specText") or ""
        holding = by_code_spec.get((code, spec))
        if not holding and code:
            matches = by_code.get(code, [])
            if len(matches) == 1:
                holding = matches[0]
        if holding:
            if not (item.get("projectName") or "").strip():
                item["projectName"] = holding.get("projectName") or ""
            if not (item.get("specText") or "").strip():
                item["specText"] = holding.get("specText") or ""
            if not (item.get("cellType") or "").strip():
                item["cellType"] = holding.get("cellType") or ""
        enriched.append(item)
    return enriched


def _row_to_dict_service_order_txn(row, db=None, tenant_no=None) -> Dict[str, Any]:
    tn = tenant_no or row.tenant_no
    d = {
        "id": row.id,
        "serviceOrderId": row.service_order_id or "",
        "txnType": row.txn_type or "",
        "amount": float(row.amount) if row.amount else 0,
        "txnStatus": row.txn_status or "",
        "productsJson": row.products_json or "",
        "remark": row.remark or "",
        "refundAttachments": row.refund_attachments or "",
        "refundAttachmentFiles": urls_to_attachment_list(row.refund_attachments, db, tn) if db and row.refund_attachments else [],
        "financeSubmitBy": row.finance_submit_by or "",
        "financeSubmitTime": row.finance_submit_time.strftime("%Y-%m-%d %H:%M:%S") if row.finance_submit_time else "",
        "financeSubmitRemark": row.finance_submit_remark or "",
        "financeSubmitAttachments": row.finance_submit_attachments or "",
        "financeSubmitAttachmentFiles": urls_to_attachment_list(row.finance_submit_attachments, db, tn) if db and row.finance_submit_attachments else [],
        "financeConfirmTime": row.finance_confirm_time.strftime("%Y-%m-%d %H:%M:%S") if row.finance_confirm_time else "",
        "financeConfirmBy": row.finance_confirm_by or "",
        "financeConfirmAttachments": row.finance_confirm_attachments or "",
        "financeConfirmAttachmentFiles": urls_to_attachment_list(row.finance_confirm_attachments, db, tn) if db and row.finance_confirm_attachments else [],
        "financeRemark": row.finance_remark or "",
        "voidBy": row.void_by or "",
        "voidTime": row.void_time.strftime("%Y-%m-%d %H:%M:%S") if row.void_time else "",
        "balanceBefore": float(row.balance_before) if row.balance_before else 0,
        "balanceAfter": float(row.balance_after) if row.balance_after else 0,
        "tenantNo": row.tenant_no or "",
        "createBy": row.create_by or "",
        "createTime": row.create_time.strftime("%Y-%m-%d %H:%M:%S") if row.create_time else "",
        "updateTime": row.update_time.strftime("%Y-%m-%d %H:%M:%S") if row.update_time else "",
    }
    if row.products_json:
        try:
            raw_products = json.loads(row.products_json)
            products = [_normalize_txn_product_item(x) for x in raw_products]
            if db and row.service_order_id:
                products = _enrich_txn_products_from_order(
                    db, row.service_order_id, products, tenant_no
                )
            d["products"] = products
        except Exception:
            d["products"] = []
    else:
        d["products"] = []
    return d


def _service_order_txn_to_response_dict(
    db,
    txn,
    tenant_no=None,
    *,
    order=None,
    name_map=None,
) -> Dict[str, Any]:
    """流水响应字典：含订单渠道信息与提交人/创建人昵称。"""
    d = _row_to_dict_service_order_txn(txn, db, tenant_no)
    if order is None and txn.service_order_id:
        order = get_service_order(db, txn.service_order_id, tenant_no)
    if order:
        d["channelName"] = order.channel_name or ""
        d["orderName"] = order.order_name or ""
        d["deptName"] = order.dept_name or ""
        d["orderType"] = order.order_type or ""
    confirm_uid = txn.finance_confirm_by or (
        txn.update_by
        if (txn.txn_status or "") == "yqr" and txn.finance_confirm_time and not txn.finance_confirm_by
        else None
    )
    void_uid = txn.void_by or (
        txn.update_by
        if (txn.txn_status or "") == "yzf"
        and txn.finance_submit_time
        and not txn.void_by
        else None
    )
    if name_map is None:
        user_ids = [x for x in (txn.create_by, txn.finance_submit_by, confirm_uid, void_uid) if x]
        name_map = _user_nick_name_map(db, user_ids) if db and user_ids else {}
    d["createByName"] = name_map.get(txn.create_by or "", "") or (txn.create_by or "")
    d["financeSubmitByName"] = name_map.get(txn.finance_submit_by or "", "") or (
        txn.finance_submit_by or ""
    )
    d["financeConfirmByName"] = name_map.get(confirm_uid or "", "") or (confirm_uid or "")
    if not d.get("voidTime") and (txn.txn_status or "") == "yzf" and txn.update_time:
        d["voidTime"] = txn.update_time.strftime("%Y-%m-%d %H:%M:%S")
    d["voidByName"] = name_map.get(void_uid or "", "") or (void_uid or "")
    if not (d.get("financeSubmitRemark") or "").strip() and (d.get("remark") or "").strip():
        d["financeSubmitRemark"] = (d.get("remark") or "").strip()
    _enrich_finance_operate_fields(d, txn)
    return d


def _enrich_finance_operate_fields(d: Dict[str, Any], txn) -> None:
    """财务页「已处理」展示：操作类型、操作人、操作时间。"""
    status = txn.txn_status or ""
    if status == "yqr" and txn.finance_confirm_time:
        d["financeOperateType"] = "confirm"
        d["financeOperateTypeLabel"] = "财务确认"
        d["financeOperateByName"] = d.get("financeConfirmByName") or d.get("financeConfirmBy") or ""
        d["financeOperateTime"] = d.get("financeConfirmTime") or ""
    elif status == "yzf" and txn.finance_submit_time:
        d["financeOperateType"] = "void"
        d["financeOperateTypeLabel"] = "财务作废"
        d["financeOperateByName"] = d.get("voidByName") or d.get("voidBy") or ""
        d["financeOperateTime"] = d.get("voidTime") or ""
    else:
        d["financeOperateType"] = ""
        d["financeOperateTypeLabel"] = ""
        d["financeOperateByName"] = ""
        d["financeOperateTime"] = ""


def list_channel_service_orders(
    db,
    *,
    page=1,
    page_size=10,
    order_type=None,
    order_status=None,
    channel_id=None,
    keyword=None,
    create_time_start=None,
    create_time_end=None,
    tenant_no=None,
    user_info=None,
):
    def _filtered():
        q0 = db.query(BizChannelServiceOrder).filter(BizChannelServiceOrder.del_flag == 0)
        if tenant_no:
            q0 = q0.filter(BizChannelServiceOrder.tenant_no == tenant_no)
        ot = (order_type or "").strip()
        if ot:
            q0 = q0.filter(BizChannelServiceOrder.order_type == ot)
        # Support comma-separated order_status
        os_raw = (order_status or "").strip()
        if os_raw:
            statuses = [s.strip() for s in os_raw.split(",") if s.strip()]
            if statuses:
                q0 = q0.filter(BizChannelServiceOrder.order_status.in_(statuses))
        if channel_id:
            q0 = q0.filter(BizChannelServiceOrder.channel_id == channel_id)
        kw = (keyword or "").strip()
        if kw:
            like = f"%{kw}%"
            q0 = q0.filter(
                or_(
                    BizChannelServiceOrder.channel_name.like(like),
                    BizChannelServiceOrder.order_name.like(like),
                    BizChannelServiceOrder.dept_name.like(like),
                )
            )
        if create_time_start and create_time_start.strip():
            q0 = q0.filter(
                BizChannelServiceOrder.create_time
                >= _normalize_date_bound(create_time_start.strip())
            )
        if create_time_end and create_time_end.strip():
            q0 = q0.filter(
                BizChannelServiceOrder.create_time
                <= _normalize_date_bound(create_time_end.strip(), end_of_day=True)
            )
        if user_info:
            from app.core.data_scope import apply_data_scope
            q0 = apply_data_scope(q0, BizChannelServiceOrder, user_info, db, tenant_no)
        return q0

    total = _filtered().order_by(None).with_entities(func.count(BizChannelServiceOrder.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        _filtered()
        .order_by(BizChannelServiceOrder.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    user_ids: List[str] = []
    for x in rows:
        if x.create_by:
            user_ids.append(x.create_by)
        if x.finance_submit_by:
            user_ids.append(x.finance_submit_by)
    name_map = _user_nick_name_map(db, user_ids) if db else {}
    gmcp_ids = [x.id for x in rows if (x.order_type or "") == "gmcp"]
    purchase_total_map = _batch_gmcp_purchase_totals(db, gmcp_ids, tenant_no) if db and gmcp_ids else {}
    holding_amount_map = (
        _batch_gmcp_current_holding_amounts(db, gmcp_ids, tenant_no) if db and gmcp_ids else {}
    )
    return [
        {
            **_row_to_dict_service_order(
                x,
                db,
                tenant_no,
                purchase_total_override=purchase_total_map.get(x.id),
                current_balance_override=holding_amount_map.get(x.id),
            ),
            "createByName": name_map.get(x.create_by or "", "") or (x.create_by or ""),
            "financeSubmitByName": name_map.get(x.finance_submit_by or "", "") or (x.finance_submit_by or ""),
        }
        for x in rows
    ], int(total)


def get_service_order(db, order_id, tenant_no=None):
    q = db.query(BizChannelServiceOrder).filter(
        and_(BizChannelServiceOrder.id == order_id, BizChannelServiceOrder.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizChannelServiceOrder.tenant_no == tenant_no)
    return q.first()


def get_service_order_products(db, service_order_id, tenant_no=None):
    q = db.query(BizChannelServiceOrderProduct).filter(
        and_(
            BizChannelServiceOrderProduct.service_order_id == service_order_id,
            BizChannelServiceOrderProduct.del_flag == 0,
        )
    )
    if tenant_no:
        q = q.filter(BizChannelServiceOrderProduct.tenant_no == tenant_no)
    rows = q.order_by(BizChannelServiceOrderProduct.create_time.asc()).all()
    return [_row_to_dict_service_order_product(x) for x in rows]


# ────────────────────────────────────────────
# Service Order: Create / Update / Delete
# ────────────────────────────────────────────

def _calc_gmcp_purchase_total(products):
    total = 0.0
    for p in products:
        price = float(p.get("agreementUnitPrice") or p.get("agreement_unit_price") or 0)
        qty = float(p.get("buyQty") or p.get("buy_qty") or 0)
        total += price * qty
    return total


def _gmcp_has_purchase_lock(db, order_id, tenant_no=None) -> bool:
    """是否存在购买记录（未作废 purchase 流水，或历史建单写入的产品行）。"""
    q = db.query(BizChannelServiceOrderTxn).filter(
        and_(
            BizChannelServiceOrderTxn.service_order_id == order_id,
            BizChannelServiceOrderTxn.txn_type == "purchase",
            BizChannelServiceOrderTxn.txn_status != "yzf",
            BizChannelServiceOrderTxn.del_flag == 0,
        )
    )
    if tenant_no:
        q = q.filter(BizChannelServiceOrderTxn.tenant_no == tenant_no)
    if q.first():
        return True
    pq = db.query(BizChannelServiceOrderProduct).filter(
        and_(
            BizChannelServiceOrderProduct.service_order_id == order_id,
            BizChannelServiceOrderProduct.del_flag == 0,
        )
    )
    if tenant_no:
        pq = pq.filter(BizChannelServiceOrderProduct.tenant_no == tenant_no)
    for row in pq.all():
        if float(row.buy_qty or 0) > 0 or float(row.total_buy_qty or 0) > 0:
            return True
    return False


def gmcp_agreement_editable(db, order, tenant_no=None) -> bool:
    if order.order_type != "gmcp" or order.order_status == "yzf":
        return False
    return not _gmcp_has_purchase_lock(db, order.id, tenant_no or order.tenant_no)


def _resolve_agreement_price_cell_type(db, channel_agreement_id, project_code, spec_text, cell_type_fallback=""):
    price_row = (
        db.query(BizChannelAgreementProductPrice)
        .filter(
            and_(
                BizChannelAgreementProductPrice.channel_agreement_id == channel_agreement_id,
                BizChannelAgreementProductPrice.project_code == (project_code or ""),
                BizChannelAgreementProductPrice.spec_text == (spec_text or ""),
                BizChannelAgreementProductPrice.del_flag == 0,
            )
        )
    ).first()
    if price_row:
        return price_row.cell_type_code or "", price_row.cell_type_name or cell_type_fallback
    return "", cell_type_fallback


def _save_gmcp_product_rows(
    db,
    order_id,
    products,
    tenant_no,
    user_id,
    now=None,
):
    """仅保存购买订单产品明细，不调整份数库存"""
    now = now or datetime.now()
    for p in products:
        project_code = p.get("projectCode") or p.get("project_code") or ""
        project_name = p.get("projectName") or p.get("project_name") or ""
        cell_type = p.get("cellType") or p.get("cell_type") or ""
        spec_text = p.get("specText") or p.get("spec_text") or ""
        agreement_unit_price = float(p.get("agreementUnitPrice") or p.get("agreement_unit_price") or 0)
        buy_qty = float(p.get("buyQty") or p.get("buy_qty") or 0)
        buy_amount = float(p.get("buyAmount") or p.get("buy_amount") or agreement_unit_price * buy_qty)

        prod_row = BizChannelServiceOrderProduct(
            id=generate_uuid(),
            service_order_id=order_id,
            project_code=project_code,
            project_name=project_name,
            cell_type=cell_type,
            spec_text=spec_text,
            agreement_unit_price=agreement_unit_price,
            buy_qty=buy_qty,
            buy_amount=buy_amount,
            tenant_no=tenant_no or "zkxw",
            order_num=0,
            remark="",
            create_by=user_id,
            create_dept_id="",
            create_time=now,
            update_by=user_id,
            update_time=now,
            del_flag=0,
        )
        db.add(prod_row)
    db.flush()


def _replace_gmcp_product_rows(db, order_id, products, tenant_no, user_id):
    now = datetime.now()
    db.query(BizChannelServiceOrderProduct).filter(
        and_(
            BizChannelServiceOrderProduct.service_order_id == order_id,
            BizChannelServiceOrderProduct.del_flag == 0,
        )
    ).update({"del_flag": 1, "update_time": now}, synchronize_session=False)
    _save_gmcp_product_rows(db, order_id, products, tenant_no, user_id, now=now)


def _apply_gmcp_order_quotas(db, order, tenant_no, user_id):
    """财务确认收款后，按订单明细增加渠道产品份数"""
    products = get_service_order_products(db, order.id, tenant_no)
    for prod in products:
        project_code = prod.get("projectCode") or ""
        spec_text = prod.get("specText") or ""
        cell_type_code, cell_type_name = _resolve_agreement_price_cell_type(
            db,
            order.channel_agreement_id,
            project_code,
            spec_text,
            prod.get("cellType") or "",
        )
        _upsert_channel_product_quota(
            db,
            channel_id=order.channel_id,
            project_code=project_code,
            project_name=prod.get("projectName") or "",
            cell_type_code=cell_type_code,
            cell_type_name=cell_type_name,
            spec_text=spec_text,
            add_qty=float(prod.get("buyQty") or 0),
            tenant_no=tenant_no or order.tenant_no,
            user_id=user_id,
        )


def _get_valid_channel_agreement(db, channel_agreement_id, channel_id, tenant_no=None):
    """校验渠道协议存在、属于当前渠道且已生效"""
    aid = (channel_agreement_id or "").strip()
    if not aid:
        raise ValueError("请选择渠道协议")
    q = db.query(BizChannelAgreement).filter(
        and_(
            BizChannelAgreement.id == aid,
            BizChannelAgreement.channel_id == channel_id,
            BizChannelAgreement.del_flag == 0,
        )
    )
    if tenant_no:
        q = q.filter(BizChannelAgreement.tenant_no == tenant_no)
    agreement = q.first()
    if not agreement:
        raise ValueError("渠道协议不存在或不属于当前渠道")
    if (agreement.agreement_status or "").strip() != "sxz":
        raise ValueError("协议未生效，无法使用")
    return agreement


def create_channel_service_order(
    db,
    *,
    channel_id,
    order_type,
    channel_name=None,
    order_name=None,
    dept_name=None,
    prepay_amount=None,
    channel_agreement_id=None,
    products=None,
    tenant_no=None,
    remark=None,
    user_id=None,
):
    # Validate channel exists
    channel = get_channel(db, channel_id, tenant_no)
    if not channel:
        raise ValueError("渠道不存在")

    now = datetime.now()
    balance_before = None
    balance_after = None
    purchase_total_amount = None
    order_status = "dsx"

    if order_type == "yfk":
        # 预存款订单：创建后为待生效，充值金额通过「发起充值」流水录入
        _get_valid_channel_agreement(db, channel_agreement_id, channel_id, tenant_no)
        order_status = "dsx"
        prepay_amount = None

    elif order_type == "gmcp":
        # 购买型：建单仅绑定渠道与协议；产品通过 purchase 流水录入并经财务确认
        if not channel_agreement_id:
            raise ValueError("请选择渠道协议")
        _get_valid_channel_agreement(db, channel_agreement_id, channel_id, tenant_no)
        purchase_total_amount = _calc_gmcp_purchase_total(products) if products else None
        order_status = "dsx"
    else:
        raise ValueError(f"不支持的订单类型: {order_type}")

    order_id = generate_uuid()
    order = BizChannelServiceOrder(
        id=order_id,
        channel_id=channel_id,
        channel_name=channel_name or channel.channel_name or "",
        order_name=(order_name or "").strip() or None,
        dept_name=dept_name or "",
        order_type=order_type,
        prepay_amount=None,
        purchase_total_amount=purchase_total_amount,
        order_status=order_status,
        channel_agreement_id=channel_agreement_id,
        balance_before=balance_before,
        balance_after=balance_after,
        tenant_no=tenant_no or "zkxw",
        order_num=0,
        remark=remark or "",
        create_by=user_id,
        create_dept_id="",
        create_time=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(order)
    db.flush()

    db.commit()
    db.refresh(order)
    return order


def update_channel_service_order(db, order_id, *, tenant_no=None, user_id=None, products=None, **kwargs):
    """更新订单基本信息；gmcp 产品仅通过 purchase 流水维护"""
    order = get_service_order(db, order_id, tenant_no)
    if not order:
        raise ValueError("订单不存在")
    if order.order_status == "yzf":
        raise ValueError("已作废订单不可编辑")
    if kwargs.get("channel_id"):
        raise ValueError("渠道不可修改")
    if products is not None and order.order_type == "gmcp":
        raise ValueError("请通过「购买产品」操作录入产品明细")

    if order.order_type in ("yfk", "gmcp"):
        if order.order_status not in ("dsx", "sxz"):
            raise ValueError("当前状态不允许编辑")
    else:
        raise ValueError("不支持的订单类型")

    if "dept_name" in kwargs and kwargs["dept_name"] is not None:
        order.dept_name = kwargs["dept_name"]
    if "order_name" in kwargs and kwargs["order_name"] is not None:
        order.order_name = (kwargs["order_name"] or "").strip() or None
    if "remark" in kwargs and kwargs["remark"] is not None:
        order.remark = kwargs["remark"]
    if order.order_type == "yfk":
        if "channel_agreement_id" in kwargs and kwargs["channel_agreement_id"] is not None:
            new_agreement_id = kwargs["channel_agreement_id"]
            if new_agreement_id != (order.channel_agreement_id or ""):
                if order.order_status != "dsx":
                    raise ValueError("已有业务流水，不可修改渠道协议")
                _get_valid_channel_agreement(
                    db, new_agreement_id, order.channel_id, tenant_no
                )
                order.channel_agreement_id = new_agreement_id
    elif order.order_type == "gmcp":
        if "channel_agreement_id" in kwargs and kwargs["channel_agreement_id"] is not None:
            new_agreement_id = kwargs["channel_agreement_id"]
            if new_agreement_id != (order.channel_agreement_id or ""):
                if _gmcp_has_purchase_lock(db, order_id, tenant_no):
                    raise ValueError("已有购买记录，不可修改渠道协议")
                _get_valid_channel_agreement(
                    db, new_agreement_id, order.channel_id, tenant_no
                )
                order.channel_agreement_id = new_agreement_id
    order.update_by = user_id
    order.update_time = datetime.now()
    db.add(order)
    db.commit()
    db.refresh(order)
    return order


def soft_delete_service_order(db, order_id, tenant_no=None):
    """仅允许待生效状态下删除订单"""
    order = get_service_order(db, order_id, tenant_no)
    if not order:
        raise ValueError("订单不存在")
    if order.order_status != "dsx":
        raise ValueError("仅待生效订单可删除")
    order.del_flag = 1
    order.update_time = datetime.now()
    db.add(order)
    db.commit()


# ────────────────────────────────────────────
# Service Order: Status Transitions
# ────────────────────────────────────────────

def submit_finance_confirm(
    db,
    order_id,
    *,
    finance_submit_remark=None,
    finance_submit_attachments=None,
    tenant_no=None,
    user_id=None,
):
    """发起财务确认：type=yfk/gmcp + status=dsx → dqr"""
    order = get_service_order(db, order_id, tenant_no)
    if not order:
        raise ValueError("订单不存在")
    if order.order_type not in ("yfk", "gmcp"):
        raise ValueError("仅预存款或购买产品订单可发起财务确认")
    if order.order_status != "dsx":
        raise ValueError("仅待生效状态可发起财务确认")

    now = datetime.now()
    order.order_status = "dqr"
    order.finance_submit_remark = finance_submit_remark or ""
    order.finance_submit_attachments = finance_submit_attachments or ""
    order.finance_submit_time = now
    order.finance_submit_by = user_id
    order.update_by = user_id
    order.update_time = now
    db.add(order)
    db.commit()
    db.refresh(order)
    return order


def confirm_receipt(db, order_id, *, finance_remark=None, finance_confirm_attachments=None, tenant_no=None, user_id=None):
    """确认到账：yfk dqr→yqr 增加余额；gmcp dqr→sxz 增加产品份数"""
    order = get_service_order(db, order_id, tenant_no)
    if not order:
        raise ValueError("订单不存在")
    if order.order_type not in ("yfk", "gmcp"):
        raise ValueError("仅预存款或购买产品订单可确认到账")
    if order.order_status != "dqr":
        raise ValueError("仅待确认状态可确认到账")

    now = datetime.now()

    if order.order_type == "yfk":
        account = _get_or_create_channel_account(db, order.channel_id, tenant_no or order.tenant_no, user_id)
        balance_before = float(account.balance_amount) if account.balance_amount else 0
        add_amount = float(order.prepay_amount) if order.prepay_amount else 0
        account.balance_amount = balance_before + add_amount
        balance_after = float(account.balance_amount)
        account.update_by = user_id
        account.update_time = now
        db.add(account)
        db.flush()

        order.order_status = "yqr"
        order.balance_before = balance_before
        order.balance_after = balance_after
    else:
        _apply_gmcp_order_quotas(db, order, tenant_no, user_id)
        order.order_status = "sxz"

    order.finance_confirm_time = now
    order.finance_confirm_attachments = finance_confirm_attachments or ""
    order.finance_remark = finance_remark or ""
    order.update_by = user_id
    order.update_time = now
    db.add(order)
    db.commit()
    db.refresh(order)
    return order


def submit_refund(db, order_id, *, refund_amount=None, refund_remark=None, refund_attachments=None, tenant_no=None, user_id=None):
    """申请退款：type=yfk + status=yqr → tfqrz"""
    order = get_service_order(db, order_id, tenant_no)
    if not order:
        raise ValueError("订单不存在")
    if order.order_type != "yfk":
        raise ValueError("仅预存款订单可申请退款")
    if order.order_status != "yqr":
        raise ValueError("仅已确认状态可申请退款")

    now = datetime.now()
    # Default refund_amount to prepay_amount
    if refund_amount is None:
        refund_amount = float(order.prepay_amount) if order.prepay_amount else 0
    else:
        refund_amount = float(refund_amount)

    prepay = float(order.prepay_amount) if order.prepay_amount else 0
    if refund_amount > prepay:
        raise ValueError(f"退款金额不能大于预存金额 {prepay}")

    order.order_status = "tfqrz"
    order.refund_apply_by = user_id
    order.refund_apply_time = now
    order.refund_amount = refund_amount
    order.refund_remark = refund_remark or ""
    order.refund_attachments = refund_attachments or ""
    order.update_by = user_id
    order.update_time = now
    db.add(order)
    db.commit()
    db.refresh(order)
    return order


def confirm_refund(db, order_id, *, refund_finance_remark=None, refund_finance_confirm_attachments=None, tenant_no=None, user_id=None):
    """确认退款：status=tfqrz → ytf，扣减余额"""
    order = get_service_order(db, order_id, tenant_no)
    if not order:
        raise ValueError("订单不存在")
    if order.order_status != "tfqrz":
        raise ValueError("仅退款待确认状态可确认退款")

    now = datetime.now()
    account = _get_or_create_channel_account(db, order.channel_id, tenant_no or order.tenant_no, user_id)
    balance_before = float(account.balance_amount) if account.balance_amount else 0
    deduct = float(order.refund_amount) if order.refund_amount else 0
    account.balance_amount = balance_before - deduct
    balance_after = float(account.balance_amount)
    account.update_by = user_id
    account.update_time = now
    db.add(account)
    db.flush()

    order.order_status = "ytf"
    order.refund_finance_confirm_time = now
    order.refund_finance_confirm_attachments = refund_finance_confirm_attachments or ""
    order.refund_finance_remark = refund_finance_remark or ""
    order.balance_before = balance_before
    order.balance_after = balance_after
    order.update_by = user_id
    order.update_time = now
    db.add(order)
    db.commit()
    db.refresh(order)
    return order


def void_service_order(db, order_id, *, tenant_no=None, user_id=None):
    """作废订单：type=gmcp + status=sxz → yzf，扣减已生效的产品份数"""
    order = get_service_order(db, order_id, tenant_no)
    if not order:
        raise ValueError("订单不存在")
    if order.order_type != "gmcp":
        raise ValueError("仅购买产品订单可作废")
    if order.order_status != "sxz":
        raise ValueError("仅生效中状态可作废")

    now = datetime.now()

    # Reduce product quotas
    products = get_service_order_products(db, order_id, tenant_no)
    for prod in products:
        cell_type_code, _ = _resolve_agreement_price_cell_type(
            db,
            order.channel_agreement_id,
            prod.get("projectCode") or "",
            prod.get("specText") or "",
            prod.get("cellType") or "",
        )

        _reduce_channel_product_quota(
            db,
            channel_id=order.channel_id,
            project_code=prod.get("projectCode") or "",
            cell_type_code=cell_type_code,
            spec_text=prod.get("specText") or "",
            reduce_qty=prod.get("buyQty") or 0,
            user_id=user_id,
        )

    order.order_status = "yzf"
    order.update_by = user_id
    order.update_time = now
    db.add(order)
    db.commit()
    db.refresh(order)
    return order


# ────────────────────────────────────────────
# Service Order Txn: 操作流水 CRUD
# ────────────────────────────────────────────

def list_service_order_txns(
    db,
    *,
    service_order_id=None,
    channel_id=None,
    txn_status=None,
    txn_type=None,
    keyword=None,
    create_time_start=None,
    create_time_end=None,
    finance_done_only=False,
    page=1,
    page_size=20,
    tenant_no=None,
):
    """查询操作流水列表。finance_done_only：财务已处理（已确认或财务侧作废）。"""
    q = db.query(BizChannelServiceOrderTxn).filter(BizChannelServiceOrderTxn.del_flag == 0)
    if tenant_no:
        q = q.filter(BizChannelServiceOrderTxn.tenant_no == tenant_no)
    if service_order_id:
        q = q.filter(BizChannelServiceOrderTxn.service_order_id == service_order_id)
    cid = (channel_id or "").strip()
    kw = (keyword or "").strip()
    if cid:
        q = q.join(
            BizChannelServiceOrder,
            and_(
                BizChannelServiceOrder.id == BizChannelServiceOrderTxn.service_order_id,
                BizChannelServiceOrder.del_flag == 0,
            ),
        ).filter(BizChannelServiceOrder.channel_id == cid)
        if kw:
            like = f"%{kw}%"
            q = q.filter(
                or_(
                    BizChannelServiceOrder.channel_name.like(like),
                    BizChannelServiceOrder.order_name.like(like),
                    BizChannelServiceOrder.dept_name.like(like),
                )
            )
    elif kw:
        like = f"%{kw}%"
        q = q.outerjoin(
            BizChannelServiceOrder,
            and_(
                BizChannelServiceOrder.id == BizChannelServiceOrderTxn.service_order_id,
                BizChannelServiceOrder.del_flag == 0,
            ),
        ).filter(
            or_(
                BizChannelServiceOrder.channel_name.like(like),
                BizChannelServiceOrder.order_name.like(like),
                BizChannelServiceOrder.dept_name.like(like),
            )
        )
    if finance_done_only:
        q = q.filter(
            or_(
                BizChannelServiceOrderTxn.txn_status == "yqr",
                and_(
                    BizChannelServiceOrderTxn.txn_status == "yzf",
                    BizChannelServiceOrderTxn.finance_submit_time.isnot(None),
                ),
            )
        )
    else:
        ss = (txn_status or "").strip()
        if ss:
            statuses = [s.strip() for s in ss.split(",") if s.strip()]
            if statuses:
                q = q.filter(BizChannelServiceOrderTxn.txn_status.in_(statuses))
    tt = (txn_type or "").strip()
    if tt:
        types = [t.strip() for t in tt.split(",") if t.strip()]
        if types:
            q = q.filter(BizChannelServiceOrderTxn.txn_type.in_(types))
    if create_time_start and create_time_start.strip():
        q = q.filter(
            BizChannelServiceOrderTxn.create_time
            >= _normalize_date_bound(create_time_start.strip())
        )
    if create_time_end and create_time_end.strip():
        q = q.filter(
            BizChannelServiceOrderTxn.create_time
            <= _normalize_date_bound(create_time_end.strip(), end_of_day=True)
        )

    total = q.order_by(None).with_entities(func.count(BizChannelServiceOrderTxn.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    if finance_done_only:
        operate_time = func.coalesce(
            BizChannelServiceOrderTxn.finance_confirm_time,
            BizChannelServiceOrderTxn.void_time,
        )
        rows = (
            q.order_by(operate_time.desc(), BizChannelServiceOrderTxn.create_time.desc())
            .offset((page - 1) * page_size)
            .limit(page_size)
            .all()
        )
    else:
        rows = (
            q.order_by(BizChannelServiceOrderTxn.create_time.desc())
            .offset((page - 1) * page_size)
            .limit(page_size)
            .all()
        )
    if not rows:
        return [], int(total)

    order_ids = list({r.service_order_id for r in rows if r.service_order_id})
    orders: Dict[str, Any] = {}
    if order_ids:
        oq = db.query(BizChannelServiceOrder).filter(
            BizChannelServiceOrder.id.in_(order_ids),
            BizChannelServiceOrder.del_flag == 0,
        )
        if tenant_no:
            oq = oq.filter(BizChannelServiceOrder.tenant_no == tenant_no)
        for o in oq.all():
            orders[o.id] = o

    user_ids: List[str] = []
    for r in rows:
        if r.create_by:
            user_ids.append(r.create_by)
        if r.finance_submit_by:
            user_ids.append(r.finance_submit_by)
        if r.finance_confirm_by:
            user_ids.append(r.finance_confirm_by)
        elif (r.txn_status or "") == "yqr" and r.update_by:
            user_ids.append(r.update_by)
        if r.void_by:
            user_ids.append(r.void_by)
        elif (r.txn_status or "") == "yzf" and r.finance_submit_time and r.update_by:
            user_ids.append(r.update_by)
    name_map = _user_nick_name_map(db, user_ids) if db and user_ids else {}

    items = [
        _service_order_txn_to_response_dict(
            db,
            r,
            tenant_no,
            order=orders.get(r.service_order_id),
            name_map=name_map,
        )
        for r in rows
    ]
    return items, int(total)


def get_service_order_txn(db, txn_id, *, tenant_no=None):
    """获取单条流水"""
    q = db.query(BizChannelServiceOrderTxn).filter(
        and_(BizChannelServiceOrderTxn.id == txn_id, BizChannelServiceOrderTxn.del_flag == 0)
    )
    if tenant_no:
        q = q.filter(BizChannelServiceOrderTxn.tenant_no == tenant_no)
    return q.first()


def create_recharge_txn(
    db,
    order_id,
    *,
    amount,
    remark=None,
    finance_submit_attachments=None,
    tenant_no=None,
    user_id=None,
):
    """创建充值流水（yfk订单），状态 dsx；附件预存于 finance_submit_attachments 供后续财务确认"""
    order = get_service_order(db, order_id, tenant_no)
    if not order:
        raise ValueError("订单不存在")
    if order.order_type != "yfk":
        raise ValueError("仅充值型订单可发起充值")
    if order.order_status not in ("dsx", "sxz"):
        raise ValueError("订单当前状态不允许充值")

    amount = float(amount)
    if amount <= 0:
        raise ValueError("充值金额必须大于0")

    now = datetime.now()
    txn = BizChannelServiceOrderTxn(
        id=generate_uuid(),
        service_order_id=order_id,
        txn_type="recharge",
        amount=amount,
        txn_status="dsx",
        remark=remark or "",
        finance_submit_attachments=(finance_submit_attachments or "").strip() or None,
        tenant_no=tenant_no or order.tenant_no,
        order_num=0,
        create_by=user_id,
        create_dept_id="",
        create_time=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(txn)
    db.commit()
    db.refresh(txn)
    return txn


def create_refund_txn(db, order_id, *, amount, remark=None, attachments=None, tenant_no=None, user_id=None):
    """创建退款流水（yfk订单），amount <= current_balance"""
    order = get_service_order(db, order_id, tenant_no)
    if not order:
        raise ValueError("订单不存在")
    if order.order_type != "yfk":
        raise ValueError("仅充值型订单可申请退款")
    if order.order_status != "sxz":
        raise ValueError("仅生效中订单可申请退款")

    amount = float(amount)
    if amount <= 0:
        raise ValueError("退款金额必须大于0")
    current_balance = float(order.current_balance) if order.current_balance else 0
    if amount > current_balance:
        raise ValueError(f"退款金额不能大于当前余额 {current_balance}")

    now = datetime.now()
    txn = BizChannelServiceOrderTxn(
        id=generate_uuid(),
        service_order_id=order_id,
        txn_type="refund",
        amount=amount,
        txn_status="dsx",
        remark=remark or "",
        refund_attachments=attachments or "",
        tenant_no=tenant_no or order.tenant_no,
        order_num=0,
        create_by=user_id,
        create_dept_id="",
        create_time=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(txn)
    db.commit()
    db.refresh(txn)
    return txn


def create_purchase_txn(db, order_id, *, products, remark=None, tenant_no=None, user_id=None):
    """创建购买产品流水（gmcp订单），同时创建/更新产品持仓"""
    order = get_service_order(db, order_id, tenant_no)
    if not order:
        raise ValueError("订单不存在")
    if order.order_type != "gmcp":
        raise ValueError("仅购买型订单可购买产品")
    if order.order_status not in ("dsx", "sxz"):
        raise ValueError("订单当前状态不允许购买")
    if not products:
        raise ValueError("产品明细不能为空")

    now = datetime.now()
    tn = tenant_no or order.tenant_no
    total_amount = 0.0

    # Upsert product holding rows
    for p in products:
        project_code = p.get("projectCode") or p.get("project_code") or ""
        project_name = p.get("projectName") or p.get("project_name") or ""
        cell_type = p.get("cellType") or p.get("cell_type") or ""
        spec_text = p.get("specText") or p.get("spec_text") or ""
        agreement_unit_price = float(p.get("agreementUnitPrice") or p.get("agreement_unit_price") or 0)
        qty = float(p.get("qty") or p.get("buyQty") or p.get("buy_qty") or 0)
        amt = float(p.get("amount") or agreement_unit_price * qty)
        total_amount += amt

        # Find or create product holding row
        existing = db.query(BizChannelServiceOrderProduct).filter(
            and_(
                BizChannelServiceOrderProduct.service_order_id == order_id,
                BizChannelServiceOrderProduct.project_code == project_code,
                BizChannelServiceOrderProduct.spec_text == (spec_text or ""),
                BizChannelServiceOrderProduct.del_flag == 0,
            )
        ).first()

        if existing:
            existing.total_buy_qty = float(existing.total_buy_qty or 0) + qty
            existing.current_qty = float(existing.current_qty or 0) + qty
            existing.update_by = user_id
            existing.update_time = now
            db.add(existing)
        else:
            prod_row = BizChannelServiceOrderProduct(
                id=generate_uuid(),
                service_order_id=order_id,
                project_code=project_code,
                project_name=project_name,
                cell_type=cell_type,
                spec_text=spec_text,
                agreement_unit_price=agreement_unit_price,
                buy_qty=qty,
                buy_amount=amt,
                total_buy_qty=qty,
                total_unsubscribe_qty=0,
                current_qty=qty,
                total_unsubscribe_amount=0,
                tenant_no=tn,
                order_num=0,
                remark="",
                create_by=user_id,
                create_dept_id="",
                create_time=now,
                update_by=user_id,
                update_time=now,
                del_flag=0,
            )
            db.add(prod_row)

    db.flush()

    products_json = json.dumps(products, ensure_ascii=False)
    txn = BizChannelServiceOrderTxn(
        id=generate_uuid(),
        service_order_id=order_id,
        txn_type="purchase",
        amount=total_amount,
        txn_status="dsx",
        products_json=products_json,
        remark=remark or "",
        tenant_no=tn,
        order_num=0,
        create_by=user_id,
        create_dept_id="",
        create_time=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    order.purchase_total_amount = float(order.purchase_total_amount or 0) + total_amount
    order.update_by = user_id
    order.update_time = now
    db.add(order)
    db.add(txn)
    db.commit()
    db.refresh(txn)
    return txn


def create_unsubscribe_txn(db, order_id, *, products, remark=None, tenant_no=None, user_id=None):
    """创建退订产品流水（gmcp订单），校验退订份数 <= current_qty"""
    order = get_service_order(db, order_id, tenant_no)
    if not order:
        raise ValueError("订单不存在")
    if order.order_type != "gmcp":
        raise ValueError("仅购买型订单可退订产品")
    if order.order_status != "sxz":
        raise ValueError("仅生效中订单可退订")
    if not products:
        raise ValueError("退订产品明细不能为空")

    now = datetime.now()
    tn = tenant_no or order.tenant_no
    total_amount = 0.0

    for p in products:
        project_code = p.get("projectCode") or p.get("project_code") or ""
        spec_text = p.get("specText") or p.get("spec_text") or ""
        qty = float(p.get("qty") or 0)
        if qty <= 0:
            raise ValueError("退订数量必须大于0")

        # Find existing product holding
        existing = db.query(BizChannelServiceOrderProduct).filter(
            and_(
                BizChannelServiceOrderProduct.service_order_id == order_id,
                BizChannelServiceOrderProduct.project_code == project_code,
                BizChannelServiceOrderProduct.spec_text == (spec_text or ""),
                BizChannelServiceOrderProduct.del_flag == 0,
            )
        ).first()
        if not existing:
            raise ValueError(f"产品 {project_code} 不存在于持仓中")

        current_qty = float(existing.current_qty or 0)
        if qty > current_qty:
            raise ValueError(f"退订数量 {qty} 超过当前持有数量 {current_qty}")

        unit_price = float(existing.agreement_unit_price or 0)
        amt = qty * unit_price
        total_amount += amt

        existing.total_unsubscribe_qty = float(existing.total_unsubscribe_qty or 0) + qty
        existing.current_qty = current_qty - qty
        existing.total_unsubscribe_amount = float(existing.total_unsubscribe_amount or 0) + amt
        existing.update_by = user_id
        existing.update_time = now
        db.add(existing)

    db.flush()

    products_json = json.dumps(products, ensure_ascii=False)
    txn = BizChannelServiceOrderTxn(
        id=generate_uuid(),
        service_order_id=order_id,
        txn_type="unsubscribe",
        amount=total_amount,
        txn_status="dsx",
        products_json=products_json,
        remark=remark or "",
        tenant_no=tn,
        order_num=0,
        create_by=user_id,
        create_dept_id="",
        create_time=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(txn)
    db.commit()
    db.refresh(txn)
    return txn


def _rollback_purchase_txn_holdings(db, order_id, products, user_id, now=None):
    now = now or datetime.now()
    for p in products:
        project_code = p.get("projectCode") or p.get("project_code") or ""
        spec_text = p.get("specText") or p.get("spec_text") or ""
        qty = float(p.get("qty") or p.get("buyQty") or p.get("buy_qty") or 0)
        existing = db.query(BizChannelServiceOrderProduct).filter(
            and_(
                BizChannelServiceOrderProduct.service_order_id == order_id,
                BizChannelServiceOrderProduct.project_code == project_code,
                BizChannelServiceOrderProduct.spec_text == (spec_text or ""),
                BizChannelServiceOrderProduct.del_flag == 0,
            )
        ).first()
        if existing:
            existing.total_buy_qty = float(existing.total_buy_qty or 0) - qty
            existing.current_qty = float(existing.current_qty or 0) - qty
            existing.update_by = user_id
            existing.update_time = now
            db.add(existing)


def _apply_purchase_txn_holdings(db, order_id, products, tenant_no, user_id, now=None):
    now = now or datetime.now()
    tn = tenant_no
    total_amount = 0.0
    for p in products:
        project_code = p.get("projectCode") or p.get("project_code") or ""
        project_name = p.get("projectName") or p.get("project_name") or ""
        cell_type = p.get("cellType") or p.get("cell_type") or ""
        spec_text = p.get("specText") or p.get("spec_text") or ""
        agreement_unit_price = float(p.get("agreementUnitPrice") or p.get("agreement_unit_price") or 0)
        qty = float(p.get("qty") or p.get("buyQty") or p.get("buy_qty") or 0)
        if qty <= 0:
            raise ValueError("购买份数必须大于0")
        amt = float(p.get("amount") or agreement_unit_price * qty)
        total_amount += amt

        existing = db.query(BizChannelServiceOrderProduct).filter(
            and_(
                BizChannelServiceOrderProduct.service_order_id == order_id,
                BizChannelServiceOrderProduct.project_code == project_code,
                BizChannelServiceOrderProduct.spec_text == (spec_text or ""),
                BizChannelServiceOrderProduct.del_flag == 0,
            )
        ).first()

        if existing:
            existing.total_buy_qty = float(existing.total_buy_qty or 0) + qty
            existing.current_qty = float(existing.current_qty or 0) + qty
            existing.update_by = user_id
            existing.update_time = now
            db.add(existing)
        else:
            prod_row = BizChannelServiceOrderProduct(
                id=generate_uuid(),
                service_order_id=order_id,
                project_code=project_code,
                project_name=project_name,
                cell_type=cell_type,
                spec_text=spec_text,
                agreement_unit_price=agreement_unit_price,
                buy_qty=qty,
                buy_amount=amt,
                total_buy_qty=qty,
                total_unsubscribe_qty=0,
                current_qty=qty,
                total_unsubscribe_amount=0,
                tenant_no=tn,
                order_num=0,
                remark="",
                create_by=user_id,
                create_dept_id="",
                create_time=now,
                update_by=user_id,
                update_time=now,
                del_flag=0,
            )
            db.add(prod_row)
    db.flush()
    return total_amount


def _rollback_unsubscribe_txn_holdings(db, order_id, products, user_id, now=None):
    now = now or datetime.now()
    for p in products:
        project_code = p.get("projectCode") or p.get("project_code") or ""
        spec_text = p.get("specText") or p.get("spec_text") or ""
        qty = float(p.get("qty") or 0)
        existing = db.query(BizChannelServiceOrderProduct).filter(
            and_(
                BizChannelServiceOrderProduct.service_order_id == order_id,
                BizChannelServiceOrderProduct.project_code == project_code,
                BizChannelServiceOrderProduct.spec_text == (spec_text or ""),
                BizChannelServiceOrderProduct.del_flag == 0,
            )
        ).first()
        if existing:
            unit_price = float(existing.agreement_unit_price or 0)
            existing.total_unsubscribe_qty = float(existing.total_unsubscribe_qty or 0) - qty
            existing.current_qty = float(existing.current_qty or 0) + qty
            existing.total_unsubscribe_amount = float(existing.total_unsubscribe_amount or 0) - (qty * unit_price)
            existing.update_by = user_id
            existing.update_time = now
            db.add(existing)


def _apply_unsubscribe_txn_holdings(db, order_id, products, user_id, now=None):
    now = now or datetime.now()
    total_amount = 0.0
    for p in products:
        project_code = p.get("projectCode") or p.get("project_code") or ""
        spec_text = p.get("specText") or p.get("spec_text") or ""
        qty = float(p.get("qty") or 0)
        if qty <= 0:
            raise ValueError("退订数量必须大于0")

        existing = db.query(BizChannelServiceOrderProduct).filter(
            and_(
                BizChannelServiceOrderProduct.service_order_id == order_id,
                BizChannelServiceOrderProduct.project_code == project_code,
                BizChannelServiceOrderProduct.spec_text == (spec_text or ""),
                BizChannelServiceOrderProduct.del_flag == 0,
            )
        ).first()
        if not existing:
            raise ValueError(f"产品 {project_code} 不存在于持仓中")

        current_qty = float(existing.current_qty or 0)
        if qty > current_qty:
            raise ValueError(f"退订数量 {qty} 超过当前持有数量 {current_qty}")

        unit_price = float(existing.agreement_unit_price or 0)
        amt = qty * unit_price
        total_amount += amt

        existing.total_unsubscribe_qty = float(existing.total_unsubscribe_qty or 0) + qty
        existing.current_qty = current_qty - qty
        existing.total_unsubscribe_amount = float(existing.total_unsubscribe_amount or 0) + amt
        existing.update_by = user_id
        existing.update_time = now
        db.add(existing)
    db.flush()
    return total_amount


def update_service_order_txn(
    db,
    txn_id,
    *,
    amount=None,
    remark=None,
    finance_submit_attachments=None,
    refund_attachments=None,
    products=None,
    tenant_no=None,
    user_id=None,
):
    """更新待生效流水：仅 dsx；充值/退款可改金额与附件，购买/退订可改产品明细与备注"""
    txn = get_service_order_txn(db, txn_id, tenant_no=tenant_no)
    if not txn:
        raise ValueError("流水不存在")
    if txn.txn_status != "dsx":
        raise ValueError("仅待生效流水可编辑")

    now = datetime.now()
    txn_type = (txn.txn_type or "").strip()

    if amount is not None:
        amt = float(amount)
        if amt <= 0:
            raise ValueError("金额必须大于0")
        if txn_type in ("recharge", "refund"):
            if txn_type == "refund":
                order = get_service_order(db, txn.service_order_id, tenant_no)
                if not order:
                    raise ValueError("关联订单不存在")
                current_balance = float(order.current_balance) if order.current_balance else 0
                if amt > current_balance:
                    raise ValueError(f"退款金额不能大于当前余额 {current_balance}")
            txn.amount = amt
        else:
            raise ValueError("购买/退订流水不可直接修改金额，请调整产品明细")

    if products is not None:
        if txn_type not in ("purchase", "unsubscribe"):
            raise ValueError("仅购买/退订流水可修改产品明细")
        if not products:
            raise ValueError("产品明细不能为空")
        order = get_service_order(db, txn.service_order_id, tenant_no)
        if not order:
            raise ValueError("关联订单不存在")
        old_products = json.loads(txn.products_json) if txn.products_json else []
        if txn_type == "purchase":
            _rollback_purchase_txn_holdings(db, order.id, old_products, user_id, now)
            total_amount = _apply_purchase_txn_holdings(
                db, order.id, products, tenant_no or order.tenant_no, user_id, now
            )
        else:
            _rollback_unsubscribe_txn_holdings(db, order.id, old_products, user_id, now)
            total_amount = _apply_unsubscribe_txn_holdings(db, order.id, products, user_id, now)
        txn.amount = total_amount
        txn.products_json = json.dumps(products, ensure_ascii=False)

    if remark is not None:
        txn.remark = remark or ""

    if finance_submit_attachments is not None:
        txn.finance_submit_attachments = (finance_submit_attachments or "").strip() or None

    if refund_attachments is not None:
        if txn_type != "refund":
            raise ValueError("仅退款流水可修改退款附件")
        txn.refund_attachments = (refund_attachments or "").strip() or None

    txn.update_by = user_id
    txn.update_time = now
    db.add(txn)
    db.commit()
    db.refresh(txn)
    return txn


def submit_txn_finance(db, txn_id, *, finance_submit_remark=None, finance_submit_attachments=None, tenant_no=None, user_id=None):
    """发起流水财务确认：dsx → dqr"""
    txn = get_service_order_txn(db, txn_id, tenant_no=tenant_no)
    if not txn:
        raise ValueError("流水不存在")
    if txn.txn_status != "dsx":
        raise ValueError("仅待生效流水可发起确认")

    now = datetime.now()
    txn.txn_status = "dqr"
    txn.finance_submit_by = user_id
    txn.finance_submit_time = now
    submit_remark = (finance_submit_remark or "").strip()
    if not submit_remark:
        submit_remark = (txn.remark or "").strip()
    txn.finance_submit_remark = submit_remark
    if finance_submit_attachments:
        txn.finance_submit_attachments = finance_submit_attachments
    elif not (txn.finance_submit_attachments or "").strip():
        if (txn.txn_type or "").strip() == "refund" and (txn.refund_attachments or "").strip():
            txn.finance_submit_attachments = txn.refund_attachments
        else:
            txn.finance_submit_attachments = ""
    txn.update_by = user_id
    txn.update_time = now
    db.add(txn)
    db.commit()
    db.refresh(txn)
    return txn


def confirm_txn(db, txn_id, *, finance_remark=None, finance_confirm_attachments=None, tenant_no=None, user_id=None):
    """财务确认流水：dqr → yqr，更新主表汇总和产品持仓，更新渠道账户余额"""
    txn = get_service_order_txn(db, txn_id, tenant_no=tenant_no)
    if not txn:
        raise ValueError("流水不存在")
    if txn.txn_status != "dqr":
        raise ValueError("仅待确认流水可进行财务确认")

    order = get_service_order(db, txn.service_order_id, tenant_no)
    if not order:
        raise ValueError("关联订单不存在")

    now = datetime.now()

    if txn.txn_type == "recharge":
        # Increase channel account balance
        account = _get_or_create_channel_account(db, order.channel_id, tenant_no or order.tenant_no, user_id)
        balance_before = float(account.balance_amount) if account.balance_amount else 0
        add_amount = float(txn.amount) if txn.amount else 0
        account.balance_amount = balance_before + add_amount
        balance_after = float(account.balance_amount)
        account.update_by = user_id
        account.update_time = now
        db.add(account)
        db.flush()

        # Update order summary
        order.total_recharge_amount = float(order.total_recharge_amount or 0) + add_amount
        order.current_balance = float(order.current_balance or 0) + add_amount
        txn.balance_before = balance_before
        txn.balance_after = balance_after

    elif txn.txn_type == "refund":
        # Deduct channel account balance
        account = _get_or_create_channel_account(db, order.channel_id, tenant_no or order.tenant_no, user_id)
        balance_before = float(account.balance_amount) if account.balance_amount else 0
        deduct_amount = float(txn.amount) if txn.amount else 0
        account.balance_amount = balance_before - deduct_amount
        balance_after = float(account.balance_amount)
        account.update_by = user_id
        account.update_time = now
        db.add(account)
        db.flush()

        # Update order summary
        order.total_refund_amount = float(order.total_refund_amount or 0) + deduct_amount
        order.current_balance = float(order.current_balance or 0) - deduct_amount
        txn.balance_before = balance_before
        txn.balance_after = balance_after

    elif txn.txn_type == "purchase":
        # Products already updated at create time, just update channel quotas
        products = json.loads(txn.products_json) if txn.products_json else []
        for p in products:
            project_code = p.get("projectCode") or p.get("project_code") or ""
            spec_text = p.get("specText") or p.get("spec_text") or ""
            qty = float(p.get("qty") or p.get("buyQty") or p.get("buy_qty") or 0)
            cell_type_code, _ = _resolve_agreement_price_cell_type(
                db, order.channel_agreement_id, project_code, spec_text, p.get("cellType") or p.get("cell_type") or ""
            )
            _upsert_channel_product_quota(
                db,
                channel_id=order.channel_id,
                project_code=project_code,
                project_name=p.get("projectName") or p.get("project_name") or "",
                cell_type_code=cell_type_code,
                cell_type_name=p.get("cellType") or p.get("cell_type") or "",
                spec_text=spec_text,
                add_qty=qty,
                tenant_no=tenant_no or order.tenant_no,
                user_id=user_id,
            )

    elif txn.txn_type == "unsubscribe":
        # Products already updated at create time, just reduce channel quotas
        products = json.loads(txn.products_json) if txn.products_json else []
        for p in products:
            project_code = p.get("projectCode") or p.get("project_code") or ""
            spec_text = p.get("specText") or p.get("spec_text") or ""
            qty = float(p.get("qty") or 0)
            cell_type_code, _ = _resolve_agreement_price_cell_type(
                db, order.channel_agreement_id, project_code, spec_text, p.get("cellType") or p.get("cell_type") or ""
            )
            _reduce_channel_product_quota(
                db,
                channel_id=order.channel_id,
                project_code=project_code,
                cell_type_code=cell_type_code,
                spec_text=spec_text,
                reduce_qty=qty,
                user_id=user_id,
            )

    # Update txn status
    txn.txn_status = "yqr"
    txn.finance_confirm_time = now
    txn.finance_confirm_by = user_id
    txn.finance_confirm_attachments = finance_confirm_attachments or ""
    txn.finance_remark = finance_remark or ""
    txn.update_by = user_id
    txn.update_time = now

    # Transition order to sxz if currently dsx
    if order.order_status == "dsx":
        order.order_status = "sxz"
    order.update_by = user_id
    order.update_time = now

    db.add(txn)
    db.add(order)
    db.commit()
    db.refresh(txn)
    return txn


def void_txn(db, txn_id, *, tenant_no=None, user_id=None):
    """作废流水：dsx/dqr → yzf。如果是 purchase/unsubscribe 还需要回滚产品持仓"""
    txn = get_service_order_txn(db, txn_id, tenant_no=tenant_no)
    if not txn:
        raise ValueError("流水不存在")
    if txn.txn_status not in ("dsx", "dqr"):
        raise ValueError("仅待生效或待财务确认流水可作废")
    if (txn.txn_type or "").strip() == "settlement":
        raise ValueError("输注计划结算流水不可作废")

    now = datetime.now()
    order = get_service_order(db, txn.service_order_id, tenant_no)

    if txn.txn_type == "purchase" and order:
        # Rollback product holdings
        products = json.loads(txn.products_json) if txn.products_json else []
        for p in products:
            project_code = p.get("projectCode") or p.get("project_code") or ""
            spec_text = p.get("specText") or p.get("spec_text") or ""
            qty = float(p.get("qty") or p.get("buyQty") or p.get("buy_qty") or 0)
            existing = db.query(BizChannelServiceOrderProduct).filter(
                and_(
                    BizChannelServiceOrderProduct.service_order_id == order.id,
                    BizChannelServiceOrderProduct.project_code == project_code,
                    BizChannelServiceOrderProduct.spec_text == (spec_text or ""),
                    BizChannelServiceOrderProduct.del_flag == 0,
                )
            ).first()
            if existing:
                existing.total_buy_qty = float(existing.total_buy_qty or 0) - qty
                existing.current_qty = float(existing.current_qty or 0) - qty
                existing.update_by = user_id
                existing.update_time = now
                db.add(existing)
        txn_amount = float(txn.amount or 0)
        if txn_amount > 0:
            order.purchase_total_amount = max(
                0.0,
                float(order.purchase_total_amount or 0) - txn_amount,
            )
            order.update_by = user_id
            order.update_time = now
            db.add(order)

    elif txn.txn_type == "unsubscribe" and order:
        # Rollback unsubscribe: restore product holdings
        products = json.loads(txn.products_json) if txn.products_json else []
        for p in products:
            project_code = p.get("projectCode") or p.get("project_code") or ""
            spec_text = p.get("specText") or p.get("spec_text") or ""
            qty = float(p.get("qty") or 0)
            unit_price = 0
            existing = db.query(BizChannelServiceOrderProduct).filter(
                and_(
                    BizChannelServiceOrderProduct.service_order_id == order.id,
                    BizChannelServiceOrderProduct.project_code == project_code,
                    BizChannelServiceOrderProduct.spec_text == (spec_text or ""),
                    BizChannelServiceOrderProduct.del_flag == 0,
                )
            ).first()
            if existing:
                unit_price = float(existing.agreement_unit_price or 0)
                existing.total_unsubscribe_qty = float(existing.total_unsubscribe_qty or 0) - qty
                existing.current_qty = float(existing.current_qty or 0) + qty
                existing.total_unsubscribe_amount = float(existing.total_unsubscribe_amount or 0) - (qty * unit_price)
                existing.update_by = user_id
                existing.update_time = now
                db.add(existing)

    txn.txn_status = "yzf"
    txn.void_by = user_id
    txn.void_time = now
    txn.update_by = user_id
    txn.update_time = now
    db.add(txn)
    db.commit()
    db.refresh(txn)
    return txn