# Rapportage use cases

> Bouw de rapporten die compliance teams nodig hebben met de Tidal Control API. Maatregeleffectiviteit per periode, verlopen maatregelen en gecombineerde status.

Source: https://tidalcontrol.com/nl/docs/api/reporting

Deze voorbeelden laten zien hoe je de operationele rapporten bouwt die compliance teams vaak nodig hebben: maatregeleffectiviteit per periode, overzichten van verlopen taken, en gecombineerde statusrapporten.

> **info:**

De API ondersteunt geen server-side datumfilters. Alle periodefiltering (bijv. "alleen februari", "Q1") doe je client-side door te filteren op het `closedDate`-veld nadat je de data hebt opgehaald.

> **tip:**

De scripts op deze pagina gebruiken `get_token()` van de pagina [Authenticatie](/docs/api/authentication). Kopieer die functie naar je script, of importeer hem vanuit een gedeelde module.

## Maatregeleffectiviteitsrapport (per periode)

Beoordelingen zijn de bron voor effectiviteit. Elke beoordeling heeft een `closedDate` (wanneer de taak is afgerond) en een `effectiveness`-waarde (`EFFECTIVE`, `INEFFECTIVE` of `UNDETERMINED`).

Om te rapporteren over een specifieke maand of kwartaal, haal je alle beoordelingen op en filter je op `closedDate` in je script.

```python

from datetime import date

GRAPHQL_URL = "https://portal.tidalcontrol.com/graphql"

QUERY = """
query ExportAssessments($first: Int, $after: String, $filter: ActivityFilter) {
  activities_paged(first: $first, after: $after, filter: $filter) {
    edges {
      node {
        id
        sequenceId
        name
        closedDate
        createdDate
        controls { id sequenceId name category }
        assets { id name }
        assignments { assignmentType user { name email } }
        ... on Assessment {
          effectiveness
        }
      }
    }
    pageInfo { hasNextPage endCursor }
  }
}
"""

def graphql(query, variables, access_token):
    r = requests.post(
        GRAPHQL_URL,
        json={"query": query, "variables": variables},
        headers={"Authorization": f"Bearer {access_token}"},
    )
    r.raise_for_status()
    result = r.json()
    if "errors" in result:
        raise RuntimeError(result["errors"])
    return result["data"]

def fetch_all_assessments(access_token):
    results = []
    cursor = None
    while True:
        data = graphql(QUERY, {"first": 50, "after": cursor, "filter": {"activityTypes": ["ASSESS"]}}, access_token)
        page = data["activities_paged"]
        results.extend(edge["node"] for edge in page["edges"])
        if not page["pageInfo"]["hasNextPage"]:
            break
        cursor = page["pageInfo"]["endCursor"]
    return results

def filter_by_period(assessments, start: date, end: date):
    """Geeft alleen beoordelingen terug die zijn afgesloten binnen [start, end] (inclusief)."""
    filtered = []
    for a in assessments:
        closed = a.get("closedDate")
        if not closed:
            continue
        closed_date = date.fromisoformat(closed[:10])
        if start <= closed_date <= end:
            filtered.append(a)
    return filtered

def to_csv(assessments, output_path):
    if not assessments:
        print("Geen beoordelingen gevonden voor deze periode.")
        return

    with open(output_path, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow([
            "Sequence ID", "Naam", "Effectiviteit",
            "Afgesloten op", "Aangemaakt op",
            "Maatregel", "Maatregelcategorie",
            "Middelen", "Uitvoerder",
        ])
        for a in assessments:
            controls = a.get("controls", [])
            control_names = ", ".join(c["name"] for c in controls)
            control_categories = ", ".join(c.get("category", "") for c in controls)
            assets = ", ".join(x["name"] for x in a.get("assets", []))
            executors = [
                x["user"]["email"]
                for x in a.get("assignments", [])
                if x["assignmentType"] == "EXECUTOR"
            ]
            writer.writerow([
                a["sequenceId"],
                a["name"],
                a["effectiveness"],
                a.get("closedDate", "")[:10] if a.get("closedDate") else "",
                a.get("createdDate", "")[:10] if a.get("createdDate") else "",
                control_names,
                control_categories,
                assets,
                ", ".join(executors),
            ])

    print(f"{len(assessments)} beoordelingen geëxporteerd naar {output_path}")

if __name__ == "__main__":
    token = get_token()
    access_token = token["access_token"]

    all_assessments = fetch_all_assessments(access_token)

    # Februari 2025
    februari = filter_by_period(all_assessments, date(2025, 2, 1), date(2025, 2, 28))
    to_csv(februari, "effectiviteit_februari_2025.csv")

    # Q1 2025
    q1 = filter_by_period(all_assessments, date(2025, 1, 1), date(2025, 3, 31))
    to_csv(q1, "effectiviteit_q1_2025.csv")
```

## Rapport van verlopen maatregelen

Haal alle taken op die momenteel verlopen zijn, inclusief de bijbehorende maatregelen.

```python
OVERDUE_QUERY = """
query OverdueTasks($first: Int, $after: String, $filter: ActivityFilter) {
  activities_paged(first: $first, after: $after, filter: $filter) {
    edges {
      node {
        sequenceId
        name
        expires
        controls { sequenceId name category }
        assets { name }
        assignments { assignmentType user { name email } }
      }
    }
    pageInfo { hasNextPage endCursor }
  }
}
"""

def fetch_overdue_tasks(access_token):
    results = []
    cursor = None
    while True:
        data = graphql(
            OVERDUE_QUERY,
            {"first": 50, "after": cursor, "filter": {"activityTypes": ["EXECUTE"], "status": ["OVERDUE"]}},
            access_token,
        )
        page = data["activities_paged"]
        results.extend(edge["node"] for edge in page["edges"])
        if not page["pageInfo"]["hasNextPage"]:
            break
        cursor = page["pageInfo"]["endCursor"]
    return results
```

## Gecombineerd statusrapport

Dit voorbeeld combineert verlopen taken en effectiviteitsresultaten in één overzicht per maatregel.

```python
from collections import defaultdict

def build_control_summary(overdue_tasks, assessments):
    """
    Geeft een dict terug per maatregel (sequenceId) met:
      - overdue_tasks: aantal momenteel verlopen taken
      - effective: aantal EFFECTIVE beoordelingen
      - ineffective: aantal INEFFECTIVE beoordelingen
      - undetermined: aantal UNDETERMINED beoordelingen
    """
    summary = defaultdict(lambda: {
        "name": "",
        "category": "",
        "overdue_tasks": 0,
        "effective": 0,
        "ineffective": 0,
        "undetermined": 0,
    })

    for task in overdue_tasks:
        for control in task.get("controls", []):
            cid = control["sequenceId"]
            summary[cid]["name"] = control["name"]
            summary[cid]["category"] = control.get("category", "")
            summary[cid]["overdue_tasks"] += 1

    for a in assessments:
        for control in a.get("controls", []):
            cid = control["sequenceId"]
            summary[cid]["name"] = control["name"]
            summary[cid]["category"] = control.get("category", "")
            eff = a["effectiveness"].lower()
            if eff in ("effective", "ineffective", "undetermined"):
                summary[cid][eff] += 1

    return summary

def summary_to_csv(summary, output_path):
    with open(output_path, "w", newline="", encoding="utf-8") as f:
        writer = csv.writer(f)
        writer.writerow([
            "Maatregel-ID", "Naam", "Categorie",
            "Verlopen taken", "Effectief", "Ineffectief", "Onbepaald",
        ])
        for cid, row in sorted(summary.items()):
            writer.writerow([
                cid, row["name"], row["category"],
                row["overdue_tasks"], row["effective"],
                row["ineffective"], row["undetermined"],
            ])
    print(f"Overzicht geschreven naar {output_path}")

if __name__ == "__main__":
    token = get_token()
    access_token = token["access_token"]

    overdue = fetch_overdue_tasks(access_token)
    all_assessments = fetch_all_assessments(access_token)
    q1 = filter_by_period(all_assessments, date(2025, 1, 1), date(2025, 3, 31))

    summary = build_control_summary(overdue, q1)
    summary_to_csv(summary, "control_overzicht_q1_2025.csv")
```

## Filteren op afdeling

Tidal Control heeft geen native "afdeling"- of "business unit"-veld op maatregelen of taken. Er zijn twee gangbare manieren om dit te modelleren:

**Via middelen**: Als jouw organisatie maatregelen koppelt aan middelen die afdelingen of business units vertegenwoordigen, filter je na het ophalen op de naam van het middel:

```python
afdeling = "Finance"
afdeling_controls = [
    c for c in controls
    if any(a["name"] == afdeling for a in c.get("assets", []))
]
```

**Via maatregelattributen**: Als jouw organisatie custom attributen gebruikt om maatregelen van een afdeling te voorzien, filter je op het `attributes`-veld:

```python
afdeling = "Finance"
afdeling_controls = [
    c for c in controls
    if any(
        attr["key"] == "department" and attr["value"] == afdeling
        for attr in c.get("attributes", [])
    )
]
```

> **tip:**

Weet je niet hoe afdelingsinformatie in jouw Tidal-omgeving is ingericht? Neem contact op via support@tidalcontrol.com.
