# Datenimport aus Connectoren

> Daten eines angebundenen Systems kommen ohne eigenen Dienst in die Datenobjekte eines Pakets, mit Zuordnung, Vorschau, Feldhoheit und Abgleich.

Ein Paket braucht oft Daten, die schon woanders stehen: Kunden und Zeiten in awork, Liegenschaften im AMS. Dafür schreibt niemand mehr einen eigenen Importer. Der Connector sagt, was sich lesen lässt, das Paket sagt, wohin es gehört, und die Plattform führt es aus. Das läuft auf jeder Instanz, auch ohne Moduldienst.

## Für Administratoren

Unter **System › Module › Datenimport** steht jeder Connector, den ein installiertes Paket nennt. Dieselbe Karte steht auch auf der Seite des Connectors unter **KI einrichten › Connectoren** und führt von dort direkt zur Zuordnung; zurück geht es über *Connector-Einstellungen*.

- **Übersicht:** ein Block je Importer: was wohin geht, in der Reihenfolge eines Laufs, dazu Import starten, jetzt abgleichen und sein automatischer Abgleich. Ein Connector kann mehrere Importer haben, etwa Stammdaten einmal am Tag und Zeiten jede Viertelstunde. Jeder hat seinen eigenen Takt und seine eigene Historie; läuft einer, wartet der nächste.
- **Zuordnung:** oben die Importer mit ihren Zuordnungen; dort legen Sie einen Importer an, benennen ihn um oder verschieben eine Zuordnung in einen anderen. Darunter links die Felder der Quelle mit echten Werten, rechts die Felder des Datenobjekts. Ziehen Sie vom Punkt auf ein Zielfeld, oder klicken Sie erst die Quelle und dann das Ziel. *Vorschlagen* ordnet nach Namen und Werten zu, *Mit KI vorschlagen* ergänzt das; die KI sieht dafür Feldnamen und kurze Wertelisten wie einen Status, keine Inhalte von Datensätzen. Die Vorschau zeigt für jede Beispielzeile, was geschrieben würde, geschrieben wird dabei nichts.
- **Läufe:** was jeder Lauf getan hat. Ein abgebrochener Lauf lässt sich fortsetzen, ein Import zurücknehmen; dann verschwinden die Datensätze, die er angelegt hat.
- **Offene Punkte:** Zeilen, die nicht übernommen werden konnten, und Felder, die auf beiden Seiten geändert wurden. Zeilen mit derselben Ursache stehen zusammen. Lösen Sie die Ursache, gehen sie beim nächsten Lauf von selbst weg.

Jede gespeicherte Zuordnung ist eine Version. Ein laufender Import behält die Zuordnung, mit der er begonnen hat.

### Wem ein Feld gehört

| Einstellung | Bedeutung |
|---|---|
| Quelle gewinnt | Jeder Lauf schreibt den Wert der Quelle. |
| Geschützt | Die Quelle schreibt, solange hier niemand geändert hat. Ändern beide, entscheiden Sie unter Offene Punkte. |
| Nur wenn leer | Geschrieben wird nur in ein leeres Feld. |
| Nur beim Anlegen | Danach gehört das Feld Ihnen. |

### Was ein Import nicht tut

- Er löscht nichts, weil eine Antwort leer ist. Verschwindet auffällig viel auf einmal, entscheiden Sie.
- Er holt keinen Datensatz zurück, den jemand hier gelöscht hat.
- Er legt nichts doppelt an: Ein zweiter Lauf ändert nur, was sich geändert hat.

Der automatische Abgleich läuft erst nach einem ersten Import und braucht den Hintergrunddienst `MODULE_CONNECTOR_SYNC`, der pro Instanz eingeschaltet wird.

## Für Modulentwickler

Der Connector beschreibt in `plugin.yml` oder im Connector-Eintrag des Pakets, was ein Lauf nicht aus den Zeilen ablesen kann. Jede lesende Aktion ohne Pflicht-Pfad zählt auch ohne Beschreibung als Datenstrom.

```yaml
streams:
  - key: projects
    action: projects
    incremental: { arg: filterby, template: "updatedOn ge datetime'{since}'" }
    enrich:
      - { as: members, action: project_members, args: { projectId: id } }
```

Das Paket ordnet zu, in `connectors[].imports` seiner `module.json`, und sagt, welcher Importer was ausführt. Ohne `importers` läuft alles in einem Importer:

```json
{
  "key": "awork",
  "catalog": true,
  "sync": { "everyMinutes": 15 },
  "imports": {
    "importers": [
      { "key": "stamm", "name": "Stammdaten", "everyMinutes": 1440 },
      { "key": "zeiten", "name": "Zeiten", "everyMinutes": 15, "after": { "service": "lagedienst", "operation": "awork" } }
    ],
    "mappings": [{
      "key": "projekte",
      "importer": "stamm",
      "stream": "projects",
      "entity": "projekt",
      "match": [{ "field": "awork_id", "expr": "id" }],
      "fields": [
        { "target": "awork_id", "expr": "id" },
        { "target": "titel", "expr": "coalesce(trim(name), id)" },
        { "target": "kunde", "expr": "companyId", "ref": "kunden", "refElse": "ohne-kunde" },
        { "target": "status", "expr": "projectStatus.type", "map": [{ "from": "closed", "to": "abgeschlossen" }], "default": "laufend" },
        { "target": "art", "expr": "count(retainers) > 0 ? 'retainer' : 'aufwand'", "policy": "create" }
      ]
    }]
  }
}
```

Ein Ausdruck ist ein Pfad (`company.name`, `teams[].name`) oder eine Formel mit festen Funktionen wie `round(duration / 3600, 2)`, `weekStart(date)` oder `coalesce(a, b)`. `derive` legt berechnete Spalten an, `group` fasst Zeilen zusammen (`sum`, `all`, `first`), `window` liest in Zeiträumen, `seed` schreibt Datensätze, für die die Quelle keine Zeile hat, und `missing: "delete"` räumt auf, was die Quelle nicht mehr nennt. `after` ruft einmal am Ende jedes Laufs dieses Importers eine Operation Ihres Dienstes auf, denn während des Laufs bekommt er keine Ereignisse zu einzelnen Datensätzen. Ein Verweis darf auf eine Zuordnung eines anderen Importers zeigen.

Manche Aktionen antworten nur für einen Datensatz auf einmal, etwa die Entnahmestellen einer Liegenschaft. Dann liest `each` den Datenstrom einmal je Zeile eines anderen:

```yaml
streams:
  - key: standorte
    action: location_search
  - key: entnahmestellen
    action: sampling_point_search
    each:
      stream: standorte
      args: { locationId: "number(_id)" }
      as: standort
      filter: "_source.samplingPointsCount > 0"
```

Jede gelesene Zeile trägt ihre Ausgangszeile unter `as`, eine Zuordnung liest also zum Beispiel `standort._id`. Das geht nur eine Ebene tief. Ein Lauf liest den Ausgangsstrom am Anfang ganz (höchstens 10.000 Zeilen), ein solcher Datenstrom hat kein `incremental`, und `missing: "delete"` räumt erst auf, wenn alle Ausgangszeilen gelesen sind. Die Vorschau im Editor liest nur eine Ausgangszeile und sagt das dazu. Blättert eine Aktion anders als der Rest ihres Connectors, bekommt sie ihr eigenes `pagination`; `{ type: none }` schaltet das Blättern für sie ab. Für Texte und Zeitpunkte gibt es außerdem `split(text, trenner)`, `now()` und `fromEpoch(ms)` für Zeitstempel in Millisekunden.

Die Regeln stehen im Paketvertrag, Abschnitt 2.14. Prüfen Sie eine Zuordnung mit `nxm validate`, Fehler in Ausdrücken fallen beim Hochladen auf, nicht erst im Lauf.