247 lines
7.4 KiB
Python
247 lines
7.4 KiB
Python
"""实验室入库编码：RK + 接收日期 YYYYMMDD + 3 位序号"""
|
||
import re
|
||
from datetime import datetime
|
||
from typing import List, Optional, Tuple
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.utils import generate_uuid
|
||
from app.models.models import BizSampleReception
|
||
|
||
PREFIX = "RK"
|
||
SEQ_DIGITS = 3
|
||
MAX_SEQ = 10**SEQ_DIGITS - 1
|
||
LAB_STORAGE_CODE_RE = re.compile(rf"^{PREFIX}(\d{{8}})(\d{{{SEQ_DIGITS}}})$")
|
||
|
||
|
||
def parse_reception_date(reception_date: str) -> str:
|
||
s = (reception_date or "").strip()
|
||
if len(s) != 10 or s[4] != "-" or s[7] != "-":
|
||
raise ValueError("接收日期格式无效，应为 YYYY-MM-DD")
|
||
y, m, d = s[:4], s[5:7], s[8:10]
|
||
if not (y.isdigit() and m.isdigit() and d.isdigit()):
|
||
raise ValueError("接收日期格式无效，应为 YYYY-MM-DD")
|
||
return f"{y}{m}{d}"
|
||
|
||
|
||
def reception_time_date_str(reception_time: datetime) -> str:
|
||
return reception_time.strftime("%Y%m%d")
|
||
|
||
|
||
def format_lab_storage_code(biz_date: str, seq: int) -> str:
|
||
return f"{PREFIX}{biz_date}{seq:0{SEQ_DIGITS}d}"
|
||
|
||
|
||
def _max_existing_seq(db: Session, tenant_no: str, biz_date: str) -> int:
|
||
prefix_pattern = f"{PREFIX}{biz_date}%"
|
||
rows = (
|
||
db.query(BizSampleReception.lab_storage_code)
|
||
.filter(
|
||
BizSampleReception.tenant_no == tenant_no,
|
||
BizSampleReception.del_flag == 0,
|
||
BizSampleReception.lab_storage_code.isnot(None),
|
||
BizSampleReception.lab_storage_code.like(prefix_pattern),
|
||
)
|
||
.all()
|
||
)
|
||
max_seq = 0
|
||
for row in rows:
|
||
code = (row.lab_storage_code or "").strip()
|
||
m = LAB_STORAGE_CODE_RE.match(code)
|
||
if m and m.group(1) == biz_date:
|
||
try:
|
||
max_seq = max(max_seq, int(m.group(2)))
|
||
except ValueError:
|
||
pass
|
||
return max_seq
|
||
|
||
|
||
def _sync_sequence_from_existing(
|
||
db: Session,
|
||
tenant_no: str,
|
||
biz_date: str,
|
||
user_id: Optional[str] = None,
|
||
) -> None:
|
||
max_existing = _max_existing_seq(db, tenant_no, biz_date)
|
||
if max_existing < 1:
|
||
return
|
||
now = datetime.now()
|
||
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, :max_existing,
|
||
0, NULL, :user_id, '', :now,
|
||
:user_id, :now, 0
|
||
)
|
||
ON DUPLICATE KEY UPDATE
|
||
last_seq = GREATEST(last_seq, :max_existing),
|
||
update_by = :user_id,
|
||
update_time = :now
|
||
"""
|
||
),
|
||
{
|
||
"id": generate_uuid(),
|
||
"tenant_no": tenant_no,
|
||
"prefix": PREFIX,
|
||
"biz_date": biz_date,
|
||
"max_existing": max_existing,
|
||
"user_id": uid,
|
||
"now": now,
|
||
},
|
||
)
|
||
|
||
|
||
def _allocate_one_seq(
|
||
db: Session,
|
||
tenant_no: str,
|
||
biz_date: str,
|
||
user_id: Optional[str] = None,
|
||
) -> int:
|
||
now = datetime.now()
|
||
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": biz_date,
|
||
"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"当日实验室入库编码序号已用尽（最大 {MAX_SEQ}）")
|
||
return seq_int
|
||
|
||
|
||
def allocate_lab_storage_codes(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
reception_date: str,
|
||
count: int = 1,
|
||
user_id: Optional[str] = None,
|
||
commit: bool = True,
|
||
) -> List[str]:
|
||
if count < 1:
|
||
return []
|
||
count = min(count, 50)
|
||
biz_date = parse_reception_date(reception_date)
|
||
_sync_sequence_from_existing(db, tenant_no, biz_date, user_id)
|
||
codes: List[str] = []
|
||
for _ in range(count):
|
||
seq = _allocate_one_seq(db, tenant_no, biz_date, user_id)
|
||
codes.append(format_lab_storage_code(biz_date, seq))
|
||
if commit:
|
||
db.commit()
|
||
return codes
|
||
|
||
|
||
def validate_and_parse_lab_storage_code(
|
||
code: str,
|
||
reception_time: Optional[datetime] = None,
|
||
) -> Tuple[str, Optional[str], Optional[int]]:
|
||
"""校验入库编码非空；若为 RK+日期+序号 格式则解析 biz_date/seq 供序号同步，否则仅返回编码本身。"""
|
||
c = (code or "").strip()
|
||
if not c:
|
||
raise ValueError("实验室入库编码不能为空")
|
||
if len(c) > 32:
|
||
raise ValueError("实验室入库编码长度不能超过 32 个字符")
|
||
m = LAB_STORAGE_CODE_RE.match(c)
|
||
if not m:
|
||
return c, None, None
|
||
biz_date, seq_str = m.group(1), m.group(2)
|
||
if reception_time is not None:
|
||
expected_date = reception_time_date_str(reception_time)
|
||
if biz_date != expected_date:
|
||
raise ValueError("实验室入库编码日期须与样本接收时间日期一致")
|
||
return c, biz_date, int(seq_str)
|
||
|
||
|
||
def assert_lab_storage_code_available(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
code: str,
|
||
exclude_reception_id: Optional[str] = None,
|
||
) -> None:
|
||
q = db.query(BizSampleReception.id).filter(
|
||
BizSampleReception.tenant_no == tenant_no,
|
||
BizSampleReception.del_flag == 0,
|
||
BizSampleReception.lab_storage_code == code,
|
||
)
|
||
if exclude_reception_id:
|
||
q = q.filter(BizSampleReception.id != exclude_reception_id)
|
||
if q.first():
|
||
raise ValueError("实验室入库编码已存在")
|
||
|
||
|
||
def sync_lab_storage_seq_after_use(
|
||
db: Session,
|
||
*,
|
||
tenant_no: str,
|
||
biz_date: str,
|
||
seq: int,
|
||
user_id: Optional[str] = None,
|
||
) -> None:
|
||
if seq < 1:
|
||
return
|
||
now = datetime.now()
|
||
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, :seq,
|
||
0, NULL, :user_id, '', :now,
|
||
:user_id, :now, 0
|
||
)
|
||
ON DUPLICATE KEY UPDATE
|
||
last_seq = GREATEST(last_seq, :seq),
|
||
update_by = :user_id,
|
||
update_time = :now
|
||
"""
|
||
),
|
||
{
|
||
"id": generate_uuid(),
|
||
"tenant_no": tenant_no,
|
||
"prefix": PREFIX,
|
||
"biz_date": biz_date,
|
||
"seq": seq,
|
||
"user_id": uid,
|
||
"now": now,
|
||
},
|
||
)
|