183 lines
6.0 KiB
Python
183 lines
6.0 KiB
Python
"""报销单号/标题生成（从 bx-backend 迁移，增加 tenant_no 隔离）。"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from datetime import date, datetime
|
|
|
|
from sqlalchemy import func, text
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.bx_models import BxReimburseApply
|
|
|
|
_AUTO_TITLE_SUFFIX_RE = re.compile(r"报销单\d{11}$")
|
|
|
|
|
|
def generate_apply_code(db: Session, prefix: str, tenant_no: str) -> str:
|
|
"""
|
|
生成顺序单号：{prefix}{YYYYMMDD}{3位序号}。
|
|
prefix 示例：BYJ（备用金）、BX（报销）。
|
|
|
|
说明：
|
|
- uk_apply_code 为全局唯一（不含 tenant / del_flag），故取号时按全库、含软删统计，
|
|
避免复用仍被唯一索引占用的号段。
|
|
- tenant_no 保留以兼容调用方；序号本身不按租户隔离。
|
|
- 并发下仍可能撞号，上层创建须在 IntegrityError 时重新取号重试。
|
|
"""
|
|
today = datetime.now().strftime("%Y%m%d")
|
|
like_pattern = f"{prefix}{today}%"
|
|
|
|
# 先锁住当日相关行，降低并发下同时读到同一 MAX 的概率
|
|
db.execute(
|
|
text(
|
|
"SELECT apply_code FROM bx_advance_apply "
|
|
"WHERE apply_code LIKE :p FOR UPDATE"
|
|
),
|
|
{"p": like_pattern},
|
|
)
|
|
db.execute(
|
|
text(
|
|
"SELECT apply_code FROM bx_reimburse_apply "
|
|
"WHERE apply_code LIKE :p FOR UPDATE"
|
|
),
|
|
{"p": like_pattern},
|
|
)
|
|
|
|
result = db.execute(
|
|
text(
|
|
"SELECT MAX(CAST(SUBSTRING(apply_code, -3) AS UNSIGNED)) FROM ("
|
|
"SELECT apply_code FROM bx_advance_apply WHERE apply_code LIKE :p1 "
|
|
"UNION ALL "
|
|
"SELECT apply_code FROM bx_reimburse_apply WHERE apply_code LIKE :p2"
|
|
") t"
|
|
),
|
|
{"p1": like_pattern, "p2": like_pattern},
|
|
).scalar()
|
|
|
|
seq = int(result or 0) + 1
|
|
if seq > 999:
|
|
raise ValueError(f"当日单号序号已用尽: {prefix}{today}")
|
|
return f"{prefix}{today}{seq:03d}"
|
|
|
|
|
|
def parse_reimburse_date(value: str | date | datetime | None) -> date:
|
|
if value is None:
|
|
return datetime.now().date()
|
|
if isinstance(value, datetime):
|
|
return value.date()
|
|
if isinstance(value, date):
|
|
return value
|
|
text_value = str(value).strip()
|
|
if not text_value:
|
|
return datetime.now().date()
|
|
return datetime.strptime(text_value, "%Y-%m-%d").date()
|
|
|
|
|
|
def parse_optional_date(value: str | date | datetime | None) -> date | None:
|
|
"""可选日期：空 → None；合法 YYYY-MM-DD / date / datetime → date。"""
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, datetime):
|
|
return value.date()
|
|
if isinstance(value, date):
|
|
return value
|
|
text_value = str(value).strip()
|
|
if not text_value:
|
|
return None
|
|
return datetime.strptime(text_value, "%Y-%m-%d").date()
|
|
|
|
|
|
def format_reimburse_title(applicant_name: str, reimburse_date: date, seq: int) -> str:
|
|
return f"{applicant_name}报销单{reimburse_date.strftime('%Y%m%d')}{seq:03d}"
|
|
|
|
|
|
def is_auto_reimburse_title(title: str) -> bool:
|
|
return bool(_AUTO_TITLE_SUFFIX_RE.search(str(title or "").strip()))
|
|
|
|
|
|
def _title_seq_max_query(
|
|
db: Session,
|
|
tenant_no: str,
|
|
reimburse_date: date,
|
|
exclude_apply_id: str | None = None,
|
|
):
|
|
"""
|
|
同租户、同报销日期下的最大标题序号。
|
|
|
|
uk_tenant_reimburse_date_title_seq 不含 del_flag，软删行仍占用序号。
|
|
取号若只统计未删除行，会再次分到已被占用的号（例如 1）并触发 1062。
|
|
"""
|
|
query = db.query(func.max(BxReimburseApply.title_seq)).filter(
|
|
BxReimburseApply.tenant_no == tenant_no,
|
|
BxReimburseApply.reimburse_date == reimburse_date,
|
|
BxReimburseApply.title_seq.isnot(None),
|
|
)
|
|
if exclude_apply_id:
|
|
query = query.filter(BxReimburseApply.id != exclude_apply_id)
|
|
return query
|
|
|
|
|
|
def preview_reimburse_title_seq(
|
|
db: Session,
|
|
tenant_no: str,
|
|
reimburse_date: date,
|
|
exclude_apply_id: str | None = None,
|
|
) -> int:
|
|
"""预览序号（不加行锁，仅用于弹窗展示）。含软删行，与正式取号一致。"""
|
|
max_seq = _title_seq_max_query(
|
|
db, tenant_no, reimburse_date, exclude_apply_id
|
|
).scalar()
|
|
return int(max_seq or 0) + 1
|
|
|
|
|
|
def allocate_reimburse_title_seq(
|
|
db: Session,
|
|
tenant_no: str,
|
|
reimburse_date: date,
|
|
exclude_apply_id: str | None = None,
|
|
) -> int:
|
|
"""
|
|
分配同租户、同报销日期下的标题序号（MAX+1，含软删）。
|
|
配合 uk_tenant_reimburse_date_title_seq 唯一约束，并发冲突时由上层重试。
|
|
"""
|
|
max_seq = _title_seq_max_query(
|
|
db, tenant_no, reimburse_date, exclude_apply_id
|
|
).with_for_update().scalar()
|
|
return int(max_seq or 0) + 1
|
|
|
|
|
|
def generate_title(
|
|
db: Session,
|
|
applicant_name: str,
|
|
tenant_no: str,
|
|
reimburse_date: str | date | datetime | None = None,
|
|
exclude_apply_id: str | None = None,
|
|
) -> str:
|
|
"""预览用标题生成（提交时仍会在服务端重新分配序号以保证并发安全）。"""
|
|
rd = parse_reimburse_date(reimburse_date)
|
|
seq = preview_reimburse_title_seq(db, tenant_no, rd, exclude_apply_id)
|
|
return format_reimburse_title(applicant_name, rd, seq)
|
|
|
|
|
|
def resolve_reimburse_title(
|
|
db: Session,
|
|
applicant_name: str,
|
|
tenant_no: str,
|
|
reimburse_date: date,
|
|
requested_title: str | None,
|
|
exclude_apply_id: str | None = None,
|
|
) -> tuple[str, int | None]:
|
|
"""
|
|
确定最终标题与 title_seq。
|
|
- 未填标题或仍为自动标题格式时：按报销日期重新分配序号并生成标题。
|
|
- 自定义标题：title_seq 为 None（不参与唯一序号约束）。
|
|
"""
|
|
custom_title = str(requested_title or "").strip()
|
|
if custom_title and not is_auto_reimburse_title(custom_title):
|
|
return custom_title, None
|
|
|
|
seq = allocate_reimburse_title_seq(
|
|
db, tenant_no, reimburse_date, exclude_apply_id=exclude_apply_id
|
|
)
|
|
title = format_reimburse_title(applicant_name, reimburse_date, seq)
|
|
return title, seq
|