zkxw/backend/app/services/market_support.py

"""移动市场表单选项、附件暂存与动作白名单。

该模块是移动市场写接口的安全边界：客户端不能选择任意字典、目录或附件 URL。
"""
from __future__ import annotations

import os
import re
import uuid
import zipfile
from decimal import Decimal
from io import BytesIO
from typing import Any, Dict, Iterable, List, Optional

from fastapi import HTTPException
from sqlalchemy import or_
from sqlalchemy.orm import Session

from app.core.config import settings
from app.core.storage import storage
from app.models.models import (
    BizAttachment,
    BizCellSampleType,
    BizCollectionProtocol,
    BizCollectionHospital,
    BizCustomer,
    BizProtocolCollectionInfo,
    BizCollectionRegion,
    BizCollectionRegionStaff,
    BizItemUnit,
    BizPresignOrder,
    BizPresignTransfer,
    BizProtocolFullmoonTodo,
    BizProtocolPayment,
    BizProtocolPaymentDetail,
    BizProtocolPaymentTransfer,
    BizProtocolStorageCert,
    BizSampleCode,
    SysDictData,
    SysEthnic,
    SysUser,
)
from app.crud import collection_region as collection_region_crud
from app.crud import customer as customer_crud
from app.core.data_scope import apply_data_scope
from app.models.research_donation import (
    BizResearchDonation,
    BizResearchDonationCollection,
)
from app.services.market_field_scope import (
    MarketFieldContext,
    apply_cell_protocol_scope,
    apply_donation_scope,
    apply_presign_scope,
    assert_cell_protocol_access,
    assert_donation_access,
    assert_presign_access,
)

FORM_MODULES = frozenset({'presign', 'protocol', 'payment', 'freight'})
UPLOAD_MODULES = frozenset({'presign', 'protocol', 'payment', 'sample'})
DICT_BY_MODULE = {
    'payment': ('zzfs', 'transferTypes'),
    'freight': ('sample_transport_mode', 'transportModes'),
}
MAX_UPLOAD_BYTES = 10 * 1024 * 1024
PRESIGN_MAX_UPLOAD_BYTES = 20 * 1024 * 1024
PROTOCOL_MAX_UPLOAD_BYTES = 20 * 1024 * 1024
MAX_ATTACHMENT_REFS = 20
MAX_ATTACHMENT_REFS_LENGTH = 4000
MAX_ATTACHMENT_REF_LENGTH = 1000
PAYMENT_TERMINAL_PROTOCOL_STATUSES = frozenset({'yzf', 'sdfq', 'zky'})
PAYMENT_WRITABLE_SCHEDULE_STATUS = 'open'
HEIC_BRANDS = frozenset(
    {b'heic', b'heix', b'hevc', b'hevx', b'heim', b'heis'}
)
HEIF_BRANDS = HEIC_BRANDS | frozenset({b'mif1', b'msf1'})
ALLOWED_MIME_BY_EXT = {
    'jpg': {'image/jpeg'},
    'jpeg': {'image/jpeg'},
    'png': {'image/png'},
    'webp': {'image/webp'},
    'heic': {'image/heic', 'image/heif', 'application/octet-stream'},
    'heif': {'image/heic', 'image/heif', 'application/octet-stream'},
    'pdf': {'application/pdf'},
    'doc': {'application/msword', 'application/octet-stream'},
    'docx': {
        'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
        'application/zip',
        'application/octet-stream',
    },
    'xls': {'application/vnd.ms-excel', 'application/octet-stream'},
    'xlsx': {
        'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
        'application/zip',
        'application/octet-stream',
    },
}
PROTOCOL_ALLOWED_MIME_BY_EXT = {
    **ALLOWED_MIME_BY_EXT,
    'gif': {'image/gif'},
    'bmp': {'image/bmp', 'image/x-ms-bmp'},
    'ppt': {'application/vnd.ms-powerpoint', 'application/octet-stream'},
    'pptx': {
        'application/vnd.openxmlformats-officedocument.presentationml.presentation',
        'application/zip',
        'application/octet-stream',
    },
    'txt': {'text/plain', 'application/octet-stream'},
    'mp4': {'video/mp4', 'application/octet-stream'},
    'webm': {'video/webm', 'application/octet-stream'},
    'ogg': {'video/ogg', 'application/ogg', 'application/octet-stream'},
    'mov': {'video/quicktime', 'application/octet-stream'},
    'avi': {'video/x-msvideo', 'application/octet-stream'},
    'zip': {'application/zip', 'application/x-zip-compressed', 'application/octet-stream'},
    'rar': {'application/vnd.rar', 'application/x-rar-compressed', 'application/octet-stream'},
}


def _option(value: Any, label: Any) -> Dict[str, str]:
    return {'value': str(value or ''), 'label': str(label or '')}


def _protocol_sample_type_option(row: BizCellSampleType) -> Dict[str, Any]:
    storage_price = (
        format(Decimal(str(row.storage_price)).quantize(Decimal('0.01')), 'f')
        if row.storage_price is not None
        else None
    )
    return {
        **_option(row.id, row.type_name),
        'storagePrice': storage_price,
        'storageYears': (
            int(row.storage_years) if row.storage_years is not None else None
        ),
        'sampleCodeFilter': (row.sample_code_filter or '').strip(),
    }


def _verify_identity(ctx: MarketFieldContext, user_info: dict) -> None:
    if not isinstance(user_info, dict):
        raise PermissionError('用户身份与现场范围上下文不一致')
    if (
        str(user_info.get('user_id') or '').strip() != ctx.user_id
        or str(user_info.get('tenant_no') or '').strip() != ctx.tenant_no
    ):
        raise PermissionError('用户身份与现场范围上下文不一致')


def authorized_region_ids(db: Session, ctx: MarketFieldContext) -> List[str]:
    query = db.query(BizCollectionRegionStaff.region_id).filter(
        BizCollectionRegionStaff.tenant_no == ctx.tenant_no,
        BizCollectionRegionStaff.del_flag == 0,
        BizCollectionRegionStaff.user_id == ctx.user_id,
    )
    if ctx.scope_kind == 'manager':
        query = query.filter(BizCollectionRegionStaff.is_leader == 1)
    return [str(row[0]) for row in query.distinct().all() if row and row[0]]


def _tenant_region_ids(db: Session, tenant_no: str) -> List[str]:
    rows = (
        db.query(BizCollectionRegion.id)
        .filter(
            BizCollectionRegion.tenant_no == tenant_no,
            BizCollectionRegion.del_flag == 0,
        )
        .all()
    )
    return [str(row[0]) for row in rows if row and row[0]]


def form_option_region_ids(
    db: Session,
    *,
    module: str,
    ctx: MarketFieldContext,
    user_info: dict,
    business_id: Optional[str] = None,
) -> List[str]:
    """表单采集地区候选：与 Web 一致，返回本租户全部采集地区。

    不依赖 ``biz_collection_region_staff`` 绑定；现场写入范围仍由各模块
    create/update（如地区经理归属校验）与列表 scope 约束。
    """
    _ = (module, user_info, business_id)
    return _tenant_region_ids(db, ctx.tenant_no)


def _dict_options(db: Session, tenant_no: str, dict_type: str) -> List[dict]:
    rows = (
        db.query(SysDictData)
        .filter(
            SysDictData.tenant_no == tenant_no,
            SysDictData.del_flag == 0,
            SysDictData.dict_type == dict_type,
        )
        .order_by(SysDictData.order_num.asc(), SysDictData.dict_label.asc())
        .all()
    )
    return [_option(row.dict_value, row.dict_label) for row in rows]


def _business_region_id(
    db: Session,
    *,
    module: str,
    business_id: str,
    ctx: MarketFieldContext,
    user_info: dict,
) -> Optional[str]:
    """仅从当前已在 scope 内的单据补充地区，不把 businessId 当越权入口。"""
    record_id = str(business_id or '').strip()
    if not record_id:
        return None
    if module == 'presign':
        return str(
            assert_presign_access(
                db, record_id, ctx, user_info=user_info
            ).region_id
            or ''
        ).strip() or None
    if module != 'protocol':
        return None
    try:
        assert_cell_protocol_access(db, record_id, ctx, user_info=user_info)
        collection = (
            db.query(BizProtocolCollectionInfo)
            .filter(
                BizProtocolCollectionInfo.collection_protocol_id == record_id,
                BizProtocolCollectionInfo.tenant_no == ctx.tenant_no,
                BizProtocolCollectionInfo.del_flag == 0,
            )
            .order_by(
                BizProtocolCollectionInfo.create_time.desc(),
                BizProtocolCollectionInfo.id.desc(),
            )
            .first()
        )
    except HTTPException:
        assert_donation_access(db, record_id, ctx, user_info=user_info)
        collection = (
            db.query(BizResearchDonationCollection)
            .filter(
                BizResearchDonationCollection.research_donation_id == record_id,
                BizResearchDonationCollection.tenant_no == ctx.tenant_no,
                BizResearchDonationCollection.del_flag == 0,
            )
            .order_by(
                BizResearchDonationCollection.create_time.desc(),
                BizResearchDonationCollection.id.desc(),
            )
            .first()
        )
    return str(collection.region_id or '').strip() if collection else None


def search_customers_for_protocol(
    db: Session,
    *,
    ctx: MarketFieldContext,
    user_info: dict,
    keyword: Optional[str] = None,
    page: int = 1,
    page_size: int = 20,
) -> Dict[str, Any]:
    """协议单姓名联想：只读检索客户档案（含回填所需联系方式/证件）。"""
    _verify_identity(ctx, user_info)
    kw = str(keyword or '').strip()
    page = max(1, int(page or 1))
    page_size = min(max(1, int(page_size or 20)), 100)
    query = db.query(BizCustomer).filter(
        BizCustomer.del_flag == 0,
        BizCustomer.tenant_no == ctx.tenant_no,
    )
    if kw:
        like = f'%{kw}%'
        query = query.filter(
            or_(
                BizCustomer.customer_name.like(like),
                BizCustomer.phone_no.like(like),
                BizCustomer.id_card_no.like(like),
                BizCustomer.customer_tags.like(like),
            ),
        )
    query = apply_data_scope(query, BizCustomer, user_info, db, ctx.tenant_no)
    total = query.count()
    rows = (
        query.order_by(BizCustomer.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    return {
        'list': [customer_crud._row_to_dict(row) for row in rows],
        'total': int(total),
        'page': page,
        'pageSize': page_size,
    }


def get_form_options(
    db: Session,
    *,
    module: str,
    ctx: MarketFieldContext,
    user_info: dict,
    region_id: Optional[str] = None,
    business_id: Optional[str] = None,
    user_keyword: Optional[str] = None,
    code_filter: Optional[str] = None,
) -> Dict[str, Any]:
    """返回移动表单最小选项集；不会返回联系方式或证件等 PII。"""
    _verify_identity(ctx, user_info)
    normalized_module = str(module or '').strip()
    if normalized_module not in FORM_MODULES:
        raise ValueError('不支持的表单选项模块')
    if normalized_module in DICT_BY_MODULE:
        dict_type, key = DICT_BY_MODULE[normalized_module]
        return {key: _dict_options(db, ctx.tenant_no, dict_type)}

    region_ids = form_option_region_ids(
        db,
        module=normalized_module,
        ctx=ctx,
        user_info=user_info,
        business_id=business_id,
    )
    selected_region = str(region_id or '').strip()
    if selected_region and selected_region not in region_ids:
        raise PermissionError('所选地区不在当前用户权限范围内')
    regions = []
    if region_ids:
        regions = (
            db.query(BizCollectionRegion)
            .filter(
                BizCollectionRegion.id.in_(region_ids),
                BizCollectionRegion.tenant_no == ctx.tenant_no,
                BizCollectionRegion.del_flag == 0,
            )
            .order_by(BizCollectionRegion.order_num.asc(), BizCollectionRegion.region_name.asc())
            .all()
        )
    active_region_ids = [str(row.id) for row in regions]
    if selected_region and selected_region not in active_region_ids:
        raise PermissionError('所选地区不存在或已停用')
    hospital_regions = (
        [selected_region] if selected_region else active_region_ids
    )
    hospitals = []
    staff_rows = []
    if hospital_regions:
        hospitals = (
            db.query(BizCollectionHospital)
            .filter(
                BizCollectionHospital.region_id.in_(hospital_regions),
                BizCollectionHospital.tenant_no == ctx.tenant_no,
                BizCollectionHospital.del_flag == 0,
            )
            .order_by(BizCollectionHospital.order_num.asc(), BizCollectionHospital.hospital_name.asc())
            .all()
        )
        staff_rows = (
            db.query(SysUser, BizCollectionRegionStaff)
            .join(
                BizCollectionRegionStaff,
                BizCollectionRegionStaff.user_id == SysUser.id,
            )
            .filter(
                BizCollectionRegionStaff.region_id.in_(hospital_regions),
                BizCollectionRegionStaff.tenant_no == ctx.tenant_no,
                BizCollectionRegionStaff.del_flag == 0,
                SysUser.tenant_no == ctx.tenant_no,
                SysUser.del_flag == 0,
                SysUser.status == '1',
            )
            .order_by(
                BizCollectionRegionStaff.is_leader.desc(),
                SysUser.nick_name.asc(),
                SysUser.id.asc(),
            )
            .all()
        )
    sample_types = (
        db.query(BizCellSampleType)
        .filter(
            BizCellSampleType.tenant_no == ctx.tenant_no,
            BizCellSampleType.del_flag == 0,
        )
        .order_by(BizCellSampleType.order_num.asc(), BizCellSampleType.type_name.asc())
        .all()
    )
    units = (
        db.query(BizItemUnit)
        .filter(BizItemUnit.tenant_no == ctx.tenant_no, BizItemUnit.del_flag == 0)
        .order_by(BizItemUnit.order_num.asc(), BizItemUnit.unit_name.asc())
        .all()
    )
    staff = []
    seen_staff_ids = set()
    for user, binding in staff_rows:
        if user.id in seen_staff_ids:
            continue
        seen_staff_ids.add(user.id)
        staff.append(
            {
                **_option(user.id, user.nick_name),
                'isLeader': bool(binding.is_leader),
            }
        )
    current_user = (
        db.query(SysUser)
        .filter(
            SysUser.id == ctx.user_id,
            SysUser.tenant_no == ctx.tenant_no,
            SysUser.del_flag == 0,
            SysUser.status == '1',
        )
        .first()
    )
    searched_users = []
    if normalized_module == 'presign' and user_keyword is not None:
        keyword = str(user_keyword or '').strip()
        user_query = db.query(SysUser).filter(
            SysUser.tenant_no == ctx.tenant_no,
            SysUser.del_flag == 0,
            SysUser.status == '1',
        )
        if keyword:
            like = f'%{keyword}%'
            user_query = user_query.filter(
                or_(SysUser.nick_name.like(like), SysUser.user_no.like(like))
            )
        searched_users = (
            user_query.order_by(SysUser.nick_name.asc(), SysUser.id.asc())
            .limit(50)
            .all()
        )
    current_user_regions = []
    if current_user:
        current_user_regions = [
            _option(item.get('id'), item.get('regionName'))
            for item in collection_region_crud.list_regions_by_staff_user(
                db,
                user_id=str(current_user.id),
                tenant_no=ctx.tenant_no,
            )
            if item.get('id')
        ]
    result = {
        'regions': [_option(row.id, row.region_name) for row in regions],
        'hospitals': [_option(row.id, row.hospital_name) for row in hospitals],
        'staff': staff,
        'sampleTypes': [
            (
                _protocol_sample_type_option(row)
                if normalized_module == 'protocol'
                else _option(row.id, row.type_name)
            )
            for row in sample_types
        ],
        'units': [_option(row.id, row.unit_name) for row in units],
        'currentUser': (
            _option(current_user.id, current_user.nick_name)
            if current_user
            else None
        ),
        # 当前登录人在 biz_collection_region_staff 的关联地区（顺序即默认取第一个）
        'currentUserRegions': current_user_regions,
    }
    if normalized_module == 'protocol':
        ethnicities = (
            db.query(SysEthnic)
            .filter(
                SysEthnic.del_flag == 0,
                or_(SysEthnic.tenant_no == '', SysEthnic.tenant_no == ctx.tenant_no),
            )
            .order_by(SysEthnic.order_num.asc(), SysEthnic.ethnic_code.asc())
            .all()
        )
        # 多取候选再过滤已占用，尽量凑满可选列表
        sample_code_query = db.query(BizSampleCode).filter(
            BizSampleCode.tenant_no == ctx.tenant_no,
            BizSampleCode.del_flag == 0,
        )
        cf = (code_filter or '').strip()
        if cf:
            sample_code_query = sample_code_query.filter(
                BizSampleCode.sample_code.like(f'%{cf}%')
            )
        sample_codes = (
            sample_code_query.order_by(
                BizSampleCode.create_time.desc(), BizSampleCode.sample_code.desc()
            )
            .limit(200)
            .all()
        )
        from app.crud import collection_protocol as cell_crud

        used_codes = cell_crud.get_sample_codes_used_in_protocol_samples(
            db,
            tenant_no=ctx.tenant_no,
            sample_codes=[(row.sample_code or '').strip() for row in sample_codes],
        )
        available_codes = [
            row
            for row in sample_codes
            if (row.sample_code or '').strip()
            and (row.sample_code or '').strip() not in used_codes
        ][:100]
        result.update(
            {
                'ethnicities': [
                    _option(row.ethnic_name, row.ethnic_name)
                    for row in ethnicities
                ],
                'sampleCodes': [
                    {
                        **_option(row.id, row.sample_code),
                        'sampleCode': row.sample_code or '',
                    }
                    for row in available_codes
                ],
                # 协议单内嵌转账与缴费模块共用转账方式字典
                'transferTypes': _dict_options(db, ctx.tenant_no, 'zzfs'),
                # 协议单内嵌运费与运费模块共用运输方式字典
                'transportModes': _dict_options(
                    db, ctx.tenant_no, 'sample_transport_mode'
                ),
            }
        )
    if normalized_module == 'presign':
        result.update(
            {
                'payModes': [
                    {'label': '趸交', 'value': 'dj'},
                    {'label': '分期', 'value': 'fq'},
                ],
                'orderStatuses': [
                    {'label': '草稿', 'value': 'draft'},
                    {'label': '进行中', 'value': 'active'},
                    {'label': '已转正式', 'value': 'converted'},
                    {'label': '已作废', 'value': 'void'},
                ],
                'transferTypes': _dict_options(db, ctx.tenant_no, 'zzfs'),
                'users': [_option(row.id, row.nick_name) for row in searched_users],
            }
        )
    return result


def _validate_upload(
    content: bytes,
    filename: str,
    content_type: str,
    *,
    max_bytes: int = MAX_UPLOAD_BYTES,
    allowed_mime_by_ext: Optional[Dict[str, set]] = None,
) -> tuple[str, str]:
    name = str(filename or '').strip()
    if not name or len(name) > 200 or re.search(r'[\x00-\x1f\x7f]', name):
        raise ValueError('文件名无效')
    if '/' in name or '\\' in name:
        raise ValueError('文件名无效')
    if not content:
        raise ValueError('上传文件不能为空')
    if len(content) > max_bytes:
        raise ValueError(f'上传文件不能超过 {max_bytes // (1024 * 1024)}MiB')
    ext = os.path.splitext(name)[1].lower().lstrip('.')
    mime = str(content_type or '').lower().split(';', 1)[0].strip()
    allowed_types = allowed_mime_by_ext or ALLOWED_MIME_BY_EXT
    if ext not in allowed_types or mime not in allowed_types[ext]:
        raise ValueError('不支持的文件类型')
    signatures = {
        'jpg': (b'\xff\xd8\xff',),
        'jpeg': (b'\xff\xd8\xff',),
        'png': (b'\x89PNG\r\n\x1a\n',),
        'webp': (b'RIFF',),
        'pdf': (b'%PDF-',),
        'doc': (b'\xd0\xcf\x11\xe0',),
        'xls': (b'\xd0\xcf\x11\xe0',),
        'docx': (b'PK\x03\x04',),
        'xlsx': (b'PK\x03\x04',),
        'gif': (b'GIF87a', b'GIF89a'),
        'bmp': (b'BM',),
        'ppt': (b'\xd0\xcf\x11\xe0',),
        'pptx': (b'PK\x03\x04',),
        'mp4': (b'\x00\x00\x00',),
        'webm': (b'\x1a\x45\xdf\xa3',),
        'ogg': (b'OggS',),
        'mov': (b'\x00\x00\x00',),
        'avi': (b'RIFF',),
        'zip': (b'PK\x03\x04',),
        'rar': (b'Rar!\x1a\x07',),
    }
    if ext not in {'heic', 'heif', 'txt'} and not any(
        content.startswith(prefix) for prefix in signatures.get(ext, ())
    ):
        raise ValueError('文件内容与扩展名不匹配')
    if ext == 'webp' and (len(content) < 12 or content[8:12] != b'WEBP'):
        raise ValueError('文件内容与扩展名不匹配')
    if ext in {'heic', 'heif'}:
        if len(content) < 16 or content[4:8] != b'ftyp':
            raise ValueError('文件内容与扩展名不匹配')
        box_size = int.from_bytes(content[:4], byteorder='big', signed=False)
        if (
            box_size < 16
            or box_size > len(content)
            or (box_size - 16) % 4 != 0
        ):
            raise ValueError('文件内容与扩展名不匹配')
        brands = {content[8:12]}
        brands.update(
            content[offset : offset + 4]
            for offset in range(16, box_size, 4)
        )
        allowed_brands = HEIC_BRANDS if ext == 'heic' else HEIF_BRANDS
        if not brands.intersection(allowed_brands):
            raise ValueError('文件内容与扩展名不匹配')
    if ext in {'docx', 'xlsx', 'pptx', 'zip'}:
        try:
            with zipfile.ZipFile(BytesIO(content)) as archive:
                names = set(archive.namelist())
        except (OSError, zipfile.BadZipFile):
            raise ValueError('文件内容与扩展名不匹配')
        if ext != 'zip':
            required_prefix = {
                'docx': 'word/',
                'xlsx': 'xl/',
                'pptx': 'ppt/',
            }[ext]
            if '[Content_Types].xml' not in names or not any(
                name.startswith(required_prefix) for name in names
            ):
                raise ValueError('文件内容与扩展名不匹配')
    if ext in {'mp4', 'mov'} and (len(content) < 12 or content[4:8] != b'ftyp'):
        raise ValueError('文件内容与扩展名不匹配')
    if ext == 'avi' and (len(content) < 12 or content[8:12] != b'AVI '):
        raise ValueError('文件内容与扩展名不匹配')
    if ext == 'txt':
        try:
            content.decode('utf-8-sig')
        except UnicodeDecodeError:
            raise ValueError('文件内容与扩展名不匹配')
    return ext, mime


def _relative_file_url(full_url: str) -> str:
    base_url = (settings.MINIO_BASE_URL or '').rstrip('/')
    if base_url and full_url.startswith(base_url + '/'):
        return full_url[len(base_url) + 1 :]
    return full_url


def _file_size_display(size: int) -> str:
    if size < 1024:
        return f'{size} B'
    if size < 1024 * 1024:
        return f'{size / 1024:.1f} KB'
    return f'{size / (1024 * 1024):.1f} MB'


def save_market_attachment(
    db: Session,
    *,
    content: bytes,
    filename: str,
    content_type: str,
    module: str,
    ctx: MarketFieldContext,
    user_info: dict,
    storage_backend=storage,
) -> Dict[str, Any]:
    _verify_identity(ctx, user_info)
    normalized_module = str(module or '').strip()
    if normalized_module not in UPLOAD_MODULES:
        raise ValueError('不支持的附件模块')
    ext, mime = _validate_upload(
        content,
        filename,
        content_type,
        max_bytes=(
            PRESIGN_MAX_UPLOAD_BYTES
            if normalized_module == 'presign'
            else PROTOCOL_MAX_UPLOAD_BYTES
            if normalized_module == 'protocol'
            else MAX_UPLOAD_BYTES
        ),
        allowed_mime_by_ext=(
            PROTOCOL_ALLOWED_MIME_BY_EXT
            if normalized_module == 'protocol'
            else ALLOWED_MIME_BY_EXT
        ),
    )
    folder = f'market/{ctx.scope_kind}/{normalized_module}'
    full_url = storage_backend.upload_file(content, filename, folder)
    file_url = _relative_file_url(full_url)
    row = BizAttachment(
        id=uuid.uuid4().hex,
        business_type=f'market:{ctx.scope_kind}:{normalized_module}:staging',
        business_id=None,
        original_name=filename,
        file_ext=ext,
        file_url=file_url,
        file_size=len(content),
        file_type=mime,
        tenant_no=ctx.tenant_no,
        create_by=ctx.user_id,
        create_dept_id=str(
            user_info.get('dept_id')
            or user_info.get('deptId')
            or user_info.get('region_id')
            or ''
        ).strip(),
        del_flag=0,
    )
    committed = False
    try:
        db.add(row)
        db.flush()
        db.refresh(row)
        result = {
            'id': row.id,
            'fileName': row.original_name,
            'fileUrl': row.file_url,
            'fileSize': row.file_size,
            'fileSizeDisplay': _file_size_display(row.file_size or 0),
            'fileType': row.file_type,
        }
        db.commit()
        committed = True
    except Exception:
        if not committed:
            db.rollback()
            try:
                storage_backend.delete_file(full_url)
            except Exception:
                pass
        raise
    return result


def _split_refs(refs: Any) -> List[str]:
    if refs is None:
        return []
    if isinstance(refs, str):
        if len(refs) > MAX_ATTACHMENT_REFS_LENGTH:
            raise ValueError('附件引用总长度超出限制')
        raw: Iterable[Any] = refs.split(',')
    elif isinstance(refs, (list, tuple, set)):
        raw = refs
    else:
        raise ValueError('附件引用格式无效')
    raw = list(raw)
    if len(raw) > MAX_ATTACHMENT_REFS:
        raise ValueError('附件数量不能超过 20 个')
    values = []
    total_length = 0
    for item in raw:
        if isinstance(item, dict):
            item = item.get('id') or item.get('fileUrl') or item.get('url')
        value = str(item or '').strip()
        if len(value) > MAX_ATTACHMENT_REF_LENGTH:
            raise ValueError('附件引用长度无效')
        total_length += len(value)
        if total_length + max(0, len(raw) - 1) > MAX_ATTACHMENT_REFS_LENGTH:
            raise ValueError('附件引用总长度超出限制')
        if value and value not in values:
            values.append(value)
    return values


def _attachment_candidates(value: str) -> List[str]:
    candidates = [value]
    base = (settings.MINIO_BASE_URL or '').rstrip('/')
    if base and value.startswith(base + '/'):
        candidates.append(value[len(base) + 1 :])
    return list(dict.fromkeys(candidates))


def _attachment_value_matches(value: Any, candidates: Iterable[str]) -> bool:
    expected = set(candidates)
    for item in str(value or '').split(','):
        normalized = str(item or '').strip()
        if normalized and expected.intersection(_attachment_candidates(normalized)):
            return True
    return False


def _attachment_columns_filter(columns, candidates: Iterable[str]):
    return or_(
        *(
            column.contains(candidate)
            for column in columns
            for candidate in candidates
        )
    )


def _rows_reference_attachment(rows, field_names, candidates) -> bool:
    return any(
        _attachment_value_matches(getattr(row, field_name, None), candidates)
        for row in rows
        for field_name in field_names
    )


def _has_accessible_protocol_cert_or_insurance_reference(
    db: Session,
    *,
    candidates: List[str],
    ctx: MarketFieldContext,
    user_info: dict,
) -> bool:
    reference_models = (
        (BizProtocolStorageCert, 'cert_image_urls'),
        (BizProtocolFullmoonTodo, 'insurance_attachments'),
    )
    for model, field_name in reference_models:
        rows = (
            apply_cell_protocol_scope(
                db.query(model).join(
                    BizCollectionProtocol,
                    BizCollectionProtocol.id == model.collection_protocol_id,
                ),
                db,
                ctx,
                user_info=user_info,
            )
            .filter(
                model.tenant_no == ctx.tenant_no,
                model.del_flag == 0,
                _attachment_columns_filter(
                    (getattr(model, field_name),), candidates
                ),
            )
            .all()
        )
        if _rows_reference_attachment(rows, (field_name,), candidates):
            return True
    return False


def _has_accessible_protocol_deposit_attachment_reference(
    db: Session,
    *,
    candidates: List[str],
    ctx: MarketFieldContext,
    user_info: dict,
) -> bool:
    """正式协议意向金凭证挂在预签转账上，经 source_presign_order_id 关联授权。"""
    transfer_rows = (
        db.query(BizPresignTransfer)
        .filter(
            BizPresignTransfer.tenant_no == ctx.tenant_no,
            BizPresignTransfer.del_flag == 0,
            _attachment_columns_filter(
                (BizPresignTransfer.transfer_attachments,), candidates
            ),
        )
        .all()
    )
    if not _rows_reference_attachment(
        transfer_rows, ('transfer_attachments',), candidates
    ):
        return False
    presign_ids = {
        str(row.presign_order_id or '').strip()
        for row in transfer_rows
        if str(row.presign_order_id or '').strip()
    }
    if not presign_ids:
        return False
    protocol_rows = (
        apply_cell_protocol_scope(
            db.query(BizCollectionProtocol),
            db,
            ctx,
            user_info=user_info,
        )
        .filter(
            BizCollectionProtocol.tenant_no == ctx.tenant_no,
            BizCollectionProtocol.del_flag == 0,
            BizCollectionProtocol.source_presign_order_id.in_(presign_ids),
        )
        .limit(1)
        .all()
    )
    return bool(protocol_rows)


def _has_accessible_market_attachment_reference(
    db: Session,
    *,
    module: str,
    candidates: List[str],
    ctx: MarketFieldContext,
    user_info: dict,
) -> bool:
    """附件必须被当前模块中调用人可访问的业务记录引用。"""
    if module == 'protocol':
        cell_fields = (
            'protocol_attachments',
            'consent_form_urls',
            'health_survey_form_urls',
        )
        cell_columns = tuple(
            getattr(BizCollectionProtocol, field) for field in cell_fields
        )
        cell_rows = (
            apply_cell_protocol_scope(
                db.query(BizCollectionProtocol),
                db,
                ctx,
                user_info=user_info,
            )
            .filter(_attachment_columns_filter(cell_columns, candidates))
            .all()
        )
        if _rows_reference_attachment(cell_rows, cell_fields, candidates):
            return True
        payment_detail_rows = (
            apply_cell_protocol_scope(
                db.query(BizProtocolPaymentDetail)
                .join(
                    BizProtocolPayment,
                    BizProtocolPayment.id == BizProtocolPaymentDetail.payment_id,
                )
                .join(
                    BizCollectionProtocol,
                    BizCollectionProtocol.id
                    == BizProtocolPayment.collection_protocol_id,
                ),
                db,
                ctx,
                user_info=user_info,
            )
            .filter(
                BizProtocolPaymentDetail.tenant_no == ctx.tenant_no,
                BizProtocolPaymentDetail.del_flag == 0,
                BizProtocolPayment.tenant_no == ctx.tenant_no,
                BizProtocolPayment.del_flag == 0,
                _attachment_columns_filter(
                    (BizProtocolPaymentDetail.transfer_attachments,), candidates
                ),
            )
            .all()
        )
        if _rows_reference_attachment(
            payment_detail_rows, ('transfer_attachments',), candidates
        ):
            return True
        if _has_accessible_protocol_cert_or_insurance_reference(
            db,
            candidates=candidates,
            ctx=ctx,
            user_info=user_info,
        ):
            return True
        if _has_accessible_protocol_deposit_attachment_reference(
            db,
            candidates=candidates,
            ctx=ctx,
            user_info=user_info,
        ):
            return True
        donation_fields = (
            'consent_form_urls',
            'attachment_urls',
            'infectious_disease_report_urls',
        )
        donation_columns = tuple(
            getattr(BizResearchDonation, field) for field in donation_fields
        )
        donation_rows = (
            apply_donation_scope(
                db.query(BizResearchDonation),
                db,
                ctx,
                user_info=user_info,
            )
            .filter(_attachment_columns_filter(donation_columns, candidates))
            .all()
        )
        return _rows_reference_attachment(
            donation_rows, donation_fields, candidates
        )

    if module == 'presign':
        rows = (
            apply_presign_scope(
                db.query(BizPresignTransfer).join(
                    BizPresignOrder,
                    BizPresignOrder.id == BizPresignTransfer.presign_order_id,
                ),
                db,
                ctx,
                user_info=user_info,
            )
            .filter(
                BizPresignTransfer.tenant_no == ctx.tenant_no,
                BizPresignTransfer.del_flag == 0,
                _attachment_columns_filter(
                    (BizPresignTransfer.transfer_attachments,), candidates
                ),
            )
            .all()
        )
        return _rows_reference_attachment(
            rows, ('transfer_attachments',), candidates
        )

    if module == 'payment':
        rows = (
            apply_cell_protocol_scope(
                db.query(BizProtocolPaymentTransfer).join(
                    BizCollectionProtocol,
                    BizCollectionProtocol.id
                    == BizProtocolPaymentTransfer.collection_protocol_id,
                ),
                db,
                ctx,
                user_info=user_info,
            )
            .filter(
                BizProtocolPaymentTransfer.tenant_no == ctx.tenant_no,
                BizProtocolPaymentTransfer.del_flag == 0,
                _attachment_columns_filter(
                    (BizProtocolPaymentTransfer.transfer_attachments,), candidates
                ),
            )
            .all()
        )
        return _rows_reference_attachment(
            rows, ('transfer_attachments',), candidates
        )

    if module == 'sample':
        return _has_accessible_protocol_cert_or_insurance_reference(
            db,
            candidates=candidates,
            ctx=ctx,
            user_info=user_info,
        )
    return False


def get_market_attachment_download_url(
    db: Session,
    *,
    file_url: str,
    file_name: Optional[str],
    module: str,
    ctx: MarketFieldContext,
    user_info: dict,
    storage_backend=storage,
) -> Dict[str, str]:
    """为市场移动端返回附件下载地址。

    新附件按市场模块隔离；历史 Web 附件没有 business_type，
    但仍必须是当前租户已登记的 BizAttachment 记录。
    """
    _verify_identity(ctx, user_info)
    normalized_module = str(module or '').strip()
    if normalized_module not in UPLOAD_MODULES:
        raise ValueError('不支持的附件模块')
    normalized_url = str(file_url or '').strip()
    if not normalized_url or len(normalized_url) > MAX_ATTACHMENT_REF_LENGTH:
        raise ValueError('文件地址无效')
    candidates = _attachment_candidates(normalized_url)
    rows = (
        db.query(BizAttachment)
        .filter(
            BizAttachment.tenant_no == ctx.tenant_no,
            BizAttachment.del_flag == 0,
            BizAttachment.file_url.in_(candidates),
        )
        .all()
    )
    expected_staging = f'market:{ctx.scope_kind}:{normalized_module}:staging'
    allowed_bound = {
        f'market:specialist:{normalized_module}:bound',
        f'market:manager:{normalized_module}:bound',
    }
    legacy_sample_bound = {
        'market:specialist:sample:bound',
        'market:manager:sample:bound',
    }
    legacy_presign_bound = {
        'market:specialist:presign:bound',
        'market:manager:presign:bound',
    }
    row = next(
        (
            item
            for item in rows
            if (
                str(item.business_type or '').strip() == expected_staging
                and str(item.create_by or '') == ctx.user_id
                and not item.business_id
            )
        ),
        None,
    )
    if row is None:
        referenced_rows = [
            item
            for item in rows
            if str(item.business_type or '').strip() in ({''} | allowed_bound)
        ]
        if referenced_rows and _has_accessible_market_attachment_reference(
            db,
            module=normalized_module,
            candidates=candidates,
            ctx=ctx,
            user_info=user_info,
        ):
            row = referenced_rows[0]
    if row is None and normalized_module == 'protocol':
        sample_bound_rows = [
            item
            for item in rows
            if str(item.business_type or '').strip() in legacy_sample_bound
        ]
        if (
            sample_bound_rows
            and _has_accessible_protocol_cert_or_insurance_reference(
                db,
                candidates=candidates,
                ctx=ctx,
                user_info=user_info,
            )
        ):
            row = sample_bound_rows[0]
    if row is None and normalized_module == 'protocol':
        deposit_bound_rows = [
            item
            for item in rows
            if str(item.business_type or '').strip() in legacy_presign_bound
        ]
        if (
            deposit_bound_rows
            and _has_accessible_protocol_deposit_attachment_reference(
                db,
                candidates=candidates,
                ctx=ctx,
                user_info=user_info,
            )
        ):
            row = deposit_bound_rows[0]
    if row is None:
        raise PermissionError('附件不存在或无权下载')
    download_name = str(row.original_name or '').strip()
    if not download_name:
        download_name = str(file_name or '').strip()
    if not download_name:
        download_name = str(row.file_url or '').rsplit('/', 1)[-1] or 'download'
    return {
        'downloadUrl': storage_backend.presigned_download_url(
            str(row.file_url), download_name
        )
    }


def validate_attachment_refs(
    db: Session,
    refs: Any,
    *,
    module: str,
    ctx: MarketFieldContext,
    creating: bool,
    existing_refs: Any = None,
    business_id: Optional[str] = None,
) -> str:
    """批量锁定并解析附件；有 business_id 时原子消费当前角色暂存记录。"""
    normalized_module = str(module or '').strip()
    if normalized_module not in UPLOAD_MODULES:
        raise ValueError('不支持的附件模块')
    values = _split_refs(refs)
    if not values:
        return ''
    existing = set(_split_refs(existing_refs)) if not creating else set()
    new_values = [value for value in values if value not in existing]
    if not new_values:
        return ','.join(values)
    normalized_business_id = str(business_id or '').strip()
    candidates_by_value = {
        value: _attachment_candidates(value) for value in new_values
    }
    all_candidates = {
        candidate
        for candidates in candidates_by_value.values()
        for candidate in candidates
    }
    rows = (
        db.query(BizAttachment)
        .filter(
            BizAttachment.tenant_no == ctx.tenant_no,
            BizAttachment.del_flag == 0,
            or_(
                BizAttachment.id.in_(new_values),
                BizAttachment.file_url.in_(all_candidates),
            ),
        )
        .with_for_update()
        .all()
    )
    by_id = {str(row.id): row for row in rows}
    by_url = {str(row.file_url): row for row in rows}
    expected_staging = f'market:{ctx.scope_kind}:{normalized_module}:staging'
    expected_bound = f'market:{ctx.scope_kind}:{normalized_module}:bound'
    resolved = []
    seen_row_ids = set()
    for value in values:
        if value in existing:
            if value not in resolved:
                resolved.append(value)
            continue
        row = by_id.get(value)
        if row is None:
            row = next(
                (
                    by_url.get(candidate)
                    for candidate in candidates_by_value[value]
                    if by_url.get(candidate)
                ),
                None,
            )
        if row is None:
            raise ValueError('附件不存在或无权引用')
        if row.id in seen_row_ids:
            continue
        seen_row_ids.add(row.id)
        is_current_staging = (
            row.business_type == expected_staging
            and row.create_by == ctx.user_id
            and not row.business_id
        )
        is_same_binding = (
            bool(normalized_business_id)
            and row.business_type == expected_bound
            and str(row.business_id or '') == normalized_business_id
        )
        if not (is_current_staging or is_same_binding):
            raise ValueError('附件不存在、已被绑定或无权引用')
        if normalized_business_id and is_current_staging:
            row.business_type = expected_bound
            row.business_id = normalized_business_id
            db.add(row)
        resolved.append(str(row.file_url))
    if normalized_business_id:
        db.flush()
    return ','.join(resolved)


def allowed_actions(
    module: str,
    *,
    status: str,
    scope_kind: str = 'specialist',
    source: str = '',
    has_sample_code: bool = False,
    has_payment_permission: bool = False,
) -> List[str]:
    """纯函数动作白名单；路由仍必须执行对应权限与状态校验。"""
    module, state = str(module or '').strip(), str(status or '').strip().lower()
    if module in {'presign-list', 'protocol-list', 'freight-list'}:
        return ['create']
    if module == 'presign':
        if state == 'converted':
            return ['edit']
        if state == 'void':
            return ['edit', 'delete']
        return ['edit', 'convert', 'delete']
    if module == 'presign-transfer':
        if state == 'dfj':
            return ['edit', 'delete', 'submit']
        if state == 'jfqrz':
            return ['withdraw']
        return []
    if module == 'protocol':
        return ['edit', 'delete']
    if module == 'payment':
        return ['register-transfer']
    if module == 'payment-transfer':
        if state == 'dfj':
            return ['edit-transfer', 'delete-payment-transfer', 'submit-finance']
        if state == 'jfqrz':
            return ['withdraw-transfer']
        return []
    if module == 'payment-chase':
        return []
    if module == 'sample':
        return []
    if module == 'freight':
        return ['edit'] if scope_kind == 'manager' else []
    return []


def decorate_allowed_actions(
    module: str,
    item: Dict[str, Any],
    *,
    scope_kind: str,
    has_payment_permission: bool = False,
    detail: bool = False,
) -> Dict[str, Any]:
    """为移动列表/详情 DTO 增加动作白名单，不把动作判断交给客户端。"""
    data = dict(item)
    basic = data.get('basic') if isinstance(data.get('basic'), dict) else {}
    status = str(
        data.get('orderStatus')
        or data.get('protocolStatus')
        or data.get('donationStatus')
        or data.get('payStatus')
        or data.get('status')
        or basic.get('protocolStatus')
        or basic.get('donationStatus')
        or basic.get('status')
        or ''
    )
    if module == 'sample':
        card = data.get('card') if isinstance(data.get('card'), dict) else data
        source = str(card.get('source') or data.get('source') or '')
        sample_code = str(card.get('sampleCode') or data.get('sampleCode') or '')
        candidates = allowed_actions(
            module,
            status=status,
            scope_kind=scope_kind,
            source=source,
            has_sample_code=bool(sample_code.strip()),
        )
        progress = data.get('progress') if isinstance(data.get('progress'), dict) else card.get('progress') or {}
        storage_ready = str((progress.get('storage') or {}).get('code') or '') == 'yrk'
        handles = next(
            (
                section
                for section in data.get('sections') or []
                if section.get('key') == 'market-handles'
            ),
            {},
        )
        storage_cert = card.get('storageCert') or handles.get('storageCert') or {}
        insurance = card.get('insurance') or handles.get('insurance') or {}
        data['allowedActions'] = [
            action
            for action in candidates
            if storage_ready
            and (
                (action == 'storage-cert' and storage_cert.get('code') != 'handled')
                or (action == 'insurance' and insurance.get('code') != 'handled')
            )
        ]
        return data
    if module == 'payment':
        schedules = []
        has_writable_schedule = False
        for schedule in data.get('schedules') or []:
            schedule_data = dict(schedule)
            try:
                arrears = Decimal(str(schedule_data.get('arrearsAmount') or 0))
            except (ArithmeticError, ValueError):
                arrears = Decimal('0')
            schedule_writable = (
                str(schedule_data.get('scheduleStatus') or '').lower() != 'void'
                and arrears > 0
            )
            schedule_data['allowedActions'] = (
                ['register-transfer'] if schedule_writable else []
            )
            has_writable_schedule = has_writable_schedule or schedule_writable
            transfers = []
            for transfer in schedule_data.get('transfers') or []:
                transfer_data = dict(transfer)
                transfer_data['allowedActions'] = (
                    allowed_actions(
                        'payment-transfer',
                        status=str(transfer_data.get('payStatus') or ''),
                    )
                )
                transfers.append(transfer_data)
            schedule_data['transfers'] = transfers
            schedules.append(schedule_data)
        data['schedules'] = schedules
        data['allowedActions'] = (
            ['register-transfer'] if has_writable_schedule else []
        )
        return data
    if module == 'presign' and detail:
        parent_writable = status.lower() in {'', 'draft', 'active'}
        data['transferAllowedActions'] = (
            ['create-transfer'] if parent_writable else []
        )
        transfers = []
        for transfer in data.get('transfers') or []:
            transfer_data = dict(transfer)
            transfer_data['allowedActions'] = (
                allowed_actions(
                    'presign-transfer',
                    status=str(transfer_data.get('payStatus') or ''),
                )
                if parent_writable
                else []
            )
            transfers.append(transfer_data)
        data['transfers'] = transfers
    if module == 'freight' and (detail or isinstance(data.get('allowedActions'), list)):
        provided = data.get('allowedActions')
        safe_actions = {'edit', 'delete', 'link', 'unlink'}
        data['allowedActions'] = (
            [action for action in provided if action in safe_actions]
            if isinstance(provided, list)
            else ['link', 'unlink']
        )
        return data
    if module == 'protocol':
        sample_payments = []
        for payment in data.get('samplePayments') or []:
            payment_data = dict(payment)
            schedules = []
            for schedule in payment_data.get('schedules') or []:
                schedule_data = dict(schedule)
                try:
                    arrears = Decimal(str(schedule_data.get('arrearsAmount') or 0))
                except (ArithmeticError, ValueError):
                    arrears = Decimal('0')
                schedule_writable = (
                    has_payment_permission
                    and str(schedule_data.get('scheduleStatus') or '').lower() != 'void'
                    and arrears > 0
                )
                schedule_data['allowedActions'] = (
                    ['register-transfer'] if schedule_writable else []
                )
                transfers = []
                for transfer in schedule_data.get('transfers') or []:
                    transfer_data = dict(transfer)
                    transfer_data['allowedActions'] = (
                        allowed_actions(
                            'payment-transfer',
                            status=str(transfer_data.get('payStatus') or ''),
                        )
                        if has_payment_permission
                        else []
                    )
                    transfers.append(transfer_data)
                schedule_data['transfers'] = transfers
                schedules.append(schedule_data)
            payment_data['schedules'] = schedules
            sample_payments.append(payment_data)
        data['samplePayments'] = sample_payments
    actions = allowed_actions(
        module,
        status=status,
        scope_kind=scope_kind,
        has_payment_permission=has_payment_permission,
    )
    data['allowedActions'] = actions
    return data