141 lines
4.3 KiB
Python
141 lines
4.3 KiB
Python
"""계정별 학습자 AI 피드백 노출 정책의 단일 경계."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Protocol
|
|
|
|
from fastapi import HTTPException, status
|
|
|
|
|
|
LEARNER_FEEDBACK_DISABLED_DETAIL = "learner_feedback_disabled"
|
|
|
|
|
|
class FeedbackPolicySession(Protocol):
|
|
learner_feedback_enabled: bool
|
|
|
|
|
|
class FeedbackPolicyPrincipal(Protocol):
|
|
role: object
|
|
learner_feedback_enabled: bool
|
|
|
|
|
|
def _role_value(role: object) -> str:
|
|
return str(getattr(role, "value", role))
|
|
|
|
|
|
def learner_feedback_enabled(session: FeedbackPolicySession) -> bool:
|
|
"""구 회기 객체는 호환성을 위해 피드백 허용으로 취급한다."""
|
|
|
|
return bool(getattr(session, "learner_feedback_enabled", True))
|
|
|
|
|
|
def effective_learner_feedback_enabled(
|
|
session: FeedbackPolicySession,
|
|
principal: FeedbackPolicyPrincipal,
|
|
) -> bool:
|
|
"""학습자는 현재 계정과 회기 스냅샷이 모두 켜져야 피드백이 활성이다."""
|
|
|
|
snapshot_enabled = learner_feedback_enabled(session)
|
|
if _role_value(principal.role) != "learner":
|
|
return snapshot_enabled
|
|
return snapshot_enabled and bool(
|
|
getattr(principal, "learner_feedback_enabled", True)
|
|
)
|
|
|
|
|
|
def can_expose_principal_learner_feedback(
|
|
session: FeedbackPolicySession,
|
|
principal: FeedbackPolicyPrincipal,
|
|
) -> bool:
|
|
"""교수자·관리자는 유지하고 학습자에게 effective AND 정책을 적용한다."""
|
|
|
|
return _role_value(principal.role) != "learner" or (
|
|
effective_learner_feedback_enabled(session, principal)
|
|
)
|
|
|
|
|
|
def can_expose_learner_feedback(
|
|
session: FeedbackPolicySession,
|
|
*,
|
|
viewer_role: str,
|
|
) -> bool:
|
|
"""관리자·교수자 검토는 유지하고 학습자에게만 스냅샷 정책을 적용한다."""
|
|
|
|
return _role_value(viewer_role) != "learner" or learner_feedback_enabled(session)
|
|
|
|
|
|
def require_learner_feedback(
|
|
session: FeedbackPolicySession,
|
|
*,
|
|
viewer_role: object,
|
|
) -> None:
|
|
"""학습자에게만 회기 스냅샷 정책을 적용하고 파생 출력을 fail-closed 한다."""
|
|
|
|
if can_expose_learner_feedback(
|
|
session,
|
|
viewer_role=_role_value(viewer_role),
|
|
):
|
|
return
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=LEARNER_FEEDBACK_DISABLED_DETAIL,
|
|
)
|
|
|
|
|
|
def require_principal_learner_feedback(principal: FeedbackPolicyPrincipal) -> None:
|
|
"""회기 ID가 없는 학습자 전용 파생 원장은 계정 정책으로 차단한다."""
|
|
|
|
if _role_value(principal.role) != "learner" or bool(
|
|
getattr(principal, "learner_feedback_enabled", True)
|
|
):
|
|
return
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=LEARNER_FEEDBACK_DISABLED_DETAIL,
|
|
)
|
|
|
|
|
|
async def can_expose_session_learner_feedback(
|
|
session_id: object,
|
|
principal: FeedbackPolicyPrincipal,
|
|
) -> bool:
|
|
"""현재 계정과 생성 당시 회기 스냅샷을 모두 확인한다.
|
|
|
|
교수자·관리자 감독 화면은 그대로 유지한다. 학습자는 관리자가 현재 피드백을
|
|
끈 경우 즉시 차단하고, 다시 켠 뒤에도 비활성 상태로 생성된 회기에서는 파생
|
|
출력을 되살리지 않는다.
|
|
"""
|
|
|
|
if _role_value(principal.role) != "learner":
|
|
return True
|
|
if not bool(getattr(principal, "learner_feedback_enabled", True)):
|
|
return False
|
|
|
|
# 순환 import를 피하면서 세션 조회 경계와 동일한 RLS/런타임 폴백을 사용한다.
|
|
from .. import session_persistence
|
|
from ..store import store
|
|
|
|
session_key = str(session_id)
|
|
session = await session_persistence.load_session(
|
|
session_key,
|
|
principal, # type: ignore[arg-type]
|
|
allow_ended=True,
|
|
)
|
|
if session is None:
|
|
session = store.get(session_key)
|
|
if session is None:
|
|
# 소유권·존재 오류는 각 도메인 저장소가 원래 계약대로 처리한다.
|
|
return True
|
|
return can_expose_principal_learner_feedback(session, principal)
|
|
|
|
|
|
async def require_session_learner_feedback(
|
|
session_id: object,
|
|
principal: FeedbackPolicyPrincipal,
|
|
) -> None:
|
|
if await can_expose_session_learner_feedback(session_id, principal):
|
|
return
|
|
raise HTTPException(
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
|
detail=LEARNER_FEEDBACK_DISABLED_DETAIL,
|
|
)
|