3222 lines
112 KiB
Python
3222 lines
112 KiB
Python
"""采集协议单 / 审核 / 缴费明细 CRUD（细胞存储 2.2.6～2.2.10）"""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import math
|
||
import calendar
|
||
from datetime import datetime, timedelta
|
||
from decimal import Decimal
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
from sqlalchemy import and_, func, or_
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.utils import generate_uuid, parse_datetime, format_date, optional_date
|
||
from app.crud.base import camel_to_snake
|
||
from app.crud.biz_no_generator import allocate_next_biz_no
|
||
from app.crud.protocol_donation_convert import (
|
||
assert_biz_no_available,
|
||
get_partner_sample_codes,
|
||
)
|
||
from app.services.protocol_payment_installment import (
|
||
calc_fq_installment_due_date,
|
||
calc_installment_due_amounts_by_ratios,
|
||
parse_installment_ratios,
|
||
resolve_installment_due_amounts,
|
||
)
|
||
from app.services.protocol_pay_mode_switch import validate_payment_details_for_save
|
||
from app.models.research_donation import BizResearchDonation, BizResearchDonationSample
|
||
from app.models.models import (
|
||
BizAttachment,
|
||
BizCollectionProtocol,
|
||
BizProtocolAudit,
|
||
BizProtocolCollectionInfo,
|
||
BizProtocolFullmoonTodo,
|
||
BizProtocolPartyCustomer,
|
||
BizProtocolPayment,
|
||
BizProtocolPaymentDetail,
|
||
BizProtocolSample,
|
||
BizProtocolSampleTransport,
|
||
BizProtocolStorageCert,
|
||
BizQcWorkOrder,
|
||
BizSampleReception,
|
||
BizStorageReleaseReview,
|
||
BizCollectionRegion,
|
||
BizCollectionHospital,
|
||
BizCellSampleType,
|
||
BizExpressCompany,
|
||
BizItemUnit,
|
||
SysSite,
|
||
SysUser,
|
||
)
|
||
|
||
_NEWBORN_SEX_LABEL = {"1": "男", "0": "女"}
|
||
_SAMPLE_STATUS_LABEL = {
|
||
"ycj": "已采集",
|
||
"jcz": "检测中",
|
||
"yrk": "已入库",
|
||
"yzf": "已作废",
|
||
}
|
||
|
||
|
||
def get_protocol_nos_used_in_protocols(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
protocol_nos: List[str],
|
||
) -> set:
|
||
"""返回已在采集协议单或科研捐赠单中占用的协议编号集合（同租户、未删除）。"""
|
||
nos = list({(n or "").strip() for n in protocol_nos if (n or "").strip()})
|
||
if not nos:
|
||
return set()
|
||
used: set = set()
|
||
proto_rows = (
|
||
db.query(BizCollectionProtocol.protocol_no)
|
||
.filter(
|
||
BizCollectionProtocol.tenant_no == tenant_no,
|
||
BizCollectionProtocol.del_flag == 0,
|
||
BizCollectionProtocol.protocol_no.in_(nos),
|
||
)
|
||
.all()
|
||
)
|
||
used.update(r[0] for r in proto_rows if r[0])
|
||
donation_rows = (
|
||
db.query(BizResearchDonation.donation_no)
|
||
.filter(
|
||
BizResearchDonation.tenant_no == tenant_no,
|
||
BizResearchDonation.del_flag == 0,
|
||
BizResearchDonation.donation_no.in_(nos),
|
||
)
|
||
.all()
|
||
)
|
||
used.update(r[0] for r in donation_rows if r[0])
|
||
return used
|
||
|
||
|
||
def get_sample_codes_used_in_protocol_samples(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
sample_codes: List[str],
|
||
) -> set:
|
||
"""返回已在采集协议单或科研捐赠单样本表中占用的样本编码集合（同租户、未删除）。"""
|
||
codes = list({(c or "").strip() for c in sample_codes if (c or "").strip()})
|
||
if not codes:
|
||
return set()
|
||
used: set = set()
|
||
proto_rows = (
|
||
db.query(BizProtocolSample.sample_code)
|
||
.filter(
|
||
BizProtocolSample.tenant_no == tenant_no,
|
||
BizProtocolSample.del_flag == 0,
|
||
BizProtocolSample.sample_code.in_(codes),
|
||
)
|
||
.all()
|
||
)
|
||
used.update(r[0] for r in proto_rows if r[0])
|
||
donation_rows = (
|
||
db.query(BizResearchDonationSample.sample_code)
|
||
.filter(
|
||
BizResearchDonationSample.tenant_no == tenant_no,
|
||
BizResearchDonationSample.del_flag == 0,
|
||
BizResearchDonationSample.sample_code.in_(codes),
|
||
)
|
||
.all()
|
||
)
|
||
used.update(r[0] for r in donation_rows if r[0])
|
||
return used
|
||
|
||
|
||
def validate_sample_codes_not_used(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
samples: list,
|
||
exclude_protocol_id: Optional[str] = None,
|
||
exclude_donation_id: Optional[str] = None,
|
||
) -> None:
|
||
"""创建/更新样本行前校验：样本编码不得已被任一业务单占用（转移配对单豁免）。"""
|
||
codes: List[str] = []
|
||
for raw_item in samples or []:
|
||
raw = _normalize_fields(raw_item)
|
||
c = (raw.get("sample_code") or "").strip()
|
||
if c:
|
||
codes.append(c)
|
||
if not codes:
|
||
return
|
||
used = get_sample_codes_used_in_protocol_samples(
|
||
db, tenant_no=tenant_no, sample_codes=codes
|
||
)
|
||
exempt = get_partner_sample_codes(
|
||
db,
|
||
tenant_no,
|
||
protocol_id=exclude_protocol_id,
|
||
donation_id=exclude_donation_id,
|
||
)
|
||
used -= exempt
|
||
conflicts = set(codes) & used
|
||
if conflicts:
|
||
first = sorted(conflicts)[0]
|
||
raise ValueError(f"样本编码「{first}」已被占用")
|
||
|
||
|
||
def total_pages(total: int, page_size: int) -> int:
|
||
if total <= 0:
|
||
return 0
|
||
return int(math.ceil(total / float(page_size)))
|
||
|
||
|
||
def _fmt_dt(t: Optional[datetime]) -> str:
|
||
return t.strftime("%Y-%m-%d %H:%M:%S") if t else ""
|
||
|
||
|
||
def _dec(v: Any) -> Optional[float]:
|
||
if v is None:
|
||
return None
|
||
if isinstance(v, Decimal):
|
||
return float(v)
|
||
return float(v)
|
||
|
||
|
||
def _optional_decimal(v: Any) -> Optional[Decimal]:
|
||
if v is None or v == "":
|
||
return None
|
||
return Decimal(str(v))
|
||
|
||
|
||
def _today_yyyymmdd() -> str:
|
||
return datetime.now().strftime("%Y%m%d")
|
||
|
||
|
||
def _normalize_yyyymmdd_date(val: Any) -> Optional[str]:
|
||
"""将 YYYYMMDD / YYYY-MM-DD 等格式规范为 8 位日期字符串。"""
|
||
if val is None:
|
||
return None
|
||
s = str(val).strip()
|
||
if not s:
|
||
return None
|
||
compact = s.replace("-", "").replace("/", "")[:8]
|
||
if len(compact) == 8 and compact.isdigit():
|
||
return compact
|
||
return None
|
||
|
||
|
||
_SNAKE_ALIASES = {
|
||
"apgar1min": "apgar_1min",
|
||
"apgar5min": "apgar_5min",
|
||
}
|
||
|
||
|
||
def _normalize_fields(fields: Optional[dict]) -> dict:
|
||
"""将前端 camelCase 字段名转为数据库 snake_case。"""
|
||
if not fields:
|
||
return {}
|
||
out: Dict[str, Any] = {}
|
||
for k, v in fields.items():
|
||
sk = _SNAKE_ALIASES.get(k) or camel_to_snake(k)
|
||
out[sk] = v
|
||
return out
|
||
|
||
|
||
def _pop_attachment_urls(raw: Dict[str, Any], *keys: str) -> Optional[str]:
|
||
"""从保存载荷中取出附件字段（支持 snake/camel、URL 串或附件对象列表）。"""
|
||
for key in keys:
|
||
if key not in raw:
|
||
continue
|
||
urls = attachments_to_urls(raw.pop(key))
|
||
if urls:
|
||
return urls
|
||
return None
|
||
|
||
|
||
def attachments_to_urls(attachments: Any) -> Optional[str]:
|
||
"""附件列表或字符串 → 逗号分隔 URL（MinIO）。"""
|
||
if attachments is None:
|
||
return None
|
||
if isinstance(attachments, str):
|
||
return attachments.strip() or None
|
||
if isinstance(attachments, list):
|
||
urls: List[str] = []
|
||
for item in attachments:
|
||
if isinstance(item, dict):
|
||
u = (item.get("fileUrl") or item.get("file_url") or "").strip()
|
||
if u:
|
||
urls.append(u)
|
||
elif isinstance(item, str) and item.strip():
|
||
urls.append(item.strip())
|
||
return ",".join(urls) if urls else None
|
||
return None
|
||
|
||
|
||
def _normalize_attachment_url_key(url: str) -> str:
|
||
u = (url or "").strip()
|
||
if not u:
|
||
return ""
|
||
if "?" in u:
|
||
u = u.split("?", 1)[0]
|
||
return u.lstrip("/")
|
||
|
||
|
||
def _format_attachment_file_size(size_bytes: int) -> str:
|
||
if size_bytes < 1024:
|
||
return f"{size_bytes} B"
|
||
if size_bytes < 1024 * 1024:
|
||
return f"{size_bytes / 1024:.1f} KB"
|
||
return f"{size_bytes / (1024 * 1024):.1f} MB"
|
||
|
||
|
||
def _lookup_attachment_by_url(
|
||
db: Session,
|
||
url: str,
|
||
tenant_no: Optional[str],
|
||
) -> Optional[BizAttachment]:
|
||
key = _normalize_attachment_url_key(url)
|
||
if not key:
|
||
return None
|
||
q = db.query(BizAttachment).filter(BizAttachment.del_flag == 0)
|
||
if tenant_no:
|
||
q = q.filter(BizAttachment.tenant_no == tenant_no)
|
||
row = q.filter(BizAttachment.file_url == key).first()
|
||
if row:
|
||
return row
|
||
tail = key.rsplit("/", 1)[-1]
|
||
if tail:
|
||
row = (
|
||
q.filter(BizAttachment.file_url.like(f"%/{tail}"))
|
||
.order_by(BizAttachment.create_time.desc())
|
||
.first()
|
||
)
|
||
return row
|
||
|
||
|
||
def urls_to_attachment_list(
|
||
urls_str: Optional[str],
|
||
db: Optional[Session] = None,
|
||
tenant_no: Optional[str] = None,
|
||
) -> List[Dict[str, Any]]:
|
||
"""逗号分隔 URL → 前端附件卡片结构；有 db 时从 biz_attachment 取原始文件名。"""
|
||
if not urls_str:
|
||
return []
|
||
result: List[Dict[str, Any]] = []
|
||
for url in urls_str.split(","):
|
||
u = url.strip()
|
||
if not u:
|
||
continue
|
||
att = _lookup_attachment_by_url(db, u, tenant_no) if db else None
|
||
if att:
|
||
size = att.file_size or 0
|
||
result.append(
|
||
{
|
||
"id": att.id,
|
||
"fileUrl": u,
|
||
"fileName": att.original_name or u.rsplit("/", 1)[-1],
|
||
"fileSize": size,
|
||
"fileSizeDisplay": _format_attachment_file_size(size),
|
||
"fileType": att.file_type or "",
|
||
}
|
||
)
|
||
else:
|
||
name = u.rsplit("/", 1)[-1] if "/" in u else u
|
||
result.append({"fileUrl": u, "fileName": name, "fileSize": 0})
|
||
return result
|
||
|
||
|
||
def _user_nick_name_map(db: Session, user_ids: List[str]) -> Dict[str, str]:
|
||
ids = list({x for x in user_ids if x})
|
||
if not ids:
|
||
return {}
|
||
rows = db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(ids)).all()
|
||
return {u.id: u.nick_name or "" for u in rows}
|
||
|
||
|
||
def _protocol_to_dict(p: BizCollectionProtocol) -> Dict[str, Any]:
|
||
return {
|
||
"id": p.id,
|
||
"protocolNoId": p.protocol_no_id or "",
|
||
"protocolNo": p.protocol_no or "",
|
||
"protocolStatus": p.protocol_status or "",
|
||
"protocolAttachments": p.protocol_attachments or "",
|
||
"registerDate": (p.register_date or "").strip() if p.register_date else "",
|
||
"effectiveTime": _fmt_dt(p.effective_time),
|
||
"effectiveDesc": p.effective_desc or "",
|
||
"expireTime": _fmt_dt(p.expire_time),
|
||
"expireDesc": p.expire_desc or "",
|
||
"voidTime": _fmt_dt(p.void_time),
|
||
"voidDesc": p.void_desc or "",
|
||
"toResearchTime": _fmt_dt(p.to_research_time),
|
||
"toResearchDesc": p.to_research_desc or "",
|
||
"collectorUserId": p.collector_user_id or "",
|
||
"tenantNo": p.tenant_no,
|
||
"orderNum": p.order_num if p.order_num is not None else 0,
|
||
"remark": p.remark or "",
|
||
"createTime": _fmt_dt(p.create_time),
|
||
"updateTime": _fmt_dt(p.update_time),
|
||
}
|
||
|
||
|
||
def _payment_to_dict(p: BizProtocolPayment) -> Dict[str, Any]:
|
||
ratios_list = parse_installment_ratios(p.installment_ratios)
|
||
return {
|
||
"id": p.id,
|
||
"collectionProtocolId": p.collection_protocol_id,
|
||
"payMode": p.pay_mode or "",
|
||
"contractAmount": _dec(p.contract_amount),
|
||
"installmentCount": p.installment_count,
|
||
"installmentIntervalMonths": _dec(p.installment_interval_months),
|
||
"lumpSumDeadline": (p.lump_sum_deadline or "").strip() if p.lump_sum_deadline else "",
|
||
"installmentStartDate": (p.installment_start_date or "").strip() if p.installment_start_date else "",
|
||
"installmentAdvanceDays": p.installment_advance_days if p.installment_advance_days is not None else 30,
|
||
"installmentRatios": ratios_list,
|
||
"tenantNo": p.tenant_no,
|
||
"orderNum": p.order_num if p.order_num is not None else 0,
|
||
"remark": p.remark or "",
|
||
"createTime": _fmt_dt(p.create_time),
|
||
"updateTime": _fmt_dt(p.update_time),
|
||
}
|
||
|
||
|
||
def _detail_to_dict(
|
||
d: BizProtocolPaymentDetail,
|
||
*,
|
||
protocol_no: str = "",
|
||
collection_protocol_id: str = "",
|
||
protocol_status: str = "",
|
||
pay_mode: str = "",
|
||
mother_name: str = "",
|
||
mother_phone: str = "",
|
||
mother_id_card: str = "",
|
||
newborn_name: str = "",
|
||
newborn_id_card: str = "",
|
||
) -> Dict[str, Any]:
|
||
return {
|
||
"id": d.id,
|
||
"paymentId": d.payment_id,
|
||
"collectionProtocolId": collection_protocol_id,
|
||
"protocolNo": protocol_no,
|
||
"protocolStatus": protocol_status,
|
||
"payMode": pay_mode,
|
||
"motherName": mother_name,
|
||
"motherPhone": mother_phone,
|
||
"motherIdCard": mother_id_card,
|
||
"newbornName": newborn_name,
|
||
"newbornIdCard": newborn_id_card,
|
||
"transferType": d.transfer_type or "",
|
||
"dueAmount": _dec(d.due_amount),
|
||
"transferAmount": _dec(d.transfer_amount),
|
||
"transferTime": _fmt_dt(d.transfer_time),
|
||
"transferAttachments": d.transfer_attachments or "",
|
||
"financeConfirmAttachments": d.finance_confirm_attachments or "",
|
||
"financeRemark": d.finance_remark or "",
|
||
"refundFinanceConfirmAttachments": d.refund_finance_confirm_attachments or "",
|
||
"refundFinanceRemark": d.refund_finance_remark or "",
|
||
"payStatus": d.pay_status or "",
|
||
"installmentNo": d.installment_no,
|
||
"installmentDueDate": (d.installment_due_date or "").strip() if d.installment_due_date else "",
|
||
"refundAmount": _dec(d.refund_amount),
|
||
"refundRemark": d.refund_remark or "",
|
||
"refundAttachments": d.refund_attachments or "",
|
||
"tenantNo": d.tenant_no,
|
||
"orderNum": d.order_num if d.order_num is not None else 0,
|
||
"remark": d.remark or "",
|
||
"createTime": _fmt_dt(d.create_time),
|
||
"updateTime": _fmt_dt(d.update_time),
|
||
}
|
||
|
||
|
||
def _audit_to_dict(
|
||
a: BizProtocolAudit,
|
||
*,
|
||
protocol_no: str = "",
|
||
auditor_user_name: str = "",
|
||
) -> Dict[str, Any]:
|
||
return {
|
||
"id": a.id,
|
||
"collectionProtocolId": a.collection_protocol_id,
|
||
"protocolNo": protocol_no,
|
||
"auditorUserId": a.auditor_user_id or "",
|
||
"auditorUserName": auditor_user_name or "",
|
||
"auditRemark": a.audit_remark or "",
|
||
"auditStatus": a.audit_status or "",
|
||
"tenantNo": a.tenant_no,
|
||
"orderNum": a.order_num if a.order_num is not None else 0,
|
||
"remark": a.remark or "",
|
||
"createTime": _fmt_dt(a.create_time),
|
||
"updateTime": _fmt_dt(a.update_time),
|
||
}
|
||
|
||
|
||
def _party_customer_to_dict(p: BizProtocolPartyCustomer) -> Dict[str, Any]:
|
||
return {
|
||
"id": p.id,
|
||
"collectionProtocolId": p.collection_protocol_id,
|
||
"customerId": p.customer_id or "",
|
||
"motherName": p.mother_name or "",
|
||
"motherPhone": p.mother_phone or "",
|
||
"motherIdCard": p.mother_id_card or "",
|
||
"motherBirthYm": (p.mother_birth_ym or "").strip() if p.mother_birth_ym else "",
|
||
"motherEthnic": p.mother_ethnic or "",
|
||
"motherEmail": p.mother_email or "",
|
||
"motherAddress": p.mother_address or "",
|
||
"fatherName": p.father_name or "",
|
||
"fatherPhone": p.father_phone or "",
|
||
"fatherIdCard": p.father_id_card or "",
|
||
"newbornName": p.newborn_name or "",
|
||
"newbornIdCard": p.newborn_id_card or "",
|
||
"expectedDeliveryDate": format_date(p.expected_delivery_date),
|
||
"newbornDeliveryTime": _fmt_dt(p.newborn_delivery_time),
|
||
"newbornSex": p.newborn_sex or "",
|
||
"deliveryMode": p.delivery_mode or "",
|
||
"birthOrder": p.birth_order,
|
||
"newbornWeightGram": p.newborn_weight_gram,
|
||
"apgar1min": _dec(p.apgar_1min),
|
||
"apgar5min": _dec(p.apgar_5min),
|
||
"collectorUserId": p.collector_user_id or "",
|
||
"tenantNo": p.tenant_no,
|
||
"orderNum": p.order_num if p.order_num is not None else 0,
|
||
"remark": p.remark or "",
|
||
"createTime": _fmt_dt(p.create_time),
|
||
"updateTime": _fmt_dt(p.update_time),
|
||
}
|
||
|
||
|
||
def _collection_info_to_dict(c: BizProtocolCollectionInfo) -> Dict[str, Any]:
|
||
return {
|
||
"id": c.id,
|
||
"collectionProtocolId": c.collection_protocol_id,
|
||
"regionId": c.region_id or "",
|
||
"hospitalId": c.hospital_id or "",
|
||
"medicalStaff": c.medical_staff or "",
|
||
"salesUserId": c.sales_user_id or "",
|
||
"sampleCollectTime": _fmt_dt(c.sample_collect_time),
|
||
"maternalBloodCollectTime": _fmt_dt(c.maternal_blood_collect_time),
|
||
"tenantNo": c.tenant_no,
|
||
"orderNum": c.order_num if c.order_num is not None else 0,
|
||
"remark": c.remark or "",
|
||
"createTime": _fmt_dt(c.create_time),
|
||
"updateTime": _fmt_dt(c.update_time),
|
||
}
|
||
|
||
|
||
def _sample_to_dict(s: BizProtocolSample) -> Dict[str, Any]:
|
||
return {
|
||
"id": s.id,
|
||
"collectionProtocolId": s.collection_protocol_id,
|
||
"sampleTypeId": s.sample_type_id or "",
|
||
"sampleCodeId": s.sample_code_id or "",
|
||
"sampleCode": s.sample_code or "",
|
||
"collectAmount": _dec(s.collect_amount),
|
||
"unitId": s.unit_id or "",
|
||
"price": _dec(s.price),
|
||
"originalPrice": _dec(s.original_price),
|
||
"storageYears": s.storage_years,
|
||
"sampleStatus": (s.sample_status or "").strip() if (s.sample_status or "").strip() else "",
|
||
"storageInDate": (s.storage_in_date or "").strip() if s.storage_in_date else "",
|
||
"detectStartDate": (s.detect_start_date or "").strip() if s.detect_start_date else "",
|
||
"voidDate": (s.void_date or "").strip() if s.void_date else "",
|
||
"storageExpireDate": (s.storage_expire_date or "").strip() if s.storage_expire_date else "",
|
||
"tenantNo": s.tenant_no,
|
||
"orderNum": s.order_num if s.order_num is not None else 0,
|
||
"remark": s.remark or "",
|
||
"createTime": _fmt_dt(s.create_time),
|
||
"updateTime": _fmt_dt(s.update_time),
|
||
}
|
||
|
||
|
||
def _transport_to_dict(t: BizProtocolSampleTransport) -> Dict[str, Any]:
|
||
return {
|
||
"id": t.id,
|
||
"collectionProtocolId": t.collection_protocol_id,
|
||
"sampleId": t.sample_id,
|
||
"transportType": t.transport_type or "",
|
||
"expressCompanyId": t.express_company_id or "",
|
||
"waybillNo": t.waybill_no or "",
|
||
"destSiteId": t.dest_site_id or "",
|
||
"expressFee": _dec(t.express_fee),
|
||
"etaTime": _fmt_dt(t.eta_time),
|
||
"carrierName": t.carrier_name or "",
|
||
"carrierPhone": t.carrier_phone or "",
|
||
"remark": t.remark or "",
|
||
"orderNum": t.order_num if t.order_num is not None else 0,
|
||
"createTime": _fmt_dt(t.create_time),
|
||
"updateTime": _fmt_dt(t.update_time),
|
||
}
|
||
|
||
|
||
def _storage_cert_to_dict(s: BizProtocolStorageCert) -> Dict[str, Any]:
|
||
return {
|
||
"id": s.id,
|
||
"collectionProtocolId": s.collection_protocol_id,
|
||
"makeStatus": s.make_status or "0",
|
||
"certImageUrls": s.cert_image_urls or "",
|
||
"makeDate": (s.make_date or "").strip() if s.make_date else "",
|
||
"tenantNo": s.tenant_no,
|
||
"orderNum": s.order_num if s.order_num is not None else 0,
|
||
"remark": s.remark or "",
|
||
"createTime": _fmt_dt(s.create_time),
|
||
"updateTime": _fmt_dt(s.update_time),
|
||
}
|
||
|
||
|
||
def _format_newborn_info(pc: Optional[BizProtocolPartyCustomer]) -> str:
|
||
if not pc:
|
||
return ""
|
||
parts: List[str] = []
|
||
name = (pc.newborn_name or "").strip()
|
||
if name:
|
||
parts.append(name)
|
||
sex = _NEWBORN_SEX_LABEL.get((pc.newborn_sex or "").strip(), "")
|
||
if sex:
|
||
parts.append(sex)
|
||
if pc.newborn_weight_gram is not None:
|
||
parts.append(f"{pc.newborn_weight_gram}g")
|
||
if pc.newborn_delivery_time:
|
||
parts.append(_fmt_dt(pc.newborn_delivery_time))
|
||
return " / ".join(parts)
|
||
|
||
|
||
def _format_sample_collect_summary(samples: List[BizProtocolSample]) -> str:
|
||
"""汇总协议单下各样本采集情况，如「1个已采集，1个未采集」。"""
|
||
collected = 0
|
||
not_collected = 0
|
||
for s in samples:
|
||
status = (s.sample_status or "").strip()
|
||
if status:
|
||
collected += 1
|
||
else:
|
||
not_collected += 1
|
||
parts: List[str] = []
|
||
if collected:
|
||
parts.append(f"{collected}个已采集")
|
||
if not_collected:
|
||
parts.append(f"{not_collected}个未采集")
|
||
return "，".join(parts)
|
||
|
||
|
||
def _format_storage_cert_sample_info(
|
||
samples: List[Dict[str, Any]], type_map: Dict[str, str]
|
||
) -> str:
|
||
parts: List[str] = []
|
||
for s in samples:
|
||
type_name = type_map.get(s.get("sampleTypeId") or "", "") or "样本"
|
||
code = (s.get("sampleCode") or "").strip()
|
||
status = (s.get("sampleStatus") or "").strip()
|
||
seg = type_name
|
||
if code:
|
||
seg += f"({code})"
|
||
if status:
|
||
seg += f"[{_SAMPLE_STATUS_LABEL.get(status, status)}]"
|
||
parts.append(seg)
|
||
return "；".join(parts)
|
||
|
||
|
||
def _pick_storage_in_date(samples: List[Dict[str, Any]]) -> str:
|
||
for s in samples:
|
||
d = (s.get("storageInDate") or "").strip()
|
||
if d:
|
||
return d
|
||
return ""
|
||
|
||
|
||
def _load_protocol_storage_in_date_map(
|
||
db: Session,
|
||
protocol_ids: List[str],
|
||
tenant_no: str,
|
||
) -> Dict[str, str]:
|
||
"""从入库放行审核记录补全入库日期（历史数据可能未写入样本 storage_in_date）。"""
|
||
if not protocol_ids:
|
||
return {}
|
||
rows = (
|
||
db.query(BizStorageReleaseReview)
|
||
.filter(
|
||
BizStorageReleaseReview.protocol_id.in_(protocol_ids),
|
||
BizStorageReleaseReview.tenant_no == tenant_no,
|
||
BizStorageReleaseReview.del_flag == 0,
|
||
BizStorageReleaseReview.storage_status == "yrk",
|
||
BizStorageReleaseReview.storage_time.isnot(None),
|
||
)
|
||
.all()
|
||
)
|
||
out: Dict[str, str] = {}
|
||
for rev in rows:
|
||
pid = (rev.protocol_id or "").strip()
|
||
if not pid or not rev.storage_time:
|
||
continue
|
||
d = rev.storage_time.strftime("%Y%m%d")
|
||
prev = out.get(pid)
|
||
if not prev or d < prev:
|
||
out[pid] = d
|
||
return out
|
||
|
||
|
||
def _load_sample_type_name_map(db: Session, samples: List[BizProtocolSample]) -> Dict[str, str]:
|
||
type_ids = list({s.sample_type_id for s in samples if s.sample_type_id})
|
||
if not type_ids:
|
||
return {}
|
||
type_map: Dict[str, str] = {}
|
||
for t in (
|
||
db.query(BizCellSampleType)
|
||
.filter(BizCellSampleType.id.in_(type_ids), BizCellSampleType.del_flag == 0)
|
||
.all()
|
||
):
|
||
type_map[t.id] = t.type_name or ""
|
||
return type_map
|
||
|
||
|
||
def _load_unit_name_map(db: Session, samples: List[BizProtocolSample]) -> Dict[str, str]:
|
||
unit_ids = list({s.unit_id for s in samples if s.unit_id})
|
||
if not unit_ids:
|
||
return {}
|
||
unit_map: Dict[str, str] = {}
|
||
for u in (
|
||
db.query(BizItemUnit)
|
||
.filter(BizItemUnit.id.in_(unit_ids), BizItemUnit.del_flag == 0)
|
||
.all()
|
||
):
|
||
unit_map[u.id] = u.unit_name or ""
|
||
return unit_map
|
||
|
||
|
||
def _load_site_name_map(db: Session, site_ids: List[str]) -> Dict[str, str]:
|
||
ids = list({sid for sid in site_ids if sid})
|
||
if not ids:
|
||
return {}
|
||
site_map: Dict[str, str] = {}
|
||
for site in (
|
||
db.query(SysSite)
|
||
.filter(SysSite.id.in_(ids), SysSite.del_flag == 0)
|
||
.all()
|
||
):
|
||
site_map[site.id] = site.site_name or ""
|
||
return site_map
|
||
|
||
|
||
def _load_express_name_map(db: Session, express_ids: List[str]) -> Dict[str, str]:
|
||
ids = list({eid for eid in express_ids if eid})
|
||
if not ids:
|
||
return {}
|
||
express_map: Dict[str, str] = {}
|
||
for row in (
|
||
db.query(BizExpressCompany)
|
||
.filter(BizExpressCompany.id.in_(ids), BizExpressCompany.del_flag == 0)
|
||
.all()
|
||
):
|
||
express_map[row.id] = row.express_name or ""
|
||
return express_map
|
||
|
||
|
||
def list_storage_certs(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
keyword: Optional[str] = None,
|
||
make_status: Optional[str] = None,
|
||
storage_in_date_start: Optional[str] = None,
|
||
storage_in_date_end: Optional[str] = None,
|
||
page: int = 1,
|
||
page_size: int = 10,
|
||
user_info=None,
|
||
) -> Tuple[List[Dict[str, Any]], int]:
|
||
q = db.query(BizCollectionProtocol).filter(
|
||
BizCollectionProtocol.del_flag == 0,
|
||
BizCollectionProtocol.tenant_no == tenant_no,
|
||
BizCollectionProtocol.protocol_status == "sxz",
|
||
)
|
||
kw = (keyword or "").strip()
|
||
if kw:
|
||
like = f"%{kw}%"
|
||
party_ids = (
|
||
db.query(BizProtocolPartyCustomer.collection_protocol_id)
|
||
.filter(
|
||
BizProtocolPartyCustomer.del_flag == 0,
|
||
BizProtocolPartyCustomer.tenant_no == tenant_no,
|
||
BizProtocolPartyCustomer.mother_name.like(like),
|
||
)
|
||
)
|
||
q = q.filter(BizCollectionProtocol.id.in_(party_ids))
|
||
if user_info:
|
||
from app.core.data_scope import apply_data_scope
|
||
q = apply_data_scope(q, BizCollectionProtocol, user_info, db, tenant_no)
|
||
|
||
rows_all = q.order_by(BizCollectionProtocol.create_time.desc()).all()
|
||
protocol_ids = [r.id for r in rows_all]
|
||
pc_map: Dict[str, BizProtocolPartyCustomer] = {}
|
||
sc_map: Dict[str, BizProtocolStorageCert] = {}
|
||
sample_map: Dict[str, List[Dict[str, Any]]] = {}
|
||
type_map: Dict[str, str] = {}
|
||
storage_in_date_map: Dict[str, str] = {}
|
||
if protocol_ids:
|
||
for pc in (
|
||
db.query(BizProtocolPartyCustomer)
|
||
.filter(
|
||
BizProtocolPartyCustomer.collection_protocol_id.in_(protocol_ids),
|
||
BizProtocolPartyCustomer.del_flag == 0,
|
||
)
|
||
.all()
|
||
):
|
||
pc_map[pc.collection_protocol_id] = pc
|
||
|
||
for sc in (
|
||
db.query(BizProtocolStorageCert)
|
||
.filter(
|
||
BizProtocolStorageCert.collection_protocol_id.in_(protocol_ids),
|
||
BizProtocolStorageCert.del_flag == 0,
|
||
)
|
||
.all()
|
||
):
|
||
sc_map[sc.collection_protocol_id] = sc
|
||
|
||
s_rows = (
|
||
db.query(BizProtocolSample)
|
||
.filter(
|
||
BizProtocolSample.collection_protocol_id.in_(protocol_ids),
|
||
BizProtocolSample.del_flag == 0,
|
||
)
|
||
.all()
|
||
)
|
||
type_map = _load_sample_type_name_map(db, s_rows)
|
||
for s in s_rows:
|
||
sample_map.setdefault(s.collection_protocol_id, []).append(_sample_to_dict(s))
|
||
|
||
storage_in_date_map = _load_protocol_storage_in_date_map(db, protocol_ids, tenant_no)
|
||
|
||
items: List[Dict[str, Any]] = []
|
||
ms_filter = (make_status or "").strip()
|
||
start_in = _yyyymmdd_from_date_param(storage_in_date_start) if storage_in_date_start else ""
|
||
end_in = _yyyymmdd_from_date_param(storage_in_date_end) if storage_in_date_end else ""
|
||
for r in rows_all:
|
||
sc = sc_map.get(r.id)
|
||
cert_ms = (sc.make_status or "0") if sc else "0"
|
||
if ms_filter and cert_ms != ms_filter:
|
||
continue
|
||
pc = pc_map.get(r.id)
|
||
samples = sample_map.get(r.id, [])
|
||
storage_in = _pick_storage_in_date(samples) or storage_in_date_map.get(r.id, "")
|
||
if start_in or end_in:
|
||
sid = _storage_in_date_to_yyyymmdd(storage_in)
|
||
if not sid:
|
||
continue
|
||
if start_in and sid < start_in:
|
||
continue
|
||
if end_in and sid > end_in:
|
||
continue
|
||
cert_dict = _storage_cert_to_dict(sc) if sc else {"makeStatus": "0", "certImageUrls": "", "makeDate": "", "remark": ""}
|
||
items.append(
|
||
{
|
||
"id": r.id,
|
||
"protocolId": r.id,
|
||
"protocolNo": r.protocol_no or "",
|
||
"protocolStatus": r.protocol_status or "",
|
||
"motherName": (pc.mother_name or "") if pc else "",
|
||
"motherPhone": (pc.mother_phone or "") if pc else "",
|
||
"motherIdCard": (pc.mother_id_card or "") if pc else "",
|
||
"newbornName": (pc.newborn_name or "") if pc else "",
|
||
"newbornIdCard": (pc.newborn_id_card or "") if pc else "",
|
||
"newbornDeliveryTime": _fmt_dt(pc.newborn_delivery_time) if pc and pc.newborn_delivery_time else "",
|
||
"newbornSex": (pc.newborn_sex or "") if pc else "",
|
||
"newbornWeightGram": pc.newborn_weight_gram if pc else None,
|
||
"newbornInfo": _format_newborn_info(pc),
|
||
"sampleInfo": _format_storage_cert_sample_info(samples, type_map),
|
||
"storageInDate": storage_in,
|
||
"samples": samples,
|
||
"makeStatus": cert_dict.get("makeStatus", "0"),
|
||
"certImageUrls": cert_dict.get("certImageUrls", ""),
|
||
"makeDate": cert_dict.get("makeDate", ""),
|
||
"remark": cert_dict.get("remark", ""),
|
||
}
|
||
)
|
||
|
||
total = len(items)
|
||
page_safe = max(1, page)
|
||
page_size_safe = min(max(1, page_size), 100)
|
||
start = (page_safe - 1) * page_size_safe
|
||
return items[start : start + page_size_safe], total
|
||
|
||
|
||
def _normalize_date_bound(value: str, *, end_of_day: bool = False) -> str:
|
||
val = (value or "").strip()
|
||
if len(val) == 10:
|
||
return f"{val} 23:59:59" if end_of_day else f"{val} 00:00:00"
|
||
return val
|
||
|
||
|
||
def _yyyymmdd_from_date_param(value: str) -> str:
|
||
val = (value or "").strip()
|
||
if len(val) == 10 and val[4] == "-" and val[7] == "-":
|
||
return val.replace("-", "")
|
||
digits = "".join(c for c in val if c.isdigit())
|
||
return digits[:8] if len(digits) >= 8 else digits
|
||
|
||
|
||
def _storage_in_date_to_yyyymmdd(value: str) -> str:
|
||
digits = "".join(c for c in (value or "").strip() if c.isdigit())
|
||
return digits[:8] if len(digits) >= 8 else ""
|
||
|
||
|
||
def list_collection_protocols(
|
||
db: Session,
|
||
*,
|
||
page: int = 1,
|
||
page_size: int = 10,
|
||
keyword: Optional[str] = None,
|
||
protocol_status: Optional[str] = None,
|
||
create_time_start: Optional[str] = None,
|
||
create_time_end: Optional[str] = None,
|
||
sample_collect_time_start: Optional[str] = None,
|
||
sample_collect_time_end: Optional[str] = None,
|
||
tenant_no: Optional[str] = None,
|
||
scope_collector_user_id: Optional[str] = None,
|
||
user_info=None,
|
||
) -> Tuple[List[Dict[str, Any]], int]:
|
||
def _filtered():
|
||
q0 = db.query(BizCollectionProtocol).filter(BizCollectionProtocol.del_flag == 0)
|
||
if tenant_no:
|
||
q0 = q0.filter(BizCollectionProtocol.tenant_no == tenant_no)
|
||
if scope_collector_user_id:
|
||
q0 = q0.filter(
|
||
BizCollectionProtocol.collector_user_id == scope_collector_user_id
|
||
)
|
||
if protocol_status:
|
||
q0 = q0.filter(BizCollectionProtocol.protocol_status == protocol_status.strip())
|
||
if create_time_start:
|
||
q0 = q0.filter(
|
||
BizCollectionProtocol.create_time >= _normalize_date_bound(create_time_start)
|
||
)
|
||
if create_time_end:
|
||
q0 = q0.filter(
|
||
BizCollectionProtocol.create_time
|
||
<= _normalize_date_bound(create_time_end, end_of_day=True)
|
||
)
|
||
if sample_collect_time_start or sample_collect_time_end:
|
||
ci_q = db.query(BizProtocolCollectionInfo.collection_protocol_id).filter(
|
||
BizProtocolCollectionInfo.del_flag == 0,
|
||
)
|
||
if tenant_no:
|
||
ci_q = ci_q.filter(BizProtocolCollectionInfo.tenant_no == tenant_no)
|
||
if sample_collect_time_start:
|
||
ci_q = ci_q.filter(
|
||
BizProtocolCollectionInfo.sample_collect_time
|
||
>= _normalize_date_bound(sample_collect_time_start)
|
||
)
|
||
if sample_collect_time_end:
|
||
ci_q = ci_q.filter(
|
||
BizProtocolCollectionInfo.sample_collect_time
|
||
<= _normalize_date_bound(sample_collect_time_end, end_of_day=True)
|
||
)
|
||
q0 = q0.filter(BizCollectionProtocol.id.in_(ci_q))
|
||
kw = (keyword or "").strip()
|
||
if kw:
|
||
like = f"%{kw}%"
|
||
party_q = db.query(BizProtocolPartyCustomer.collection_protocol_id).filter(
|
||
BizProtocolPartyCustomer.del_flag == 0,
|
||
or_(
|
||
BizProtocolPartyCustomer.mother_name.like(like),
|
||
BizProtocolPartyCustomer.mother_id_card.like(like),
|
||
BizProtocolPartyCustomer.mother_phone.like(like),
|
||
),
|
||
)
|
||
if tenant_no:
|
||
party_q = party_q.filter(BizProtocolPartyCustomer.tenant_no == tenant_no)
|
||
kw_conds = [
|
||
BizCollectionProtocol.id.in_(party_q),
|
||
]
|
||
if tenant_no:
|
||
from app.crud import sample_reception as reception_svc
|
||
|
||
proto_ids = reception_svc.protocol_ids_matching_lab_storage_keyword(
|
||
db, tenant_no, kw
|
||
)
|
||
if proto_ids:
|
||
kw_conds.append(BizCollectionProtocol.id.in_(proto_ids))
|
||
q0 = q0.filter(or_(*kw_conds))
|
||
if user_info:
|
||
from app.core.data_scope import apply_data_scope
|
||
q0 = apply_data_scope(q0, BizCollectionProtocol, user_info, db, tenant_no)
|
||
return q0
|
||
|
||
total = _filtered().order_by(None).with_entities(func.count(BizCollectionProtocol.id)).scalar() or 0
|
||
page = max(1, page)
|
||
page_size = min(max(1, page_size), 100)
|
||
rows = (
|
||
_filtered()
|
||
.order_by(BizCollectionProtocol.create_time.desc())
|
||
.offset((page - 1) * page_size)
|
||
.limit(page_size)
|
||
.all()
|
||
)
|
||
|
||
# 批量加载子表数据，避免 N+1 查询
|
||
protocol_ids = [r.id for r in rows]
|
||
pc_map: Dict[str, Dict[str, Any]] = {}
|
||
ci_map: Dict[str, Dict[str, Any]] = {}
|
||
if protocol_ids:
|
||
pc_rows = (
|
||
db.query(BizProtocolPartyCustomer)
|
||
.filter(
|
||
BizProtocolPartyCustomer.collection_protocol_id.in_(protocol_ids),
|
||
BizProtocolPartyCustomer.del_flag == 0,
|
||
)
|
||
.all()
|
||
)
|
||
for pc in pc_rows:
|
||
pc_map[pc.collection_protocol_id] = _party_customer_to_dict(pc)
|
||
|
||
ci_rows = (
|
||
db.query(BizProtocolCollectionInfo)
|
||
.filter(
|
||
BizProtocolCollectionInfo.collection_protocol_id.in_(protocol_ids),
|
||
BizProtocolCollectionInfo.del_flag == 0,
|
||
)
|
||
.all()
|
||
)
|
||
for ci in ci_rows:
|
||
ci_map[ci.collection_protocol_id] = _collection_info_to_dict(ci)
|
||
|
||
# 批量解析地区/医院名称
|
||
region_ids = list({ci_map.get(pid, {}).get("regionId", "") for pid in protocol_ids} - {""})
|
||
hospital_ids = list({ci_map.get(pid, {}).get("hospitalId", "") for pid in protocol_ids} - {""})
|
||
region_name_map: Dict[str, str] = {}
|
||
hospital_name_map: Dict[str, str] = {}
|
||
if region_ids:
|
||
for r in db.query(BizCollectionRegion.id, BizCollectionRegion.region_name).filter(BizCollectionRegion.id.in_(region_ids)).all():
|
||
region_name_map[r.id] = r.region_name or ""
|
||
if hospital_ids:
|
||
for h in db.query(BizCollectionHospital.id, BizCollectionHospital.hospital_name).filter(BizCollectionHospital.id.in_(hospital_ids)).all():
|
||
hospital_name_map[h.id] = h.hospital_name or ""
|
||
|
||
# 批量解析销售人员名称
|
||
sales_user_ids = list({ci_map.get(pid, {}).get("salesUserId", "") for pid in protocol_ids} - {""})
|
||
collector_user_ids = list({_protocol_to_dict(r).get("collectorUserId", "") for r in rows} - {""})
|
||
all_user_ids = list(set(sales_user_ids + collector_user_ids))
|
||
user_name_map: Dict[str, str] = {}
|
||
if all_user_ids:
|
||
for u in db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(all_user_ids)).all():
|
||
user_name_map[u.id] = u.nick_name or ""
|
||
|
||
# 批量加载协议样本（列表展示样本类型、样本编码）
|
||
sample_map: Dict[str, List[BizProtocolSample]] = {}
|
||
sample_type_name_map: Dict[str, str] = {}
|
||
if protocol_ids:
|
||
s_rows = (
|
||
db.query(BizProtocolSample)
|
||
.filter(
|
||
BizProtocolSample.collection_protocol_id.in_(protocol_ids),
|
||
BizProtocolSample.del_flag == 0,
|
||
)
|
||
.order_by(BizProtocolSample.order_num.asc(), BizProtocolSample.create_time.asc())
|
||
.all()
|
||
)
|
||
sample_type_name_map = _load_sample_type_name_map(db, s_rows)
|
||
for s in s_rows:
|
||
sample_map.setdefault(s.collection_protocol_id, []).append(s)
|
||
|
||
all_sample_codes: List[str] = []
|
||
for samples in sample_map.values():
|
||
for s in samples:
|
||
c = (s.sample_code or "").strip()
|
||
if c:
|
||
all_sample_codes.append(c)
|
||
lab_code_map: Dict[str, str] = {}
|
||
if tenant_no and all_sample_codes:
|
||
from app.crud import sample_reception as reception_svc
|
||
|
||
lab_code_map = reception_svc.get_lab_storage_code_map_by_sample_codes(
|
||
db, tenant_no, all_sample_codes, reception_source="cell_storage"
|
||
)
|
||
|
||
transport_by_sample_id: Dict[str, BizProtocolSampleTransport] = {}
|
||
if protocol_ids:
|
||
for t in (
|
||
db.query(BizProtocolSampleTransport)
|
||
.filter(
|
||
BizProtocolSampleTransport.collection_protocol_id.in_(protocol_ids),
|
||
BizProtocolSampleTransport.tenant_no == tenant_no,
|
||
BizProtocolSampleTransport.del_flag == 0,
|
||
)
|
||
.all()
|
||
):
|
||
transport_by_sample_id[t.sample_id] = t
|
||
site_name_map = _load_site_name_map(
|
||
db, [t.dest_site_id for t in transport_by_sample_id.values() if t.dest_site_id]
|
||
)
|
||
express_name_map = _load_express_name_map(
|
||
db,
|
||
[t.express_company_id for t in transport_by_sample_id.values() if t.express_company_id],
|
||
)
|
||
|
||
from app.services.qc_work_order_fee_summary import build_protocol_payment_summaries
|
||
|
||
payment_summary_map = build_protocol_payment_summaries(db, protocol_ids, tenant_no)
|
||
|
||
# 合并主表 + 子表数据
|
||
result = []
|
||
for r in rows:
|
||
d = _protocol_to_dict(r)
|
||
pc = pc_map.get(r.id)
|
||
ci = ci_map.get(r.id)
|
||
d["collectorUserName"] = user_name_map.get(d.get("collectorUserId", ""), "")
|
||
if pc:
|
||
d["motherName"] = pc.get("motherName", "")
|
||
d["motherPhone"] = pc.get("motherPhone", "")
|
||
d["motherIdCard"] = pc.get("motherIdCard", "")
|
||
d["motherBirthYm"] = pc.get("motherBirthYm", "")
|
||
d["motherEthnic"] = pc.get("motherEthnic", "")
|
||
d["fatherName"] = pc.get("fatherName", "")
|
||
d["fatherPhone"] = pc.get("fatherPhone", "")
|
||
d["newbornName"] = pc.get("newbornName", "")
|
||
d["expectedDeliveryDate"] = pc.get("expectedDeliveryDate", "")
|
||
d["newbornDeliveryTime"] = pc.get("newbornDeliveryTime", "")
|
||
d["newbornSex"] = pc.get("newbornSex", "")
|
||
d["deliveryMode"] = pc.get("deliveryMode", "")
|
||
d["birthOrder"] = pc.get("birthOrder")
|
||
d["newbornWeightGram"] = pc.get("newbornWeightGram")
|
||
else:
|
||
d["motherName"] = ""
|
||
d["motherPhone"] = ""
|
||
d["motherIdCard"] = ""
|
||
d["motherBirthYm"] = ""
|
||
d["motherEthnic"] = ""
|
||
d["fatherName"] = ""
|
||
d["fatherPhone"] = ""
|
||
d["newbornName"] = ""
|
||
d["expectedDeliveryDate"] = ""
|
||
d["newbornDeliveryTime"] = ""
|
||
d["newbornSex"] = ""
|
||
d["deliveryMode"] = ""
|
||
d["birthOrder"] = None
|
||
d["newbornWeightGram"] = None
|
||
if ci:
|
||
d["regionName"] = region_name_map.get(ci.get("regionId", ""), "")
|
||
d["hospitalName"] = hospital_name_map.get(ci.get("hospitalId", ""), "")
|
||
d["salesUserName"] = user_name_map.get(ci.get("salesUserId", ""), "")
|
||
d["sampleCollectTime"] = ci.get("sampleCollectTime", "")
|
||
else:
|
||
d["regionName"] = ""
|
||
d["hospitalName"] = ""
|
||
d["salesUserName"] = ""
|
||
d["sampleCollectTime"] = ""
|
||
protocol_samples = sample_map.get(r.id, [])
|
||
type_names: List[str] = []
|
||
codes: List[str] = []
|
||
sample_items: List[Dict[str, Any]] = []
|
||
for s in protocol_samples:
|
||
type_name = sample_type_name_map.get(s.sample_type_id or "", "")
|
||
code = (s.sample_code or "").strip()
|
||
if type_name:
|
||
type_names.append(type_name)
|
||
if code:
|
||
codes.append(code)
|
||
transport_obj = transport_by_sample_id.get(s.id)
|
||
transport_dict: Optional[Dict[str, Any]] = None
|
||
if transport_obj and (transport_obj.transport_type or "").strip():
|
||
dest_site_id = transport_obj.dest_site_id or ""
|
||
express_company_id = transport_obj.express_company_id or ""
|
||
transport_dict = {
|
||
"transportType": transport_obj.transport_type or "",
|
||
"destSiteId": dest_site_id,
|
||
"destSiteName": site_name_map.get(dest_site_id, ""),
|
||
"expressCompanyId": express_company_id,
|
||
"expressCompanyName": express_name_map.get(express_company_id, ""),
|
||
"waybillNo": transport_obj.waybill_no or "",
|
||
"carrierName": transport_obj.carrier_name or "",
|
||
"carrierPhone": transport_obj.carrier_phone or "",
|
||
"etaTime": _fmt_dt(transport_obj.eta_time),
|
||
}
|
||
sample_items.append(
|
||
{
|
||
"sampleTypeName": type_name,
|
||
"sampleCode": code,
|
||
"sampleStatus": (s.sample_status or "").strip(),
|
||
"labStorageCode": lab_code_map.get(code, "") if code else "",
|
||
"transport": transport_dict,
|
||
}
|
||
)
|
||
d["sampleItems"] = sample_items
|
||
d["sampleTypeNames"] = ",".join(type_names)
|
||
d["sampleCodes"] = ",".join(codes)
|
||
d["labStorageCodes"] = ",".join((lab_code_map.get(c) or "") for c in codes)
|
||
d["sampleCollectSummary"] = _format_sample_collect_summary(protocol_samples)
|
||
d["paymentSummary"] = payment_summary_map.get(r.id)
|
||
result.append(d)
|
||
return result, int(total)
|
||
|
||
|
||
def get_collection_protocol(
|
||
db: Session,
|
||
protocol_id: str,
|
||
tenant_no: str,
|
||
*,
|
||
scope_collector_user_id: Optional[str] = None,
|
||
) -> Optional[BizCollectionProtocol]:
|
||
filters = [
|
||
BizCollectionProtocol.id == protocol_id,
|
||
BizCollectionProtocol.del_flag == 0,
|
||
BizCollectionProtocol.tenant_no == tenant_no,
|
||
]
|
||
if scope_collector_user_id:
|
||
filters.append(
|
||
BizCollectionProtocol.collector_user_id == scope_collector_user_id
|
||
)
|
||
return db.query(BizCollectionProtocol).filter(and_(*filters)).first()
|
||
|
||
|
||
def get_payment_by_protocol_id(db: Session, collection_protocol_id: str, tenant_no: str) -> Optional[BizProtocolPayment]:
|
||
return (
|
||
db.query(BizProtocolPayment)
|
||
.filter(
|
||
and_(
|
||
BizProtocolPayment.collection_protocol_id == collection_protocol_id,
|
||
BizProtocolPayment.tenant_no == tenant_no,
|
||
BizProtocolPayment.del_flag == 0,
|
||
)
|
||
)
|
||
.first()
|
||
)
|
||
|
||
|
||
def get_protocol_detail_bundle(
|
||
db: Session,
|
||
protocol_id: str,
|
||
tenant_no: str,
|
||
*,
|
||
scope_collector_user_id: Optional[str] = None,
|
||
) -> Optional[Dict[str, Any]]:
|
||
p = get_collection_protocol(
|
||
db, protocol_id, tenant_no, scope_collector_user_id=scope_collector_user_id
|
||
)
|
||
if not p:
|
||
return None
|
||
pay = get_payment_by_protocol_id(db, protocol_id, tenant_no)
|
||
pc = get_party_customer(db, protocol_id, tenant_no)
|
||
ci = get_collection_info(db, protocol_id, tenant_no)
|
||
s_rows = list_samples(db, protocol_id, tenant_no)
|
||
sc = get_storage_cert(db, protocol_id, tenant_no)
|
||
protocol_dict = _protocol_to_dict(p)
|
||
ci_dict = _collection_info_to_dict(ci) if ci else None
|
||
user_ids: List[str] = []
|
||
if protocol_dict.get("collectorUserId"):
|
||
user_ids.append(str(protocol_dict["collectorUserId"]))
|
||
if ci_dict and ci_dict.get("salesUserId"):
|
||
user_ids.append(str(ci_dict["salesUserId"]))
|
||
user_name_map = _user_nick_name_map(db, user_ids)
|
||
protocol_dict["collectorUserName"] = user_name_map.get(
|
||
str(protocol_dict.get("collectorUserId") or ""), ""
|
||
)
|
||
if ci_dict:
|
||
ci_dict["salesUserName"] = user_name_map.get(str(ci_dict.get("salesUserId") or ""), "")
|
||
region_id = str(ci_dict.get("regionId") or "")
|
||
if region_id:
|
||
r = (
|
||
db.query(BizCollectionRegion.region_name)
|
||
.filter(BizCollectionRegion.id == region_id)
|
||
.first()
|
||
)
|
||
ci_dict["regionName"] = r.region_name if r else ""
|
||
hospital_id = str(ci_dict.get("hospitalId") or "")
|
||
if hospital_id:
|
||
h = (
|
||
db.query(BizCollectionHospital.hospital_name)
|
||
.filter(BizCollectionHospital.id == hospital_id)
|
||
.first()
|
||
)
|
||
ci_dict["hospitalName"] = h.hospital_name if h else ""
|
||
out: Dict[str, Any] = {"protocol": protocol_dict}
|
||
out["payment"] = _payment_to_dict(pay) if pay else None
|
||
out["partyCustomer"] = _party_customer_to_dict(pc) if pc else None
|
||
out["collectionInfo"] = ci_dict
|
||
transport_rows = list_sample_transports(db, protocol_id, tenant_no)
|
||
site_name_map = _load_site_name_map(
|
||
db, [t.dest_site_id for t in transport_rows if t.dest_site_id]
|
||
)
|
||
transport_by_sample: Dict[str, Dict[str, Any]] = {}
|
||
for t in transport_rows:
|
||
td = _transport_to_dict(t)
|
||
td["destSiteName"] = site_name_map.get(t.dest_site_id or "", "")
|
||
transport_by_sample[t.sample_id] = td
|
||
from app.crud import sample_reception as sample_reception_svc
|
||
|
||
sample_code_by_id = {s.id: s.sample_code or "" for s in s_rows}
|
||
reception_by_sample = sample_reception_svc.get_receptions_map_for_samples(
|
||
db,
|
||
tenant_no,
|
||
sample_code_by_id=sample_code_by_id,
|
||
)
|
||
from app.crud import qc_work_order as qc_work_order_svc
|
||
from app.crud import production_work_order as production_work_order_svc
|
||
from app.crud import storage_release_review as storage_release_review_svc
|
||
|
||
qc_by_sample = qc_work_order_svc.get_qc_info_map_for_samples(
|
||
db,
|
||
[s.id for s in s_rows],
|
||
tenant_no,
|
||
sample_code_by_id=sample_code_by_id,
|
||
)
|
||
production_by_sample = production_work_order_svc.get_production_info_map_for_samples(
|
||
db,
|
||
[s.id for s in s_rows],
|
||
tenant_no,
|
||
sample_code_by_id=sample_code_by_id,
|
||
)
|
||
storage_release_by_sample = storage_release_review_svc.get_storage_release_map_for_samples(
|
||
db,
|
||
[s.id for s in s_rows],
|
||
tenant_no,
|
||
sample_code_by_id=sample_code_by_id,
|
||
)
|
||
unit_name_map = _load_unit_name_map(db, s_rows)
|
||
type_name_map = _load_sample_type_name_map(db, s_rows)
|
||
samples_out: List[Dict[str, Any]] = []
|
||
for s in s_rows:
|
||
item = _sample_to_dict(s)
|
||
item["sampleTypeName"] = type_name_map.get(s.sample_type_id or "", "")
|
||
item["unitName"] = unit_name_map.get(s.unit_id or "", "")
|
||
item["transport"] = transport_by_sample.get(s.id)
|
||
item["sampleReception"] = reception_by_sample.get(s.id)
|
||
item["qcInfo"] = qc_by_sample.get(s.id)
|
||
item["productionInfo"] = production_by_sample.get(s.id)
|
||
item["storageReleaseInfo"] = storage_release_by_sample.get(s.id)
|
||
samples_out.append(item)
|
||
out["samples"] = samples_out
|
||
out["storageCert"] = _storage_cert_to_dict(sc) if sc else None
|
||
ft = get_fullmoon_todo(db, protocol_id, tenant_no)
|
||
delivery_dt = pc.newborn_delivery_time if pc else None
|
||
out["fullmoonTodo"] = _fullmoon_todo_detail_dict(ft, newborn_delivery_time=delivery_dt)
|
||
out["attachments"] = urls_to_attachment_list(p.protocol_attachments, db, tenant_no)
|
||
if pay:
|
||
detail_rows = list_payment_details(db, pay.id, tenant_no)
|
||
details_out: List[Dict[str, Any]] = []
|
||
for d in detail_rows:
|
||
item = _detail_to_dict(
|
||
d,
|
||
protocol_no=p.protocol_no or "",
|
||
collection_protocol_id=protocol_id,
|
||
)
|
||
item["attachments"] = urls_to_attachment_list(d.transfer_attachments, db, tenant_no)
|
||
item["financeConfirmAttachmentFiles"] = urls_to_attachment_list(
|
||
d.finance_confirm_attachments, db, tenant_no
|
||
)
|
||
item["refundFinanceConfirmAttachmentFiles"] = urls_to_attachment_list(
|
||
d.refund_finance_confirm_attachments, db, tenant_no
|
||
)
|
||
item["refundAttachmentFiles"] = urls_to_attachment_list(
|
||
d.refund_attachments, db, tenant_no
|
||
)
|
||
details_out.append(item)
|
||
out["paymentDetails"] = details_out
|
||
else:
|
||
out["paymentDetails"] = []
|
||
return out
|
||
|
||
|
||
def list_audits(
|
||
db: Session,
|
||
*,
|
||
page: int = 1,
|
||
page_size: int = 10,
|
||
keyword: Optional[str] = None,
|
||
audit_status: Optional[str] = None,
|
||
tenant_no: Optional[str] = None,
|
||
user_info=None,
|
||
) -> Tuple[List[Dict[str, Any]], int]:
|
||
q_base = (
|
||
db.query(BizProtocolAudit, BizCollectionProtocol.protocol_no)
|
||
.join(
|
||
BizCollectionProtocol,
|
||
and_(
|
||
BizProtocolAudit.collection_protocol_id == BizCollectionProtocol.id,
|
||
BizCollectionProtocol.del_flag == 0,
|
||
),
|
||
)
|
||
.filter(BizProtocolAudit.del_flag == 0)
|
||
)
|
||
if tenant_no:
|
||
q_base = q_base.filter(BizProtocolAudit.tenant_no == tenant_no)
|
||
if audit_status:
|
||
q_base = q_base.filter(BizProtocolAudit.audit_status == audit_status.strip())
|
||
kw = (keyword or "").strip()
|
||
if kw:
|
||
like = f"%{kw}%"
|
||
q_base = q_base.filter(BizProtocolAudit.audit_remark.like(like))
|
||
if user_info:
|
||
from app.core.data_scope import apply_data_scope
|
||
q_base = apply_data_scope(q_base, BizProtocolAudit, user_info, db, tenant_no)
|
||
|
||
total = q_base.order_by(None).with_entities(func.count(BizProtocolAudit.id)).scalar() or 0
|
||
page = max(1, page)
|
||
page_size = min(max(1, page_size), 100)
|
||
rows = (
|
||
q_base.order_by(BizProtocolAudit.create_time.desc())
|
||
.offset((page - 1) * page_size)
|
||
.limit(page_size)
|
||
.all()
|
||
)
|
||
auditor_ids = list({a.auditor_user_id for a, _ in rows if a.auditor_user_id})
|
||
auditor_name_map: Dict[str, str] = {}
|
||
if auditor_ids:
|
||
for u in db.query(SysUser.id, SysUser.nick_name).filter(SysUser.id.in_(auditor_ids)).all():
|
||
auditor_name_map[u.id] = u.nick_name or ""
|
||
out = [
|
||
_audit_to_dict(
|
||
a,
|
||
protocol_no=pn or "",
|
||
auditor_user_name=auditor_name_map.get(a.auditor_user_id or "", ""),
|
||
)
|
||
for a, pn in rows
|
||
]
|
||
return out, int(total)
|
||
|
||
|
||
def create_audit(
|
||
db: Session,
|
||
*,
|
||
collection_protocol_id: str,
|
||
audit_status: str,
|
||
audit_remark: Optional[str],
|
||
tenant_no: str,
|
||
user_id: Optional[str],
|
||
) -> BizProtocolAudit:
|
||
p = get_collection_protocol(db, collection_protocol_id, tenant_no)
|
||
if not p:
|
||
raise ValueError("协议单不存在")
|
||
st = (audit_status or "").strip()
|
||
if st not in ("shtg", "shth"):
|
||
raise ValueError("审核状态无效")
|
||
obj = BizProtocolAudit(
|
||
id=generate_uuid(),
|
||
collection_protocol_id=collection_protocol_id,
|
||
auditor_user_id=user_id,
|
||
audit_remark=(audit_remark or "").strip() or None,
|
||
audit_status=st,
|
||
tenant_no=tenant_no,
|
||
order_num=0,
|
||
remark=None,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=datetime.now(),
|
||
update_by=user_id,
|
||
update_time=datetime.now(),
|
||
del_flag=0,
|
||
)
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def _detail_query_with_protocol(
|
||
db: Session,
|
||
tenant_no: str,
|
||
pay_status_filter: str,
|
||
):
|
||
return (
|
||
db.query(BizProtocolPaymentDetail, BizProtocolPayment, BizCollectionProtocol)
|
||
.join(BizProtocolPayment, BizProtocolPaymentDetail.payment_id == BizProtocolPayment.id)
|
||
.join(BizCollectionProtocol, BizProtocolPayment.collection_protocol_id == BizCollectionProtocol.id)
|
||
.filter(
|
||
BizProtocolPaymentDetail.del_flag == 0,
|
||
BizProtocolPayment.del_flag == 0,
|
||
BizCollectionProtocol.del_flag == 0,
|
||
BizProtocolPaymentDetail.tenant_no == tenant_no,
|
||
BizProtocolPaymentDetail.pay_status == pay_status_filter,
|
||
)
|
||
)
|
||
|
||
|
||
def _apply_payment_detail_list_filters(
|
||
q,
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
keyword: Optional[str] = None,
|
||
installment_due_date_start: Optional[str] = None,
|
||
installment_due_date_end: Optional[str] = None,
|
||
):
|
||
kw = (keyword or "").strip()
|
||
if kw:
|
||
like = f"%{kw}%"
|
||
party_q = db.query(BizProtocolPartyCustomer.collection_protocol_id).filter(
|
||
BizProtocolPartyCustomer.del_flag == 0,
|
||
or_(
|
||
BizProtocolPartyCustomer.mother_name.like(like),
|
||
BizProtocolPartyCustomer.mother_id_card.like(like),
|
||
BizProtocolPartyCustomer.newborn_name.like(like),
|
||
BizProtocolPartyCustomer.newborn_id_card.like(like),
|
||
),
|
||
)
|
||
if tenant_no:
|
||
party_q = party_q.filter(BizProtocolPartyCustomer.tenant_no == tenant_no)
|
||
q = q.filter(BizCollectionProtocol.id.in_(party_q))
|
||
if installment_due_date_start:
|
||
start_due = _yyyymmdd_from_date_param(installment_due_date_start)
|
||
if start_due:
|
||
q = q.filter(BizProtocolPaymentDetail.installment_due_date >= start_due)
|
||
if installment_due_date_end:
|
||
end_due = _yyyymmdd_from_date_param(installment_due_date_end)
|
||
if end_due:
|
||
q = q.filter(BizProtocolPaymentDetail.installment_due_date <= end_due)
|
||
return q
|
||
|
||
|
||
def _rows_to_payment_detail_dicts(
|
||
db: Session,
|
||
rows: List[Any],
|
||
tenant_no: str,
|
||
*,
|
||
post_process: Optional[Any] = None,
|
||
) -> List[Dict[str, Any]]:
|
||
"""缴费明细行 → 前端字典（含协议、客户、转账附件）。"""
|
||
protocol_ids = [
|
||
pay.collection_protocol_id for _, pay, _ in rows if pay and pay.collection_protocol_id
|
||
]
|
||
pc_map = _party_customer_map_by_protocol_ids(db, protocol_ids)
|
||
out: List[Dict[str, Any]] = []
|
||
for d, pay, prot in rows:
|
||
cpid = pay.collection_protocol_id if pay else ""
|
||
pc = pc_map.get(cpid)
|
||
item = _detail_to_dict(
|
||
d,
|
||
protocol_no=(prot.protocol_no if prot else "") or "",
|
||
collection_protocol_id=cpid,
|
||
protocol_status=(prot.protocol_status if prot else "") or "",
|
||
pay_mode=(pay.pay_mode if pay else "") or "",
|
||
mother_name=(pc.mother_name if pc else "") or "",
|
||
mother_id_card=(pc.mother_id_card if pc else "") or "",
|
||
newborn_name=(pc.newborn_name if pc else "") or "",
|
||
newborn_id_card=(pc.newborn_id_card if pc else "") or "",
|
||
)
|
||
item["attachments"] = urls_to_attachment_list(d.transfer_attachments, db, tenant_no)
|
||
if post_process:
|
||
post_process(item, d)
|
||
out.append(item)
|
||
return out
|
||
|
||
|
||
def list_receipt_confirm_details(
|
||
db: Session,
|
||
*,
|
||
page: int = 1,
|
||
page_size: int = 10,
|
||
keyword: Optional[str] = None,
|
||
installment_due_date_start: Optional[str] = None,
|
||
installment_due_date_end: Optional[str] = None,
|
||
tenant_no: Optional[str] = None,
|
||
user_info=None,
|
||
) -> Tuple[List[Dict[str, Any]], int]:
|
||
if not tenant_no:
|
||
return [], 0
|
||
q_base = _detail_query_with_protocol(db, tenant_no, "jfqrz")
|
||
q_base = _apply_payment_detail_list_filters(
|
||
q_base,
|
||
db,
|
||
tenant_no=tenant_no,
|
||
keyword=keyword,
|
||
installment_due_date_start=installment_due_date_start,
|
||
installment_due_date_end=installment_due_date_end,
|
||
)
|
||
if user_info:
|
||
from app.core.data_scope import apply_data_scope
|
||
q_base = apply_data_scope(q_base, BizProtocolPaymentDetail, user_info, db, tenant_no)
|
||
total = q_base.order_by(None).with_entities(func.count(BizProtocolPaymentDetail.id)).scalar() or 0
|
||
page = max(1, page)
|
||
page_size = min(max(1, page_size), 100)
|
||
rows = (
|
||
q_base.order_by(BizProtocolPaymentDetail.create_time.desc())
|
||
.offset((page - 1) * page_size)
|
||
.limit(page_size)
|
||
.all()
|
||
)
|
||
return _rows_to_payment_detail_dicts(db, rows, tenant_no), int(total)
|
||
|
||
|
||
def list_receipt_confirmed_details(
|
||
db: Session,
|
||
*,
|
||
page: int = 1,
|
||
page_size: int = 10,
|
||
keyword: Optional[str] = None,
|
||
installment_due_date_start: Optional[str] = None,
|
||
installment_due_date_end: Optional[str] = None,
|
||
tenant_no: Optional[str] = None,
|
||
user_info=None,
|
||
) -> Tuple[List[Dict[str, Any]], int]:
|
||
if not tenant_no:
|
||
return [], 0
|
||
q_base = _detail_query_with_protocol(db, tenant_no, "yjf")
|
||
q_base = _apply_payment_detail_list_filters(
|
||
q_base,
|
||
db,
|
||
tenant_no=tenant_no,
|
||
keyword=keyword,
|
||
installment_due_date_start=installment_due_date_start,
|
||
installment_due_date_end=installment_due_date_end,
|
||
)
|
||
if user_info:
|
||
from app.core.data_scope import apply_data_scope
|
||
q_base = apply_data_scope(q_base, BizProtocolPaymentDetail, user_info, db, tenant_no)
|
||
total = q_base.order_by(None).with_entities(func.count(BizProtocolPaymentDetail.id)).scalar() or 0
|
||
page = max(1, page)
|
||
page_size = min(max(1, page_size), 100)
|
||
rows = (
|
||
q_base.order_by(BizProtocolPaymentDetail.update_time.desc())
|
||
.offset((page - 1) * page_size)
|
||
.limit(page_size)
|
||
.all()
|
||
)
|
||
|
||
def _add_finance_files(item: Dict[str, Any], d: BizProtocolPaymentDetail) -> None:
|
||
item["financeConfirmAttachmentFiles"] = urls_to_attachment_list(
|
||
d.finance_confirm_attachments, db, tenant_no
|
||
)
|
||
|
||
return (
|
||
_rows_to_payment_detail_dicts(db, rows, tenant_no, post_process=_add_finance_files),
|
||
int(total),
|
||
)
|
||
|
||
|
||
def confirm_receipt(
|
||
db: Session,
|
||
*,
|
||
detail_id: str,
|
||
tenant_no: str,
|
||
user_id: Optional[str],
|
||
finance_confirm_attachments: Optional[str] = None,
|
||
finance_remark: Optional[str] = None,
|
||
) -> BizProtocolPaymentDetail:
|
||
d = (
|
||
db.query(BizProtocolPaymentDetail)
|
||
.filter(
|
||
and_(
|
||
BizProtocolPaymentDetail.id == detail_id,
|
||
BizProtocolPaymentDetail.tenant_no == tenant_no,
|
||
BizProtocolPaymentDetail.del_flag == 0,
|
||
)
|
||
)
|
||
.first()
|
||
)
|
||
if not d:
|
||
raise ValueError("缴费明细不存在")
|
||
if d.pay_status != "jfqrz":
|
||
if d.pay_status == "yjf":
|
||
raise ValueError("该明细已确认收款")
|
||
raise ValueError("当前状态不可确认收款")
|
||
d.pay_status = "yjf"
|
||
if finance_confirm_attachments is not None:
|
||
d.finance_confirm_attachments = finance_confirm_attachments.strip() or None
|
||
if finance_remark is not None:
|
||
d.finance_remark = finance_remark.strip() or None
|
||
d.update_by = user_id
|
||
d.update_time = datetime.now()
|
||
db.add(d)
|
||
db.commit()
|
||
db.refresh(d)
|
||
return d
|
||
|
||
|
||
def list_refund_confirm_details(
|
||
db: Session,
|
||
*,
|
||
page: int = 1,
|
||
page_size: int = 10,
|
||
keyword: Optional[str] = None,
|
||
installment_due_date_start: Optional[str] = None,
|
||
installment_due_date_end: Optional[str] = None,
|
||
tenant_no: Optional[str] = None,
|
||
user_info=None,
|
||
) -> Tuple[List[Dict[str, Any]], int]:
|
||
if not tenant_no:
|
||
return [], 0
|
||
q_base = _detail_query_with_protocol(db, tenant_no, "tfqrz")
|
||
q_base = _apply_payment_detail_list_filters(
|
||
q_base,
|
||
db,
|
||
tenant_no=tenant_no,
|
||
keyword=keyword,
|
||
installment_due_date_start=installment_due_date_start,
|
||
installment_due_date_end=installment_due_date_end,
|
||
)
|
||
if user_info:
|
||
from app.core.data_scope import apply_data_scope
|
||
q_base = apply_data_scope(q_base, BizProtocolPaymentDetail, user_info, db, tenant_no)
|
||
total = q_base.order_by(None).with_entities(func.count(BizProtocolPaymentDetail.id)).scalar() or 0
|
||
page = max(1, page)
|
||
page_size = min(max(1, page_size), 100)
|
||
rows = (
|
||
q_base.order_by(BizProtocolPaymentDetail.create_time.desc())
|
||
.offset((page - 1) * page_size)
|
||
.limit(page_size)
|
||
.all()
|
||
)
|
||
def _add_refund_apply_files(item: Dict[str, Any], d: BizProtocolPaymentDetail) -> None:
|
||
item["refundAttachmentFiles"] = urls_to_attachment_list(
|
||
d.refund_attachments, db, tenant_no
|
||
)
|
||
|
||
return (
|
||
_rows_to_payment_detail_dicts(db, rows, tenant_no, post_process=_add_refund_apply_files),
|
||
int(total),
|
||
)
|
||
|
||
|
||
def list_refund_confirmed_details(
|
||
db: Session,
|
||
*,
|
||
page: int = 1,
|
||
page_size: int = 10,
|
||
keyword: Optional[str] = None,
|
||
installment_due_date_start: Optional[str] = None,
|
||
installment_due_date_end: Optional[str] = None,
|
||
tenant_no: Optional[str] = None,
|
||
user_info=None,
|
||
) -> Tuple[List[Dict[str, Any]], int]:
|
||
if not tenant_no:
|
||
return [], 0
|
||
q_base = _detail_query_with_protocol(db, tenant_no, "ytf")
|
||
q_base = _apply_payment_detail_list_filters(
|
||
q_base,
|
||
db,
|
||
tenant_no=tenant_no,
|
||
keyword=keyword,
|
||
installment_due_date_start=installment_due_date_start,
|
||
installment_due_date_end=installment_due_date_end,
|
||
)
|
||
if user_info:
|
||
from app.core.data_scope import apply_data_scope
|
||
q_base = apply_data_scope(q_base, BizProtocolPaymentDetail, user_info, db, tenant_no)
|
||
total = q_base.order_by(None).with_entities(func.count(BizProtocolPaymentDetail.id)).scalar() or 0
|
||
page = max(1, page)
|
||
page_size = min(max(1, page_size), 100)
|
||
rows = (
|
||
q_base.order_by(BizProtocolPaymentDetail.update_time.desc())
|
||
.offset((page - 1) * page_size)
|
||
.limit(page_size)
|
||
.all()
|
||
)
|
||
def _add_refund_confirmed_files(item: Dict[str, Any], d: BizProtocolPaymentDetail) -> None:
|
||
item["refundAttachmentFiles"] = urls_to_attachment_list(
|
||
d.refund_attachments, db, tenant_no
|
||
)
|
||
item["refundFinanceConfirmAttachmentFiles"] = urls_to_attachment_list(
|
||
d.refund_finance_confirm_attachments, db, tenant_no
|
||
)
|
||
|
||
return (
|
||
_rows_to_payment_detail_dicts(db, rows, tenant_no, post_process=_add_refund_confirmed_files),
|
||
int(total),
|
||
)
|
||
|
||
|
||
def confirm_refund(
|
||
db: Session,
|
||
*,
|
||
detail_id: str,
|
||
tenant_no: str,
|
||
user_id: Optional[str],
|
||
refund_finance_confirm_attachments: Optional[str] = None,
|
||
refund_finance_remark: Optional[str] = None,
|
||
) -> BizProtocolPaymentDetail:
|
||
d = (
|
||
db.query(BizProtocolPaymentDetail)
|
||
.filter(
|
||
and_(
|
||
BizProtocolPaymentDetail.id == detail_id,
|
||
BizProtocolPaymentDetail.tenant_no == tenant_no,
|
||
BizProtocolPaymentDetail.del_flag == 0,
|
||
)
|
||
)
|
||
.first()
|
||
)
|
||
if not d:
|
||
raise ValueError("缴费明细不存在")
|
||
if d.pay_status != "tfqrz":
|
||
if d.pay_status == "ytf":
|
||
raise ValueError("该明细已确认退费")
|
||
raise ValueError("当前状态不可确认退费")
|
||
d.pay_status = "ytf"
|
||
if refund_finance_confirm_attachments is not None:
|
||
d.refund_finance_confirm_attachments = refund_finance_confirm_attachments.strip() or None
|
||
if refund_finance_remark is not None:
|
||
d.refund_finance_remark = refund_finance_remark.strip() or None
|
||
d.update_by = user_id
|
||
d.update_time = datetime.now()
|
||
db.add(d)
|
||
db.commit()
|
||
db.refresh(d)
|
||
return d
|
||
|
||
|
||
def _get_payment_detail_or_raise(
|
||
db: Session,
|
||
*,
|
||
detail_id: str,
|
||
tenant_no: str,
|
||
) -> BizProtocolPaymentDetail:
|
||
d = (
|
||
db.query(BizProtocolPaymentDetail)
|
||
.filter(
|
||
and_(
|
||
BizProtocolPaymentDetail.id == detail_id,
|
||
BizProtocolPaymentDetail.tenant_no == tenant_no,
|
||
BizProtocolPaymentDetail.del_flag == 0,
|
||
)
|
||
)
|
||
.first()
|
||
)
|
||
if not d:
|
||
raise ValueError("缴费明细不存在")
|
||
return d
|
||
|
||
|
||
def _apply_submit_receipt_finance(
|
||
d: BizProtocolPaymentDetail,
|
||
*,
|
||
user_id: Optional[str],
|
||
) -> None:
|
||
if d.pay_status != "dfj":
|
||
raise ValueError("仅待缴费状态可提交财务确认")
|
||
d.pay_status = "jfqrz"
|
||
d.update_by = user_id
|
||
d.update_time = datetime.now()
|
||
|
||
|
||
def submit_receipt_finance(
|
||
db: Session,
|
||
*,
|
||
detail_id: str,
|
||
tenant_no: str,
|
||
user_id: Optional[str],
|
||
) -> BizProtocolPaymentDetail:
|
||
d = _get_payment_detail_or_raise(db, detail_id=detail_id, tenant_no=tenant_no)
|
||
_apply_submit_receipt_finance(d, user_id=user_id)
|
||
db.add(d)
|
||
db.commit()
|
||
db.refresh(d)
|
||
return d
|
||
|
||
|
||
def batch_submit_receipt_finance(
|
||
db: Session,
|
||
*,
|
||
detail_ids: List[str],
|
||
tenant_no: str,
|
||
user_id: Optional[str],
|
||
) -> int:
|
||
ids = [str(i).strip() for i in (detail_ids or []) if str(i).strip()]
|
||
if not ids:
|
||
raise ValueError("请选择缴费明细")
|
||
for detail_id in ids:
|
||
d = _get_payment_detail_or_raise(db, detail_id=detail_id, tenant_no=tenant_no)
|
||
_apply_submit_receipt_finance(d, user_id=user_id)
|
||
db.add(d)
|
||
db.commit()
|
||
return len(ids)
|
||
|
||
|
||
def _apply_submit_refund_finance(
|
||
d: BizProtocolPaymentDetail,
|
||
*,
|
||
user_id: Optional[str],
|
||
refund_amount: Any,
|
||
refund_remark: Optional[str] = None,
|
||
refund_attachments: Optional[str] = None,
|
||
) -> None:
|
||
if d.pay_status != "yjf":
|
||
raise ValueError("仅已缴费状态可提交退费财务确认")
|
||
if refund_amount is None or Decimal(str(refund_amount)) <= 0:
|
||
raise ValueError("请填写有效的退费金额")
|
||
d.pay_status = "tfqrz"
|
||
d.refund_amount = Decimal(str(refund_amount))
|
||
d.refund_remark = (refund_remark or "").strip() or None
|
||
d.refund_attachments = (refund_attachments or "").strip() or None
|
||
d.update_by = user_id
|
||
d.update_time = datetime.now()
|
||
|
||
|
||
def submit_refund_finance(
|
||
db: Session,
|
||
*,
|
||
detail_id: str,
|
||
tenant_no: str,
|
||
user_id: Optional[str],
|
||
refund_amount: Any,
|
||
refund_remark: Optional[str] = None,
|
||
refund_attachments: Optional[str] = None,
|
||
) -> BizProtocolPaymentDetail:
|
||
d = _get_payment_detail_or_raise(db, detail_id=detail_id, tenant_no=tenant_no)
|
||
_apply_submit_refund_finance(
|
||
d,
|
||
user_id=user_id,
|
||
refund_amount=refund_amount,
|
||
refund_remark=refund_remark,
|
||
refund_attachments=refund_attachments,
|
||
)
|
||
db.add(d)
|
||
db.commit()
|
||
db.refresh(d)
|
||
return d
|
||
|
||
|
||
def batch_submit_refund_finance(
|
||
db: Session,
|
||
*,
|
||
detail_ids: List[str],
|
||
tenant_no: str,
|
||
user_id: Optional[str],
|
||
refund_amount: Any,
|
||
refund_remark: Optional[str] = None,
|
||
refund_attachments: Optional[str] = None,
|
||
) -> int:
|
||
ids = [str(i).strip() for i in (detail_ids or []) if str(i).strip()]
|
||
if not ids:
|
||
raise ValueError("请选择缴费明细")
|
||
for detail_id in ids:
|
||
d = _get_payment_detail_or_raise(db, detail_id=detail_id, tenant_no=tenant_no)
|
||
_apply_submit_refund_finance(
|
||
d,
|
||
user_id=user_id,
|
||
refund_amount=refund_amount,
|
||
refund_remark=refund_remark,
|
||
refund_attachments=refund_attachments,
|
||
)
|
||
db.add(d)
|
||
db.commit()
|
||
return len(ids)
|
||
|
||
|
||
def _party_customer_map_by_protocol_ids(
|
||
db: Session,
|
||
protocol_ids: List[str],
|
||
) -> Dict[str, BizProtocolPartyCustomer]:
|
||
if not protocol_ids:
|
||
return {}
|
||
rows = (
|
||
db.query(BizProtocolPartyCustomer)
|
||
.filter(
|
||
BizProtocolPartyCustomer.collection_protocol_id.in_(protocol_ids),
|
||
BizProtocolPartyCustomer.del_flag == 0,
|
||
)
|
||
.all()
|
||
)
|
||
return {r.collection_protocol_id: r for r in rows}
|
||
|
||
|
||
def _yyyymmdd_from_date_param(value: str) -> str:
|
||
val = (value or "").strip()
|
||
if len(val) == 10 and val[4] == "-" and val[7] == "-":
|
||
return val.replace("-", "")
|
||
return val
|
||
|
||
|
||
def list_payment_details_page(
|
||
db: Session,
|
||
*,
|
||
page: int = 1,
|
||
page_size: int = 10,
|
||
pay_status: Optional[str] = None,
|
||
keyword: Optional[str] = None,
|
||
transfer_type: Optional[str] = None,
|
||
installment_due_date_start: Optional[str] = None,
|
||
installment_due_date_end: Optional[str] = None,
|
||
tenant_no: Optional[str] = None,
|
||
user_info=None,
|
||
) -> Tuple[List[Dict[str, Any]], int]:
|
||
if not tenant_no:
|
||
return [], 0
|
||
q = (
|
||
db.query(BizProtocolPaymentDetail, BizProtocolPayment, BizCollectionProtocol)
|
||
.join(BizProtocolPayment, BizProtocolPaymentDetail.payment_id == BizProtocolPayment.id)
|
||
.join(BizCollectionProtocol, BizProtocolPayment.collection_protocol_id == BizCollectionProtocol.id)
|
||
.filter(
|
||
BizProtocolPaymentDetail.del_flag == 0,
|
||
BizProtocolPayment.del_flag == 0,
|
||
BizCollectionProtocol.del_flag == 0,
|
||
BizProtocolPaymentDetail.tenant_no == tenant_no,
|
||
)
|
||
)
|
||
if pay_status:
|
||
q = q.filter(BizProtocolPaymentDetail.pay_status == pay_status)
|
||
if keyword and keyword.strip():
|
||
like = f"%{keyword.strip()}%"
|
||
party_q = db.query(BizProtocolPartyCustomer.collection_protocol_id).filter(
|
||
BizProtocolPartyCustomer.del_flag == 0,
|
||
or_(
|
||
BizProtocolPartyCustomer.mother_name.like(like),
|
||
BizProtocolPartyCustomer.mother_id_card.like(like),
|
||
BizProtocolPartyCustomer.newborn_name.like(like),
|
||
BizProtocolPartyCustomer.newborn_id_card.like(like),
|
||
),
|
||
)
|
||
if tenant_no:
|
||
party_q = party_q.filter(BizProtocolPartyCustomer.tenant_no == tenant_no)
|
||
q = q.filter(BizCollectionProtocol.id.in_(party_q))
|
||
if transfer_type and transfer_type.strip():
|
||
q = q.filter(BizProtocolPaymentDetail.transfer_type == transfer_type.strip())
|
||
if installment_due_date_start:
|
||
start_due = _yyyymmdd_from_date_param(installment_due_date_start)
|
||
if start_due:
|
||
q = q.filter(BizProtocolPaymentDetail.installment_due_date >= start_due)
|
||
if installment_due_date_end:
|
||
end_due = _yyyymmdd_from_date_param(installment_due_date_end)
|
||
if end_due:
|
||
q = q.filter(BizProtocolPaymentDetail.installment_due_date <= end_due)
|
||
if user_info:
|
||
from app.core.data_scope import apply_data_scope
|
||
q = apply_data_scope(q, BizProtocolPaymentDetail, user_info, db, tenant_no)
|
||
total = q.order_by(None).with_entities(func.count(BizProtocolPaymentDetail.id)).scalar() or 0
|
||
page = max(1, page)
|
||
page_size = min(max(1, page_size), 100)
|
||
due_col = BizProtocolPaymentDetail.installment_due_date
|
||
# 缴费信息 Tab：待缴费/缴费确认中/退费确认中 → 应缴日期正序；已缴费/已退费 → 倒序
|
||
due_order = due_col.desc() if pay_status in ("yjf", "ytf") else due_col.asc()
|
||
rows = (
|
||
q.order_by(due_order)
|
||
.offset((page - 1) * page_size)
|
||
.limit(page_size)
|
||
.all()
|
||
)
|
||
protocol_ids = [
|
||
pay.collection_protocol_id for _, pay, _ in rows if pay and pay.collection_protocol_id
|
||
]
|
||
pc_map = _party_customer_map_by_protocol_ids(db, protocol_ids)
|
||
out = []
|
||
for d, pay, prot in rows:
|
||
cpid = pay.collection_protocol_id if pay else ""
|
||
pc = pc_map.get(cpid)
|
||
out.append(
|
||
_detail_to_dict(
|
||
d,
|
||
protocol_no=(prot.protocol_no if prot else "") or "",
|
||
collection_protocol_id=cpid,
|
||
protocol_status=(prot.protocol_status if prot else "") or "",
|
||
pay_mode=(pay.pay_mode if pay else "") or "",
|
||
mother_name=(pc.mother_name if pc else "") or "",
|
||
mother_phone=(pc.mother_phone if pc else "") or "",
|
||
mother_id_card=(pc.mother_id_card if pc else "") or "",
|
||
newborn_name=(pc.newborn_name if pc else "") or "",
|
||
newborn_id_card=(pc.newborn_id_card if pc else "") or "",
|
||
)
|
||
)
|
||
return out, int(total)
|
||
|
||
|
||
def _classify_installment_warning(due: str, today: str) -> str:
|
||
"""overdue | warn | upcoming — due 为 YYYYMMDD"""
|
||
if not due or len(due) != 8 or not due.isdigit():
|
||
return "unknown"
|
||
if due < today:
|
||
return "overdue"
|
||
last_warn = (datetime.strptime(today, "%Y%m%d") + timedelta(days=15)).strftime("%Y%m%d")
|
||
if today <= due <= last_warn:
|
||
return "warn"
|
||
return "upcoming"
|
||
|
||
|
||
def list_installment_warning(
|
||
db: Session,
|
||
*,
|
||
page: int = 1,
|
||
page_size: int = 10,
|
||
warning_type: str = "all",
|
||
tenant_no: Optional[str] = None,
|
||
user_info=None,
|
||
) -> Tuple[List[Dict[str, Any]], int]:
|
||
if not tenant_no:
|
||
return [], 0
|
||
today = _today_yyyymmdd()
|
||
q0 = _detail_query_with_protocol(db, tenant_no, "dfj").filter(
|
||
BizProtocolPaymentDetail.installment_no.isnot(None),
|
||
)
|
||
if user_info:
|
||
from app.core.data_scope import apply_data_scope
|
||
q0 = apply_data_scope(q0, BizProtocolPaymentDetail, user_info, db, tenant_no)
|
||
|
||
# 拉取当前租户下候选（分期号非空）；日期过滤在内存中完成以保证与 classify 一致
|
||
rows_all = q0.order_by(BizProtocolPaymentDetail.installment_due_date.asc()).all()
|
||
flattened: List[Dict[str, Any]] = []
|
||
for d, pay, prot in rows_all:
|
||
due_raw = (d.installment_due_date or "").strip() if d.installment_due_date else ""
|
||
wt = _classify_installment_warning(due_raw, today)
|
||
cpid = pay.collection_protocol_id if pay else ""
|
||
item = _detail_to_dict(
|
||
d,
|
||
protocol_no=(prot.protocol_no if prot else "") or "",
|
||
collection_protocol_id=cpid,
|
||
protocol_status=(prot.protocol_status if prot else "") or "",
|
||
pay_mode=(pay.pay_mode if pay else "") or "",
|
||
)
|
||
item["warningType"] = wt
|
||
wt_filter = (warning_type or "all").strip().lower()
|
||
if wt_filter == "overdue" and wt != "overdue":
|
||
continue
|
||
if wt_filter == "warn" and wt != "warn":
|
||
continue
|
||
if wt_filter == "all":
|
||
pass
|
||
elif wt_filter not in ("all", "warn", "overdue"):
|
||
pass
|
||
flattened.append(item)
|
||
|
||
total = len(flattened)
|
||
page = max(1, page)
|
||
page_size = min(max(1, page_size), 100)
|
||
start = (page - 1) * page_size
|
||
page_rows = flattened[start : start + page_size]
|
||
return page_rows, total
|
||
|
||
|
||
def create_collection_protocol(
|
||
db: Session,
|
||
*,
|
||
protocol_no_id: Optional[str] = None,
|
||
protocol_no: str = "",
|
||
protocol_status: str = "dsx",
|
||
protocol_attachments: Optional[str] = None,
|
||
register_date: Optional[str] = None,
|
||
collector_user_id: Optional[str] = None,
|
||
remark: Optional[str] = None,
|
||
tenant_no: str = "zkxw",
|
||
user_id: Optional[str] = None,
|
||
) -> BizCollectionProtocol:
|
||
pn = (protocol_no or "").strip()
|
||
if not pn:
|
||
pn = allocate_next_biz_no(
|
||
db, tenant_no=tenant_no, kind="collection", user_id=user_id
|
||
)
|
||
assert_biz_no_available(db, tenant_no=tenant_no, biz_no=pn)
|
||
obj = BizCollectionProtocol(
|
||
id=generate_uuid(),
|
||
protocol_no_id=(protocol_no_id or "").strip() or None,
|
||
protocol_no=pn,
|
||
protocol_status=protocol_status.strip() or "dsx",
|
||
protocol_attachments=(protocol_attachments or "").strip() or None,
|
||
register_date=(register_date or "").strip() or None,
|
||
collector_user_id=(collector_user_id or "").strip() or None,
|
||
tenant_no=tenant_no,
|
||
order_num=0,
|
||
remark=(remark or "").strip() or None,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=datetime.now(),
|
||
update_by=user_id,
|
||
update_time=datetime.now(),
|
||
del_flag=0,
|
||
)
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def update_collection_protocol(
|
||
db: Session,
|
||
protocol_id: str,
|
||
*,
|
||
tenant_no: str,
|
||
user_id: Optional[str] = None,
|
||
scope_collector_user_id: Optional[str] = None,
|
||
protocol_no: Optional[str] = None,
|
||
protocol_status: Optional[str] = None,
|
||
protocol_attachments: Optional[str] = None,
|
||
register_date: Optional[str] = None,
|
||
effective_time: Optional[datetime] = None,
|
||
effective_desc: Optional[str] = None,
|
||
expire_time: Optional[datetime] = None,
|
||
expire_desc: Optional[str] = None,
|
||
void_time: Optional[datetime] = None,
|
||
void_desc: Optional[str] = None,
|
||
to_research_time: Optional[datetime] = None,
|
||
to_research_desc: Optional[str] = None,
|
||
collector_user_id: Optional[str] = None,
|
||
remark: Optional[str] = None,
|
||
) -> BizCollectionProtocol:
|
||
obj = get_collection_protocol(
|
||
db, protocol_id, tenant_no, scope_collector_user_id=scope_collector_user_id
|
||
)
|
||
if not obj:
|
||
raise ValueError("协议单不存在或无权操作")
|
||
if protocol_no is not None:
|
||
pn = protocol_no.strip()
|
||
if pn != (obj.protocol_no or ""):
|
||
raise ValueError("协议号不可修改")
|
||
if protocol_status is not None:
|
||
obj.protocol_status = protocol_status.strip()
|
||
if protocol_attachments is not None:
|
||
obj.protocol_attachments = protocol_attachments.strip() or None
|
||
if register_date is not None:
|
||
obj.register_date = register_date.strip() or None
|
||
if effective_time is not None:
|
||
obj.effective_time = effective_time
|
||
if effective_desc is not None:
|
||
obj.effective_desc = effective_desc.strip() or None
|
||
if expire_time is not None:
|
||
obj.expire_time = expire_time
|
||
if expire_desc is not None:
|
||
obj.expire_desc = expire_desc.strip() or None
|
||
if void_time is not None:
|
||
obj.void_time = void_time
|
||
if void_desc is not None:
|
||
obj.void_desc = void_desc.strip() or None
|
||
if to_research_time is not None:
|
||
obj.to_research_time = to_research_time
|
||
if to_research_desc is not None:
|
||
obj.to_research_desc = to_research_desc.strip() or None
|
||
if collector_user_id is not None:
|
||
obj.collector_user_id = collector_user_id.strip() or None
|
||
if remark is not None:
|
||
obj.remark = remark.strip() or None
|
||
obj.update_by = user_id
|
||
obj.update_time = datetime.now()
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def soft_delete_collection_protocol(
|
||
db: Session,
|
||
protocol_id: str,
|
||
tenant_no: str,
|
||
user_id: Optional[str] = None,
|
||
scope_collector_user_id: Optional[str] = None,
|
||
) -> None:
|
||
obj = get_collection_protocol(
|
||
db, protocol_id, tenant_no, scope_collector_user_id=scope_collector_user_id
|
||
)
|
||
if not obj:
|
||
raise ValueError("协议单不存在或无权操作")
|
||
obj.del_flag = 1
|
||
obj.update_by = user_id
|
||
obj.update_time = datetime.now()
|
||
db.add(obj)
|
||
db.commit()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 子表：当事人客户 / 采集信息 / 样本 / 储存证书
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def get_party_customer(db: Session, collection_protocol_id: str, tenant_no: str) -> Optional[BizProtocolPartyCustomer]:
|
||
return (
|
||
db.query(BizProtocolPartyCustomer)
|
||
.filter(
|
||
and_(
|
||
BizProtocolPartyCustomer.collection_protocol_id == collection_protocol_id,
|
||
BizProtocolPartyCustomer.tenant_no == tenant_no,
|
||
BizProtocolPartyCustomer.del_flag == 0,
|
||
)
|
||
)
|
||
.first()
|
||
)
|
||
|
||
|
||
def upsert_party_customer(
|
||
db: Session,
|
||
collection_protocol_id: str,
|
||
tenant_no: str,
|
||
user_id: Optional[str],
|
||
**fields,
|
||
) -> BizProtocolPartyCustomer:
|
||
fields = _normalize_fields(fields)
|
||
obj = get_party_customer(db, collection_protocol_id, tenant_no)
|
||
_STR_FIELDS = [
|
||
"customer_id", "mother_name", "mother_phone", "mother_id_card",
|
||
"mother_birth_ym", "mother_ethnic", "mother_email", "mother_address",
|
||
"father_name", "father_phone", "father_id_card",
|
||
"newborn_name", "newborn_id_card",
|
||
"newborn_sex", "delivery_mode", "collector_user_id",
|
||
]
|
||
_INT_FIELDS = ["birth_order", "newborn_weight_gram"]
|
||
_DEC_FIELDS = ["apgar_1min", "apgar_5min"]
|
||
_DATE_FIELDS = ["expected_delivery_date"]
|
||
_DT_FIELDS = ["newborn_delivery_time"]
|
||
if obj:
|
||
for k in _STR_FIELDS:
|
||
if k in fields:
|
||
setattr(obj, k, (fields[k] or "").strip() or None)
|
||
for k in _INT_FIELDS:
|
||
if k in fields:
|
||
setattr(obj, k, fields[k])
|
||
for k in _DEC_FIELDS:
|
||
if k in fields:
|
||
setattr(obj, k, _optional_decimal(fields[k]))
|
||
for k in _DATE_FIELDS:
|
||
if k in fields:
|
||
setattr(obj, k, optional_date(fields[k]))
|
||
for k in _DT_FIELDS:
|
||
if k in fields:
|
||
setattr(obj, k, fields[k])
|
||
if "remark" in fields:
|
||
obj.remark = (fields["remark"] or "").strip() or None
|
||
obj.update_by = user_id
|
||
obj.update_time = datetime.now()
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
obj = BizProtocolPartyCustomer(
|
||
id=generate_uuid(),
|
||
collection_protocol_id=collection_protocol_id,
|
||
tenant_no=tenant_no,
|
||
order_num=0,
|
||
remark=(fields.get("remark") or "").strip() or None,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=datetime.now(),
|
||
update_by=user_id,
|
||
update_time=datetime.now(),
|
||
del_flag=0,
|
||
)
|
||
for k in _STR_FIELDS:
|
||
if k in fields:
|
||
setattr(obj, k, (fields[k] or "").strip() or None)
|
||
for k in _INT_FIELDS:
|
||
if k in fields:
|
||
setattr(obj, k, fields[k])
|
||
for k in _DEC_FIELDS:
|
||
if k in fields:
|
||
setattr(obj, k, _optional_decimal(fields[k]))
|
||
for k in _DATE_FIELDS:
|
||
if k in fields:
|
||
setattr(obj, k, optional_date(fields[k]))
|
||
for k in _DT_FIELDS:
|
||
if k in fields:
|
||
setattr(obj, k, fields[k])
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def get_collection_info(db: Session, collection_protocol_id: str, tenant_no: str) -> Optional[BizProtocolCollectionInfo]:
|
||
return (
|
||
db.query(BizProtocolCollectionInfo)
|
||
.filter(
|
||
and_(
|
||
BizProtocolCollectionInfo.collection_protocol_id == collection_protocol_id,
|
||
BizProtocolCollectionInfo.tenant_no == tenant_no,
|
||
BizProtocolCollectionInfo.del_flag == 0,
|
||
)
|
||
)
|
||
.first()
|
||
)
|
||
|
||
|
||
def upsert_collection_info(
|
||
db: Session,
|
||
collection_protocol_id: str,
|
||
tenant_no: str,
|
||
user_id: Optional[str],
|
||
**fields,
|
||
) -> BizProtocolCollectionInfo:
|
||
fields = _normalize_fields(fields)
|
||
obj = get_collection_info(db, collection_protocol_id, tenant_no)
|
||
_STR_FIELDS = ["region_id", "hospital_id", "medical_staff", "sales_user_id"]
|
||
_DT_FIELDS = ["sample_collect_time", "maternal_blood_collect_time"]
|
||
if obj:
|
||
for k in _STR_FIELDS:
|
||
if k in fields:
|
||
setattr(obj, k, (fields[k] or "").strip() or None)
|
||
for k in _DT_FIELDS:
|
||
if k in fields:
|
||
setattr(obj, k, fields[k])
|
||
if "remark" in fields:
|
||
obj.remark = (fields["remark"] or "").strip() or None
|
||
obj.update_by = user_id
|
||
obj.update_time = datetime.now()
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
obj = BizProtocolCollectionInfo(
|
||
id=generate_uuid(),
|
||
collection_protocol_id=collection_protocol_id,
|
||
tenant_no=tenant_no,
|
||
order_num=0,
|
||
remark=(fields.get("remark") or "").strip() or None,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=datetime.now(),
|
||
update_by=user_id,
|
||
update_time=datetime.now(),
|
||
del_flag=0,
|
||
)
|
||
for k in _STR_FIELDS:
|
||
if k in fields:
|
||
setattr(obj, k, (fields[k] or "").strip() or None)
|
||
for k in _DT_FIELDS:
|
||
if k in fields:
|
||
setattr(obj, k, fields[k])
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def list_samples(db: Session, collection_protocol_id: str, tenant_no: str) -> List[BizProtocolSample]:
|
||
return (
|
||
db.query(BizProtocolSample)
|
||
.filter(
|
||
and_(
|
||
BizProtocolSample.collection_protocol_id == collection_protocol_id,
|
||
BizProtocolSample.tenant_no == tenant_no,
|
||
BizProtocolSample.del_flag == 0,
|
||
)
|
||
)
|
||
.order_by(BizProtocolSample.order_num.asc(), BizProtocolSample.create_time.asc())
|
||
.all()
|
||
)
|
||
|
||
|
||
def _migrate_sample_child_refs_on_replace(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
old_id_to_code: Dict[str, str],
|
||
new_code_to_id: Dict[str, str],
|
||
user_id: Optional[str],
|
||
) -> None:
|
||
"""协议样本行重建后，将接收/质检工单的 protocol_sample_id 迁移到新样本 ID。"""
|
||
if not old_id_to_code or not new_code_to_id:
|
||
return
|
||
old_ids = list(old_id_to_code.keys())
|
||
now = datetime.now()
|
||
for model in (BizSampleReception, BizQcWorkOrder):
|
||
rows = (
|
||
db.query(model)
|
||
.filter(
|
||
model.protocol_sample_id.in_(old_ids),
|
||
model.tenant_no == tenant_no,
|
||
model.del_flag == 0,
|
||
)
|
||
.all()
|
||
)
|
||
for row in rows:
|
||
old_sid = (row.protocol_sample_id or "").strip()
|
||
code = old_id_to_code.get(old_sid) or (row.sample_code or "").strip()
|
||
new_sid = new_code_to_id.get(code) if code else None
|
||
if not new_sid:
|
||
continue
|
||
row.protocol_sample_id = new_sid
|
||
row.update_by = user_id
|
||
row.update_time = now
|
||
db.add(row)
|
||
|
||
|
||
def save_samples(
|
||
db: Session,
|
||
collection_protocol_id: str,
|
||
tenant_no: str,
|
||
user_id: Optional[str],
|
||
samples: list,
|
||
) -> List[BizProtocolSample]:
|
||
existing = list_samples(db, collection_protocol_id, tenant_no)
|
||
old_id_to_code = {
|
||
s.id: (s.sample_code or "").strip()
|
||
for s in existing
|
||
if (s.sample_code or "").strip()
|
||
}
|
||
for s in existing:
|
||
s.del_flag = 1
|
||
s.update_by = user_id
|
||
s.update_time = datetime.now()
|
||
db.add(s)
|
||
if existing:
|
||
db.flush()
|
||
validate_sample_codes_not_used(
|
||
db,
|
||
tenant_no=tenant_no,
|
||
samples=samples,
|
||
exclude_protocol_id=collection_protocol_id,
|
||
)
|
||
_STR_FIELDS = [
|
||
"sample_type_id", "sample_code_id", "sample_code", "unit_id",
|
||
"sample_status", "storage_in_date", "detect_start_date",
|
||
"void_date", "storage_expire_date",
|
||
]
|
||
_DEC_FIELDS = ["collect_amount", "price", "original_price"]
|
||
_INT_FIELDS = ["storage_years"]
|
||
result = []
|
||
for idx, raw in enumerate(samples):
|
||
raw = _normalize_fields(raw)
|
||
obj = BizProtocolSample(
|
||
id=generate_uuid(),
|
||
collection_protocol_id=collection_protocol_id,
|
||
tenant_no=tenant_no,
|
||
order_num=idx,
|
||
remark=(raw.get("remark") or "").strip() or None,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=datetime.now(),
|
||
update_by=user_id,
|
||
update_time=datetime.now(),
|
||
del_flag=0,
|
||
)
|
||
for k in _STR_FIELDS:
|
||
if k == "sample_status":
|
||
continue
|
||
if k in raw:
|
||
setattr(obj, k, (raw[k] or "").strip() or None)
|
||
# 样本状态：显式支持 null/空字符串，避免 ORM/库默认值 ycj 回填
|
||
if "sample_status" in raw:
|
||
status_val = raw.get("sample_status")
|
||
obj.sample_status = (
|
||
str(status_val).strip() if status_val is not None and str(status_val).strip() else None
|
||
)
|
||
for k in _DEC_FIELDS:
|
||
if k in raw and raw[k] is not None:
|
||
setattr(obj, k, Decimal(str(raw[k])))
|
||
for k in _INT_FIELDS:
|
||
if k in raw:
|
||
setattr(obj, k, raw[k])
|
||
db.add(obj)
|
||
result.append(obj)
|
||
new_code_to_id = {
|
||
(o.sample_code or "").strip(): o.id
|
||
for o in result
|
||
if (o.sample_code or "").strip()
|
||
}
|
||
_migrate_sample_child_refs_on_replace(
|
||
db,
|
||
tenant_no=tenant_no,
|
||
old_id_to_code=old_id_to_code,
|
||
new_code_to_id=new_code_to_id,
|
||
user_id=user_id,
|
||
)
|
||
db.commit()
|
||
for o in result:
|
||
db.refresh(o)
|
||
save_sample_transports(db, collection_protocol_id, tenant_no, user_id, result, samples)
|
||
return result
|
||
|
||
|
||
def list_sample_transports(
|
||
db: Session,
|
||
collection_protocol_id: str,
|
||
tenant_no: str,
|
||
) -> List[BizProtocolSampleTransport]:
|
||
return (
|
||
db.query(BizProtocolSampleTransport)
|
||
.filter(
|
||
and_(
|
||
BizProtocolSampleTransport.collection_protocol_id == collection_protocol_id,
|
||
BizProtocolSampleTransport.tenant_no == tenant_no,
|
||
BizProtocolSampleTransport.del_flag == 0,
|
||
)
|
||
)
|
||
.order_by(BizProtocolSampleTransport.order_num.asc(), BizProtocolSampleTransport.create_time.asc())
|
||
.all()
|
||
)
|
||
|
||
|
||
def _soft_delete_transports_for_protocol(
|
||
db: Session,
|
||
collection_protocol_id: str,
|
||
tenant_no: str,
|
||
user_id: Optional[str],
|
||
) -> None:
|
||
rows = list_sample_transports(db, collection_protocol_id, tenant_no)
|
||
for r in rows:
|
||
r.del_flag = 1
|
||
r.update_by = user_id
|
||
r.update_time = datetime.now()
|
||
db.add(r)
|
||
if rows:
|
||
db.flush()
|
||
|
||
|
||
def save_sample_transports(
|
||
db: Session,
|
||
collection_protocol_id: str,
|
||
tenant_no: str,
|
||
user_id: Optional[str],
|
||
saved_samples: List[BizProtocolSample],
|
||
input_samples: list,
|
||
) -> None:
|
||
_soft_delete_transports_for_protocol(db, collection_protocol_id, tenant_no, user_id)
|
||
for idx, sample_obj in enumerate(saved_samples):
|
||
if idx >= len(input_samples):
|
||
break
|
||
raw = _normalize_fields(input_samples[idx] if isinstance(input_samples[idx], dict) else {})
|
||
transport_raw = raw.get("transport")
|
||
if not isinstance(transport_raw, dict):
|
||
continue
|
||
tnorm = _normalize_fields(transport_raw)
|
||
transport_type = (tnorm.get("transport_type") or "").strip()
|
||
if not transport_type:
|
||
continue
|
||
|
||
eta_time = tnorm.get("eta_time")
|
||
if isinstance(eta_time, str):
|
||
eta_time = parse_datetime(eta_time)
|
||
|
||
express_fee = tnorm.get("express_fee")
|
||
if express_fee is not None and express_fee != "":
|
||
express_fee = Decimal(str(express_fee))
|
||
else:
|
||
express_fee = None
|
||
|
||
obj = BizProtocolSampleTransport(
|
||
id=generate_uuid(),
|
||
collection_protocol_id=collection_protocol_id,
|
||
sample_id=sample_obj.id,
|
||
sample_code=sample_obj.sample_code or None,
|
||
transport_type=transport_type,
|
||
express_company_id=(tnorm.get("express_company_id") or "").strip() or None,
|
||
waybill_no=(tnorm.get("waybill_no") or "").strip() or None,
|
||
dest_site_id=(tnorm.get("dest_site_id") or "").strip() or None,
|
||
express_fee=express_fee,
|
||
eta_time=eta_time,
|
||
carrier_name=(tnorm.get("carrier_name") or "").strip() or None,
|
||
carrier_phone=(tnorm.get("carrier_phone") or "").strip() or None,
|
||
tenant_no=tenant_no,
|
||
order_num=idx,
|
||
remark=(tnorm.get("remark") or "").strip() or None,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=datetime.now(),
|
||
update_by=user_id,
|
||
update_time=datetime.now(),
|
||
del_flag=0,
|
||
)
|
||
db.add(obj)
|
||
db.commit()
|
||
|
||
|
||
def get_storage_cert(db: Session, collection_protocol_id: str, tenant_no: str) -> Optional[BizProtocolStorageCert]:
|
||
return (
|
||
db.query(BizProtocolStorageCert)
|
||
.filter(
|
||
and_(
|
||
BizProtocolStorageCert.collection_protocol_id == collection_protocol_id,
|
||
BizProtocolStorageCert.tenant_no == tenant_no,
|
||
BizProtocolStorageCert.del_flag == 0,
|
||
)
|
||
)
|
||
.first()
|
||
)
|
||
|
||
|
||
def _add_months_yyyymmdd(date_str: str, months: int) -> str:
|
||
raw = (date_str or "").strip().replace("-", "")
|
||
if len(raw) != 8:
|
||
return ""
|
||
y = int(raw[0:4])
|
||
m = int(raw[4:6])
|
||
d = int(raw[6:8])
|
||
month_index = m - 1 + months
|
||
ny = y + month_index // 12
|
||
nm = month_index % 12 + 1
|
||
last_day = calendar.monthrange(ny, nm)[1]
|
||
nd = min(d, last_day)
|
||
return f"{ny:04d}{nm:02d}{nd:02d}"
|
||
|
||
|
||
def _calc_installment_due_amounts(
|
||
contract_amount: Optional[Decimal],
|
||
count: int,
|
||
) -> List[Decimal]:
|
||
if contract_amount is None or count < 1:
|
||
return []
|
||
total_cents = int(round(float(contract_amount) * 100))
|
||
base_cents = total_cents // count
|
||
amounts: List[Decimal] = []
|
||
sum_cents = 0
|
||
for i in range(count):
|
||
if i == count - 1:
|
||
cents = total_cents - sum_cents
|
||
else:
|
||
cents = base_cents
|
||
sum_cents += base_cents
|
||
amounts.append(Decimal(cents) / Decimal(100))
|
||
return amounts
|
||
|
||
|
||
def align_fq_payment_details_payload(
|
||
pay: BizProtocolPayment,
|
||
details: Optional[list],
|
||
) -> list:
|
||
if pay.pay_mode != "fq" or not pay.installment_count or int(pay.installment_count) < 1:
|
||
return list(details or [])
|
||
count = int(pay.installment_count)
|
||
rows = [_normalize_fields(dict(d)) for d in (details or [])]
|
||
if len(rows) >= count:
|
||
needed = set(range(1, count + 1))
|
||
have = {
|
||
int(r["installment_no"])
|
||
for r in rows
|
||
if r.get("installment_no") is not None
|
||
}
|
||
if needed - have == set():
|
||
return rows[:count]
|
||
|
||
by_no: Dict[int, Dict[str, Any]] = {}
|
||
unnumbered: List[Dict[str, Any]] = []
|
||
for r in rows:
|
||
no = r.get("installment_no")
|
||
if no is not None:
|
||
by_no[int(no)] = r
|
||
else:
|
||
unnumbered.append(r)
|
||
|
||
ratios = parse_installment_ratios(pay.installment_ratios)
|
||
amounts = resolve_installment_due_amounts(pay.contract_amount, ratios, count)
|
||
start = (pay.installment_start_date or "").strip()
|
||
interval = pay.installment_interval_months
|
||
interval_val = float(interval) if interval is not None else 0
|
||
advance_days = int(pay.installment_advance_days or 30)
|
||
result: List[Dict[str, Any]] = []
|
||
for i in range(count):
|
||
no = i + 1
|
||
if no in by_no:
|
||
result.append(by_no[no])
|
||
continue
|
||
if i < len(unnumbered):
|
||
raw = dict(unnumbered[i])
|
||
raw["installment_no"] = no
|
||
if not raw.get("installment_due_date") and start and interval_val:
|
||
raw["installment_due_date"] = calc_fq_installment_due_date(
|
||
start, i, interval_val, advance_days
|
||
)
|
||
result.append(raw)
|
||
continue
|
||
due_date = (
|
||
calc_fq_installment_due_date(start, i, interval_val, advance_days)
|
||
if start and interval_val
|
||
else ""
|
||
)
|
||
due = amounts[i] if i < len(amounts) else None
|
||
result.append(
|
||
{
|
||
"installment_no": no,
|
||
"installment_due_date": due_date or None,
|
||
"due_amount": float(due) if due is not None else None,
|
||
"pay_status": "dfj",
|
||
"transfer_type": "",
|
||
"transfer_amount": None,
|
||
"attachments": [],
|
||
}
|
||
)
|
||
return result
|
||
|
||
|
||
def list_payment_details(
|
||
db: Session,
|
||
payment_id: str,
|
||
tenant_no: str,
|
||
) -> List[BizProtocolPaymentDetail]:
|
||
return (
|
||
db.query(BizProtocolPaymentDetail)
|
||
.filter(
|
||
and_(
|
||
BizProtocolPaymentDetail.payment_id == payment_id,
|
||
BizProtocolPaymentDetail.tenant_no == tenant_no,
|
||
BizProtocolPaymentDetail.del_flag == 0,
|
||
)
|
||
)
|
||
.order_by(
|
||
BizProtocolPaymentDetail.installment_no.asc(),
|
||
BizProtocolPaymentDetail.order_num.asc(),
|
||
BizProtocolPaymentDetail.create_time.asc(),
|
||
)
|
||
.all()
|
||
)
|
||
|
||
|
||
def upsert_payment(
|
||
db: Session,
|
||
collection_protocol_id: str,
|
||
tenant_no: str,
|
||
user_id: Optional[str],
|
||
**fields,
|
||
) -> BizProtocolPayment:
|
||
fields = _normalize_fields(fields)
|
||
obj = get_payment_by_protocol_id(db, collection_protocol_id, tenant_no)
|
||
_STR_FIELDS = ["pay_mode"]
|
||
_STR_DATE_FIELDS = ["lump_sum_deadline", "installment_start_date"]
|
||
_DEC_FIELDS = ["contract_amount", "installment_interval_months"]
|
||
_INT_FIELDS = ["installment_count", "installment_advance_days"]
|
||
|
||
def _apply(target: BizProtocolPayment) -> None:
|
||
for k in _STR_FIELDS:
|
||
if k in fields:
|
||
setattr(target, k, (fields[k] or "").strip() or None)
|
||
for k in _STR_DATE_FIELDS:
|
||
if k in fields:
|
||
setattr(target, k, _normalize_yyyymmdd_date(fields[k]))
|
||
for k in _DEC_FIELDS:
|
||
if k in fields and fields[k] is not None:
|
||
setattr(target, k, Decimal(str(fields[k])))
|
||
for k in _INT_FIELDS:
|
||
if k in fields:
|
||
setattr(target, k, fields[k])
|
||
if "installment_ratios" in fields:
|
||
val = fields["installment_ratios"]
|
||
if val is None:
|
||
target.installment_ratios = None
|
||
elif isinstance(val, str):
|
||
target.installment_ratios = val.strip() or None
|
||
else:
|
||
target.installment_ratios = json.dumps(list(val), ensure_ascii=False)
|
||
if "remark" in fields:
|
||
target.remark = (fields["remark"] or "").strip() or None
|
||
|
||
if obj:
|
||
_apply(obj)
|
||
obj.update_by = user_id
|
||
obj.update_time = datetime.now()
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
pay_mode = (fields.get("pay_mode") or "").strip() or "dj"
|
||
obj = BizProtocolPayment(
|
||
id=generate_uuid(),
|
||
collection_protocol_id=collection_protocol_id,
|
||
pay_mode=pay_mode,
|
||
tenant_no=tenant_no,
|
||
order_num=0,
|
||
remark=(fields.get("remark") or "").strip() or None,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=datetime.now(),
|
||
update_by=user_id,
|
||
update_time=datetime.now(),
|
||
del_flag=0,
|
||
)
|
||
_apply(obj)
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
def _compare_money(val: Any) -> Optional[str]:
|
||
if val is None or val == "":
|
||
return None
|
||
try:
|
||
return f"{Decimal(str(val)).quantize(Decimal('0.01'))}"
|
||
except Exception:
|
||
return str(val).strip() or None
|
||
|
||
|
||
def _compare_datetime(val: Any) -> Optional[str]:
|
||
if val is None or val == "":
|
||
return None
|
||
if isinstance(val, datetime):
|
||
return _fmt_dt(val)
|
||
if isinstance(val, str):
|
||
dt = parse_datetime(val)
|
||
return _fmt_dt(dt) if dt else val.strip()
|
||
return str(val)
|
||
|
||
|
||
def _payment_detail_compare_row(raw: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""Normalize one payment detail row for equality compare."""
|
||
raw = _normalize_fields(dict(raw))
|
||
transfer_att = _pop_attachment_urls(
|
||
raw, "attachments", "transfer_attachments", "transferAttachments"
|
||
)
|
||
finance_att = _pop_attachment_urls(
|
||
raw,
|
||
"finance_confirm_attachments",
|
||
"financeConfirmAttachments",
|
||
"finance_confirm_attachment_files",
|
||
"financeConfirmAttachmentFiles",
|
||
)
|
||
refund_att = _pop_attachment_urls(
|
||
raw,
|
||
"refund_attachments",
|
||
"refundAttachments",
|
||
"refund_attachment_files",
|
||
"refundAttachmentFiles",
|
||
)
|
||
refund_finance_att = _pop_attachment_urls(
|
||
raw,
|
||
"refund_finance_confirm_attachments",
|
||
"refundFinanceConfirmAttachments",
|
||
"refund_finance_confirm_attachment_files",
|
||
"refundFinanceConfirmAttachmentFiles",
|
||
)
|
||
transfer_time = raw.get("transfer_time")
|
||
if isinstance(transfer_time, str):
|
||
transfer_time = parse_datetime(transfer_time)
|
||
|
||
pay_status = (raw.get("pay_status") or "").strip() or "dfj"
|
||
|
||
return {
|
||
"installment_no": raw.get("installment_no"),
|
||
"installment_due_date": _normalize_yyyymmdd_date(raw.get("installment_due_date")),
|
||
"due_amount": _compare_money(raw.get("due_amount")),
|
||
"transfer_type": (raw.get("transfer_type") or "").strip() or None,
|
||
"transfer_amount": _compare_money(raw.get("transfer_amount")),
|
||
"transfer_time": _compare_datetime(transfer_time),
|
||
"pay_status": pay_status,
|
||
"remark": (raw.get("remark") or "").strip() or None,
|
||
"finance_remark": (raw.get("finance_remark") or "").strip() or None,
|
||
"refund_remark": (raw.get("refund_remark") or "").strip() or None,
|
||
"refund_finance_remark": (raw.get("refund_finance_remark") or "").strip() or None,
|
||
"refund_amount": _compare_money(raw.get("refund_amount")),
|
||
"transfer_attachments": transfer_att or None,
|
||
"finance_confirm_attachments": finance_att or None,
|
||
"refund_attachments": refund_att or None,
|
||
"refund_finance_confirm_attachments": refund_finance_att or None,
|
||
}
|
||
|
||
|
||
def _payment_detail_compare_from_model(d: BizProtocolPaymentDetail) -> Dict[str, Any]:
|
||
pay_status = (d.pay_status or "").strip() or "dfj"
|
||
return {
|
||
"installment_no": d.installment_no,
|
||
"installment_due_date": _normalize_yyyymmdd_date(d.installment_due_date),
|
||
"due_amount": _compare_money(d.due_amount),
|
||
"transfer_type": (d.transfer_type or "").strip() or None,
|
||
"transfer_amount": _compare_money(d.transfer_amount),
|
||
"transfer_time": _compare_datetime(d.transfer_time),
|
||
"pay_status": pay_status,
|
||
"remark": (d.remark or "").strip() or None,
|
||
"finance_remark": (d.finance_remark or "").strip() or None,
|
||
"refund_remark": (d.refund_remark or "").strip() or None,
|
||
"refund_finance_remark": (d.refund_finance_remark or "").strip() or None,
|
||
"refund_amount": _compare_money(d.refund_amount),
|
||
"transfer_attachments": (d.transfer_attachments or "").strip() or None,
|
||
"finance_confirm_attachments": (d.finance_confirm_attachments or "").strip() or None,
|
||
"refund_attachments": (d.refund_attachments or "").strip() or None,
|
||
"refund_finance_confirm_attachments": (
|
||
(d.refund_finance_confirm_attachments or "").strip() or None
|
||
),
|
||
}
|
||
|
||
|
||
def _payment_detail_sort_key(row: Dict[str, Any]) -> tuple:
|
||
no = row.get("installment_no")
|
||
if no is not None:
|
||
return (0, int(no))
|
||
return (1, 0)
|
||
|
||
|
||
def _payment_details_payload_unchanged(
|
||
existing: List[BizProtocolPaymentDetail],
|
||
incoming: list,
|
||
) -> bool:
|
||
incoming_rows = [_payment_detail_compare_row(dict(d)) for d in (incoming or [])]
|
||
existing_rows = [_payment_detail_compare_from_model(d) for d in existing]
|
||
if len(existing_rows) != len(incoming_rows):
|
||
return False
|
||
existing_sorted = sorted(existing_rows, key=_payment_detail_sort_key)
|
||
incoming_sorted = sorted(incoming_rows, key=_payment_detail_sort_key)
|
||
return existing_sorted == incoming_sorted
|
||
|
||
|
||
def save_payment_details(
|
||
db: Session,
|
||
payment_id: str,
|
||
tenant_no: str,
|
||
user_id: Optional[str],
|
||
details: list,
|
||
) -> List[BizProtocolPaymentDetail]:
|
||
existing = list_payment_details(db, payment_id, tenant_no)
|
||
if _payment_details_payload_unchanged(existing, details):
|
||
return existing
|
||
|
||
pay = (
|
||
db.query(BizProtocolPayment)
|
||
.filter(
|
||
and_(
|
||
BizProtocolPayment.id == payment_id,
|
||
BizProtocolPayment.tenant_no == tenant_no,
|
||
BizProtocolPayment.del_flag == 0,
|
||
)
|
||
)
|
||
.first()
|
||
)
|
||
if pay:
|
||
normalized = [_normalize_fields(dict(d)) for d in (details or [])]
|
||
validate_payment_details_for_save(
|
||
pay.pay_mode or "",
|
||
pay.contract_amount,
|
||
pay.installment_count,
|
||
normalized,
|
||
installment_ratios=parse_installment_ratios(pay.installment_ratios),
|
||
installment_advance_days=pay.installment_advance_days,
|
||
)
|
||
|
||
for d in existing:
|
||
d.del_flag = 1
|
||
d.update_by = user_id
|
||
d.update_time = datetime.now()
|
||
db.add(d)
|
||
if existing:
|
||
db.flush()
|
||
|
||
_STR_FIELDS = [
|
||
"transfer_type",
|
||
"pay_status",
|
||
"finance_remark",
|
||
"refund_finance_remark",
|
||
"refund_remark",
|
||
]
|
||
_STR_DATE_FIELDS = ["installment_due_date"]
|
||
_DEC_FIELDS = ["due_amount", "transfer_amount", "refund_amount"]
|
||
_INT_FIELDS = ["installment_no"]
|
||
result: List[BizProtocolPaymentDetail] = []
|
||
for idx, raw in enumerate(details or []):
|
||
raw = _normalize_fields(dict(raw))
|
||
transfer_att = _pop_attachment_urls(
|
||
raw, "attachments", "transfer_attachments", "transferAttachments"
|
||
)
|
||
finance_confirm_att = _pop_attachment_urls(
|
||
raw,
|
||
"finance_confirm_attachments",
|
||
"financeConfirmAttachments",
|
||
"finance_confirm_attachment_files",
|
||
"financeConfirmAttachmentFiles",
|
||
)
|
||
refund_att = _pop_attachment_urls(
|
||
raw,
|
||
"refund_attachments",
|
||
"refundAttachments",
|
||
"refund_attachment_files",
|
||
"refundAttachmentFiles",
|
||
)
|
||
refund_finance_att = _pop_attachment_urls(
|
||
raw,
|
||
"refund_finance_confirm_attachments",
|
||
"refundFinanceConfirmAttachments",
|
||
"refund_finance_confirm_attachment_files",
|
||
"refundFinanceConfirmAttachmentFiles",
|
||
)
|
||
|
||
transfer_time = raw.get("transfer_time")
|
||
if isinstance(transfer_time, str):
|
||
transfer_time = parse_datetime(transfer_time)
|
||
|
||
installment_no = raw.get("installment_no")
|
||
if installment_no is None and raw.get("installment_due_date"):
|
||
installment_no = idx + 1
|
||
|
||
obj = BizProtocolPaymentDetail(
|
||
id=generate_uuid(),
|
||
payment_id=payment_id,
|
||
tenant_no=tenant_no,
|
||
order_num=idx,
|
||
remark=(raw.get("remark") or "").strip() or None,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=datetime.now(),
|
||
update_by=user_id,
|
||
update_time=datetime.now(),
|
||
del_flag=0,
|
||
transfer_attachments=transfer_att,
|
||
finance_confirm_attachments=finance_confirm_att,
|
||
refund_attachments=refund_att,
|
||
refund_finance_confirm_attachments=refund_finance_att,
|
||
transfer_time=transfer_time,
|
||
installment_no=installment_no,
|
||
)
|
||
for k in _STR_FIELDS:
|
||
if k in raw:
|
||
setattr(obj, k, (raw[k] or "").strip() or None)
|
||
for k in _STR_DATE_FIELDS:
|
||
if k in raw:
|
||
setattr(obj, k, _normalize_yyyymmdd_date(raw[k]))
|
||
for k in _DEC_FIELDS:
|
||
if k in raw and raw[k] is not None:
|
||
setattr(obj, k, Decimal(str(raw[k])))
|
||
for k in _INT_FIELDS:
|
||
if k in raw and raw[k] is not None:
|
||
setattr(obj, k, raw[k])
|
||
if "pay_status" not in raw or not (raw.get("pay_status") or "").strip():
|
||
obj.pay_status = "dfj"
|
||
db.add(obj)
|
||
result.append(obj)
|
||
db.commit()
|
||
for o in result:
|
||
db.refresh(o)
|
||
return result
|
||
|
||
|
||
def upsert_storage_cert(
|
||
db: Session,
|
||
collection_protocol_id: str,
|
||
tenant_no: str,
|
||
user_id: Optional[str],
|
||
**fields,
|
||
) -> BizProtocolStorageCert:
|
||
fields = _normalize_fields(fields)
|
||
obj = get_storage_cert(db, collection_protocol_id, tenant_no)
|
||
_STR_FIELDS = ["make_status", "cert_image_urls", "make_date"]
|
||
if obj:
|
||
for k in _STR_FIELDS:
|
||
if k in fields:
|
||
setattr(obj, k, (fields[k] or "").strip() or None)
|
||
if "remark" in fields:
|
||
obj.remark = (fields["remark"] or "").strip() or None
|
||
obj.update_by = user_id
|
||
obj.update_time = datetime.now()
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
obj = BizProtocolStorageCert(
|
||
id=generate_uuid(),
|
||
collection_protocol_id=collection_protocol_id,
|
||
tenant_no=tenant_no,
|
||
order_num=0,
|
||
remark=(fields.get("remark") or "").strip() or None,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=datetime.now(),
|
||
update_by=user_id,
|
||
update_time=datetime.now(),
|
||
del_flag=0,
|
||
)
|
||
for k in _STR_FIELDS:
|
||
if k in fields:
|
||
setattr(obj, k, (fields[k] or "").strip() or None)
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 子表：满月待办
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _fullmoon_todo_empty_dict() -> Dict[str, Any]:
|
||
return {
|
||
"todoStatus": "0",
|
||
"handleDate": "",
|
||
"handleRemark": "",
|
||
"insuranceAttachments": "",
|
||
"insuranceCompanyId": "",
|
||
"insuranceCompanyName": "",
|
||
"insuranceType": "",
|
||
"insurancePolicyNo": "",
|
||
"insuranceEffectiveDate": "",
|
||
"insuranceExpiryDate": "",
|
||
"insuranceSubscribed": "0",
|
||
"insuranceFee": None,
|
||
"fullmoonDate": "",
|
||
"isFullmoonPassed": False,
|
||
}
|
||
|
||
|
||
def _fullmoon_todo_detail_dict(
|
||
ft: Optional[BizProtocolFullmoonTodo],
|
||
*,
|
||
newborn_delivery_time: Optional[datetime] = None,
|
||
) -> Dict[str, Any]:
|
||
"""协议详情 bundle 用：满月待办字段 + 由分娩日推算的满月日期。"""
|
||
base = _fullmoon_todo_to_dict(ft) if ft else dict(_fullmoon_todo_empty_dict())
|
||
fullmoon_date_str = ""
|
||
is_fullmoon_passed = False
|
||
if newborn_delivery_time:
|
||
fullmoon_date = newborn_delivery_time + timedelta(days=30)
|
||
fullmoon_date_str = fullmoon_date.strftime("%Y-%m-%d")
|
||
is_fullmoon_passed = fullmoon_date <= datetime.now()
|
||
base["fullmoonDate"] = fullmoon_date_str
|
||
base["isFullmoonPassed"] = is_fullmoon_passed
|
||
return base
|
||
|
||
|
||
def _fullmoon_todo_to_dict(ft: BizProtocolFullmoonTodo) -> Dict[str, Any]:
|
||
return {
|
||
"id": ft.id,
|
||
"collectionProtocolId": ft.collection_protocol_id,
|
||
"todoStatus": ft.todo_status or "0",
|
||
"handleDate": (ft.handle_date or "").strip() if ft.handle_date else "",
|
||
"handleRemark": ft.handle_remark or "",
|
||
"insuranceAttachments": ft.insurance_attachments or "",
|
||
"insuranceCompanyId": ft.insurance_company_id or "",
|
||
"insuranceCompanyName": ft.insurance_company_name or "",
|
||
"insuranceType": ft.insurance_type or "",
|
||
"insurancePolicyNo": ft.insurance_policy_no or "",
|
||
"insuranceEffectiveDate": (ft.insurance_effective_date or "").strip() if ft.insurance_effective_date else "",
|
||
"insuranceExpiryDate": (ft.insurance_expiry_date or "").strip() if ft.insurance_expiry_date else "",
|
||
"insuranceSubscribed": ft.insurance_subscribed or "0",
|
||
"insuranceFee": _dec(ft.insurance_fee),
|
||
"tenantNo": ft.tenant_no,
|
||
"orderNum": ft.order_num if ft.order_num is not None else 0,
|
||
"remark": ft.remark or "",
|
||
"createTime": _fmt_dt(ft.create_time),
|
||
"updateTime": _fmt_dt(ft.update_time),
|
||
}
|
||
|
||
|
||
def get_fullmoon_todo(
|
||
db: Session,
|
||
collection_protocol_id: str,
|
||
tenant_no: str,
|
||
) -> Optional[BizProtocolFullmoonTodo]:
|
||
return (
|
||
db.query(BizProtocolFullmoonTodo)
|
||
.filter(
|
||
and_(
|
||
BizProtocolFullmoonTodo.collection_protocol_id == collection_protocol_id,
|
||
BizProtocolFullmoonTodo.tenant_no == tenant_no,
|
||
BizProtocolFullmoonTodo.del_flag == 0,
|
||
)
|
||
)
|
||
.first()
|
||
)
|
||
|
||
|
||
def upsert_fullmoon_todo(
|
||
db: Session,
|
||
collection_protocol_id: str,
|
||
tenant_no: str,
|
||
user_id: Optional[str],
|
||
**fields,
|
||
) -> BizProtocolFullmoonTodo:
|
||
fields = _normalize_fields(fields)
|
||
obj = get_fullmoon_todo(db, collection_protocol_id, tenant_no)
|
||
_STR_FIELDS = [
|
||
"todo_status", "handle_date", "handle_remark",
|
||
"insurance_attachments", "insurance_company_id",
|
||
"insurance_company_name", "insurance_type",
|
||
"insurance_policy_no", "insurance_effective_date",
|
||
"insurance_expiry_date", "insurance_subscribed",
|
||
]
|
||
_DEC_FIELDS = ["insurance_fee"]
|
||
if obj:
|
||
for k in _STR_FIELDS:
|
||
if k in fields:
|
||
setattr(obj, k, (fields[k] or "").strip() or None)
|
||
for k in _DEC_FIELDS:
|
||
if k in fields and fields[k] is not None:
|
||
setattr(obj, k, Decimal(str(fields[k])))
|
||
if "remark" in fields:
|
||
obj.remark = (fields["remark"] or "").strip() or None
|
||
obj.update_by = user_id
|
||
obj.update_time = datetime.now()
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|
||
obj = BizProtocolFullmoonTodo(
|
||
id=generate_uuid(),
|
||
collection_protocol_id=collection_protocol_id,
|
||
tenant_no=tenant_no,
|
||
order_num=0,
|
||
remark=(fields.get("remark") or "").strip() or None,
|
||
create_by=user_id,
|
||
create_dept_id="",
|
||
create_time=datetime.now(),
|
||
update_by=user_id,
|
||
update_time=datetime.now(),
|
||
del_flag=0,
|
||
)
|
||
for k in _STR_FIELDS:
|
||
if k in fields:
|
||
setattr(obj, k, (fields[k] or "").strip() or None)
|
||
for k in _DEC_FIELDS:
|
||
if k in fields and fields[k] is not None:
|
||
setattr(obj, k, Decimal(str(fields[k])))
|
||
db.add(obj)
|
||
db.commit()
|
||
db.refresh(obj)
|
||
return obj
|