"""EssentialCode generation service. Encoding rule: {proposal_code}:E{seq:05x} Where: - ``proposal_code`` is the parent Proposal's code (e.g. ``PROJ01:P00001``) - ``E`` is the fixed Essential prefix - ``seq`` is a 5-digit zero-padded hex sequence scoped per Proposal Sequence assignment: Uses the max existing ``essential_code`` suffix under the same Proposal to derive the next value. No separate counter table is needed because Essentials are always scoped to a single Proposal and created one at a time (or in a small batch during Proposal Accept). Examples: PROJ01:P00001:E00001 PROJ01:P00001:E00002 HRBFRG:P00003:E0000a See: NEXT_WAVE_DEV_DIRECTION.md §8.5 / §8.6 """ from __future__ import annotations import re from typing import TYPE_CHECKING from sqlalchemy import func as sa_func from app.models.essential import Essential if TYPE_CHECKING: from sqlalchemy.orm import Session from app.models.proposal import Proposal # Matches the trailing hex portion after ":E" _SUFFIX_RE = re.compile(r":E([0-9a-fA-F]+)$") # Fixed prefix letter for Essential codes ESSENTIAL_PREFIX = "E" # Width of the hex sequence portion SEQ_WIDTH = 5 def _extract_seq(essential_code: str) -> int: """Extract the numeric sequence from an EssentialCode string. Returns 0 if the code doesn't match the expected pattern. """ m = _SUFFIX_RE.search(essential_code) if m: return int(m.group(1), 16) return 0 def _max_seq_for_proposal(db: "Session", proposal_id: int) -> int: """Return the highest existing sequence number for a given Proposal. Returns 0 if no Essentials exist yet. """ essentials = ( db.query(Essential.essential_code) .filter(Essential.proposal_id == proposal_id) .all() ) if not essentials: return 0 return max(_extract_seq(row[0]) for row in essentials) def generate_essential_code( db: "Session", proposal: "Proposal", *, batch_offset: int = 0, ) -> str: """Generate the next EssentialCode for *proposal*. Parameters ---------- db: Active SQLAlchemy session (must be inside a transaction so the caller can flush/commit to avoid race conditions). proposal: The parent Proposal ORM instance. Its ``proposal_code`` (hybrid property over ``propose_code``) is used as the prefix. batch_offset: When creating multiple Essentials in a single transaction (e.g. during Proposal Accept), pass an incrementing offset (0, 1, 2, …) so each call returns a unique code without needing intermediate flushes. Returns ------- str A unique EssentialCode such as ``PROJ01:P00001:E00001``. Raises ------ ValueError If the parent Proposal has no code assigned. """ proposal_code = proposal.proposal_code if not proposal_code: raise ValueError( f"Proposal id={proposal.id} has no proposal_code; " "cannot generate EssentialCode" ) current_max = _max_seq_for_proposal(db, proposal.id) next_seq = current_max + 1 + batch_offset suffix = format(next_seq, "x").upper().zfill(SEQ_WIDTH) return f"{proposal_code}:{ESSENTIAL_PREFIX}{suffix}" def validate_essential_code(code: str) -> bool: """Check whether *code* conforms to the EssentialCode format. Expected format: ``{any}:E{hex_digits}`` """ return bool(_SUFFIX_RE.search(code))