track-LICA/backend/app/services/counter_service.py

"""序列号计数器 — 基于 PostgreSQL Sequence，生成定长序列号（产品身份证）"""
from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.config import settings

SEQUENCE_NAME = "product_hex_counter"

# 序列号总长度（与 products.serial_number String(16)、前端的 16 位校验一致）
SERIAL_LENGTH = 16


async def ensure_sequence(db: AsyncSession) -> None:
    """确保 counter sequence 存在（幂等）"""
    await db.execute(
        text(f"CREATE SEQUENCE IF NOT EXISTS {SEQUENCE_NAME} START 1;")
    )


async def next_hex_id(db: AsyncSession, length: int = SERIAL_LENGTH) -> str:
    """
    生成下一个序列号：{部门前缀}{递增计数器的 16 进制}，总长固定为 length。

    前缀来自 settings.SERIAL_PREFIX（LICA 是 "L"，IRIS 留空），
    用来在一眼扫号时区分部门来源；计数器部分保持 HEX，便于人工核对数量。

    示例（SERIAL_PREFIX = "L"）:
        1   → "L000000000000001"
        255 → "L0000000000000FF"
    示例（SERIAL_PREFIX = ""，IRIS 实例）:
        1   → "0000000000000001"
        255 → "00000000000000FF"

    ⚠️ 长度必须严格等于 length：products.serial_number 是 String(16)，
       且 /products/qrcode/{sn} 接口会校验 16 位，前缀是从 HEX 位里让出来的。
    """
    result = await db.execute(text(f"SELECT nextval('{SEQUENCE_NAME}');"))
    counter: int = result.scalar_one()

    prefix = settings.SERIAL_PREFIX or ""
    hex_length = length - len(prefix)
    if hex_length <= 0:
        raise ValueError(
            f"SERIAL_PREFIX({prefix!r}) 过长，没有给计数器留下位数"
        )
    return f"{prefix}{format(counter, f'0{hex_length}X')}"