Python을 사용한 데이터 이동권 처리 시스템 구현 예제입니다.
"""
데이터 이동권(Data Portability) 처리 시스템
GDPR 제20조 및 개인정보보호법 제35조의2 준수
"""
import json
import csv
from datetime import datetime
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
from enum import Enum
import hashlib
import zipfile
import io
class DataFormat(Enum):
"""지원되는 데이터 포맷"""
JSON = "application/json"
CSV = "text/csv"
XML = "application/xml"
class TransferStatus(Enum):
"""전송 상태"""
PENDING = "pending"
PROCESSING = "processing"
COMPLETED = "completed"
FAILED = "failed"
EXPIRED = "expired"
@dataclass
class PortabilityRequest:
"""이동권 요청 데이터 구조"""
request_id: str
subject_id: str
requested_at: datetime
data_categories: List[str]
target_format: DataFormat
recipient_id: Optional[str] # 직접 전송 시 수신자 ID
status: TransferStatus
expiry_date: datetime
class DataPortabilityHandler:
"""
데이터 이동권 처리 핸들러
GDPR Article 20 준수 구현
"""
# 이동 가능한 데이터 카테고리 (직접 제공 + 관찰 데이터)
PORTABLE_CATEGORIES = {
"profile": "사용자 프로필 정보",
"activity": "활동 기록 및 거래 내역",
"preferences": "설정 및 선호도",
"content": "사용자 생성 콘텐츠",
"communications": "메시지 및 커뮤니케이션",
"sensor_data": "센서/디바이스 관찰 데이터"
}
# 제외 카테고리 (추론 데이터)
EXCLUDED_CATEGORIES = {
"credit_score": "AI 기반 신용 점수",
"recommendations": "추천 알고리즘 결과",
"risk_assessment": "위험 평가 결과",
"derived_insights": "파생 분석 데이터"
}
RESPONSE_DEADLINE_DAYS = 30 # GDPR 기준 응답 기한
def __init__(self, user_data_store, audit_logger):
self.user_data_store = user_data_store
self.audit_logger = audit_logger
self.pending_requests: Dict[str, PortabilityRequest] = {}
def create_portability_request(
self,
subject_id: str,
data_categories: List[str],
target_format: DataFormat = DataFormat.JSON,
recipient_id: Optional[str] = None
) -> PortabilityRequest:
"""
이동권 요청 생성
Args:
subject_id: 정보주체 식별자
data_categories: 요청 데이터 카테고리
target_format: 출력 포맷
recipient_id: 직접 전송 시 수신자 (GDPR Article 20(2))
"""
# 요청 ID 생성
request_id = hashlib.sha256(
f"{subject_id}:{datetime.now().isoformat()}".encode()
).hexdigest()[:16]
# 유효한 카테고리만 필터링
valid_categories = [
cat for cat in data_categories
if cat in self.PORTABLE_CATEGORIES
]
# 제외 카테고리 경고
excluded = [
cat for cat in data_categories
if cat in self.EXCLUDED_CATEGORIES
]
if excluded:
self.audit_logger.log_warning(
f"Excluded derived data categories: {excluded}",
subject_id=subject_id,
request_id=request_id
)
request = PortabilityRequest(
request_id=request_id,
subject_id=subject_id,
requested_at=datetime.now(),
data_categories=valid_categories,
target_format=target_format,
recipient_id=recipient_id,
status=TransferStatus.PENDING,
expiry_date=datetime.now() + timedelta(days=self.RESPONSE_DEADLINE_DAYS)
)
self.pending_requests[request_id] = request
# 감사 로그 기록
self.audit_logger.log_event(
event_type="PORTABILITY_REQUEST_CREATED",
subject_id=subject_id,
request_id=request_id,
details={
"categories": valid_categories,
"format": target_format.value,
"has_recipient": recipient_id is not None
}
)
return request
def process_request(self, request_id: str) -> bytes:
"""
이동권 요청 처리 및 데이터 패키징
Returns:
bytes: 구조화된 포터블 데이터 (ZIP 아카이브)
"""
request = self.pending_requests.get(request_id)
if not request:
raise ValueError(f"Request not found: {request_id}")
# 상태 업데이트
request.status = TransferStatus.PROCESSING
try:
# 데이터 수집
collected_data = self._collect_portable_data(
request.subject_id,
request.data_categories
)
# 포맷 변환
formatted_data = self._format_data(
collected_data,
request.target_format
)
# 메타데이터 추가
package = self._create_data_package(
formatted_data,
request
)
request.status = TransferStatus.COMPLETED
# 직접 전송 요청인 경우
if request.recipient_id:
self._transfer_to_recipient(
request.recipient_id,
package,
request
)
self.audit_logger.log_event(
event_type="PORTABILITY_REQUEST_COMPLETED",
subject_id=request.subject_id,
request_id=request_id,
details={"data_size_bytes": len(package)}
)
return package
except Exception as e:
request.status = TransferStatus.FAILED
self.audit_logger.log_error(
f"Portability request failed: {e}",
request_id=request_id
)
raise
def _collect_portable_data(
self,
subject_id: str,
categories: List[str]
) -> Dict:
"""카테고리별 포터블 데이터 수집"""
data = {}
for category in categories:
if category == "profile":
data["profile"] = self.user_data_store.get_profile(subject_id)
elif category == "activity":
data["activity"] = self.user_data_store.get_activities(subject_id)
elif category == "preferences":
data["preferences"] = self.user_data_store.get_preferences(subject_id)
elif category == "content":
data["content"] = self.user_data_store.get_user_content(subject_id)
elif category == "communications":
data["communications"] = self.user_data_store.get_communications(subject_id)
elif category == "sensor_data":
data["sensor_data"] = self.user_data_store.get_sensor_data(subject_id)
return data
def _format_data(self, data: Dict, format: DataFormat) -> Dict[str, bytes]:
"""데이터를 요청된 포맷으로 변환"""
formatted = {}
for category, content in data.items():
if format == DataFormat.JSON:
formatted[f"{category}.json"] = json.dumps(
content,
ensure_ascii=False,
indent=2,
default=str
).encode('utf-8')
elif format == DataFormat.CSV:
formatted[f"{category}.csv"] = self._to_csv(content)
return formatted
def _to_csv(self, data: List[Dict]) -> bytes:
"""리스트 데이터를 CSV로 변환"""
if not data:
return b""
output = io.StringIO()
writer = csv.DictWriter(output, fieldnames=data[0].keys())
writer.writeheader()
writer.writerows(data)
return output.getvalue().encode('utf-8')
def _create_data_package(
self,
formatted_data: Dict[str, bytes],
request: PortabilityRequest
) -> bytes:
"""데이터 패키지(ZIP) 생성 with 메타데이터"""
buffer = io.BytesIO()
with zipfile.ZipFile(buffer, 'w', zipfile.ZIP_DEFLATED) as zf:
# 메타데이터 파일 추가
metadata = {
"export_version": "1.0",
"gdpr_article": "20",
"exported_at": datetime.now().isoformat(),
"subject_id_hash": hashlib.sha256(
request.subject_id.encode()
).hexdigest(),
"categories": request.data_categories,
"format": request.target_format.value,
"data_controller": {
"name": "KAITRUST AI Platform",
"contact": "privacy@kaitrust.ai"
},
"interoperability": {
"standard": "Data Transfer Project",
"schema_version": "2024.1"
}
}
zf.writestr(
"metadata.json",
json.dumps(metadata, ensure_ascii=False, indent=2)
)
# 데이터 파일 추가
for filename, content in formatted_data.items():
zf.writestr(f"data/{filename}", content)
# 무결성 체크섬
checksums = {}
for filename, content in formatted_data.items():
checksums[filename] = hashlib.sha256(content).hexdigest()
zf.writestr(
"checksums.json",
json.dumps(checksums, indent=2)
)
return buffer.getvalue()
def _transfer_to_recipient(
self,
recipient_id: str,
package: bytes,
request: PortabilityRequest
):
"""
다른 컨트롤러에게 직접 전송 (Article 20(2))
기술적으로 가능한 경우에만 실행
"""
# 수신자 엔드포인트 조회
recipient = self._lookup_recipient_endpoint(recipient_id)
if recipient and recipient.get("supports_portability"):
# 보안 전송 수행
self._secure_transfer(
recipient["endpoint"],
package,
request.request_id
)
else:
self.audit_logger.log_warning(
"Direct transfer not supported by recipient",
request_id=request.request_id,
recipient_id=recipient_id
)
# 사용 예시
if __name__ == "__main__":
from datetime import timedelta
# 목업 데이터 저장소
class MockUserDataStore:
def get_profile(self, uid):
return {"name": "홍길동", "email": "hong@example.com"}
def get_activities(self, uid):
return [{"action": "login", "timestamp": "2024-01-15T10:00:00"}]
def get_preferences(self, uid):
return {"language": "ko", "theme": "dark"}
def get_user_content(self, uid):
return [{"type": "post", "content": "AI 관련 글"}]
def get_communications(self, uid):
return []
def get_sensor_data(self, uid):
return []
class MockAuditLogger:
def log_event(self, **kwargs): print(f"EVENT: {kwargs}")
def log_warning(self, msg, **kwargs): print(f"WARN: {msg}")
def log_error(self, msg, **kwargs): print(f"ERROR: {msg}")
handler = DataPortabilityHandler(MockUserDataStore(), MockAuditLogger())
# 이동권 요청 생성
request = handler.create_portability_request(
subject_id="user_12345",
data_categories=["profile", "activity", "preferences", "credit_score"],
target_format=DataFormat.JSON
)
print(f"Request ID: {request.request_id}")
print(f"Valid categories: {request.data_categories}")
print(f"Deadline: {request.expiry_date}")
# 요청 처리
package = handler.process_request(request.request_id)
print(f"Package size: {len(package)} bytes")