zkxw/backend/app/crud/sample_reception_lab_code.py

"""实验室入库编码校验与（遗留）RK 自动占号分配。"""
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 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_lab_storage_code(code: str) -> str:
    """校验入库编码：非空、长度不超过 32；不限制具体格式。"""
    c = (code or "").strip()
    if not c:
        raise ValueError("实验室入库编码不能为空")
    if len(c) > 32:
        raise ValueError("实验室入库编码长度不能超过 32 个字符")
    return c


def validate_and_parse_lab_storage_code(
    code: str,
) -> Tuple[str, Optional[str], Optional[int]]:
    """兼容旧调用：校验编码；若为 RK+日期+序号 则额外解析供占号逻辑使用。"""
    c = validate_lab_storage_code(code)
    m = LAB_STORAGE_CODE_RE.match(c)
    if not m:
        return c, None, None
    biz_date, seq_str = m.group(1), m.group(2)
    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,
        },
    )