from __future__ import annotations import re import secrets from datetime import UTC, datetime from typing import Callable, Literal from sqlalchemy import select from sqlalchemy.orm import Session from app.models.financial_record import ExpenseClaim DocumentNumberKind = Literal["application", "reimbursement", "audit"] DOCUMENT_NUMBER_PREFIXES: dict[DocumentNumberKind, str] = { "application": "AP", "reimbursement": "RE", "audit": "AD", } DOCUMENT_NUMBER_TOKEN_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" DOCUMENT_NUMBER_TOKEN_LENGTH = 8 DOCUMENT_NUMBER_PATTERN = re.compile( rf"^(?:AP|RE|AD)-\d{{14}}-[{DOCUMENT_NUMBER_TOKEN_ALPHABET}]{{{DOCUMENT_NUMBER_TOKEN_LENGTH}}}$", flags=re.IGNORECASE, ) DOCUMENT_NUMBER_EXTRACT_PATTERN = re.compile( rf"(?:AP|RE|AD)-\d{{14}}-[{DOCUMENT_NUMBER_TOKEN_ALPHABET}]{{{DOCUMENT_NUMBER_TOKEN_LENGTH}}}" r"|APP-\d{8}-[A-Z0-9]{6}" r"|EXP-\d{6}-\d{3}", flags=re.IGNORECASE, ) def generate_document_token() -> str: return "".join( secrets.choice(DOCUMENT_NUMBER_TOKEN_ALPHABET) for _ in range(DOCUMENT_NUMBER_TOKEN_LENGTH) ) def build_document_number( kind: DocumentNumberKind, *, timestamp: datetime | None = None, token: str | None = None, ) -> str: prefix = DOCUMENT_NUMBER_PREFIXES[kind] generated_at = timestamp or datetime.now(UTC) if generated_at.tzinfo is None: generated_at = generated_at.replace(tzinfo=UTC) normalized_token = (token or generate_document_token()).strip().upper() if not re.fullmatch( rf"[{DOCUMENT_NUMBER_TOKEN_ALPHABET}]{{{DOCUMENT_NUMBER_TOKEN_LENGTH}}}", normalized_token, ): raise ValueError("document number token must be 8 chars from the configured alphabet") return f"{prefix}-{generated_at.astimezone(UTC):%Y%m%d%H%M%S}-{normalized_token}" def generate_unique_expense_claim_no( db: Session, kind: DocumentNumberKind, *, timestamp: datetime | None = None, token_factory: Callable[[], str] = generate_document_token, max_attempts: int = 8, ) -> str: for _ in range(max_attempts): candidate = build_document_number( kind, timestamp=timestamp, token=token_factory(), ) exists = db.scalar( select(ExpenseClaim.id) .where(ExpenseClaim.claim_no == candidate) .limit(1) ) if exists is None: return candidate raise RuntimeError(f"failed to generate a unique {kind} document number") def is_application_claim_no(value: object) -> bool: normalized = str(value or "").strip().upper() return normalized.startswith(("AP-", "APP-"))