86 lines
2.5 KiB
Python
86 lines
2.5 KiB
Python
"""业务协议号自动生成：XBCC / JZCC + YYYYMMDD + 6 位序号"""
|
||
from datetime import datetime
|
||
from typing import Literal, Optional
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.utils import generate_uuid
|
||
from app.crud.protocol_donation_convert import assert_biz_no_available
|
||
|
||
PREFIX_COLLECTION = "XBCC"
|
||
PREFIX_DONATION = "JZCC"
|
||
SEQ_DIGITS = 6
|
||
MAX_SEQ = 10**SEQ_DIGITS - 1
|
||
|
||
BizNoKind = Literal["collection", "donation"]
|
||
|
||
|
||
def _prefix_for(kind: BizNoKind) -> str:
|
||
return PREFIX_COLLECTION if kind == "collection" else PREFIX_DONATION
|
||
|
||
|
||
def _format_biz_no(prefix: str, biz_date: str, seq: int) -> str:
|
||
return f"{prefix}{biz_date}{seq:0{SEQ_DIGITS}d}"
|
||
|
||
|
||
def allocate_next_biz_no(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
kind: BizNoKind,
|
||
user_id: Optional[str] = None,
|
||
biz_date: Optional[datetime] = None,
|
||
) -> str:
|
||
"""
|
||
原子分配下一个业务协议号。
|
||
须在 create 事务内调用；使用 MySQL LAST_INSERT_ID 保证并发安全。
|
||
"""
|
||
prefix = _prefix_for(kind)
|
||
now = biz_date or datetime.now()
|
||
date_str = now.strftime("%Y%m%d")
|
||
uid = user_id or ""
|
||
|
||
db.execute(
|
||
text(
|
||
"""
|
||
INSERT INTO biz_biz_no_sequence (
|
||
id, tenant_no, prefix, biz_date, last_seq,
|
||
order_num, remark, create_by, create_dept_id, create_time,
|
||
update_by, update_time, del_flag
|
||
) VALUES (
|
||
:id, :tenant_no, :prefix, :biz_date, LAST_INSERT_ID(1),
|
||
0, NULL, :user_id, '', :now,
|
||
:user_id, :now, 0
|
||
)
|
||
ON DUPLICATE KEY UPDATE
|
||
last_seq = LAST_INSERT_ID(last_seq + 1),
|
||
update_by = :user_id,
|
||
update_time = :now
|
||
"""
|
||
),
|
||
{
|
||
"id": generate_uuid(),
|
||
"tenant_no": tenant_no,
|
||
"prefix": prefix,
|
||
"biz_date": date_str,
|
||
"user_id": uid,
|
||
"now": now,
|
||
},
|
||
)
|
||
seq = db.execute(text("SELECT LAST_INSERT_ID()")).scalar()
|
||
if seq is None or int(seq) < 1:
|
||
raise ValueError("协议号序号分配失败")
|
||
seq_int = int(seq)
|
||
if seq_int > MAX_SEQ:
|
||
raise ValueError(f"当日{prefix}协议号序号已用尽（最大 {MAX_SEQ}）")
|
||
|
||
biz_no = _format_biz_no(prefix, date_str, seq_int)
|
||
assert_biz_no_available(
|
||
db,
|
||
tenant_no=tenant_no,
|
||
biz_no=biz_no,
|
||
for_donation=(kind == "donation"),
|
||
)
|
||
return biz_no
|