137 lines
4.6 KiB
Python
137 lines
4.6 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（报销）。
|
|
"""
|
|
today = datetime.now().strftime("%Y%m%d")
|
|
like_pattern = f"{prefix}{today}%"
|
|
|
|
result = db.execute(
|
|
text(
|
|
"SELECT MAX(CAST(SUBSTRING(apply_code, -3) AS UNSIGNED)) FROM ("
|
|
"SELECT apply_code FROM bx_advance_apply "
|
|
"WHERE tenant_no = :tenant_no AND apply_code LIKE :p1 AND del_flag = 0 "
|
|
"UNION ALL "
|
|
"SELECT apply_code FROM bx_reimburse_apply "
|
|
"WHERE tenant_no = :tenant_no AND apply_code LIKE :p2 AND del_flag = 0"
|
|
") t"
|
|
),
|
|
{"tenant_no": tenant_no, "p1": like_pattern, "p2": like_pattern},
|
|
).scalar()
|
|
|
|
seq = (result or 0) + 1
|
|
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 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 preview_reimburse_title_seq(
|
|
db: Session,
|
|
tenant_no: str,
|
|
reimburse_date: date,
|
|
exclude_apply_id: str | None = None,
|
|
) -> int:
|
|
"""预览序号（不加行锁，仅用于弹窗展示）。"""
|
|
query = db.query(func.max(BxReimburseApply.title_seq)).filter(
|
|
BxReimburseApply.tenant_no == tenant_no,
|
|
BxReimburseApply.reimburse_date == reimburse_date,
|
|
BxReimburseApply.del_flag == 0,
|
|
BxReimburseApply.title_seq.isnot(None),
|
|
)
|
|
if exclude_apply_id:
|
|
query = query.filter(BxReimburseApply.id != exclude_apply_id)
|
|
max_seq = query.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 唯一约束，并发冲突时由上层重试。
|
|
"""
|
|
query = db.query(func.max(BxReimburseApply.title_seq)).filter(
|
|
BxReimburseApply.tenant_no == tenant_no,
|
|
BxReimburseApply.reimburse_date == reimburse_date,
|
|
BxReimburseApply.del_flag == 0,
|
|
BxReimburseApply.title_seq.isnot(None),
|
|
)
|
|
if exclude_apply_id:
|
|
query = query.filter(BxReimburseApply.id != exclude_apply_id)
|
|
max_seq = query.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
|