|
|
@@ -0,0 +1,79 @@
|
|
|
+"""质检物料入库编号生成:RK + YYYYMMDD + 3位序号。"""
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import re
|
|
|
+from datetime import date
|
|
|
+from typing import Iterable, List
|
|
|
+
|
|
|
+from sqlalchemy.orm import Session
|
|
|
+
|
|
|
+from app.models.qc_material import BizQcMaterialStockBatch
|
|
|
+
|
|
|
+_SUFFIX_RE = re.compile(r"^(\d{3})$")
|
|
|
+
|
|
|
+
|
|
|
+def format_stock_in_no(d: date, seq: int) -> str:
|
|
|
+ if seq < 1 or seq > 999:
|
|
|
+ raise ValueError(f"入库编号序号超出范围: {seq}")
|
|
|
+ return f"RK{d.strftime('%Y%m%d')}{seq:03d}"
|
|
|
+
|
|
|
+
|
|
|
+def prefix_for_date(d: date) -> str:
|
|
|
+ return f"RK{d.strftime('%Y%m%d')}"
|
|
|
+
|
|
|
+
|
|
|
+def max_seq_from_stock_in_nos(prefix: str, nos: Iterable[str]) -> int:
|
|
|
+ max_seq = 0
|
|
|
+ for raw in nos:
|
|
|
+ s = (raw or "").strip()
|
|
|
+ if not s.startswith(prefix):
|
|
|
+ continue
|
|
|
+ suffix = s[len(prefix) :]
|
|
|
+ m = _SUFFIX_RE.match(suffix)
|
|
|
+ if not m:
|
|
|
+ continue
|
|
|
+ max_seq = max(max_seq, int(m.group(1)))
|
|
|
+ return max_seq
|
|
|
+
|
|
|
+
|
|
|
+def next_stock_in_nos(
|
|
|
+ d: date,
|
|
|
+ *,
|
|
|
+ count: int,
|
|
|
+ existing_nos: Iterable[str],
|
|
|
+) -> List[str]:
|
|
|
+ if count <= 0:
|
|
|
+ return []
|
|
|
+ prefix = prefix_for_date(d)
|
|
|
+ start = max_seq_from_stock_in_nos(prefix, existing_nos) + 1
|
|
|
+ if start + count - 1 > 999:
|
|
|
+ raise ValueError("当日入库编号序号已用尽(上限 999)")
|
|
|
+ return [format_stock_in_no(d, start + i) for i in range(count)]
|
|
|
+
|
|
|
+
|
|
|
+def list_existing_stock_in_nos_for_prefix(
|
|
|
+ db: Session, *, tenant_no: str, prefix: str
|
|
|
+) -> List[str]:
|
|
|
+ like = f"{prefix}%"
|
|
|
+ rows = (
|
|
|
+ db.query(BizQcMaterialStockBatch.stock_in_no)
|
|
|
+ .filter(
|
|
|
+ BizQcMaterialStockBatch.tenant_no == tenant_no,
|
|
|
+ BizQcMaterialStockBatch.del_flag == 0,
|
|
|
+ BizQcMaterialStockBatch.stock_in_no.like(like),
|
|
|
+ )
|
|
|
+ .all()
|
|
|
+ )
|
|
|
+ return [r[0] for r in rows if r[0]]
|
|
|
+
|
|
|
+
|
|
|
+def allocate_stock_in_nos(
|
|
|
+ db: Session,
|
|
|
+ *,
|
|
|
+ tenant_no: str,
|
|
|
+ stock_in_date: date,
|
|
|
+ count: int,
|
|
|
+) -> List[str]:
|
|
|
+ prefix = prefix_for_date(stock_in_date)
|
|
|
+ existing = list_existing_stock_in_nos_for_prefix(db, tenant_no=tenant_no, prefix=prefix)
|
|
|
+ return next_stock_in_nos(stock_in_date, count=count, existing_nos=existing)
|