1// flexylead-sync — port of the local sync.js into one Apteva function. 2// 3// What it does on each invocation: 4// 1. Read existing rapports + appel claims from Tables (preserve agent 5// ownership; collect evaluated-contract set for re-linking on ID drift) 6// 2. Drive discovery (Composio): audios, transcript docs, sheets, folders 7// 3. Build ventes from sales spreadsheets (per-sheet column mappings, 8// multi-contract splitting for Reda/Elite/TINGIS) 9// 4. Build appels from the Drive audio inventory + transcript docs. 10// Phone-number fallback recovers audios whose filename has no contract 11// ID but matches a vente's client.telephone (Reda Protected line). 12// 5. Link ventes ↔ appels by contract overlap; derive appel.status from 13// transcript/audio presence + rapport coverage. 14// 5.5 Backfill missing transcript plaintext via parallel Google Docs reads 15// (only for appels whose carried-over transcript_text is empty). 16// 6. Upsert every sync-managed table (centres, offres, grilles, criteres, 17// ventes, appels) by local_id — inserts new rows, updates rows whose 18// diff-fields changed, deletes rows that no longer exist upstream. 19// Returns per-table {added, modified, removed} deltas, plus a `work` 20// block listing newly-inserted appel ids ready for transcribe / eval 21// agents to pick up without polling the whole table. 22// 23// Heavy AI work stays outside the function — separate agents watch 24// appels.status in ('pending_transcript', 'pending_eval') and claim rows 25// via status_owner. The sync preserves those claims across runs. 26 27import { Composio } from '@composio/core'; 28 29// ─── Composio ──────────────────────────────────────────────────────── 30const composio = new Composio({ 31 apiKey: 'ak_cpsw-4oewu9zim8vwS6K', 32 toolkitVersions: { googledrive: '20260506_00', googledocs: '20260506_00', googlesheets: '20260506_00' }, 33}); 34const COMPOSIO_USER = 'pg-test-2M6ZsbRqWTRFywqEAdRixc5ID9Ph7YaV'; 35async function execComposio(slug, args) { 36 const r = await composio.tools.execute(slug, { userId: COMPOSIO_USER, arguments: args }); 37 if (!r.successful) throw new Error(slug + ': ' + r.error); 38 return r.data; 39} 40async function findAll(q, fields = 'files(id,name,mimeType,size,modifiedTime,createdTime,parents,webViewLink)') { 41 const all = []; 42 let pageToken; 43 do { 44 const args = { q, pageSize: 100, fields: 'nextPageToken,' + fields, corpora: 'allDrives' }; 45 if (pageToken) args.pageToken = pageToken; 46 const data = await execComposio('GOOGLEDRIVE_FIND_FILE', args); 47 all.push(...(data.files || [])); 48 pageToken = data.nextPageToken; 49 } while (pageToken); 50 return all; 51} 52async function getSheetTabs(spreadsheetId, ranges) { 53 const r = await execComposio('GOOGLESHEETS_BATCH_GET', { spreadsheet_id: spreadsheetId, ranges }); 54 const vrs = r.valueRanges || r.value_ranges || []; 55 return vrs.map(vr => vr.values || []); 56} 57async function getSheetNames(spreadsheetId) { 58 const r = await execComposio('GOOGLESHEETS_GET_SPREADSHEET_INFO', { spreadsheet_id: spreadsheetId }); 59 return (r.sheets || []).map(s => s.properties?.title || s.title).filter(Boolean); 60} 61 ··· 1961// Userspace upsert by key. Computes which rows are new, which changed (per 1962// diffFields, deep-equal-ish via JSON), which are missing. Executes: 1963// insert new rows in batches, update changed rows in parallel, delete 1964// missing rows in parallel. Returns the delta for downstream notification. 1965// 1966// existingRows: optional pre-fetched array (avoids a re-read if the caller 1967// already has them — handler does for appels). 1968// Canonical JSON: sorts object keys, normalizes null/undefined, so two 1969// semantically-equal values produce the same string. Critical for diffing 1970// rows round-tripped through the DB, where Apteva may reorder JSON keys 1971// or normalize types. 1972function canon(v) { 1973 if (v === null || v === undefined) return null; 1974 if (Array.isArray(v)) return v.map(canon); 1975 if (typeof v === 'object') { 1976 const out = {}; 1977 for (const k of Object.keys(v).sort()) { 1978 const cv = canon(v[k]); 1979 if (cv !== null) out[k] = cv; // drop null/undefined fields — both sides treat absent as null 1980 } 1981 return out; 1982 } 1983 if (typeof v === 'string') return v === '' ? null : v; // empty string == null at our boundary 1984 return v; 1985} 1986 1987async function upsertTable(ctx, name, newRows, { 1988 keyField = 'local_id', 1989 diffFields = null, // null = always treat as changed; [] = never; ['status','x'] = check these 1990 existingRows = null, 1991 deleteMissing = true, 1992 updateConcurrency = 10, 1993} = {}) { 1994 const existing = existingRows ?? await loadAllRows(ctx, name); 1995 const byKey = new Map(existing.map(r => [String(r[keyField]), r])); 1996 const newByKey = new Map(newRows.map(r => [String(r[keyField]), r])); 1997 1998 const toInsert = []; 1999 const toUpdate = []; // { id, fields } 2000 const added = []; // local_ids that are new 2001 const modified = []; // local_ids that changed 2002 2003 for (const r of newRows) { 2004 const key = String(r[keyField]); 2005 const ex = byKey.get(key); 2006 if (!ex) { 2007 toInsert.push(r); 2008 added.push(key); 2009 continue; 2010 } 2011 if (diffFields === null) { 2012 // Always update — safe but wasteful. Used when we don't want to 2013 // diff complex JSON shapes (ventes/criteres). 2014 toUpdate.push({ id: ex.id, fields: r }); 2015 continue; 2016 } 2017 if (diffFields.length === 0) continue; // never update 2018 let changed = false; 2019 for (const f of diffFields) { 2020 if (JSON.stringify(canon(ex[f])) !== JSON.stringify(canon(r[f]))) { changed = true; break; } 2021 } ··· 2363 trackers: c.trackers ?? null, 2364 audio_path_matchers: c.audio_path_matchers ?? null, 2365 transcripts_path_matchers: c.transcripts_path_matchers ?? null, 2366 call_id_patterns: c.call_id_patterns ?? null, 2367 process_for_evaluation: c.process_for_evaluation !== false, 2368 partner_token: PARTNER_TOKEN_BY_CENTRE[c.id] || null, 2369 })); 2370 const offreRows = OFFRES.map(o => ({ 2371 local_id: o.id, nom: o.nom, fournisseur: o.fournisseur ?? null, 2372 famille: o.famille ?? null, criteres_grille_id: o.criteres_grille_id ?? null, 2373 })); 2374 2375 const deltas = {}; 2376 // centres + criteres: dashboard UI is the source of truth once a row 2377 // exists. Sync still INSERTS new rows from inlined config / rubric sheets 2378 // (bootstrap), but never updates or deletes — so UI edits and ops-added 2379 // centres survive every hourly run. 2380 deltas.centres = await upsertTable(context, 'centres', centreRows, 2381 { diffFields: [], deleteMissing: false }); 2382 // offres + grilles: dashboard/manual edits are the source of truth once a 2383 // row exists. Sync still INSERTS new rows from inlined config (bootstrap), 2384 // but never updates or deletes — so manually-imported grilles (e.g. the 2385 // local .numbers imports for ConsoPilote / Météore / FREE v2) survive 2386 // every hourly run. 2387 deltas.offres = await upsertTable(context, 'offres', offreRows, 2388 { diffFields: [], deleteMissing: false }); 2389 deltas.grilles = await upsertTable(context, 'grilles', grilles, 2390 { diffFields: [], deleteMissing: false }); 2391 deltas.criteres = await upsertTable(context, 'criteres', criteres, 2392 { diffFields: [], deleteMissing: false }); 2393 deltas.ventes = await upsertTable(context, 'ventes', ventes, 2394 { diffFields: ['etat', 'date_activation', 'appel_id', 'score', 'client', 'iban', 'commercial', 'date_vente', 'offre_id'] }); 2395 // appels: pass pre-loaded existing rows (we have them from carry-forward read) 2396 deltas.appels = await upsertTable(context, 'appels', appels, { 2397 existingRows: existingAppels, 2398 diffFields: ['status', 'audio_files', 'transcript', 'transcript_text', 'status_owner', 'status_error', 'attempt_number', 'contrat_numeros'], 2399 }); 2400 2401 timings.push_ms = Date.now() - tStart; 2402 2403 // Release the concurrency lock (best-effort; the TTL covers crashes). 2404 if (lockId) { 2405 try { await context.call('tables', 'rows_update', { table: 'sync_locks', id: lockId, fields: { locked_at: null } }); } 2406 catch (e) { log('lock release failed (TTL will clear it): ' + String(e?.message || e).slice(0, 100)); } 2407 } 2408 2409 return { 2410 ok: true, 2411 total_ms: Date.now() - t0, 2412 counts, 2413 timings, 2414 drive: { audios: inventory.audios.length, docs: inventory.docs.length, sheets: inventory.sheets.length }, 2415 transcripts: { to_fetch: toFetch.length, fetched, errors: fetchErrors, skipped }, 2416 audio_mirror: audioMirror, 2417 audio_orphans: audioOrphans, 2418 // Per-table deltas — counts only here; the lists below carry the actual 2419 // local_ids agents need to act on. 2420 deltas: Object.fromEntries(Object.entries(deltas).map(([t, d]) => [t, { 2421 inserted: d.inserted, updated: d.updated, deleted: d.deleted, 2422 }])), 2423 // Appel-level changes that an agent should pick up. New appels with