from transforms.api import transform_df, Input, Output
from pyspark.sql import functions as F

@transform_df(
    Output("/Company/datasets/customers_safe"),
    customers=Input("/Company/datasets/raw_customers"),
)
def redact_pii(customers):
    """Create an analytics-safe view with PII removed."""
    return (
        customers
        .withColumn("email", F.sha2(F.col("email"), 256))           # Hash email
        .withColumn("phone", F.lit("***-***-****"))                   # Mask phone
        .withColumn("ssn", F.lit(None).cast("string"))                # Remove SSN
        .withColumn("name", F.concat(
            F.substring("first_name", 1, 1), F.lit("***")            # First initial only
        ))
        .drop("first_name", "last_name", "address", "date_of_birth")
    )

def delete_user_data(client, user_id: str):
    """GDPR Article 17: delete all data for a specific user."""
    datasets_with_pii = [
        "/Company/datasets/raw_customers",
        "/Company/datasets/raw_orders",
        "/Company/datasets/customer_communications",
    ]
    for dataset_path in datasets_with_pii:
        # Trigger a transform that filters out the user
        client.ontologies.Action.apply(
            ontology="my-company",
            action_type="gdprDeleteUser",
            parameters={"userId": user_id, "datasetPath": dataset_path},
        )
    # Log the deletion for compliance
    client.ontologies.Action.apply(
        ontology="my-company",
        action_type="logDeletionRequest",
        parameters={
            "userId": user_id,
            "requestedAt": datetime.utcnow().isoformat(),
            "status": "completed",
        },
    )

# Define object type with restricted properties
# In Ontology Manager:
# - fullName: marking = CONFIDENTIAL
# - email: marking = CONFIDENTIAL  
# - department: marking = INTERNAL
# - employeeId: marking = INTERNAL

# Users without CONFIDENTIAL marking see:
# employeeId, department (but NOT fullName, email)

@transform_df(
    Output("/Company/datasets/events_retained"),
    events=Input("/Company/datasets/raw_events"),
)
def apply_retention(events):
    """Keep only last 2 years of data per retention policy."""
    from pyspark.sql import functions as F
    from datetime import datetime, timedelta

    cutoff = (datetime.utcnow() - timedelta(days=730)).strftime("%Y-%m-%d")
    return events.filter(F.col("event_date") >= cutoff)