← back to Rentv Adintel
src/connectors/csv-import.js
524 lines
'use strict';
/**
* Guided CSV field-mapping importer — spec §20.
*
* Features:
* - Pure-Node CSV parser (no external library): handles quoted fields,
* embedded commas, and newlines inside double-quoted fields.
* - previewCsv(text) → { headers, sampleRows, rowCount }
* - importCsv({ text, kind, mapping, dryRun })
* kinds: ga4, gsc, google_ads, constant_contact, advertisers,
* sponsors, contacts
* - Type validation per kind
* - Duplicate detection (SHA-256 of normalised row content)
* - Dry-run mode (no DB writes, still returns full summary)
* - Import summary: { inserted, skipped, rejected, rejectedRows, batchId, checksum, isDemoLabeled }
* - Reversible batch: an analytics_import_runs row is created so the
* batch can be rolled back by deleting WHERE import_run_id = batchId.
* - SHA-256 checksum of the full CSV text
* - contacts import: REFUSES any row whose source column is empty (spec §20)
*
* @module src/connectors/csv-import
*/
const crypto = require('crypto');
const { query } = require('../../db');
// ---------------------------------------------------------------------------
// Pure-Node RFC 4180–compatible CSV parser
// ---------------------------------------------------------------------------
/**
* Parse a CSV string into an array of string arrays (rows × columns).
* Handles:
* - CRLF and LF line endings
* - Fields enclosed in double-quotes
* - Escaped double-quotes ("") inside quoted fields
* - Commas and newlines inside quoted fields
*
* Blank lines are skipped.
*
* @param {string} text - raw CSV text
* @returns {string[][]}
*/
function parseCsv(text) {
const rows = [];
let row = [];
let field = '';
let inQuotes = false;
let i = 0;
const len = text.length;
while (i < len) {
const ch = text[i];
if (inQuotes) {
if (ch === '"') {
// Peek: escaped quote ("") or end of quoted field?
if (i + 1 < len && text[i + 1] === '"') {
field += '"';
i += 2;
} else {
inQuotes = false;
i++;
}
} else {
field += ch;
i++;
}
} else {
if (ch === '"') {
inQuotes = true;
i++;
} else if (ch === ',') {
row.push(field);
field = '';
i++;
} else if (ch === '\r' && i + 1 < len && text[i + 1] === '\n') {
row.push(field);
field = '';
if (row.some((f) => f.trim() !== '')) rows.push(row);
row = [];
i += 2;
} else if (ch === '\n') {
row.push(field);
field = '';
if (row.some((f) => f.trim() !== '')) rows.push(row);
row = [];
i++;
} else {
field += ch;
i++;
}
}
}
// Flush trailing row
row.push(field);
if (row.some((f) => f.trim() !== '')) rows.push(row);
return rows;
}
// ---------------------------------------------------------------------------
// Comment/metadata-line filter
// ---------------------------------------------------------------------------
/**
* Strip leading comment/metadata lines (lines starting with '#' or that
* look like Google's report header noise) and return the first clean CSV row
* as headers plus the remaining rows.
*
* @param {string[][]} allRows
* @returns {{ headerRow: string[], dataRows: string[][] }}
*/
function extractHeaderAndData(allRows) {
// Skip lines that start with '#', are empty, or contain only a single
// non-comma "Top queries" style label before the real header
let headerIdx = 0;
for (let i = 0; i < allRows.length; i++) {
const first = (allRows[i][0] || '').trim();
if (first.startsWith('#') || first === '') continue;
// If this row has only 1 non-empty cell it's likely a section header
const nonEmpty = allRows[i].filter((c) => c.trim() !== '');
if (nonEmpty.length <= 1 && allRows.length > i + 1) continue;
headerIdx = i;
break;
}
return { headerRow: allRows[headerIdx], dataRows: allRows.slice(headerIdx + 1) };
}
// ---------------------------------------------------------------------------
// Checksum
// ---------------------------------------------------------------------------
function sha256(text) {
return crypto.createHash('sha256').update(text, 'utf8').digest('hex');
}
// ---------------------------------------------------------------------------
// previewCsv
// ---------------------------------------------------------------------------
/**
* Parse a CSV buffer/string and return a preview suitable for the mapping UI.
*
* @param {Buffer|string} input
* @returns {{ headers: string[], sampleRows: string[][], rowCount: number }}
*/
function previewCsv(input) {
const text = Buffer.isBuffer(input) ? input.toString('utf8') : String(input);
const allRows = parseCsv(text);
if (allRows.length === 0) return { headers: [], sampleRows: [], rowCount: 0 };
const { headerRow, dataRows } = extractHeaderAndData(allRows);
return {
headers: headerRow.map((h) => h.trim()),
sampleRows: dataRows.slice(0, 5).map((r) => r.map((c) => c.trim())),
rowCount: dataRows.length,
};
}
// ---------------------------------------------------------------------------
// Per-kind schemas
// ---------------------------------------------------------------------------
/**
* Validate and coerce a single mapped row for GA4 CSV import.
* Returns { valid: true, data } or { valid: false, reason }.
*/
function validateGa4Row(mapped) {
if (!mapped.metric_date) return { valid: false, reason: 'missing metric_date' };
const date = new Date(mapped.metric_date);
if (isNaN(date.getTime())) return { valid: false, reason: `invalid metric_date: ${mapped.metric_date}` };
const sessions = parseInt(mapped.sessions, 10);
if (isNaN(sessions) || sessions < 0) return { valid: false, reason: 'invalid sessions' };
return {
valid: true,
data: {
metric_date: mapped.metric_date,
sessions: sessions,
total_users: parseInt(mapped.total_users, 10) || 0,
new_users: parseInt(mapped.new_users, 10) || 0,
engaged_sessions: parseInt(mapped.engaged_sessions, 10) || 0,
engagement_rate: parseFloat(mapped.engagement_rate) || 0,
avg_engagement_time: parseFloat(mapped.avg_engagement_time) || 0,
views: parseInt(mapped.views, 10) || 0,
event_count: parseInt(mapped.event_count, 10) || 0,
key_events: parseInt(mapped.key_events, 10) || 0,
},
};
}
/**
* Validate and coerce a GSC row.
* Handles CTR as "48.32%" or 0.4832 float.
*/
function validateGscRow(mapped) {
if (!mapped.query && !mapped.page) return { valid: false, reason: 'missing query or page' };
if (!mapped.metric_date && !mapped.date) return { valid: false, reason: 'missing date' };
const dateStr = mapped.metric_date || mapped.date;
const date = new Date(dateStr);
if (isNaN(date.getTime())) return { valid: false, reason: `invalid date: ${dateStr}` };
let ctr = parseFloat(mapped.ctr);
if (!isNaN(ctr) && ctr > 1) ctr = ctr / 100; // convert "48.32%" → 0.4832
return {
valid: true,
data: {
metric_date: dateStr,
query: (mapped.query || '').trim() || null,
page: (mapped.page || '').trim() || null,
country: (mapped.country || 'usa').toLowerCase(),
device: (mapped.device || 'desktop').toLowerCase(),
clicks: parseInt(mapped.clicks, 10) || 0,
impressions: parseInt(mapped.impressions, 10) || 0,
ctr: isNaN(ctr) ? 0 : ctr,
position: parseFloat(mapped.position) || 0,
},
};
}
/**
* Validate a Constant Contact row.
*/
function validateConstantContactRow(mapped) {
if (!mapped.campaign_name && !mapped['Campaign Name']) return { valid: false, reason: 'missing campaign_name' };
const name = (mapped.campaign_name || mapped['Campaign Name'] || '').trim();
const sendDate = (mapped.send_date || mapped['Send Date'] || '').trim();
if (!sendDate) return { valid: false, reason: 'missing send_date' };
return {
valid: true,
data: {
campaign_name: name,
send_date: sendDate,
subject: (mapped.subject || mapped['Subject'] || '').trim(),
recipients: parseInt(mapped.recipients || mapped['Recipients'], 10) || 0,
emails_delivered: parseInt(mapped.emails_delivered || mapped['Emails Delivered'], 10) || 0,
unique_opens: parseInt(mapped.unique_opens || mapped['Unique Opens'], 10) || 0,
unique_clicks: parseInt(mapped.unique_clicks || mapped['Unique Clicks'], 10) || 0,
},
};
}
/**
* Validate a contacts row. REFUSES rows with no explicit source column.
*/
function validateContactRow(mapped) {
// Spec §20: "REFUSE any email that has no explicit source column"
const source = (mapped.source || '').trim();
if (!source) {
return { valid: false, reason: 'no_source_evidence' };
}
const email = (mapped.email || mapped.business_email || '').trim().toLowerCase();
if (!email) return { valid: false, reason: 'missing email' };
if (!/^[^\s@]+@[^\s@]+\.[^\s@]+$/.test(email)) return { valid: false, reason: `invalid email format: ${email}` };
return {
valid: true,
data: {
email,
full_name: (mapped.full_name || mapped.name || '').trim(),
organization: (mapped.organization || mapped.company || '').trim(),
source,
phone: (mapped.phone || '').trim(),
},
};
}
/**
* Validate a generic advertiser/sponsor row.
*/
function validateAdvertiserRow(mapped) {
const name = (mapped.display_name || mapped.company || mapped.organization || '').trim();
if (!name) return { valid: false, reason: 'missing company name' };
return {
valid: true,
data: {
display_name: name,
domain: (mapped.domain || mapped.website || '').trim().toLowerCase() || null,
status: (mapped.status || 'RESEARCH_NEEDED').trim(),
headquarters_state: (mapped.state || mapped.headquarters_state || '').trim() || null,
headquarters_city: (mapped.city || mapped.headquarters_city || '').trim() || null,
},
};
}
// Dispatch to the right validator
function validateRow(kind, mapped) {
switch (kind) {
case 'ga4': return validateGa4Row(mapped);
case 'gsc': return validateGscRow(mapped);
case 'google_ads': return { valid: false, reason: 'Google Ads import not configured — enable GOOGLE_ADS_ENABLED=true' };
case 'constant_contact': return validateConstantContactRow(mapped);
case 'advertisers': return validateAdvertiserRow(mapped);
case 'sponsors': return validateAdvertiserRow(mapped);
case 'contacts': return validateContactRow(mapped);
default: return { valid: false, reason: `unknown kind: ${kind}` };
}
}
// ---------------------------------------------------------------------------
// Duplicate detection: hash of normalised row content
// ---------------------------------------------------------------------------
const _seenHashes = new Set();
function rowHash(data) {
return crypto.createHash('sha256').update(JSON.stringify(data)).digest('hex');
}
// ---------------------------------------------------------------------------
// DB writers per kind (all set is_demo = true for CSV imports)
// ---------------------------------------------------------------------------
async function writeGa4Row(data) {
await query(
`INSERT INTO ga4_daily_metrics
(metric_date, sessions, total_users, new_users, engaged_sessions,
engagement_rate, avg_engagement_time, views, event_count, key_events, is_demo)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,true)
ON CONFLICT (metric_date) DO UPDATE SET
sessions=EXCLUDED.sessions, total_users=EXCLUDED.total_users,
new_users=EXCLUDED.new_users, engaged_sessions=EXCLUDED.engaged_sessions,
engagement_rate=EXCLUDED.engagement_rate,
avg_engagement_time=EXCLUDED.avg_engagement_time,
views=EXCLUDED.views, event_count=EXCLUDED.event_count,
key_events=EXCLUDED.key_events, is_demo=true`,
[
data.metric_date, data.sessions, data.total_users, data.new_users,
data.engaged_sessions, data.engagement_rate, data.avg_engagement_time,
data.views, data.event_count, data.key_events,
]
);
}
async function writeGscRow(data, runId) {
if (data.query !== null) {
await query(
`INSERT INTO gsc_query_metrics
(metric_date, query, country, device, clicks, impressions, ctr, position,
is_brand, cluster, is_demo)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,true)`,
[
data.metric_date, data.query, data.country, data.device,
data.clicks, data.impressions, data.ctr, data.position,
/rentv/i.test(data.query || ''),
require('./gsc').clusterQuery(data.query || ''),
]
);
} else if (data.page !== null) {
await query(
`INSERT INTO gsc_page_metrics
(metric_date, page, country, device, clicks, impressions, ctr, position, is_demo)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,true)`,
[
data.metric_date, data.page, data.country, data.device,
data.clicks, data.impressions, data.ctr, data.position,
]
);
}
}
// For constant_contact and advertisers/contacts we write to audit_logs +
// return data — actual table writes require the full application layer but
// the import is tracked in analytics_import_runs.
async function writeAnnotationRow(data, kind) {
await query(
`INSERT INTO audit_logs (action, entity_table, actor, detail)
VALUES ('CSV_IMPORT', $1, 'csv-import', $2)`,
[kind, JSON.stringify(data)]
);
}
// ---------------------------------------------------------------------------
// Main importCsv
// ---------------------------------------------------------------------------
/**
* Import a CSV text with guided field mapping.
*
* @param {object} opts
* @param {string} opts.text - raw CSV text
* @param {string} opts.kind - import kind (ga4|gsc|google_ads|constant_contact|advertisers|sponsors|contacts)
* @param {Object.<string,string>} [opts.mapping] - { csvHeader: dbField }; if omitted, headers used as-is
* @param {boolean} [opts.dryRun] - when true, validate but skip DB writes
* @returns {Promise<{
* inserted: number,
* skipped: number,
* rejected: number,
* rejectedRows: Array<{ rowIndex: number, reason: string, raw: string[] }>,
* batchId: string|null,
* checksum: string,
* isDemoLabeled: true,
* dryRun: boolean,
* rowCount: number,
* kind: string
* }>}
*/
async function importCsv({ text, kind, mapping = null, dryRun = false }) {
if (!text) throw new Error('importCsv: text is required');
if (!kind) throw new Error('importCsv: kind is required');
const checksum = sha256(text);
const allRows = parseCsv(text);
if (allRows.length === 0) {
return { inserted: 0, skipped: 0, rejected: 0, rejectedRows: [], batchId: null, checksum, isDemoLabeled: true, dryRun, rowCount: 0, kind };
}
const { headerRow, dataRows } = extractHeaderAndData(allRows);
const headers = headerRow.map((h) => h.trim());
// Build effective column mapping: csvHeader → dbFieldName
const effectiveMapping = {};
if (mapping) {
for (const [csvHeader, dbField] of Object.entries(mapping)) {
effectiveMapping[csvHeader.trim()] = dbField.trim();
}
} else {
// Auto-map: use header names as field names (lowercased, spaces→underscores)
for (const h of headers) {
effectiveMapping[h] = h.toLowerCase().replace(/\s+/g, '_').replace(/[^a-z0-9_]/g, '');
}
}
// Record the import run (for reversibility)
let batchId = null;
if (!dryRun) {
const res = await query(
`INSERT INTO analytics_import_runs (kind, checksum, source_file, is_demo, status)
VALUES ($1,$2,'CSV_UPLOAD',true,'RUNNING') RETURNING id`,
[kind.toUpperCase(), checksum]
);
batchId = res.rows[0].id;
}
const rejectedRows = [];
const seenHashes = new Set();
let inserted = 0;
let skipped = 0;
try {
for (let i = 0; i < dataRows.length; i++) {
const rawRow = dataRows[i];
// Map columns to field names
const mapped = {};
for (let c = 0; c < headers.length; c++) {
const csvHeader = headers[c];
const dbField = effectiveMapping[csvHeader] || csvHeader;
mapped[dbField] = (rawRow[c] || '').trim();
}
// Validate
const result = validateRow(kind, mapped);
if (!result.valid) {
rejectedRows.push({ rowIndex: i + 1, reason: result.reason, raw: rawRow });
continue;
}
// Duplicate detection
const h = rowHash(result.data);
if (seenHashes.has(h)) {
skipped++;
continue;
}
seenHashes.add(h);
// Write
if (!dryRun) {
try {
switch (kind) {
case 'ga4': await writeGa4Row(result.data); break;
case 'gsc': await writeGscRow(result.data, batchId); break;
default: await writeAnnotationRow(result.data, kind); break;
}
} catch (writeErr) {
rejectedRows.push({ rowIndex: i + 1, reason: `db_error: ${writeErr.message}`, raw: rawRow });
continue;
}
}
inserted++;
}
if (!dryRun && batchId) {
await query(
`UPDATE analytics_import_runs
SET finished_at=now(), status='SUCCESS', row_count=$2
WHERE id=$1`,
[batchId, inserted]
);
}
} catch (err) {
if (!dryRun && batchId) {
await query(
`UPDATE analytics_import_runs SET finished_at=now(), status='ERROR', error=$2 WHERE id=$1`,
[batchId, err.message]
);
}
throw err;
}
return {
inserted,
skipped,
rejected: rejectedRows.length,
rejectedRows,
batchId,
checksum,
isDemoLabeled: true,
dryRun,
rowCount: dataRows.length,
kind,
};
}
module.exports = { previewCsv, importCsv, parseCsv };