Spaces:
Running
Running
File size: 3,707 Bytes
3eae4cc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | """
state_machine.py — Gov Workflow OpenEnv
Deterministic workflow transition engine aligned with Phase 1 schemas.
"""
from __future__ import annotations
from app.models import ApplicationCase, InternalSubstate, StageType
INTERNAL_TO_PUBLIC_STAGE: dict[InternalSubstate, StageType] = {
InternalSubstate.PRE_SCRUTINY: StageType.SUBMISSION,
InternalSubstate.DOC_VALIDATION: StageType.DOCUMENT_VERIFICATION,
InternalSubstate.SERVICE_SPECIFIC_VALIDATION: StageType.DOCUMENT_VERIFICATION,
InternalSubstate.FIELD_VERIFICATION_PENDING: StageType.FIELD_VERIFICATION,
InternalSubstate.DECISION_PENDING: StageType.APPROVAL,
InternalSubstate.ISSUANCE_READY: StageType.ISSUANCE,
InternalSubstate.BLOCKED_MISSING_DOCS: StageType.DOCUMENT_VERIFICATION,
InternalSubstate.COMPLETED: StageType.ISSUANCE,
InternalSubstate.REJECTED: StageType.APPROVAL,
}
def build_public_stage(substate: InternalSubstate) -> StageType:
return INTERNAL_TO_PUBLIC_STAGE.get(substate, StageType.SUBMISSION)
def transition_case(case: ApplicationCase, new_substate: InternalSubstate) -> None:
case.internal_substate = new_substate
case.public_stage = build_public_stage(new_substate)
case.days_in_current_stage = 0
def can_advance(case: ApplicationCase) -> bool:
if case.completed or case.rejected:
return False
if case.internal_substate == InternalSubstate.BLOCKED_MISSING_DOCS:
return False
return True
def advance_case(case: ApplicationCase, rng: object = None) -> tuple[bool, bool]:
"""
Returns (progressed, completed).
"""
if not can_advance(case):
return False, False
early_stages = {
InternalSubstate.PRE_SCRUTINY,
InternalSubstate.DOC_VALIDATION,
}
if case.has_missing_docs and case.internal_substate in early_stages:
transition_case(case, InternalSubstate.BLOCKED_MISSING_DOCS)
return True, False
current = case.internal_substate
if current == InternalSubstate.PRE_SCRUTINY:
transition_case(case, InternalSubstate.DOC_VALIDATION)
return True, False
if current == InternalSubstate.DOC_VALIDATION:
if case.field_verification_required:
transition_case(case, InternalSubstate.FIELD_VERIFICATION_PENDING)
else:
transition_case(case, InternalSubstate.DECISION_PENDING)
return True, False
if current == InternalSubstate.SERVICE_SPECIFIC_VALIDATION:
if case.field_verification_required:
transition_case(case, InternalSubstate.FIELD_VERIFICATION_PENDING)
else:
transition_case(case, InternalSubstate.DECISION_PENDING)
return True, False
if current == InternalSubstate.FIELD_VERIFICATION_PENDING:
return False, False
if current == InternalSubstate.DECISION_PENDING:
transition_case(case, InternalSubstate.ISSUANCE_READY)
return True, False
if current == InternalSubstate.ISSUANCE_READY:
transition_case(case, InternalSubstate.COMPLETED)
case.completed = True
return True, True
return False, False
def unblock_missing_docs(case: ApplicationCase) -> bool:
if case.internal_substate != InternalSubstate.BLOCKED_MISSING_DOCS:
return False
case.has_missing_docs = False
case.doc_resolution_day = None
transition_case(case, InternalSubstate.DOC_VALIDATION)
return True
def complete_field_verification(case: ApplicationCase) -> bool:
if case.internal_substate != InternalSubstate.FIELD_VERIFICATION_PENDING:
return False
case.field_verification_completion_day = None
transition_case(case, InternalSubstate.DECISION_PENDING)
return True |