Benjamin Admin
78aeedafae
Step 2 of the convergence sequence. The Navigator sits over the CanonicalProductRegulatoryProfile (prefilled from company-profile / ProductWizard) and reports ONLY which facts are still missing + prioritized questions to collect them. It decides which facts are needed, NEVER what applies — that stays with the Scope Engine (step 3). No regulation logic, no UI, no Go, no RAG. - NavigatorQuestion (interaction type, NOT a compliance-meta-model class — freeze v1.0 untouched): question_id, target_field, label, why_needed, regulatory_domains_unblocked (static metadata), answer_type, options, priority. - QUESTION_CATALOG: 12 questions over canonical gaps — P0 (markets, role, lifecycle, machine/component), P1 (radio, usage-data, security-function, environmental wastewater/air/chemicals triggers), P2 (structured BOM). - engine: navigate() -> missing_facts + suggested_questions (priority-sorted) + completeness_summary (ready_for_scope = no P0 missing); apply_answers() -> updated profile. Pure field-presence; no scope import. - 8 tests: <=10 questions for a filled company-profile, known facts not re-asked, environmental = trigger questions only (no law evaluation), apply round-trip, P0 ordering, ready_for_scope. 41 tests green, mypy clean, LOC ok. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
117 lines
3.9 KiB
Python
117 lines
3.9 KiB
Python
"""Product Regulatory Navigator engine — missing-facts only.
|
|
|
|
`navigate(profile)` reports which canonical fields are still unknown and the
|
|
prioritized questions to fill them. `apply_answers(profile, answers)` returns the
|
|
updated profile. It NEVER decides what applies — that is the Scope Engine (step 3).
|
|
Pure field-presence checking; no scope-engine import, no regulation evaluation.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, List, Type
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
from compliance.profile.canonical import (
|
|
CanonicalLifecyclePhase,
|
|
CanonicalProductRegulatoryProfile,
|
|
EconomicOperatorRole,
|
|
ProductComponent,
|
|
)
|
|
|
|
from .questions import QUESTION_CATALOG, NavigatorQuestion, QuestionPriority
|
|
|
|
_ENUM_FIELDS: Dict[str, Type[Any]] = {
|
|
"economic_operator_role": EconomicOperatorRole,
|
|
"lifecycle_phase": CanonicalLifecyclePhase,
|
|
}
|
|
|
|
|
|
class CompletenessSummary(BaseModel):
|
|
total_relevant: int
|
|
answered: int
|
|
missing: int
|
|
missing_by_priority: Dict[str, int] = Field(default_factory=dict)
|
|
ready_for_scope: bool # True once no P0 fact is missing
|
|
note: str = ""
|
|
|
|
|
|
class NavigatorResult(BaseModel):
|
|
missing_facts: List[str] = Field(default_factory=list) # canonical target fields
|
|
suggested_questions: List[NavigatorQuestion] = Field(default_factory=list)
|
|
completeness_summary: CompletenessSummary
|
|
|
|
|
|
def _value(profile: CanonicalProductRegulatoryProfile, dotted: str) -> Any:
|
|
if "." in dotted:
|
|
head, tail = dotted.split(".", 1)
|
|
return getattr(getattr(profile, head), tail, None)
|
|
return getattr(profile, dotted, None)
|
|
|
|
|
|
def _is_unknown(profile: CanonicalProductRegulatoryProfile, q: NavigatorQuestion) -> bool:
|
|
value = _value(profile, q.target_field)
|
|
if value is None:
|
|
return True
|
|
if isinstance(value, list) and not value:
|
|
return True
|
|
return False
|
|
|
|
|
|
def navigate(profile: CanonicalProductRegulatoryProfile) -> NavigatorResult:
|
|
missing = [q for q in QUESTION_CATALOG if _is_unknown(profile, q)]
|
|
missing.sort(key=lambda q: q.order())
|
|
|
|
by_priority: Dict[str, int] = {}
|
|
for q in missing:
|
|
by_priority[q.priority.value] = by_priority.get(q.priority.value, 0) + 1
|
|
ready = QuestionPriority.P0.value not in by_priority
|
|
|
|
total = len(QUESTION_CATALOG)
|
|
summary = CompletenessSummary(
|
|
total_relevant=total,
|
|
answered=total - len(missing),
|
|
missing=len(missing),
|
|
missing_by_priority=by_priority,
|
|
ready_for_scope=ready,
|
|
note=(
|
|
"%d von %d Fakten vorhanden; %d offen. Scope-Engine startklar: %s."
|
|
% (total - len(missing), total, len(missing), "ja" if ready else "nein (P0 fehlt)")
|
|
),
|
|
)
|
|
return NavigatorResult(
|
|
missing_facts=[q.target_field for q in missing],
|
|
suggested_questions=missing,
|
|
completeness_summary=summary,
|
|
)
|
|
|
|
|
|
def _coerce(q: NavigatorQuestion, value: Any) -> Any:
|
|
if q.target_field in _ENUM_FIELDS:
|
|
return _ENUM_FIELDS[q.target_field](value)
|
|
if q.target_field == "components":
|
|
return [c if isinstance(c, ProductComponent) else ProductComponent(**c) for c in (value or [])]
|
|
if q.answer_type.value in {"country_list", "multiselect"}:
|
|
return list(value or [])
|
|
if q.answer_type.value == "bool":
|
|
return bool(value)
|
|
return value
|
|
|
|
|
|
def apply_answers(
|
|
profile: CanonicalProductRegulatoryProfile, answers: Dict[str, Any]
|
|
) -> CanonicalProductRegulatoryProfile:
|
|
updated = profile.model_copy(deep=True)
|
|
by_id = {q.question_id: q for q in QUESTION_CATALOG}
|
|
for question_id, raw in answers.items():
|
|
q = by_id.get(question_id)
|
|
if q is None or raw is None:
|
|
continue
|
|
value = _coerce(q, raw)
|
|
if "." in q.target_field:
|
|
head, tail = q.target_field.split(".", 1)
|
|
setattr(getattr(updated, head), tail, value)
|
|
else:
|
|
setattr(updated, q.target_field, value)
|
|
return updated
|