# CSV, XLSX and Parquet upload

Source module: `app/connectors/generic_csv.py`. Kind: **both**. Display name: "CSV / XLSX / Parquet upload". Auth: **upload** (no external system).

## Purpose

Ingests any vendor's claims export and any facts export that has no built-in connector, using a saved column mapping. This is the path for Decagon, Sierra, Salesforce Agentforce usage, Zendesk automated-resolution exports, HubSpot Breeze resolutions, and for supplying conversation text to transcript assist.

## Auth and scopes

None. Files are uploaded by an `analyst` or higher through `POST /v1/connections/upload` (multipart: `file`, `kind` = `claims` | `facts`, `mapping` JSON, optional `save_mapping_as`, optional `connection_id` to reuse an upload connection). `POST /v1/connections/upload/preview` returns headers, the first 10 rows, the row count and a suggested mapping. Uploads are limited to 50 MB. Claims uploads are checked against the plan's `outcomes_per_month` limit before ingestion.

## Supported formats

| Extension | Parser |
|---|---|
| `.csv` (any delimiter among `, ; tab |`, sniffed; UTF-8 with or without BOM) | Python `csv` |
| `.xlsx`, `.xlsm` | `openpyxl` (active sheet, first non-empty row = headers) |
| `.parquet` | DuckDB `read_parquet` |

Empty rows are dropped.

## Mapping

`mapping.columns` maps canonical field → file header. Other mapping keys set defaults.

**Claims** (`kind = claims`):

| Canonical column | Required | Notes |
|---|---|---|
| `subject_id` | yes | Verbatim; must match the fact subjects it should be evaluated against (e.g. `zendesk_ticket_123`) or be linkable |
| `claimed_at` | yes | ISO-8601 (with `Z`, an offset, or a space instead of `T`), `YYYY-MM-DD`, `MM/DD/YYYY [HH:MM[:SS]]`, `DD/MM/YYYY`, `YYYY/MM/DD [HH:MM:SS]`, or epoch seconds/milliseconds; normalized to UTC. Naive values are treated as UTC. An unrecognized value aborts the upload with `unrecognized timestamp: …` |
| `claim_id` | no | If present, the stored id is `sha256(tenant|value)`-derived; otherwise derived from file name + row |
| `outcome_type` | no | Default from `mapping.outcome_type` or `resolution` |
| `unit_price`, `currency` | no | Defaults `mapping.unit_price` / `0`, `USD` |
| `channel` | no | For `channel_in` exclusions |
| `billing_period` | no | `YYYY-MM`; defaults to the month of `claimed_at` |
| `customer_email` | no | HMAC-hashed into `customer_hid`; never stored raw |

Mapping-level defaults: `contract_id` (default `generic_outcome_csv`), `vendor` (default `generic`), `subject_kind` (default `conversation`), `outcome_type`, `unit_price`.

**Facts** (`kind = facts`):

| Canonical column | Required | Notes |
|---|---|---|
| `subject_id` | yes | Verbatim |
| `occurred_at` | yes | Same formats as above |
| `fact_type` | yes (or `mapping.fact_type` default) | e.g. `customer_message`, `human_reply`, `status_change`, `reopen`, `escalation`, `csat_response`, `customer_confirmation`, `refund`, `credit`, `opportunity_stage`, `meeting_event` |
| `subject_kind` | no | Default `mapping.subject_kind` or `conversation` |
| `from`, `to`, `actor` | no | Status change attributes (lower-cased) |
| `stage`, `status` | no | Opportunity/meeting attributes (lower-cased) |
| `score`, `sentiment` | no | CSAT; if `sentiment` is absent it is derived from `score` (≥4 positive, ≤2 negative, else neutral) |
| `amount` | no | Refund/credit amount |
| `intent` | no | For `customer_message` (`confirmed_resolved` satisfies `customer_confirmed_resolved`) |
| `text` | no | Stored **only** when `mapping.keep_text = true`; required for transcript assist |
| `customer_email` | no | HMAC-hashed into `customer_hid` |

Mapping-level: `source` (stored as the fact `source`, default `generic_csv`), `provides` (list of source categories this upload supplies, e.g. `["conversations", "csat"]`; merged into the connection's `config.provides` and used for coverage disclosure and `missing_fact_source`).

Suggested mappings are produced from common header names (`conversation_id`, `ticket_id`, `resolved_at`, `rating`, `body`, …); see `SUGGEST` in the module.

## What is hashed / never stored

- `customer_email` → `customer_hid` (HMAC-SHA256, per-tenant key). The raw email is not stored in claims or facts.
- Unmapped columns are ignored and never persisted in `claims`/`facts`.
- The uploaded file itself is landed in the raw landing zone (below) and deleted after `RAW_LANDING_RETENTION_DAYS` (90).
- `text` is stored only with `keep_text = true`, is excluded from the evidence API, is redacted before any transcript-assist call, and excerpts derived from it are purged after 30 days.

## Raw landing, incremental behaviour, watermark

- Every upload is written to `tenants/{tenant_id}/generic_csv/{YYYY-MM-DD}/{filename}` before parsing.
- There is no incremental sync and no watermark: `sync()` is a no-op and `POST /v1/connections/{id}/sync` returns `not_syncable`. Each upload is processed in full. Because ids derive from `sha256(tenant_id | csv:{filename}:{claim_id or row})`, re-uploading the **same file name** updates the same rows; uploading the same data under a different file name creates new rows unless `claim_id`/`fact_id` columns are mapped (then ids derive from those values and stay stable across file names).
- A `sync_runs` row is recorded per upload with row counts; `connection.upload` is audited with the file name and counts.

## Rate-limit handling

Not applicable.

## Fact and claim types produced

Whatever the mapping declares. Claims always get `raw_ref = csv:{filename}:{claim_id or row}`; facts `raw_ref = csv:{filename}:{fact_id or row}`.

## Subject id format

Whatever the file contains, verbatim. To be evaluated against facts from another connector, use that connector's format (`intercom_conv_{id}`, `zendesk_ticket_{id}`, `sf_case_{Id}`, `hubspot_ticket_{id}`, `stripe_customer_{id}`), or supply `customer_email` on both sides so hashed-email linking can connect them.

## Known limitations

- 50 MB per upload; the file is parsed in memory.
- Parquet parsing writes a temporary file on the host.
- Only the active worksheet of an XLSX is read.
- Time zones: naive timestamps are treated as UTC.
- Rows with an **empty** `subject_id`, `claimed_at`/`occurred_at` (or, for facts, `fact_type`) are skipped silently. A **non-empty but unrecognized** timestamp raises `unrecognized timestamp: …` and aborts the whole upload; fix the column or the mapping and re-upload (the file lands in the raw zone before parsing, so nothing partial is persisted).
- No de-duplication across differently named files unless `claim_id`/`fact_id` columns are mapped.