730 lines
25 KiB
Python
730 lines
25 KiB
Python
"""移动市场表单选项、附件暂存与动作白名单。
|
|
|
|
该模块是移动市场写接口的安全边界：客户端不能选择任意字典、目录或附件 URL。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import uuid
|
|
import zipfile
|
|
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,
|
|
BizCollectionHospital,
|
|
BizProtocolCollectionInfo,
|
|
BizCollectionRegion,
|
|
BizCollectionRegionStaff,
|
|
BizItemUnit,
|
|
SysDictData,
|
|
SysUser,
|
|
)
|
|
from app.models.research_donation import BizResearchDonationCollection
|
|
from app.services.market_field_scope import (
|
|
MarketFieldContext,
|
|
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
|
|
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',
|
|
},
|
|
}
|
|
|
|
|
|
def _option(value: Any, label: Any) -> Dict[str, str]:
|
|
return {'value': str(value or ''), 'label': str(label or '')}
|
|
|
|
|
|
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 _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 get_form_options(
|
|
db: Session,
|
|
*,
|
|
module: str,
|
|
ctx: MarketFieldContext,
|
|
user_info: dict,
|
|
region_id: Optional[str] = None,
|
|
business_id: Optional[str] = None,
|
|
) -> Dict[str, List[dict]]:
|
|
"""返回移动表单最小选项集；不会返回联系方式或证件等 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 = authorized_region_ids(db, ctx)
|
|
document_region = _business_region_id(
|
|
db,
|
|
module=normalized_module,
|
|
business_id=str(business_id or ''),
|
|
ctx=ctx,
|
|
user_info=user_info,
|
|
)
|
|
if document_region and document_region not in region_ids:
|
|
region_ids.append(document_region)
|
|
selected_region = str(region_id or '').strip()
|
|
if selected_region and selected_region not in region_ids:
|
|
raise PermissionError('所选地区不在当前用户权限范围内')
|
|
hospital_regions = [selected_region] if selected_region else region_ids
|
|
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()
|
|
)
|
|
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),
|
|
}
|
|
)
|
|
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': [_option(row.id, row.type_name) for row in sample_types],
|
|
'units': [_option(row.id, row.unit_name) for row in units],
|
|
}
|
|
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'),
|
|
}
|
|
)
|
|
return result
|
|
|
|
|
|
def _validate_upload(
|
|
content: bytes,
|
|
filename: str,
|
|
content_type: str,
|
|
*,
|
|
max_bytes: int = MAX_UPLOAD_BYTES,
|
|
) -> 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()
|
|
if ext not in ALLOWED_MIME_BY_EXT or mime not in ALLOWED_MIME_BY_EXT[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',),
|
|
}
|
|
if ext not in {'heic', 'heif'} and not any(
|
|
content.startswith(prefix) for prefix in signatures[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'}:
|
|
try:
|
|
with zipfile.ZipFile(BytesIO(content)) as archive:
|
|
names = set(archive.namelist())
|
|
except (OSError, zipfile.BadZipFile):
|
|
raise ValueError('文件内容与扩展名不匹配')
|
|
required_prefix = 'word/' if ext == 'docx' else 'xl/'
|
|
if '[Content_Types].xml' not in names or not any(
|
|
name.startswith(required_prefix) for name in names
|
|
):
|
|
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 MAX_UPLOAD_BYTES
|
|
),
|
|
)
|
|
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 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'] if state in {'', 'draft', 'dsx', 'dqr', 'sxz', 'ygq'} else []
|
|
if module == 'payment':
|
|
return [] if state in PAYMENT_TERMINAL_PROTOCOL_STATUSES else ['register-transfer']
|
|
if module == 'payment-transfer':
|
|
if state == 'dfj':
|
|
return ['edit-transfer', 'submit-finance']
|
|
if state == 'jfqrz':
|
|
return ['withdraw-transfer']
|
|
return []
|
|
if module == 'payment-chase':
|
|
return ['register-transfer'] if has_payment_permission else []
|
|
if module == 'sample':
|
|
if source == 'cell_storage' and has_sample_code:
|
|
return ['storage-cert', 'insurance']
|
|
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('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 '')
|
|
data['allowedActions'] = allowed_actions(
|
|
module,
|
|
status=status,
|
|
scope_kind=scope_kind,
|
|
source=source,
|
|
has_sample_code=bool(sample_code.strip()),
|
|
)
|
|
return data
|
|
if module == 'payment':
|
|
parent_writable = status.lower() not in PAYMENT_TERMINAL_PROTOCOL_STATUSES
|
|
schedules = []
|
|
has_writable_schedule = False
|
|
for schedule in data.get('schedules') or []:
|
|
schedule_data = dict(schedule)
|
|
schedule_writable = (
|
|
parent_writable
|
|
and str(schedule_data.get('scheduleStatus') or '').lower()
|
|
== PAYMENT_WRITABLE_SCHEDULE_STATUS
|
|
)
|
|
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 ''),
|
|
)
|
|
if schedule_writable
|
|
else []
|
|
)
|
|
transfers.append(transfer_data)
|
|
schedule_data['transfers'] = transfers
|
|
schedules.append(schedule_data)
|
|
data['schedules'] = schedules
|
|
data['allowedActions'] = (
|
|
allowed_actions(module, status=status)
|
|
if parent_writable and has_writable_schedule
|
|
else []
|
|
)
|
|
return data
|
|
if module == 'presign' and detail:
|
|
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 ''),
|
|
)
|
|
transfers.append(transfer_data)
|
|
data['transfers'] = transfers
|
|
if module == 'freight' and detail:
|
|
provided = data.get('allowedActions')
|
|
safe_actions = {'edit', 'link', 'unlink'}
|
|
data['allowedActions'] = (
|
|
[action for action in provided if action in safe_actions]
|
|
if isinstance(provided, list)
|
|
else ['link', 'unlink']
|
|
)
|
|
return data
|
|
data['allowedActions'] = allowed_actions(
|
|
module,
|
|
status=status,
|
|
scope_kind=scope_kind,
|
|
has_payment_permission=has_payment_permission,
|
|
)
|
|
return data
|