feat(cra): CRA Readiness Check lead-magnet on /sdk/cra (Track A)
Low-friction, stateless readiness check (no project/DB): business-scope answers (internet / parameter app / remote maintenance / updates / firmware / personal data / critical infra) -> Annex III/IV classification (reuses _classify) + a high-level guideline grouped Code / Prozess / Dokumentation (via Annex I evidence_type) + conformity path + deadlines + rough effort + the "we implement" hook and a CTA into the existing project workflow. Endpoint POST /api/v1/cra/ readiness. Reuse + reframe of the existing CRA module — no duplicate questionnaire. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -18,6 +18,8 @@ from compliance.services.cra_finding_mapper import assess_findings_payload
|
||||
from compliance.services.cra_snapshot_store import save_snapshot, list_snapshots, get_snapshot
|
||||
from compliance.services.cra_use_case_controls import enrich_findings_with_breadth
|
||||
from compliance.services.cra_component_findings import findings_from_components
|
||||
from compliance.api.cra_annex_i_data import ANNEX_I_REQUIREMENTS, MEASURES, DEADLINES
|
||||
from compliance.api.cra_routes import _classify # reuse the deterministic Annex III/IV classifier
|
||||
from database import SessionLocal
|
||||
from .tenant_utils import get_tenant_id
|
||||
|
||||
@@ -113,3 +115,64 @@ async def get_assess_snapshot(snapshot_id: str, tenant_id: str = Depends(get_ten
|
||||
if not snap:
|
||||
raise HTTPException(status_code=404, detail="Snapshot not found")
|
||||
return snap
|
||||
|
||||
|
||||
# --- Lead-magnet readiness check (stateless, no project, no DB) ---
|
||||
|
||||
class ReadinessRequest(BaseModel):
|
||||
intended_use: Optional[str] = ""
|
||||
connected_to_internet: Optional[bool] = False
|
||||
has_software_updates: Optional[bool] = False
|
||||
processes_personal_data: Optional[bool] = False
|
||||
is_critical_infra_supplier: Optional[bool] = False
|
||||
has_firmware: Optional[bool] = False
|
||||
remote_maintenance: Optional[bool] = False # implies connectivity + updates
|
||||
user_parameter_app: Optional[bool] = False # implies connectivity + updates
|
||||
|
||||
|
||||
# CRA Annex I evidence_type -> guideline bucket (Code / Prozess / Dokumentation).
|
||||
_GUIDELINE_BUCKET = {"code": "code", "hybrid": "code", "process": "process", "document": "document"}
|
||||
_PATH_HINT = {
|
||||
"CRITICAL": "Konformitaet ueber benannte Stelle / EUCC (Modul H/C)",
|
||||
"IMPORTANT_II": "Modul B+C oder harmonisierte Norm",
|
||||
"IMPORTANT_I": "Self-Assessment bei harmonisierten Normen, sonst Modul B",
|
||||
"STANDARD": "Self-Assessment (Modul A)",
|
||||
"NOT_IN_SCOPE": "—",
|
||||
}
|
||||
|
||||
|
||||
@router.post("/readiness")
|
||||
async def readiness(body: ReadinessRequest):
|
||||
"""Low-friction CRA readiness check: business-scope answers -> Annex III/IV
|
||||
classification + a high-level guideline grouped Code / Prozess / Dokumentation.
|
||||
Reuses the deterministic classifier + Annex I spine. No project, no DB."""
|
||||
intake = {
|
||||
"intended_use": body.intended_use,
|
||||
"connected_to_internet": bool(body.connected_to_internet or body.remote_maintenance or body.user_parameter_app),
|
||||
"has_software_updates": bool(body.has_software_updates or body.remote_maintenance or body.user_parameter_app),
|
||||
"processes_personal_data": bool(body.processes_personal_data),
|
||||
"is_critical_infra_supplier": bool(body.is_critical_infra_supplier),
|
||||
}
|
||||
classification, rationale = _classify(intake)
|
||||
in_scope = classification != "NOT_IN_SCOPE"
|
||||
groups = {"code": [], "process": [], "document": []}
|
||||
if in_scope:
|
||||
for req in ANNEX_I_REQUIREMENTS:
|
||||
bucket = _GUIDELINE_BUCKET.get(req.get("evidence_type", "process"), "process")
|
||||
groups[bucket].append({
|
||||
"req_id": req["req_id"], "title": req["title"], "category": req["category"],
|
||||
"annex_anchor": req["annex_anchor"], "severity": req["severity"],
|
||||
"effort_days": req.get("effort_days"),
|
||||
"measures": [{"id": m, "name": MEASURES.get(m, m)} for m in req.get("mapped_measures", [])],
|
||||
})
|
||||
total_effort = sum(r["effort_days"] for g in groups.values() for r in g if r.get("effort_days"))
|
||||
return {
|
||||
"in_scope": in_scope,
|
||||
"classification": classification,
|
||||
"rationale": rationale,
|
||||
"conformity_path_hint": _PATH_HINT.get(classification, ""),
|
||||
"guideline": groups,
|
||||
"counts": {k: len(v) for k, v in groups.items()},
|
||||
"total_effort_days": total_effort,
|
||||
"deadlines": list(DEADLINES),
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user