Python으로 처리정지권 요청을 처리하고 데이터 처리 파이프라인에서 제한을 적용하는 시스템입니다.
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from enum import Enum
from typing import Optional, Callable
import uuid
class RestrictionGround(Enum):
"""GDPR 제18조 처리정지 사유"""
ACCURACY_CONTESTED = "accuracy_contested" # 제18(1)(a): 정확성 이의
UNLAWFUL_PROCESSING = "unlawful_processing" # 제18(1)(b): 불법 처리
LEGAL_CLAIMS = "legal_claims" # 제18(1)(c): 법적 청구 목적 보존 필요
OBJECTION_PENDING = "objection_pending" # 제18(1)(d): 이의권 검증 중
class RestrictionStatus(Enum):
"""처리정지 상태"""
REQUESTED = "requested"
ACTIVE = "active" # 처리정지 적용 중
LIFTED = "lifted" # 해제됨
REJECTED = "rejected"
class AllowedProcessing(Enum):
"""처리정지 중 허용되는 처리 (제18조 제2항)"""
STORAGE = "storage_only" # 저장
DATA_SUBJECT_CONSENT = "data_subject_consent" # 정보주체 동의
LEGAL_CLAIMS = "legal_claims_establishment" # 법적 청구
RIGHTS_PROTECTION = "other_persons_rights" # 타인 권리 보호
PUBLIC_INTEREST = "important_public_interest" # 중요 공익
@dataclass
class RestrictionRequest:
"""처리정지 요청"""
request_id: str = field(default_factory=lambda: f"RST-{uuid.uuid4().hex[:8].upper()}")
data_subject_id: str = ""
ground: RestrictionGround = RestrictionGround.ACCURACY_CONTESTED
affected_data_categories: list[str] = field(default_factory=list)
reason_description: str = ""
submitted_at: datetime = field(default_factory=datetime.now)
status: RestrictionStatus = RestrictionStatus.REQUESTED
restricted_at: Optional[datetime] = None
lifted_at: Optional[datetime] = None
lift_notification_sent: bool = False
@dataclass
class DataRecord:
"""데이터 레코드 (처리정지 플래그 포함)"""
record_id: str
data_subject_id: str
data_category: str
content: dict
is_restricted: bool = False
restriction_request_id: Optional[str] = None
restricted_since: Optional[datetime] = None
class ProcessingRestrictionHandler:
"""처리정지권 처리 시스템"""
# GDPR 응답 기한: 1개월 (연장 시 최대 3개월)
GDPR_RESPONSE_DEADLINE_DAYS = 30
# 개인정보보호법 응답 기한: 10일
PIPA_RESPONSE_DEADLINE_DAYS = 10
def __init__(self, jurisdiction: str = "GDPR"):
self.jurisdiction = jurisdiction
self.requests: dict[str, RestrictionRequest] = {}
self.data_store: dict[str, DataRecord] = {}
self.processing_hooks: list[Callable] = []
def submit_restriction_request(
self,
data_subject_id: str,
ground: RestrictionGround,
data_categories: list[str],
reason: str
) -> RestrictionRequest:
"""처리정지 요청 제출"""
request = RestrictionRequest(
data_subject_id=data_subject_id,
ground=ground,
affected_data_categories=data_categories,
reason_description=reason
)
self.requests[request.request_id] = request
# 정확성 이의나 이의권 행사 중인 경우 즉시 처리정지 적용
if ground in [RestrictionGround.ACCURACY_CONTESTED,
RestrictionGround.OBJECTION_PENDING]:
self._apply_restriction(request)
return request
def _apply_restriction(self, request: RestrictionRequest) -> int:
"""처리정지 적용"""
restricted_count = 0
for record_id, record in self.data_store.items():
if record.data_subject_id == request.data_subject_id:
if not request.affected_data_categories or \
record.data_category in request.affected_data_categories:
record.is_restricted = True
record.restriction_request_id = request.request_id
record.restricted_since = datetime.now()
restricted_count += 1
request.status = RestrictionStatus.ACTIVE
request.restricted_at = datetime.now()
return restricted_count
def process_data(
self,
record_id: str,
processing_purpose: AllowedProcessing,
has_consent: bool = False
) -> dict:
"""데이터 처리 시도 (처리정지 확인)"""
if record_id not in self.data_store:
return {"allowed": False, "reason": "Record not found"}
record = self.data_store[record_id]
if not record.is_restricted:
return {"allowed": True, "data": record.content}
# 처리정지 중인 경우 허용되는 처리만 가능 (제18조 제2항)
allowed_reasons = {
AllowedProcessing.STORAGE: True, # 저장은 항상 허용
AllowedProcessing.DATA_SUBJECT_CONSENT: has_consent,
AllowedProcessing.LEGAL_CLAIMS: True, # 법적 청구는 별도 검증 필요
AllowedProcessing.RIGHTS_PROTECTION: True, # 별도 검증 필요
AllowedProcessing.PUBLIC_INTEREST: True # 별도 검증 필요
}
if allowed_reasons.get(processing_purpose, False):
return {
"allowed": True,
"restricted": True,
"processing_basis": processing_purpose.value,
"data": record.content,
"warning": "Data is under restriction - only limited processing allowed"
}
return {
"allowed": False,
"restricted": True,
"reason": f"Processing restricted since {record.restricted_since}",
"restriction_id": record.restriction_request_id,
"allowed_processing": [p.value for p in AllowedProcessing]
}
def lift_restriction(
self,
request_id: str,
lift_reason: str
) -> dict:
"""처리정지 해제 (사전 통지 필수 - 제18조 제3항)"""
if request_id not in self.requests:
return {"success": False, "error": "Request not found"}
request = self.requests[request_id]
if request.status != RestrictionStatus.ACTIVE:
return {"success": False, "error": "Restriction not active"}
# 정보주체에게 해제 전 통지 (필수)
if not request.lift_notification_sent:
notification = self._send_lift_notification(request)
request.lift_notification_sent = True
return {
"success": False,
"pending": True,
"reason": "Data subject must be notified before lifting",
"notification_sent": notification
}
# 처리정지 해제
lifted_count = 0
for record in self.data_store.values():
if record.restriction_request_id == request_id:
record.is_restricted = False
record.restriction_request_id = None
record.restricted_since = None
lifted_count += 1
request.status = RestrictionStatus.LIFTED
request.lifted_at = datetime.now()
return {
"success": True,
"request_id": request_id,
"lifted_records": lifted_count,
"lift_reason": lift_reason,
"lifted_at": request.lifted_at.isoformat()
}
def _send_lift_notification(self, request: RestrictionRequest) -> dict:
"""처리정지 해제 전 정보주체 통지"""
return {
"notification_type": "restriction_lift_notice",
"data_subject_id": request.data_subject_id,
"message": "처리정지가 해제될 예정입니다. 이의가 있으시면 연락주세요.",
"sent_at": datetime.now().isoformat(),
"legal_basis": "GDPR Article 18(3)"
}
def get_response_deadline(self, request_id: str) -> dict:
"""응답 기한 계산"""
request = self.requests.get(request_id)
if not request:
return {"error": "Request not found"}
if self.jurisdiction == "GDPR":
deadline_days = self.GDPR_RESPONSE_DEADLINE_DAYS
else: # PIPA
deadline_days = self.PIPA_RESPONSE_DEADLINE_DAYS
deadline = request.submitted_at + timedelta(days=deadline_days)
remaining = (deadline - datetime.now()).days
return {
"request_id": request_id,
"jurisdiction": self.jurisdiction,
"submitted_at": request.submitted_at.isoformat(),
"deadline": deadline.isoformat(),
"remaining_days": max(0, remaining),
"overdue": remaining < 0
}
def filter_for_ai_training(self, dataset: list[str]) -> list[str]:
"""AI 학습 데이터셋에서 처리정지 데이터 제외"""
return [
record_id for record_id in dataset
if record_id in self.data_store
and not self.data_store[record_id].is_restricted
]
# 사용 예시
if __name__ == "__main__":
handler = ProcessingRestrictionHandler(jurisdiction="GDPR")
# 테스트 데이터 추가
handler.data_store["REC001"] = DataRecord(
record_id="REC001",
data_subject_id="USER123",
data_category="credit_history",
content={"score": 720, "history": "good"}
)
# 정확성 이의로 처리정지 요청
request = handler.submit_restriction_request(
data_subject_id="USER123",
ground=RestrictionGround.ACCURACY_CONTESTED,
data_categories=["credit_history"],
reason="신용 점수가 실제와 다름"
)
print(f"처리정지 요청: {request.request_id}, 상태: {request.status.value}")
# 데이터 처리 시도 (일반 목적)
result = handler.process_data("REC001", AllowedProcessing.STORAGE)
print(f"저장 처리: {result['allowed']}")
# AI 학습용 필터링
filtered = handler.filter_for_ai_training(["REC001", "REC002"])
print(f"AI 학습 가능 데이터: {filtered}")