breakpilot-compliance/backend-compliance/compliance/services/vendor_vvt_mapper.py

"""
Vendor VVT Mapper — map TCF vendors to VVT entries.

Converts resolved TCF vendor data (from GVL) into the format
needed for the Verarbeitungsverzeichnis (VVT) and for DSI
cross-checking.
"""

import logging

logger = logging.getLogger(__name__)

# IAB TCF v2.2 Purpose definitions (German)
TCF_PURPOSE_LABELS = {
    1: "Speicherung/Zugriff auf Endgeraet",
    2: "Auswahl einfacher Anzeigen",
    3: "Personalisiertes Anzeigenprofil erstellen",
    4: "Personalisierte Anzeigen auswaehlen",
    5: "Personalisiertes Inhaltsprofil erstellen",
    6: "Personalisierte Inhalte auswaehlen",
    7: "Anzeigenleistung messen",
    8: "Inhaltsleistung messen",
    9: "Marktforschung",
    10: "Produkte entwickeln und verbessern",
    11: "Geraeteeigenschaften zur Identifizierung nutzen",
}

# Purpose → Banner-Kategorie Mapping
PURPOSE_CATEGORY = {
    1: "necessary",
    2: "marketing", 3: "marketing", 4: "marketing",
    5: "marketing", 6: "marketing",
    7: "statistics", 8: "statistics",
    9: "statistics", 10: "functional", 11: "functional",
}

# EWR countries
_EU_EWR = {
    "AT", "BE", "BG", "HR", "CY", "CZ", "DK", "EE", "FI", "FR",
    "DE", "GR", "HU", "IE", "IT", "LV", "LT", "LU", "MT", "NL",
    "PL", "PT", "RO", "SK", "SI", "ES", "SE", "IS", "LI", "NO",
    "CH", "GB",
}


def tcf_vendor_to_vvt(vendor: dict) -> dict:
    """Map a resolved TCF vendor to a VVT entry.

    Args:
        vendor: Resolved GVL vendor dict with name, purposes, country, etc.

    Returns:
        VVT-compatible dict with name, zweck, rechtsgrundlage, drittland, etc.
    """
    purposes = vendor.get("purposes", [])
    country = vendor.get("country")
    is_eu = vendor.get("is_eu", country in _EU_EWR if country else None)

    # Determine primary category from purposes
    categories = set()
    for p in purposes:
        cat = PURPOSE_CATEGORY.get(p, "functional")
        categories.add(cat)

    # Rechtsgrundlage depends on category
    if "marketing" in categories or "statistics" in categories:
        rechtsgrundlage = "Einwilligung (Art. 6(1)(a) DSGVO, §25 Abs. 1 TDDDG)"
    else:
        rechtsgrundlage = "Berechtigtes Interesse (Art. 6(1)(f) DSGVO, §25 Abs. 2 TDDDG)"

    return {
        "vendor_id": vendor.get("vendor_id"),
        "name": vendor.get("name", ""),
        "zweck": [TCF_PURPOSE_LABELS.get(p, f"Zweck {p}") for p in purposes],
        "zweck_kurz": _summarize_purposes(purposes),
        "kategorie": sorted(categories)[0] if categories else "functional",
        "rechtsgrundlage": rechtsgrundlage,
        "drittland": not is_eu if is_eu is not None else None,
        "land": country,
        "transfermechanismus": "SCC/DPF" if (not is_eu and is_eu is not None) else None,
        "speicherdauer_tage": vendor.get("retention_days"),
        "policy_url": vendor.get("policy_url", ""),
        "uses_cookies": vendor.get("uses_cookies", False),
    }


def map_vendors_to_vvt(vendors: list[dict]) -> list[dict]:
    """Map a list of TCF vendors to VVT entries."""
    return [tcf_vendor_to_vvt(v) for v in vendors]


def _summarize_purposes(purposes: list[int]) -> str:
    """Short German summary of purposes."""
    if not purposes:
        return "Keine Zwecke angegeben"
    cats = set(PURPOSE_CATEGORY.get(p, "sonstig") for p in purposes)
    labels = {
        "marketing": "Marketing/Werbung",
        "statistics": "Analyse/Messung",
        "functional": "Funktional",
        "necessary": "Technisch notwendig",
        "sonstig": "Sonstige",
    }
    return ", ".join(labels.get(c, c) for c in sorted(cats))