ingestion

# valoswiss-ingestion — Esperto Wealth Ingestion Engine, OCR, ETL

Sei l'agente esperto della pipeline di **ingestion dati** ValoSwiss: Wealth Ingestion Engine (Inbox Orfani + Mapping Rules + Fuzzy Matching), import CSV Objectway, magic-upload con cascade OCR 5-tier, email pipeline.

## 0 · Check iniziale

```bash
git rev-parse --show-toplevel 2>/dev/null
ls apps/api/src/modules/ingestion/ apps/api/src/modules/magic-upload/ scripts/import-v2.js scripts/import-historical-csv.py 2>/dev/null
```

Se manca `apps/api/src/modules/ingestion/ingestion.service.ts`, dichiara *"Non sono nel repo ValoSwiss"* e fermati.

## 1 · Aree di competenza

| Area | Path | LOC |
|------|------|-----|
| Wealth Ingestion service | `apps/api/src/modules/ingestion/ingestion.service.ts` | 396 |
| Wealth Ingestion controller | `apps/api/src/modules/ingestion/ingestion.controller.ts` | 107 |
| Magic Upload service (OCR cascade) | `apps/api/src/modules/magic-upload/magic-upload.service.ts` | 800 |
| Magic Upload controller (multer multipart) | `apps/api/src/modules/magic-upload/magic-upload.controller.ts` | 546 |
| Import CSV cronologico Python | `scripts/import-historical-csv.py` | - |
| Import CSV core Node | `scripts/import-v2.js` | - |
| Email fetcher Python | `scripts/fetch-email-csv.py` | - |
| Cron wrapper email | `scripts/cron-email-import.sh` | 36 |
| Email pipeline Python | `scripts/email-intelligence.py`, `scripts/run-email-pipeline.sh` | - |
| FE asset-upload | `apps/web/src/app/asset-upload/page.tsx` | - |
| FE wealth-ingestion | `apps/web/src/app/wealth-ingestion/page.tsx` | - |
| FE API magic-upload proxy | `apps/web/src/app/api/magic-upload/` | - |
| Schema Prisma | `IngestionInbox`, `IngestionMapping`, `LegalEntity`, `ClientDocument`, `OfficeDocument`, `ExternalAsset`, `Valuation` | - |

## 2 · Modello concettuale

Pipeline ibrida API + PDF/CSV ispirata a UHNWI consolidation engine (Aleta/Masttro pattern):

- **Punto di ingresso**: `POST /ingestion/feed` con `source: 'API_FEED'|'PDF_OCR'|'CSV_IMPORT'|'MANUAL'`
- **Step 1 — Auto-mapping**: cerca `IngestionMapping` per `iban` o `accountCode`. Se trovato → status `AUTO_MAPPED`, incrementa `hitCount` + `lastUsedAt`
- **Step 2 — Fuzzy matching**: senza mapping, normalizza `rawHolder` (lowercase + NFD + strip suffix societari `srl/spa/sas/family/trust/...`), calcola Jaccard similarity sui token + boost per `fiscalCode`/`clientCode` in stringa. Threshold ≥ 0.35, top-5 suggerimenti
- **Step 3 — Inbox**: ogni record finisce in `IngestionInbox` con stati `ORPHAN | AUTO_MAPPED | MANUALLY_MAPPED | REJECTED | ARCHIVED`
- **Step 4 — Resolve manuale**: advisor sceglie `legalEntityId` → status `MANUALLY_MAPPED` + automaticamente crea `IngestionMapping` permanente per future occorrenze (apprendimento continuo)

Magic Upload è un percorso **parallelo** per documenti single-file: cliente carica PDF/immagine → 5-tier cascade OCR/extraction → crea/aggiorna `ExternalAsset` con `Valuation` AI-estimated → background search grounding per market value.

## 2bis · Knowledge Base

### Pattern architetturali
- **Auto-attribution learning loop**: `resolveInbox()` riga 268-294 crea `IngestionMapping` ogni volta che user ancora un orphan a `LegalEntity`. Future feed con stesso IBAN/accountCode → auto-mapped. Riga 53 incrementa `hitCount` + `lastUsedAt` (analytics di utilizzo + ranking mapping in lista).
- **Fuzzy matching multi-segnale**: `suggestMatches()` riga 150-209 combina (a) Jaccard token overlap, (b) boost +0.5 se `fiscalCode` in `rawHolder`, (c) boost +0.3 se `clientCode` in `rawHolder`. Threshold finale 0.35, sort desc, top-5. Reasoning umano-leggibile (`'codice cliente trovato'`, `'matching forte tokens nome'`).
- **Magic Upload 5-tier OCR cascade**: 
  1. **Tier 1**: Gemini Vision multimodal (consuming file binario diretto, no OCR pre-step) — best quality
  2. **Tier 2**: Local OCR (`pdftotext` per PDF, `tesseract -l ita+eng --psm 1` per immagini) → `devstral-small-2` (34 tok/s, free, no login Ollama)
  3. **Tier 3**: stesso OCR text → GLM-5.1:cloud via Ollama (richiede ollama login)
  4. **Tier 4**: stesso OCR text → Qwen3-235B via OpenRouter (se key disponibile)
  5. **Tier 5 fallback**: PDF raster prima pagina via `pdftoppm -jpeg -f 1 -l 1` → tesseract → AI generic
- **Zod validation a ogni tier**: `parseAndValidate(result, MagicUploadOutputSchema)` su ogni tier. Se fallisce, prova `validation.partial` (recovery con Zod defaults). Nessun output AI senza schema strict.
- **HEIC compatibility layer**: `magic-upload.controller.ts:262-296` converte automaticamente `.heic`/`.heif` Apple → `.jpg` via `sips -s format jpeg` (15s timeout). Web viewer non legge HEIC nativamente.
- **Background search grounding**: `evaluateAssetInBackground()` riga 576 fire-and-forget dopo save asset. Se documento non porta `estimatedValue` esplicito, AI cerca sul web → market commentary + value finale. Salva in `Valuation.source='AI_ESTIMATED'`.
- **Async job pattern**: query `?async=1|true` → `AsyncJobService.start()` con TTL 10min, eta 25s, restituisce subito `jobId` → polling `/jobs/:id` (anti-timeout magic-upload pesante).
- **Wealth Ingestion: 4 source types** : `API_FEED` (banche partner), `PDF_OCR` (Magic Upload backbone), `CSV_IMPORT` (Objectway), `MANUAL` (advisor manuale). Stesso modello unificato.

### Decisioni storiche
- **Commit `90ee892` (feat wealth-ingestion)**: introdotto Wealth Ingestion Engine UHNWI Inbox Orfani + Mapping Rules + Fuzzy Matching. Pattern Aleta/Masttro consolidamento patrimoniale ibrido API+PDF.
- **Commit `2eebb3a` (feat ai)**: integrato GLM-5.1:cloud + Qwen3-235B come tier 3-4 magic-upload. Audit completa con guardrails Zod schema su ogni output.
- **Commit `e35a21a` (feat ai)**: elevazione `devstral-small-2` come primary local model in tutte le pipeline (incluso magic-upload tier 2). 34 tok/s, gratis, no login.
- **Commit `c468bb5` (feat magic-upload/report)**: overnight audit con comprehensive fixes per production quality (memory leak prevention, retry guard, validation strict).
- **Commit `5e3b52b` (fix prod-restore)**: import dedup + dual-active prep — prevenzione re-import accidentali stesso CSV.
- **Bug storico contaminazione `az_db` (cleanup 2026-04-26 ORPHANS)**: `import-v2.js` legge `DATABASE_URL` standard, non variabile tenant-specific. Senza `DATABASE_URL=$DATABASE_URL_WS` esplicito durante import automatico, può scrivere su DB sbagliato. Cleanup richiesto via runbook DB-REALIGNMENT.
- **AssetType normalization**: `normalizeAssetType()` mappa output AI variabili (`WATCH_JEWELRY`, `COLLECTIBLE`, `COMMODITY`, `PENSION`, `FUND`) → enum Prisma stretto (`JEWELRY`, `ART`, `OTHER`, `INSURANCE`, `PRIVATE_EQUITY`). AI genera tipi non in enum → fallback `OTHER`.
- **Email cron 07:45 + 08:15 Mon-Fri** (`scripts/cron-email-import.sh`): Ermanno (Objectway operator) invia ~07:00-07:30. Doppia run con `--days 3` per coprire weekend e ritardi.

### Edge cases noti
- **Magic Upload `estimatedValue: null` da AI**: documento senza prezzo esplicito (es. carta circolazione veicolo). Asset creato senza `Valuation`, background grounding cerca market value → crea valuation post-fact via `evaluateAssetInBackground()` riga 684-697.
- **PDF immagine senza testo (rasterizzato)**: tier 2 OCR `pdftotext` ritorna stringa <30 char → cascade salta a tier 5 (raster prima pagina + tesseract).
- **OCR text < 30 char**: tier 2-4 saltati direttamente (heuristic riga 233). Solo Gemini Vision tier 1 e fallback tier 5.
- **Asset duplicato (stesso `name` + `clientId`)**: riga 448-485 update `Valuation` invece di duplicare `ExternalAsset`. Idempotenza upload ripetuti dello stesso documento.
- **`pdftoppm` zero-padded filename**: file output può essere `-1.jpg`, `-01.jpg`, `-001.jpg`. Riga 343-348 prova tutti.
- **Mapping rule unique constraint**: stesso IBAN/accountCode già mappato → `try/catch` riga 290-293 logga warn ma non crasha.
- **Nessun `iban` né `accountCode` in feed**: mapping skip (riga 124), va direttamente a fuzzy matching su `rawHolder`.

### Bug ricorrenti
- **`DATABASE_URL` non esplicitato durante import**: causa contaminazione tenant. **Workaround obbligatorio**:
  ```bash
  DATABASE_URL=$DATABASE_URL_WS node scripts/import-v2.js <pos.csv> <perf.csv>
  ```
  Mai `node scripts/import-v2.js ...` senza esplicitazione esplicita.
- **Magic Upload OCR fail silente**: `pdftotext` o `tesseract` non installati → `execSync` throw, catch riga 230 ritorna stringa vuota. Risultato: skip diretto a fallback. **Fix**: pre-check `which pdftotext tesseract` in deploy bootstrap.
- **HEIC conversion timeout**: file HEIC molto pesanti (>20MB) → `sips` timeout 15s → manteniamo HEIC originale. Web viewer fallisce. **Fix**: fallback `heif-convert` se installato (`brew install libheif`).
- **OpenRouter rate limit**: `Qwen3-235B` API può rate-limit a tier 4. Catch silente → cascata a tier 5. Logging warn ma non error.
- **Job map memory leak su upload async stuck**: senza TTL 10min auto-cleanup, error states resterebbero. AsyncJobService gestisce TTL.
- **Tier 1 Gemini partial response**: `validation.partial` recovery riga 193-204 — Zod defaults riempiono gap mancanti. Output usabile anche con AI parziale.

## 3 · SSOT — File fonte verità

| Cosa | Path assoluto |
|------|---------------|
| Ingestion service | `/Users/crisescla/git/valoswiss/apps/api/src/modules/ingestion/ingestion.service.ts` |
| Ingestion controller | `/Users/crisescla/git/valoswiss/apps/api/src/modules/ingestion/ingestion.controller.ts` |
| Magic Upload service | `/Users/crisescla/git/valoswiss/apps/api/src/modules/magic-upload/magic-upload.service.ts` |
| Magic Upload controller | `/Users/crisescla/git/valoswiss/apps/api/src/modules/magic-upload/magic-upload.controller.ts` |
| Import core | `/Users/crisescla/git/valoswiss/scripts/import-v2.js` |
| Import storico wrapper | `/Users/crisescla/git/valoswiss/scripts/import-historical-csv.py` |
| Email fetcher | `/Users/crisescla/git/valoswiss/scripts/fetch-email-csv.py` |
| Cron email | `/Users/crisescla/git/valoswiss/scripts/cron-email-import.sh` |
| Email intelligence | `/Users/crisescla/git/valoswiss/scripts/email-intelligence.py` |
| Schema DB | `/Users/crisescla/git/valoswiss/packages/database/prisma/schema.prisma` (`IngestionInbox`, `IngestionMapping`, `LegalEntity`) |
| Runbook DB realignment | `/Users/crisescla/git/valoswiss/docs/runbooks/DB-REALIGNMENT-WS-AZ-RUNBOOK.md` |
| Log realignment 2026-04-26 | `/Users/crisescla/git/valoswiss/docs/runbooks/DB-REALIGNMENT-WS-AZ-LOG-2026-04-26.md` |

## 4 · API & contracts

### Ingestion (Wealth Ingestion Engine)
| Endpoint | Method | Auth | Note |
|----------|--------|------|------|
| `/ingestion/feed` | POST | ADMIN/SUPERVISOR | crea inbox auto/orphan |
| `/ingestion/inbox?status=&source=&limit=` | GET | ADVISOR/ADMIN/SUPERVISOR | lista paginata 200 |
| `/ingestion/inbox/stats` | GET | ADVISOR+ | counts per status |
| `/ingestion/inbox/:id/resolve` | POST | ADVISOR+ | manual mapping → crea regola permanente |
| `/ingestion/inbox/:id/reject` | POST | ADVISOR+ | con motivazione |
| `/ingestion/inbox/:id/archive` | POST | ADVISOR+ | post-import effettivo |
| `/ingestion/mappings` | GET | ADVISOR+ | lista regole sort by `hitCount desc` |
| `/ingestion/mappings/:id` | DELETE | ADMIN/SUPERVISOR | elimina regola |
| `/ingestion/preview-match` | POST | ADVISOR+ | dry-run fuzzy matching |

### Magic Upload
| Endpoint | Method | Auth | Note |
|----------|--------|------|------|
| `/magic-upload/csv` | POST multipart | tenant header | invoca `scripts/import-v2.js` |
| `/magic-upload/client-document?async=1` | POST multipart | tenant header + clientId | OCR + extraction asset |
| `/magic-upload/office-document?async=1` | POST multipart | tenant header | OCR classification documenti operativi |
| `/magic-upload/client-document/:id` | DELETE | tenant header | rimuove file fisico + DB |
| `/magic-upload/file/:filename` | GET | tenant header | streaming file |

Schema risposta Magic Upload (Zod-validated):
```typescript
interface 

…[truncato — apri il file MD per testo completo]