Automates the Privacy Impact Assessment (PIA) workflow including data flow mapping, privacy risk scoring matrices, GDPR Article 35 DPIA and CCPA/CPRA alignment
复制下面这句话,粘贴给 Claude Code、Codex、Cursor 等 AI 编程工具,它会读取安装说明并在你确认后完成安装。
请阅读 https://ai.atlankj.com/install/asset/gh-performing-privacy-impact-assessment-2b37925dddc8 ,按照其中的说明把「performing-privacy-impact-assessment」安装到你(当前 AI 工具)中。执行前先告诉我将运行的命令和写入的位置,等我确认。
查看 AI 将读取的安装说明正在读取 GitHub 原文…
内容来自 GitHub 原始文件,由原作者维护。在 GitHub 查看
Build a complete inventory of personal data processing activities. Each record of processing activity (ROPA) entry must capture the data categories, legal basis, retention periods, and data subjects involved.
from agent import PrivacyImpactAssessmentEngine
engine = PrivacyImpactAssessmentEngine()
# Register a processing activity for assessment
activity = engine.register_processing_activity(
name="Customer Analytics Platform",
description="Collects browsing behavior and purchase history for personalization",
data_controller="Acme Corp",
data_processor="CloudAnalytics Inc",
data_categories=["browsing_history", "purchase_records", "ip_address", "device_id"],
data_subjects=["customers", "website_visitors"],
legal_basis="consent",
retention_period_days=730,
cross_border_transfer=True,
transfer_destinations=["US", "IN"],
automated_decision_making=True,
)
print(f"Registered activity: {activity['activity_id']}")
Map all data flows from collection to deletion, identifying every touchpoint, transformation, and storage location. This reveals hidden privacy risks in data movement across systems.
# Build the data flow map
flow_map = engine.map_data_flows(
activity_id=activity["activity_id"],
flows=[
{
"stage": "collection",
"source": "Web browser cookie + form submission",
"destination": "CDN edge server",
"data_elements": ["ip_address", "device_id", "browsing_history"],
"encryption_in_transit": True,
"protocol": "TLS 1.3",
},
{
"stage": "processing",
"source": "CDN edge server",
"destination": "Analytics data warehouse (US-East)",
"data_elements": ["browsing_history", "purchase_records", "device_id"],
"encryption_in_transit": True,
"encryption_at_rest": True,
"protocol": "mTLS",
},
{
"stage": "storage",
"source": "Analytics data warehouse",
"destination": "S3 encrypted bucket",
"data_elements": ["browsing_history", "purchase_records"],
"encryption_at_rest": True,
"retention_days": 730,
"access_controls": "IAM role-based, MFA required",
},
{
"stage": "sharing",
"source": "Analytics data warehouse",
"destination": "Third-party ML provider (IN)",
"data_elements": ["browsing_history", "purchase_records"],
"encryption_in_transit": True,
"data_processing_agreement": True,
"cross_border": True,
},
{
"stage": "deletion",
"source": "S3 bucket + data warehouse",
"destination": "Secure erasure",
"method": "Cryptographic erasure + lifecycle policy",
"verification": "Automated deletion audit log",
},
],
)
engine.render_data_flow_diagram(flow_map)
Apply a structured risk scoring methodology evaluating likelihood and impact across multiple privacy risk dimensions. The matrix aligns with both the NIST PRAM and ICO DPIA risk assessment approaches.
# Run the risk assessment
risk_report = engine.assess_privacy_risks(
activity_id=activity["activity_id"],
assessment_type="full_dpia",
)
# Display risk matrix results
for risk in risk_report["risks"]:
print(f"[{risk['severity']}] {risk['category']}: {risk['description']}")
print(f" Likelihood: {risk['likelihood']}/5 | Impact: {risk['impact']}/5 | Score: {risk['risk_score']}/25")
print(f" Mitigation: {risk['recommended_mitigation']}")
Risk categories evaluated include:
Run automated compliance checks against specific regulatory requirements. The engine maps each processing activity against article-level GDPR obligations and CCPA/CPRA consumer rights requirements.
# GDPR compliance check
gdpr_report = engine.check_gdpr_compliance(activity_id=activity["activity_id"])
print(f"GDPR Score: {gdpr_report['compliance_score']}/100")
for finding in gdpr_report["findings"]:
print(f" [{finding['status']}] Art.{finding['article']}: {finding['description']}")
# CCPA/CPRA compliance check
ccpa_report = engine.check_ccpa_compliance(activity_id=activity["activity_id"])
print(f"CCPA Score: {ccpa_report['compliance_score']}/100")
for finding in ccpa_report["findings"]:
print(f" [{finding['status']}] Sec.{finding['section']}: {finding['description']}")
Generate a prioritized remediation plan with specific action items, responsible parties, deadlines, and generate the formal PIA/DPIA report document.
# Generate remediation plan
remediation = engine.generate_remediation_plan(
activity_id=activity["activity_id"],
risk_report=risk_report,
gdpr_report=gdpr_report,
ccpa_report=ccpa_report,
)
for item in remediation["action_items"]:
print(f"[{item['priority']}] {item['action']}")
print(f" Owner: {item['owner']} | Deadline: {item['deadline']}")
print(f" Addresses: {', '.join(item['addresses_risks'])}")
# Generate formal DPIA report
engine.generate_dpia_report(
activity_id=activity["activity_id"],
output_path="dpia_report_customer_analytics.json",
format="json",
)
print("[+] DPIA report generated")
Determine whether a full DPIA is required using the ICO screening checklist:
engine = PrivacyImpactAssessmentEngine()
screening = engine.run_screening_checklist(
uses_special_category_data=False,
large_scale_processing=True,
systematic_monitoring=True,
automated_decision_making=True,
cross_border_transfer=True,
vulnerable_data_subjects=False,
innovative_technology=True,
denial_of_service_or_rights=False,
)
print(f"DPIA Required: {screening['dpia_required']}")
print(f"Triggers: {screening['triggers']}")
# Output: DPIA Required: True
# Triggers: ['large_scale_processing', 'systematic_monitoring',
# 'automated_decision_making', 'cross_border_transfer',
# 'innovative_technology']
engine = PrivacyImpactAssessmentEngine()
activities = [
{"name": "Email Marketing", "data_categories": ["email", "name"],
"legal_basis": "consent", "cross_border_transfer": False},
{"name": "HR Analytics", "data_categories": ["employee_id", "performance_scores",
"health_data"], "legal_basis": "legitimate_interest", "cross_border_transfer": True},
{"name": "Fraud Detection", "data_categories": ["transaction_data", "ip_address",
"device_fingerprint"], "legal_basis": "legitimate_interest",
"automated_decision_making": True, "cross_border_transfer": False},
]
for act_def in activities:
activity = engine.register_processing_activity(**act_def)
risk = engine.assess_privacy_risks(activity_id=activity["activity_id"])
print(f"{act_def['name']}: Overall Risk={risk['overall_risk_level']} "
f"({risk['risk_count_by_severity']})")
engine = PrivacyImpactAssessmentEngine()
profile = engine.generate_nist_privacy_profile(
activity_id=activity["activity_id"],
target_tier="tier_3", # Repeatable
)
for function_id, outcomes in profile["functions"].items():
print(f"\n{function_id}:")
for outcome in outcomes:
status = "PASS" if outcome["implemented"] else "GAP"
print(f" [{status}] {outcome['subcategory']}: {outcome['description']}")