shanxia2.0/backend/app/utils/bx_codegen.py

"""报销单号/标题生成（从 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