Gdpr Dsgvo Expert

#!/usr/bin/env python3
"""
Data Subject Rights Tracker

Tracks and manages data subject rights requests under GDPR Articles 15-22.
Monitors deadlines, generates response templates, and produces compliance reports.

Usage:
    python data_subject_rights_tracker.py list
    python data_subject_rights_tracker.py add --type access --subject "John Doe"
    python data_subject_rights_tracker.py status --id REQ-001
    python data_subject_rights_tracker.py report --output compliance_report.json
"""

import argparse
import json
import os
import sys
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, List, Optional
from uuid import uuid4


def add_months(dt: datetime, months: int) -> datetime:
    """Add calendar months per GDPR Art. 12(3) (one month, not 30 days).

    If the target month has no equivalent day (e.g., Jan 31 + 1 month),
    clamp to the last day of the target month.
    """
    month_index = dt.month - 1 + months
    year = dt.year + month_index // 12
    month = month_index % 12 + 1
    # last day of target month
    if month == 12:
        next_month_first = datetime(year + 1, 1, 1)
    else:
        next_month_first = datetime(year, month + 1, 1)
    last_day = (next_month_first - timedelta(days=1)).day
    return dt.replace(year=year, month=month, day=min(dt.day, last_day))


# GDPR Articles for each right
RIGHTS_TYPES = {
    "access": {
        "article": "Art. 15",
        "name": "Right of Access",
        "deadline_months": 1,  # Art. 12(3): one month of receipt
        "description": "Data subject has the right to obtain confirmation of processing and access to their data",
        "response_includes": [
            "Purposes of processing",
            "Categories of personal data",
            "Recipients or categories of recipients",
            "Retention period or criteria",
            "Right to lodge complaint",
            "Source of data (if not collected from subject)",
            "Existence of automated decision-making"
        ]
    },
    "rectification": {
        "article": "Art. 16",
        "name": "Right to Rectification",
        "deadline_months": 1,  # Art. 12(3): one month of receipt
        "description": "Data subject has the right to have inaccurate personal data corrected",
        "response_includes": [
            "Confirmation of correction",
            "Details of corrected data",
            "Notification to recipients"
        ]
    },
    "erasure": {
        "article": "Art. 17",
        "name": "Right to Erasure (Right to be Forgotten)",
        "deadline_months": 1,  # Art. 12(3): one month of receipt
        "description": "Data subject has the right to have their personal data erased",
        "grounds": [
            "Data no longer necessary for original purpose",
            "Consent withdrawn",
            "Objection to processing (no overriding grounds)",
            "Unlawful processing",
            "Legal obligation to erase",
            "Data collected from child"
        ],
        "exceptions": [
            "Freedom of expression",
            "Legal obligation to retain",
            "Public health reasons",
            "Archiving in public interest",
            "Legal claims"
        ]
    },
    "restriction": {
        "article": "Art. 18",
        "name": "Right to Restriction of Processing",
        "deadline_months": 1,  # Art. 12(3): one month of receipt
        "description": "Data subject has the right to restrict processing of their data",
        "grounds": [
            "Accuracy contested (during verification)",
            "Processing is unlawful (erasure opposed)",
            "Controller no longer needs data (subject needs for legal claims)",
            "Objection pending verification"
        ]
    },
    "portability": {
        "article": "Art. 20",
        "name": "Right to Data Portability",
        "deadline_months": 1,  # Art. 12(3): one month of receipt
        "description": "Data subject has the right to receive their data in a portable format",
        "conditions": [
            "Processing based on consent or contract",
            "Processing carried out by automated means"
        ],
        "format_requirements": [
            "Structured format",
            "Commonly used format",
            "Machine-readable format"
        ]
    },
    "objection": {
        "article": "Art. 21",
        "name": "Right to Object",
        "deadline_months": 1,  # Art. 12(3): one month of receipt
        "description": "Data subject has the right to object to processing",
        "applies_to": [
            "Processing based on legitimate interests",
            "Processing for direct marketing",
            "Processing for research/statistics"
        ]
    },
    "automated": {
        "article": "Art. 22",
        "name": "Rights Related to Automated Decision-Making",
        "deadline_months": 1,  # Art. 12(3): one month of receipt
        "description": "Data subject has the right not to be subject to solely automated decisions",
        "includes": [
            "Right to human intervention",
            "Right to express point of view",
            "Right to contest decision"
        ]
    }
}

# Request statuses
STATUSES = {
    "received": "Request received, pending identity verification",
    "verified": "Identity verified, processing request",
    "in_progress": "Gathering data / processing request",
    "pending_info": "Awaiting additional information from subject",
    "extended": "Deadline extended (complex request)",
    "completed": "Request completed and response sent",
    "refused": "Request refused (with justification)",
    "escalated": "Escalated to DPO/legal"
}


class RightsTracker:
    """Manages data subject rights requests."""

    def __init__(self, data_file: str = "dsr_requests.json"):
        self.data_file = Path(data_file)
        self.requests = self._load_requests()

    def _load_requests(self) -> Dict:
        """Load requests from file."""
        if self.data_file.exists():
            with open(self.data_file, "r") as f:
                return json.load(f)
        return {"requests": [], "metadata": {"created": datetime.now().isoformat()}}

    def _save_requests(self):
        """Save requests to file."""
        self.requests["metadata"]["updated"] = datetime.now().isoformat()
        with open(self.data_file, "w") as f:
            json.dump(self.requests, f, indent=2)

    def _generate_id(self) -> str:
        """Generate unique request ID."""
        count = len(self.requests["requests"]) + 1
        return f"DSR-{datetime.now().strftime('%Y%m')}-{count:04d}"

    def add_request(
        self,
        right_type: str,
        subject_name: str,
        subject_email: str,
        details: str = ""
    ) -> Dict:
        """Add a new data subject request."""
        if right_type not in RIGHTS_TYPES:
            raise ValueError(f"Invalid right type. Must be one of: {list(RIGHTS_TYPES.keys())}")

        right_info = RIGHTS_TYPES[right_type]
        now = datetime.now()
        # Art. 12(3): respond within one calendar month of receipt
        deadline = add_months(now, right_info["deadline_months"])

        request = {
            "id": self._generate_id(),
            "type": right_type,
            "article": right_info["article"],
            "right_name": right_info["name"],
            "subject": {
                "name": subject_name,
                "email": subject_email,
                "verified": False
            },
            "details": details,
            "status": "received",
            "status_description": STATUSES["received"],
            "dates": {
                "received": now.isoformat(),
                "deadline": deadline.isoformat(),
                "verified": None,
                "completed": None
            },
            "notes": [],
            "response": None
        }

        self.requests["requests"].append(request)
        self._save_requests()
        return request

    def update_status(
        self,
        request_id: str,
        new_status: str,
        note: str = ""
    ) -> Optional[Dict]:
        """Update request status."""
        if new_status not in STATUSES:
            raise ValueError(f"Invalid status. Must be one of: {list(STATUSES.keys())}")

        for req in self.requests["requests"]:
            if req["id"] == request_id:
                req["status"] = new_status
                req["status_description"] = STATUSES[new_status]

                if new_status == "verified":
                    req["subject"]["verified"] = True
                    req["dates"]["verified"] = datetime.now().isoformat()
                elif new_status == "completed":
                    req["dates"]["completed"] = datetime.now().isoformat()
                elif new_status == "extended":
                    # Art. 12(3): extendable by two further calendar months for
                    # complex/numerous requests (data subject informed within month 1)
                    original_deadline = datetime.fromisoformat(req["dates"]["deadline"])
                    req["dates"]["deadline"] = add_months(original_deadline, 2).isoformat()

                if note:
                    req["notes"].append({
                        "timestamp": datetime.now().isoformat(),
                        "note": note
                    })

                self._save_requests()
                return req

        return None

    def get_request(self, request_id: str) -> Optional[Dict]:
        """Get request by ID."""
        for req in self.requests["requests"]:
            if req["id"] == request_id:
                return req
        return None

    def list_requests(
        self,
        status_filter: Optional[str] = None,
        overdue_only: bool = False
    ) -> List[Dict]:
        """List requests with optional filtering."""
        results = []
        now = datetime.now()

        for req in self.requests["requests"]:
            if status_filter and req["status"] != status_filter:
                continue

            deadline = datetime.fromisoformat(req["dates"]["deadline"])
            is_overdue = deadline < now and req["status"] not in ["completed", "refused"]

            if overdue_only and not is_overdue:
                continue

            req_summary = {
                **req,
                "is_overdue": is_overdue,
                "days_remaining": (deadline - now).days if not is_overdue else 0
            }
            results.append(req_summary)

        return results

    def generate_report(self) -> Dict:
        """Generate compliance report."""
        now = datetime.now()
        total = len(self.requests["requests"])

        status_counts = {}
        for status in STATUSES:
            status_counts[status] = sum(1 for r in self.requests["requests"] if r["status"] == status)

        type_counts = {}
        for right_type in RIGHTS_TYPES:
            type_counts[right_type] = sum(1 for r in self.requests["requests"] if r["type"] == right_type)

        overdue = []
        completed_on_time = 0
        completed_late = 0

        for req in self.requests["requests"]:
            deadline = datetime.fromisoformat(req["dates"]["deadline"])

            if req["status"] in ["completed", "refused"]:
                completed_date = datetime.fromisoformat(req["dates"]["completed"])
                if completed_date <= deadline:
                    completed_on_time += 1
                else:
                    completed_late += 1
            elif deadline < now:
                overdue.append({
                    "id": req["id"],
                    "type": req["type"],
                    "subject": req["subject"]["name"],
                    "days_overdue": (now - deadline).days
                })

        compliance_rate = (completed_on_time / (completed_on_time + completed_late) * 100) if (completed_on_time + completed_late) > 0 else 100

        return {
            "report_date": now.isoformat(),
            "summary": {
                "total_requests": total,
                "open_requests": total - status_counts.get("completed", 0) - status_counts.get("refused", 0),
                "overdue_requests": len(overdue),
                "compliance_rate": round(compliance_rate, 1)
            },
            "by_status": status_counts,
            "by_type": type_counts,
            "overdue_details": overdue,
            "performance": {
                "completed_on_time": completed_on_time,
                "completed_late": completed_late,
                "average_response_days": self._calculate_avg_response_time()
            }
        }

    def _calculate_avg_response_time(self) -> float:
        """Calculate average response time for completed requests."""
        response_times = []

        for req in self.requests["requests"]:
            if req["status"] == "completed" and req["dates"]["completed"]:
                received = datetime.fromisoformat(req["dates"]["received"])
                completed = datetime.fromisoformat(req["dates"]["completed"])
                response_times.append((completed - received).days)

        return round(sum(response_times) / len(response_times), 1) if response_times else 0

    def generate_response_template(self, request_id: str) -> Optional[str]:
        """Generate response template for a request."""
        req = self.get_request(request_id)
        if not req:
            return None

        right_info = RIGHTS_TYPES.get(req["type"], {})
        template = f"""
Subject: Response to Your {right_info.get('name', 'Data Subject')} Request ({req['id']})

Dear {req['subject']['name']},

Thank you for your request dated {req['dates']['received'][:10]} exercising your {right_info.get('name', 'data protection right')} under {right_info.get('article', 'GDPR')}.

We have processed your request and respond as follows:

[RESPONSE DETAILS HERE]

"""
        if req["type"] == "access":
            template += """
As required under Article 15, we provide the following information:

1. Purposes of Processing:
   [List purposes]

2. Categories of Personal Data:
   [List categories]

3. Recipients:
   [List recipients or categories]

4. Retention Period:
   [Specify period or criteria]

5. Your Rights:
   - Right to rectification (Art. 16)
   - Right to erasure (Art. 17)
   - Right to restriction (Art. 18)
   - Right to object (Art. 21)
   - Right to lodge complaint with supervisory authority

6. Source of Data:
   [Specify if not collected from you directly]

7. Automated Decision-Making:
   [Confirm if applicable and provide meaningful information]

Enclosed: Copy of your personal data
"""
        elif req["type"] == "erasure":
            template += """
We confirm that your personal data has been erased from our systems, except where:
- We are legally required to retain it
- It is necessary for legal claims
- [Other applicable exceptions]

We have also notified the following recipients of the erasure:
[List recipients]
"""
        elif req["type"] == "portability":
            template += """
Please find attached your personal data in [JSON/CSV] format.

This includes all data:
- Provided by you
- Processed based on your consent or contract
- Processed by automated means

You may transmit this data to another controller or request direct transmission where technically feasible.
"""

        template += f"""
If you have any questions about this response, please contact our Data Protection Officer at [DPO EMAIL].

If you are not satisfied with our response, you have the right to lodge a complaint with the supervisory authority:
[SUPERVISORY AUTHORITY DETAILS]

Yours sincerely,
[CONTROLLER NAME]
Data Protection Team

Reference: {req['id']}
"""
        return template


def main():
    parser = argparse.ArgumentParser(
        description="Track and manage data subject rights requests"
    )
    parser.add_argument(
        "--data-file",
        default="dsr_requests.json",
        help="Path to requests data file (default: dsr_requests.json)"
    )

    subparsers = parser.add_subparsers(dest="command", help="Commands")

    # Add command
    add_parser = subparsers.add_parser("add", help="Add new request")
    add_parser.add_argument("--type", "-t", required=True, choices=RIGHTS_TYPES.keys())
    add_parser.add_argument("--subject", "-s", required=True, help="Subject name")
    add_parser.add_argument("--email", "-e", required=True, help="Subject email")
    add_parser.add_argument("--details", "-d", default="", help="Request details")

    # List command
    list_parser = subparsers.add_parser("list", help="List requests")
    list_parser.add_argument("--status", choices=STATUSES.keys(), help="Filter by status")
    list_parser.add_argument("--overdue", action="store_true", help="Show only overdue")
    list_parser.add_argument("--json", action="store_true", help="JSON output")

    # Status command
    status_parser = subparsers.add_parser("status", help="Get/update request status")
    status_parser.add_argument("--id", required=True, help="Request ID")
    status_parser.add_argument("--update", choices=STATUSES.keys(), help="Update status")
    status_parser.add_argument("--note", default="", help="Add note")

    # Report command
    report_parser = subparsers.add_parser("report", help="Generate compliance report")
    report_parser.add_argument("--output", "-o", help="Output file")

    # Template command
    template_parser = subparsers.add_parser("template", help="Generate response template")
    template_parser.add_argument("--id", required=True, help="Request ID")

    # Types command
    subparsers.add_parser("types", help="List available request types")

    args = parser.parse_args()

    tracker = RightsTracker(args.data_file)

    if args.command == "add":
        request = tracker.add_request(
            args.type, args.subject, args.email, args.details
        )
        print(f"Request created: {request['id']}")
        print(f"Type: {request['right_name']} ({request['article']})")
        print(f"Deadline: {request['dates']['deadline'][:10]}")

    elif args.command == "list":
        requests = tracker.list_requests(args.status, args.overdue)
        if args.json:
            print(json.dumps(requests, indent=2))
        else:
            if not requests:
                print("No requests found.")
                return
            print(f"{'ID':<20} {'Type':<15} {'Subject':<20} {'Status':<15} {'Deadline':<12} {'Overdue'}")
            print("-" * 95)
            for req in requests:
                overdue_flag = "YES" if req.get("is_overdue") else ""
                print(f"{req['id']:<20} {req['type']:<15} {req['subject']['name'][:20]:<20} {req['status']:<15} {req['dates']['deadline'][:10]:<12} {overdue_flag}")

    elif args.command == "status":
        if args.update:
            req = tracker.update_status(args.id, args.update, args.note)
            if req:
                print(f"Updated {args.id} to status: {args.update}")
            else:
                print(f"Request not found: {args.id}")
        else:
            req = tracker.get_request(args.id)
            if req:
                print(json.dumps(req, indent=2))
            else:
                print(f"Request not found: {args.id}")

    elif args.command == "report":
        report = tracker.generate_report()
        output = json.dumps(report, indent=2)
        if args.output:
            with open(args.output, "w") as f:
                f.write(output)
            print(f"Report written to {args.output}")
        else:
            print(output)

    elif args.command == "template":
        template = tracker.generate_response_template(args.id)
        if template:
            print(template)
        else:
            print(f"Request not found: {args.id}")

    elif args.command == "types":
        print("Available Request Types:")
        print("-" * 60)
        for key, info in RIGHTS_TYPES.items():
            print(f"\n{key} ({info['article']})")
            print(f"  {info['name']}")
            print(f"  Deadline: {info['deadline_months']} calendar month(s) (Art. 12(3))")

    else:
        parser.print_help()


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
DPIA Generator

Generates Data Protection Impact Assessment documentation based on
processing activity inputs. Creates structured DPIA reports following
GDPR Article 35 requirements.

Usage:
    python dpia_generator.py --interactive
    python dpia_generator.py --input processing_activity.json --output dpia_report.md
    python dpia_generator.py --template > template.json
"""

import argparse
import json
import sys
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional


# DPIA threshold criteria (Art. 35(3) and WP29 Guidelines)
DPIA_TRIGGERS = {
    "systematic_monitoring": {
        "description": "Systematic monitoring of publicly accessible area",
        "article": "Art. 35(3)(c)",
        "weight": 10
    },
    "large_scale_special_category": {
        "description": "Large-scale processing of special category data (Art. 9)",
        "article": "Art. 35(3)(b)",
        "weight": 10
    },
    "automated_decision_making": {
        "description": "Automated decision-making with legal/significant effects",
        "article": "Art. 35(3)(a)",
        "weight": 10
    },
    "evaluation_scoring": {
        "description": "Evaluation or scoring of individuals",
        "article": "WP29 Guidelines",
        "weight": 7
    },
    "sensitive_data": {
        "description": "Processing of sensitive data or highly personal data",
        "article": "WP29 Guidelines",
        "weight": 7
    },
    "large_scale": {
        "description": "Data processed on a large scale",
        "article": "WP29 Guidelines",
        "weight": 6
    },
    "data_matching": {
        "description": "Matching or combining datasets",
        "article": "WP29 Guidelines",
        "weight": 5
    },
    "vulnerable_subjects": {
        "description": "Data concerning vulnerable data subjects",
        "article": "WP29 Guidelines",
        "weight": 7
    },
    "innovative_technology": {
        "description": "Innovative use or applying new technological solutions",
        "article": "WP29 Guidelines",
        "weight": 5
    },
    "cross_border_transfer": {
        "description": "Transfer of data outside the EU/EEA",
        "article": "GDPR Chapter V",
        "weight": 5
    }
}

# Risk categories and mitigation measures
RISK_CATEGORIES = {
    "unauthorized_access": {
        "description": "Risk of unauthorized access to personal data",
        "impact": "high",
        "mitigations": [
            "Implement access controls and authentication",
            "Use encryption for data at rest and in transit",
            "Maintain audit logs of access",
            "Implement least privilege principle"
        ]
    },
    "data_breach": {
        "description": "Risk of data breach or unauthorized disclosure",
        "impact": "high",
        "mitigations": [
            "Implement intrusion detection systems",
            "Establish incident response procedures",
            "Regular security assessments",
            "Employee security training"
        ]
    },
    "excessive_collection": {
        "description": "Risk of collecting more data than necessary",
        "impact": "medium",
        "mitigations": [
            "Implement data minimization principles",
            "Regular review of data collected",
            "Privacy by design approach",
            "Document purpose for each data element"
        ]
    },
    "purpose_creep": {
        "description": "Risk of using data for purposes beyond original scope",
        "impact": "medium",
        "mitigations": [
            "Clear purpose limitation policies",
            "Consent management for new purposes",
            "Technical controls on data access",
            "Regular purpose review"
        ]
    },
    "retention_violation": {
        "description": "Risk of retaining data longer than necessary",
        "impact": "medium",
        "mitigations": [
            "Implement retention schedules",
            "Automated deletion processes",
            "Regular data inventory audits",
            "Document retention justification"
        ]
    },
    "rights_violation": {
        "description": "Risk of failing to fulfill data subject rights",
        "impact": "high",
        "mitigations": [
            "Implement subject access request process",
            "Technical capability for data portability",
            "Deletion/erasure procedures",
            "Staff training on rights requests"
        ]
    },
    "inaccurate_data": {
        "description": "Risk of processing inaccurate or outdated data",
        "impact": "medium",
        "mitigations": [
            "Data quality checks at collection",
            "Regular data verification",
            "Easy update mechanisms for subjects",
            "Automated accuracy validation"
        ]
    },
    "third_party_risk": {
        "description": "Risk from third-party processors",
        "impact": "high",
        "mitigations": [
            "Due diligence on processors",
            "Data Processing Agreements",
            "Regular processor audits",
            "Clear processor instructions"
        ]
    }
}

# Legal bases under Article 6
LEGAL_BASES = {
    "consent": {
        "article": "Art. 6(1)(a)",
        "description": "Data subject has given consent",
        "requirements": [
            "Consent must be freely given",
            "Specific to the purpose",
            "Informed consent with clear information",
            "Unambiguous indication of wishes",
            "Easy to withdraw"
        ]
    },
    "contract": {
        "article": "Art. 6(1)(b)",
        "description": "Processing necessary for contract performance",
        "requirements": [
            "Contract must exist or be in negotiation",
            "Processing must be necessary for the contract",
            "Cannot process more than contractually needed"
        ]
    },
    "legal_obligation": {
        "article": "Art. 6(1)(c)",
        "description": "Processing necessary for legal obligation",
        "requirements": [
            "Legal obligation must be binding",
            "Must be EU or Member State law",
            "Processing must be necessary to comply"
        ]
    },
    "vital_interests": {
        "article": "Art. 6(1)(d)",
        "description": "Processing necessary to protect vital interests",
        "requirements": [
            "Life-threatening situation",
            "No other legal basis available",
            "Typically emergency situations"
        ]
    },
    "public_interest": {
        "article": "Art. 6(1)(e)",
        "description": "Processing necessary for public interest task",
        "requirements": [
            "Task in public interest or official authority",
            "Legal basis in EU or Member State law",
            "Processing must be necessary"
        ]
    },
    "legitimate_interests": {
        "article": "Art. 6(1)(f)",
        "description": "Processing necessary for legitimate interests",
        "requirements": [
            "Identify the legitimate interest",
            "Show processing is necessary",
            "Balance against data subject rights",
            "Not available for public authorities"
        ]
    }
}


def get_template() -> Dict:
    """Return a blank DPIA input template."""
    return {
        "project_name": "",
        "version": "1.0",
        "date": datetime.now().strftime("%Y-%m-%d"),
        "controller": {
            "name": "",
            "contact": "",
            "dpo_contact": ""
        },
        "processing_activity": {
            "description": "",
            "purposes": [],
            "legal_basis": "",
            "legal_basis_justification": ""
        },
        "data_subjects": {
            "categories": [],
            "estimated_number": "",
            "vulnerable_groups": False,
            "vulnerable_groups_details": ""
        },
        "personal_data": {
            "categories": [],
            "special_categories": [],
            "source": "",
            "retention_period": ""
        },
        "processing_operations": {
            "collection_method": "",
            "storage_location": "",
            "access_controls": "",
            "automated_decisions": False,
            "profiling": False
        },
        "data_recipients": {
            "internal": [],
            "external_processors": [],
            "third_countries": []
        },
        "dpia_triggers": [],
        "identified_risks": [],
        "mitigations_planned": []
    }


def assess_dpia_requirement(input_data: Dict) -> Dict:
    """Assess whether DPIA is required based on triggers."""
    triggers_present = input_data.get("dpia_triggers", [])
    total_weight = 0
    triggered_criteria = []

    for trigger in triggers_present:
        if trigger in DPIA_TRIGGERS:
            trigger_info = DPIA_TRIGGERS[trigger]
            total_weight += trigger_info["weight"]
            triggered_criteria.append({
                "trigger": trigger,
                "description": trigger_info["description"],
                "article": trigger_info["article"]
            })

    # Also check data characteristics
    if input_data.get("data_subjects", {}).get("vulnerable_groups"):
        if "vulnerable_subjects" not in triggers_present:
            total_weight += DPIA_TRIGGERS["vulnerable_subjects"]["weight"]
            triggered_criteria.append({
                "trigger": "vulnerable_subjects",
                "description": DPIA_TRIGGERS["vulnerable_subjects"]["description"],
                "article": DPIA_TRIGGERS["vulnerable_subjects"]["article"]
            })

    if input_data.get("personal_data", {}).get("special_categories"):
        if "sensitive_data" not in triggers_present:
            total_weight += DPIA_TRIGGERS["sensitive_data"]["weight"]
            triggered_criteria.append({
                "trigger": "sensitive_data",
                "description": DPIA_TRIGGERS["sensitive_data"]["description"],
                "article": DPIA_TRIGGERS["sensitive_data"]["article"]
            })

    if input_data.get("data_recipients", {}).get("third_countries"):
        if "cross_border_transfer" not in triggers_present:
            total_weight += DPIA_TRIGGERS["cross_border_transfer"]["weight"]
            triggered_criteria.append({
                "trigger": "cross_border_transfer",
                "description": DPIA_TRIGGERS["cross_border_transfer"]["description"],
                "article": DPIA_TRIGGERS["cross_border_transfer"]["article"]
            })

    # DPIA required if 2+ triggers or weight >= 10
    dpia_required = len(triggered_criteria) >= 2 or total_weight >= 10

    return {
        "dpia_required": dpia_required,
        "risk_score": total_weight,
        "triggered_criteria": triggered_criteria,
        "recommendation": "DPIA is mandatory" if dpia_required else "DPIA recommended as best practice"
    }


def assess_risks(input_data: Dict) -> List[Dict]:
    """Assess risks based on processing characteristics."""
    risks = []

    # Check each risk category
    processing = input_data.get("processing_operations", {})
    recipients = input_data.get("data_recipients", {})
    personal_data = input_data.get("personal_data", {})

    # Unauthorized access risk
    if processing.get("storage_location") or processing.get("collection_method"):
        risks.append({
            **RISK_CATEGORIES["unauthorized_access"],
            "likelihood": "medium",
            "residual_risk": "low" if processing.get("access_controls") else "medium"
        })

    # Data breach risk (always present)
    risks.append({
        **RISK_CATEGORIES["data_breach"],
        "likelihood": "medium",
        "residual_risk": "medium"
    })

    # Third party risk
    if recipients.get("external_processors") or recipients.get("third_countries"):
        risks.append({
            **RISK_CATEGORIES["third_party_risk"],
            "likelihood": "medium",
            "residual_risk": "medium"
        })

    # Rights violation risk
    risks.append({
        **RISK_CATEGORIES["rights_violation"],
        "likelihood": "low",
        "residual_risk": "low"
    })

    # Retention violation risk
    if not personal_data.get("retention_period"):
        risks.append({
            **RISK_CATEGORIES["retention_violation"],
            "likelihood": "high",
            "residual_risk": "high"
        })

    # Automated decision risk
    if processing.get("automated_decisions") or processing.get("profiling"):
        risks.append({
            "description": "Risk of unfair automated decisions affecting individuals",
            "impact": "high",
            "likelihood": "medium",
            "residual_risk": "medium",
            "mitigations": [
                "Human review of automated decisions",
                "Transparency about logic involved",
                "Right to contest decisions",
                "Regular algorithm audits"
            ]
        })

    return risks


def generate_dpia_report(input_data: Dict) -> str:
    """Generate DPIA report in Markdown format."""
    requirement = assess_dpia_requirement(input_data)
    risks = assess_risks(input_data)

    project = input_data.get("project_name", "Unnamed Project")
    controller = input_data.get("controller", {})
    processing = input_data.get("processing_activity", {})
    subjects = input_data.get("data_subjects", {})
    personal_data = input_data.get("personal_data", {})
    operations = input_data.get("processing_operations", {})
    recipients = input_data.get("data_recipients", {})

    legal_basis = processing.get("legal_basis", "")
    legal_info = LEGAL_BASES.get(legal_basis, {})

    report = f"""# Data Protection Impact Assessment (DPIA)

## Project: {project}

| Field | Value |
|-------|-------|
| Version | {input_data.get('version', '1.0')} |
| Date | {input_data.get('date', datetime.now().strftime('%Y-%m-%d'))} |
| Controller | {controller.get('name', 'N/A')} |
| DPO Contact | {controller.get('dpo_contact', 'N/A')} |

---

## 1. DPIA Threshold Assessment

**Result: {requirement['recommendation']}**

Risk Score: {requirement['risk_score']}/100

### Triggered Criteria

"""
    if requirement['triggered_criteria']:
        for criteria in requirement['triggered_criteria']:
            report += f"- **{criteria['description']}** ({criteria['article']})\n"
    else:
        report += "- No mandatory triggers identified\n"

    report += f"""
---

## 2. Description of Processing

### Purpose of Processing

{processing.get('description', 'Not specified')}

### Purposes

"""
    for purpose in processing.get('purposes', ['Not specified']):
        report += f"- {purpose}\n"

    report += f"""
### Legal Basis

**{legal_info.get('article', 'Not specified')}**: {legal_info.get('description', processing.get('legal_basis', 'Not specified'))}

**Justification**: {processing.get('legal_basis_justification', 'Not provided')}

"""
    if legal_info.get('requirements'):
        report += "**Requirements to satisfy:**\n"
        for req in legal_info['requirements']:
            report += f"- {req}\n"

    report += f"""
---

## 3. Data Subjects

| Aspect | Details |
|--------|---------|
| Categories | {', '.join(subjects.get('categories', ['Not specified']))} |
| Estimated Number | {subjects.get('estimated_number', 'Not specified')} |
| Vulnerable Groups | {'Yes - ' + subjects.get('vulnerable_groups_details', '') if subjects.get('vulnerable_groups') else 'No'} |

---

## 4. Personal Data Processed

### Data Categories

"""
    for category in personal_data.get('categories', ['Not specified']):
        report += f"- {category}\n"

    if personal_data.get('special_categories'):
        report += "\n### Special Category Data (Art. 9)\n\n"
        for category in personal_data['special_categories']:
            report += f"- **{category}** - Requires Art. 9(2) exception\n"

    report += f"""
### Data Source

{personal_data.get('source', 'Not specified')}

### Retention Period

{personal_data.get('retention_period', 'Not specified')}

---

## 5. Processing Operations

| Operation | Details |
|-----------|---------|
| Collection Method | {operations.get('collection_method', 'Not specified')} |
| Storage Location | {operations.get('storage_location', 'Not specified')} |
| Access Controls | {operations.get('access_controls', 'Not specified')} |
| Automated Decisions | {'Yes' if operations.get('automated_decisions') else 'No'} |
| Profiling | {'Yes' if operations.get('profiling') else 'No'} |

---

## 6. Data Recipients

### Internal Recipients

"""
    for recipient in recipients.get('internal', ['Not specified']):
        report += f"- {recipient}\n"

    report += "\n### External Processors\n\n"
    for processor in recipients.get('external_processors', ['None']):
        report += f"- {processor}\n"

    if recipients.get('third_countries'):
        report += "\n### Third Country Transfers\n\n"
        report += "**Warning**: Transfers require Chapter V safeguards\n\n"
        for country in recipients['third_countries']:
            report += f"- {country}\n"

    report += """
---

## 7. Risk Assessment

"""
    for i, risk in enumerate(risks, 1):
        report += f"""### Risk {i}: {risk['description']}

| Aspect | Assessment |
|--------|------------|
| Impact | {risk.get('impact', 'medium').upper()} |
| Likelihood | {risk.get('likelihood', 'medium').upper()} |
| Residual Risk | {risk.get('residual_risk', 'medium').upper()} |

**Recommended Mitigations:**

"""
        for mitigation in risk.get('mitigations', []):
            report += f"- {mitigation}\n"
        report += "\n"

    report += """---

## 8. Necessity and Proportionality

### Assessment Questions

1. **Is the processing necessary for the stated purpose?**
   - [ ] Yes, no less intrusive alternative exists
   - [ ] Alternative considered: _______________

2. **Is the data collection proportionate?**
   - [ ] Only necessary data is collected
   - [ ] Data minimization applied

3. **Are retention periods justified?**
   - [ ] Retention period is necessary
   - [ ] Deletion procedures in place

---

## 9. DPO Consultation

| Aspect | Details |
|--------|---------|
| DPO Consulted | [ ] Yes / [ ] No |
| DPO Name | |
| Consultation Date | |
| DPO Opinion | |

---

## 10. Sign-Off

| Role | Name | Signature | Date |
|------|------|-----------|------|
| Project Owner | | | |
| Data Protection Officer | | | |
| Controller Representative | | | |

---

## 11. Review Schedule

This DPIA should be reviewed:
- [ ] Annually
- [ ] When processing changes significantly
- [ ] Following a data incident
- [ ] As required by supervisory authority

Next Review Date: _______________

---

*Generated by DPIA Generator - This document requires completion and review by qualified personnel.*
"""
    return report


def main():
    parser = argparse.ArgumentParser(
        description="Generate DPIA documentation"
    )
    parser.add_argument(
        "--input", "-i",
        help="Path to JSON input file with processing activity details"
    )
    parser.add_argument(
        "--output", "-o",
        help="Path to output file (default: stdout)"
    )
    parser.add_argument(
        "--template",
        action="store_true",
        help="Output a blank JSON template"
    )
    parser.add_argument(
        "--interactive",
        action="store_true",
        help="Run in interactive mode"
    )

    args = parser.parse_args()

    if args.template:
        print(json.dumps(get_template(), indent=2))
        return

    if args.interactive:
        print("DPIA Generator - Interactive Mode")
        print("=" * 40)
        print("\nTo use this tool:")
        print("1. Generate a template: python dpia_generator.py --template > input.json")
        print("2. Fill in the template with your processing details")
        print("3. Generate DPIA: python dpia_generator.py --input input.json --output dpia.md")
        return

    if not args.input:
        print("Error: --input required (or use --template to get started)")
        sys.exit(1)

    input_path = Path(args.input)
    if not input_path.exists():
        print(f"Error: Input file not found: {input_path}")
        sys.exit(1)

    with open(input_path, "r") as f:
        input_data = json.load(f)

    report = generate_dpia_report(input_data)

    if args.output:
        with open(args.output, "w") as f:
            f.write(report)
        print(f"DPIA report written to {args.output}")
    else:
        print(report)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
GDPR Compliance Checker

Scans codebases, configurations, and data handling patterns for potential
GDPR compliance issues. Identifies personal data processing, consent gaps,
and documentation requirements.

Usage:
    python gdpr_compliance_checker.py /path/to/project
    python gdpr_compliance_checker.py . --json
    python gdpr_compliance_checker.py /path/to/project --output report.json
"""

import argparse
import json
import os
import re
import sys
from pathlib import Path
from typing import Dict, List, Optional, Tuple


# Personal data patterns to detect
PERSONAL_DATA_PATTERNS = {
    "email": {
        "pattern": r"[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}",
        "category": "contact_data",
        "gdpr_article": "Art. 4(1)",
        "risk": "medium"
    },
    "ip_address": {
        "pattern": r"\b(?:\d{1,3}\.){3}\d{1,3}\b",
        "category": "online_identifier",
        "gdpr_article": "Art. 4(1), Recital 30",
        "risk": "medium"
    },
    "phone_number": {
        "pattern": r"(?:\+\d{1,3}[-.\s]?)?\(?\d{3}\)?[-.\s]?\d{3}[-.\s]?\d{4}",
        "category": "contact_data",
        "gdpr_article": "Art. 4(1)",
        "risk": "medium"
    },
    "credit_card": {
        "pattern": r"\b(?:\d{4}[-\s]?){3}\d{4}\b",
        "category": "financial_data",
        "gdpr_article": "Art. 4(1)",
        "risk": "high"
    },
    "iban": {
        "pattern": r"\b[A-Z]{2}\d{2}[A-Z0-9]{4}\d{7}(?:[A-Z0-9]?){0,16}\b",
        "category": "financial_data",
        "gdpr_article": "Art. 4(1)",
        "risk": "high"
    },
    "german_id": {
        "pattern": r"\b[A-Z0-9]{9}\b",
        "category": "government_id",
        "gdpr_article": "Art. 4(1)",
        "risk": "high"
    },
    "date_of_birth": {
        "pattern": r"\b(?:birth|dob|geboren|geburtsdatum)\b",
        "category": "demographic_data",
        "gdpr_article": "Art. 4(1)",
        "risk": "medium"
    },
    "health_data": {
        "pattern": r"\b(?:diagnosis|treatment|medication|patient|medical|health|symptom|disease)\b",
        "category": "special_category",
        "gdpr_article": "Art. 9(1)",
        "risk": "critical"
    },
    "biometric": {
        "pattern": r"\b(?:fingerprint|facial|retina|biometric|voice_print)\b",
        "category": "special_category",
        "gdpr_article": "Art. 9(1)",
        "risk": "critical"
    },
    "religion": {
        "pattern": r"\b(?:religion|religious|faith|church|mosque|synagogue)\b",
        "category": "special_category",
        "gdpr_article": "Art. 9(1)",
        "risk": "critical"
    }
}

# Code patterns indicating GDPR concerns
CODE_PATTERNS = {
    "logging_personal_data": {
        "pattern": r"(?:log|print|console)\s*\.\s*(?:info|debug|warn|error)\s*\([^)]*(?:email|user|name|address|phone)",
        "issue": "Potential logging of personal data",
        "gdpr_article": "Art. 5(1)(c) - Data minimization",
        "recommendation": "Review logging to ensure personal data is not logged or is properly pseudonymized",
        "severity": "high"
    },
    "missing_consent": {
        "pattern": r"(?:track|analytics|marketing|cookie)(?!.*consent)",
        "issue": "Tracking without apparent consent mechanism",
        "gdpr_article": "Art. 6(1)(a) - Consent",
        "recommendation": "Implement consent management before tracking",
        "severity": "high"
    },
    "hardcoded_retention": {
        "pattern": r"(?:retention|expire|ttl|lifetime)\s*[=:]\s*(?:null|undefined|0|never|forever)",
        "issue": "Indefinite data retention detected",
        "gdpr_article": "Art. 5(1)(e) - Storage limitation",
        "recommendation": "Define and implement data retention periods",
        "severity": "medium"
    },
    "third_party_transfer": {
        "pattern": r"(?:api|http|fetch|request)\s*\.\s*(?:post|put|send)\s*\([^)]*(?:user|personal|data)",
        "issue": "Potential third-party data transfer",
        "gdpr_article": "Art. 28 - Processor requirements",
        "recommendation": "Ensure Data Processing Agreement exists with third parties",
        "severity": "medium"
    },
    "encryption_missing": {
        "pattern": r"(?:password|secret|token|key)\s*[=:]\s*['\"][^'\"]+['\"]",
        "issue": "Potentially unencrypted sensitive data",
        "gdpr_article": "Art. 32(1)(a) - Encryption",
        "recommendation": "Encrypt sensitive data at rest and in transit",
        "severity": "critical"
    },
    "no_deletion": {
        "pattern": r"(?:delete|remove|erase).*(?:disabled|false|TODO|FIXME)",
        "issue": "Data deletion may be disabled or incomplete",
        "gdpr_article": "Art. 17 - Right to erasure",
        "recommendation": "Implement complete data deletion functionality",
        "severity": "high"
    }
}

# Configuration files to check for GDPR-relevant settings
CONFIG_PATTERNS = {
    "analytics_config": {
        "files": ["analytics.json", "gtag.js", "google-analytics.js"],
        "check": "anonymize_ip",
        "issue": "IP anonymization should be enabled for analytics",
        "gdpr_article": "Art. 5(1)(c)"
    },
    "cookie_config": {
        "files": ["cookie.config.js", "cookies.json"],
        "check": "consent_required",
        "issue": "Cookie consent should be required before non-essential cookies",
        "gdpr_article": "Art. 6(1)(a)"
    }
}

# File extensions to scan
SCANNABLE_EXTENSIONS = {
    ".py", ".js", ".ts", ".jsx", ".tsx", ".java", ".kt",
    ".go", ".rb", ".php", ".cs", ".swift", ".json", ".yaml",
    ".yml", ".xml", ".html", ".env", ".config"
}

# Files/directories to skip
SKIP_PATTERNS = {
    "node_modules", "vendor", ".git", "__pycache__", "dist",
    "build", ".venv", "venv", "env"
}


def should_skip(path: Path) -> bool:
    """Check if path should be skipped."""
    return any(skip in path.parts for skip in SKIP_PATTERNS)


def scan_file_for_patterns(
    filepath: Path,
    patterns: Dict
) -> List[Dict]:
    """Scan a file for pattern matches."""
    findings = []

    try:
        with open(filepath, "r", encoding="utf-8", errors="ignore") as f:
            content = f.read()
            lines = content.split("\n")

        for pattern_name, pattern_info in patterns.items():
            regex = re.compile(pattern_info["pattern"], re.IGNORECASE)

            for line_num, line in enumerate(lines, 1):
                matches = regex.findall(line)
                if matches:
                    findings.append({
                        "file": str(filepath),
                        "line": line_num,
                        "pattern": pattern_name,
                        "matches": len(matches) if isinstance(matches, list) else 1,
                        **{k: v for k, v in pattern_info.items() if k != "pattern"}
                    })

    except Exception as e:
        pass  # Skip files that can't be read

    return findings


def analyze_project(project_path: Path) -> Dict:
    """Analyze project for GDPR compliance issues."""
    personal_data_findings = []
    code_issue_findings = []
    config_findings = []
    files_scanned = 0

    # Scan all relevant files
    for filepath in project_path.rglob("*"):
        if filepath.is_file() and not should_skip(filepath):
            if filepath.suffix.lower() in SCANNABLE_EXTENSIONS:
                files_scanned += 1

                # Check for personal data patterns
                personal_data_findings.extend(
                    scan_file_for_patterns(filepath, PERSONAL_DATA_PATTERNS)
                )

                # Check for code issues
                code_issue_findings.extend(
                    scan_file_for_patterns(filepath, CODE_PATTERNS)
                )

    # Check for specific config files
    for config_name, config_info in CONFIG_PATTERNS.items():
        for config_file in config_info["files"]:
            config_path = project_path / config_file
            if config_path.exists():
                try:
                    with open(config_path, "r") as f:
                        content = f.read()
                    if config_info["check"] not in content.lower():
                        config_findings.append({
                            "file": str(config_path),
                            "config": config_name,
                            "issue": config_info["issue"],
                            "gdpr_article": config_info["gdpr_article"]
                        })
                except Exception:
                    pass

    # Calculate risk scores
    critical_count = sum(1 for f in personal_data_findings if f.get("risk") == "critical")
    critical_count += sum(1 for f in code_issue_findings if f.get("severity") == "critical")

    high_count = sum(1 for f in personal_data_findings if f.get("risk") == "high")
    high_count += sum(1 for f in code_issue_findings if f.get("severity") == "high")

    medium_count = sum(1 for f in personal_data_findings if f.get("risk") == "medium")
    medium_count += sum(1 for f in code_issue_findings if f.get("severity") == "medium")

    # Determine compliance score (100 = compliant, 0 = critical issues)
    score = 100
    score -= critical_count * 20
    score -= high_count * 10
    score -= medium_count * 5
    score -= len(config_findings) * 5
    score = max(0, score)

    # Determine compliance status
    if score >= 80:
        status = "compliant"
        status_description = "Low risk - minor improvements recommended"
    elif score >= 60:
        status = "needs_attention"
        status_description = "Medium risk - action required"
    elif score >= 40:
        status = "non_compliant"
        status_description = "High risk - immediate action required"
    else:
        status = "critical"
        status_description = "Critical risk - significant GDPR violations detected"

    return {
        "summary": {
            "files_scanned": files_scanned,
            "compliance_score": score,
            "status": status,
            "status_description": status_description,
            "issue_counts": {
                "critical": critical_count,
                "high": high_count,
                "medium": medium_count,
                "config_issues": len(config_findings)
            }
        },
        "personal_data_findings": personal_data_findings[:50],  # Limit output
        "code_issues": code_issue_findings[:50],
        "config_issues": config_findings,
        "recommendations": generate_recommendations(
            personal_data_findings, code_issue_findings, config_findings
        )
    }


def generate_recommendations(
    personal_data: List[Dict],
    code_issues: List[Dict],
    config_issues: List[Dict]
) -> List[Dict]:
    """Generate prioritized recommendations."""
    recommendations = []
    seen_issues = set()

    # Critical issues first
    for finding in code_issues:
        if finding.get("severity") == "critical":
            issue_key = finding.get("issue", "")
            if issue_key not in seen_issues:
                recommendations.append({
                    "priority": "P0",
                    "issue": finding.get("issue"),
                    "gdpr_article": finding.get("gdpr_article"),
                    "action": finding.get("recommendation"),
                    "affected_files": [finding.get("file")]
                })
                seen_issues.add(issue_key)

    # Special category data
    special_category_files = set()
    for finding in personal_data:
        if finding.get("category") == "special_category":
            special_category_files.add(finding.get("file"))

    if special_category_files:
        recommendations.append({
            "priority": "P0",
            "issue": "Special category personal data (Art. 9) detected",
            "gdpr_article": "Art. 9(1)",
            "action": "Ensure explicit consent or other Art. 9(2) legal basis exists",
            "affected_files": list(special_category_files)[:5]
        })

    # High priority issues
    for finding in code_issues:
        if finding.get("severity") == "high":
            issue_key = finding.get("issue", "")
            if issue_key not in seen_issues:
                recommendations.append({
                    "priority": "P1",
                    "issue": finding.get("issue"),
                    "gdpr_article": finding.get("gdpr_article"),
                    "action": finding.get("recommendation"),
                    "affected_files": [finding.get("file")]
                })
                seen_issues.add(issue_key)

    # Config issues
    for finding in config_issues:
        recommendations.append({
            "priority": "P1",
            "issue": finding.get("issue"),
            "gdpr_article": finding.get("gdpr_article"),
            "action": f"Update configuration in {finding.get('file')}",
            "affected_files": [finding.get("file")]
        })

    return recommendations[:15]


def print_report(analysis: Dict) -> None:
    """Print human-readable report."""
    summary = analysis["summary"]

    print("=" * 60)
    print("GDPR COMPLIANCE ASSESSMENT REPORT")
    print("=" * 60)
    print()
    print(f"Compliance Score: {summary['compliance_score']}/100")
    print(f"Status: {summary['status'].upper()}")
    print(f"Assessment: {summary['status_description']}")
    print(f"Files Scanned: {summary['files_scanned']}")
    print()

    counts = summary["issue_counts"]
    print("--- ISSUE SUMMARY ---")
    print(f"  Critical: {counts['critical']}")
    print(f"  High: {counts['high']}")
    print(f"  Medium: {counts['medium']}")
    print(f"  Config Issues: {counts['config_issues']}")
    print()

    if analysis["recommendations"]:
        print("--- PRIORITIZED RECOMMENDATIONS ---")
        for i, rec in enumerate(analysis["recommendations"][:10], 1):
            print(f"\n{i}. [{rec['priority']}] {rec['issue']}")
            print(f"   GDPR Article: {rec['gdpr_article']}")
            print(f"   Action: {rec['action']}")

    print()
    print("=" * 60)
    print("Note: This is an automated assessment. Manual review by a")
    print("qualified Data Protection Officer is recommended.")
    print("=" * 60)


def main():
    parser = argparse.ArgumentParser(
        description="Scan project for GDPR compliance issues"
    )
    parser.add_argument(
        "project_path",
        nargs="?",
        default=".",
        help="Path to project directory (default: current directory)"
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="Output in JSON format"
    )
    parser.add_argument(
        "--output", "-o",
        help="Write output to file"
    )

    args = parser.parse_args()

    project_path = Path(args.project_path).resolve()
    if not project_path.exists():
        print(f"Error: Path does not exist: {project_path}", file=sys.stderr)
        sys.exit(1)

    analysis = analyze_project(project_path)

    if args.json:
        output = json.dumps(analysis, indent=2)
        if args.output:
            with open(args.output, "w") as f:
                f.write(output)
            print(f"Report written to {args.output}")
        else:
            print(output)
    else:
        print_report(analysis)
        if args.output:
            with open(args.output, "w") as f:
                json.dump(analysis, f, indent=2)
            print(f"\nDetailed JSON report written to {args.output}")


if __name__ == "__main__":
    main()