app/services/essential_code.py

@@ -0,0 +1,123 @@
+"""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))