138 lines
4.2 KiB
JavaScript
138 lines
4.2 KiB
JavaScript
/**
|
|
* Upsert en masse dans PostgreSQL (chunks de 500 lignes).
|
|
* @param {import('pg').Pool} pg
|
|
* @param {string} table - nom de la table PostgreSQL (lowercase)
|
|
* @param {object[]} rows - lignes avec les valeurs à insérer
|
|
* @param {string[]} pk - colonnes formant la clé primaire
|
|
* @param {string[]} cols - toutes les colonnes à upsert
|
|
* @returns {number} total de lignes traitées
|
|
*/
|
|
async function batchUpsert(pg, table, rows, pk, cols) {
|
|
if (!rows.length) return 0;
|
|
const CHUNK = 500;
|
|
let total = 0;
|
|
for (let i = 0; i < rows.length; i += CHUNK) {
|
|
const chunk = rows.slice(i, i + CHUNK);
|
|
const values = [];
|
|
const placeholders = chunk.map((row, ri) => {
|
|
const ph = cols.map((col, ci) => {
|
|
values.push(row[col] ?? null);
|
|
return `$${ri * cols.length + ci + 1}`;
|
|
});
|
|
return `(${ph.join(', ')})`;
|
|
});
|
|
const updateCols = cols.filter(c => !pk.includes(c));
|
|
const updateSet = updateCols.map(c => `${c} = EXCLUDED.${c}`).join(', ');
|
|
const sql = `
|
|
INSERT INTO ${table} (${cols.join(', ')})
|
|
VALUES ${placeholders.join(', ')}
|
|
ON CONFLICT (${pk.join(', ')}) DO UPDATE SET ${updateSet}
|
|
`;
|
|
await pg.query(sql, values);
|
|
total += chunk.length;
|
|
}
|
|
return total;
|
|
}
|
|
|
|
/**
|
|
* Full refresh : TRUNCATE puis INSERT en masse.
|
|
*/
|
|
async function fullRefresh(pg, table, rows, cols) {
|
|
if (!rows.length) {
|
|
await pg.query(`TRUNCATE TABLE ${table}`);
|
|
return 0;
|
|
}
|
|
const CHUNK = 500;
|
|
await pg.query(`TRUNCATE TABLE ${table}`);
|
|
let total = 0;
|
|
for (let i = 0; i < rows.length; i += CHUNK) {
|
|
const chunk = rows.slice(i, i + CHUNK);
|
|
const values = [];
|
|
const placeholders = chunk.map((row, ri) => {
|
|
const ph = cols.map((col, ci) => {
|
|
values.push(row[col] ?? null);
|
|
return `$${ri * cols.length + ci + 1}`;
|
|
});
|
|
return `(${ph.join(', ')})`;
|
|
});
|
|
const sql = `INSERT INTO ${table} (${cols.join(', ')}) VALUES ${placeholders.join(', ')}`;
|
|
await pg.query(sql, values);
|
|
total += chunk.length;
|
|
}
|
|
return total;
|
|
}
|
|
|
|
/**
|
|
* Lit la date du dernier sync réussi pour une table.
|
|
*/
|
|
async function getLastSync(pg, tableName) {
|
|
const res = await pg.query(
|
|
`SELECT last_sync FROM sync_log WHERE table_name = $1 AND status = 'ok'`,
|
|
[tableName]
|
|
);
|
|
return res.rows[0]?.last_sync || null;
|
|
}
|
|
|
|
/**
|
|
* Enregistre le résultat d'un sync dans sync_log.
|
|
*/
|
|
async function logSync(pg, tableName, rowsSynced, status, errorMsg = null) {
|
|
await pg.query(`
|
|
INSERT INTO sync_log (table_name, last_sync, rows_synced, status, error_msg)
|
|
VALUES ($1, NOW(), $2, $3, $4)
|
|
ON CONFLICT (table_name) DO UPDATE SET
|
|
last_sync = NOW(),
|
|
rows_synced = $2,
|
|
status = $3,
|
|
error_msg = $4
|
|
`, [tableName, rowsSynced, status, errorMsg]);
|
|
}
|
|
|
|
// ── Helpers de conversion SQL Server → PostgreSQL ───────────
|
|
|
|
/** Nettoie une chaîne : supprime null bytes (0x00), retourne null si vide */
|
|
function safeStr(v) {
|
|
if (v == null) return null;
|
|
return String(v).replace(/\0/g, '') || null;
|
|
}
|
|
|
|
/** Convertit en nombre, retourne null si NaN ou non numérique */
|
|
function safeNum(v) {
|
|
if (v == null) return null;
|
|
if (typeof v === 'number') return isNaN(v) ? null : v;
|
|
const n = parseFloat(String(v).replace(/[^0-9.\-]/g, ''));
|
|
return isNaN(n) ? null : n;
|
|
}
|
|
|
|
/** Convertit en entier, retourne null si NaN */
|
|
function safeInt(v) {
|
|
if (v == null) return null;
|
|
if (typeof v === 'number') return Math.round(v);
|
|
const n = parseInt(v, 10);
|
|
return isNaN(n) ? null : n;
|
|
}
|
|
|
|
/** Convertit un BIT SQL Server (true/false) en SMALLINT (1/0) */
|
|
function safeBit(v) {
|
|
if (v == null) return null;
|
|
return v ? 1 : 0;
|
|
}
|
|
|
|
/**
|
|
* Convertit une valeur DECIMAL/MONEY qui peut être retournée par mssql
|
|
* sous forme d'objet, tableau, ou string "[date,valeur]".
|
|
*/
|
|
function safeDecimal(v) {
|
|
if (v == null) return null;
|
|
if (typeof v === 'number') return isNaN(v) ? null : v;
|
|
if (Array.isArray(v)) {
|
|
const last = v[v.length - 1];
|
|
return typeof last === 'number' ? last : safeNum(last);
|
|
}
|
|
// ex: "[01/01/1901,163.000]" → 163
|
|
const m = String(v).match(/([\d]+\.?[\d]*)[\]]*$/);
|
|
return m ? parseFloat(m[1]) : null;
|
|
}
|
|
|
|
module.exports = { batchUpsert, fullRefresh, getLastSync, logSync, safeStr, safeNum, safeInt, safeBit, safeDecimal };
|