cangku/backend/scripts/backfillPickupHistory.js

672 lines
23 KiB
JavaScript

#!/usr/bin/env node
require('dotenv').config();
const dayjs = require('dayjs');
const database = require('../src/config/database');
const cloudDb = require('../src/config/cloudDatabase');
const redis = require('../src/config/redis');
const { refreshAllCache } = require('../src/jobs/dailyRefresh');
const localDb = database.local;
const PICKUP_EXCLUDED_LABEL_KEYWORDS = [
'\u914d\u4ef6',
'\u7ecf\u7ba1\u5927\u5355',
'\u4e0d\u53d1',
'\u7b49\u901a\u77e5',
'\u5ef6\u671f',
'\u5ef6\u8fdf',
'\u6b66\u8fdb\u5b89\u80fd',
'\u4e2d\u94c1',
'\u7269\u6d41',
'\u81ea\u63d0',
'\u5916\u8d38',
'\u91c7\u8d2d\u5355',
'\u5907\u8d27',
'\u5b9a\u5236',
'\u4e2d\u8f6c\u96c6\u8fd0',
'\u4e2d\u8f6c\u8ba2\u5355',
'\u65b0\u7586\u897f\u85cf\u5355',
'ECLP\u8ba2\u5355',
];
const PICKUP_EXCLUDED_COURIER_KEYWORDS = [
'\u6b66\u8fdb\u5b89\u80fd',
'\u4e2d\u94c1',
'\u7269\u6d41',
'\u81ea\u63d0',
'\u7ebf\u4e0b\u5feb\u9012',
'\u4e13\u7ebf',
];
const PICKUP_EXCLUDED_REMARK_KEYWORDS = [
'bbff',
'ccdd',
'\u8f6cX00006',
];
const PICKUP_TIMELY_NO_FETCH_COURIER_KEYWORDS = [
'\u4eac\u4e1c\u5feb\u8fd0',
'\u4eac\u4e1c\u81ea\u8425',
'\u987a\u4e30',
];
const PICKUP_TIMELY_LABEL_KEYWORDS = [
'CLPS\u8ba2\u5355',
'ECLP\u8ba2\u5355',
];
const UNKNOWN_WAREHOUSE = '\u672a\u77e5\u4ed3\u5e93';
const UNKNOWN_SUPPLIER = '\u672a\u77e5\u4f9b\u5e94\u5546';
const UNKNOWN_PROVINCE = '\u672a\u77e5\u7701\u4efd';
const UNKNOWN_COURIER = '\u672a\u77e5\u5feb\u9012';
const NO_AFTERSALE = '\u6682\u65e0\u552e\u540e';
const NO_REFUND_REQUEST = '\u672a\u7533\u8bf7\u9000\u6b3e';
const q = (value) => `N'${String(value).replace(/'/g, "''")}'`;
const buildSqlLikeAny = (column, keywords) => keywords
.map((keyword) => `${column} LIKE ${q(`%${keyword}%`)}`)
.join(' OR ');
const excludedLabelSql = buildSqlLikeAny(`ISNULL(src.order_labels, N'')`, PICKUP_EXCLUDED_LABEL_KEYWORDS);
const excludedCourierSql = buildSqlLikeAny(`ISNULL(src.courier_name, N'')`, PICKUP_EXCLUDED_COURIER_KEYWORDS);
const excludedRemarkSql = buildSqlLikeAny(
`ISNULL(src.buyer_message, N'') + N' ' + ISNULL(src.remark, N'') + N' ' + ISNULL(src.remark_content, N'')`,
PICKUP_EXCLUDED_REMARK_KEYWORDS
);
const timelyNoFetchCourierSql = buildSqlLikeAny(`ISNULL(src.courier_name, N'')`, PICKUP_TIMELY_NO_FETCH_COURIER_KEYWORDS);
const timelyLabelSql = buildSqlLikeAny(`ISNULL(src.order_labels, N'')`, PICKUP_TIMELY_LABEL_KEYWORDS);
const BACKFILL_PICKUP_SQL = `
SET NOCOUNT ON;
SELECT
NULLIF(LTRIM(RTRIM(undefined1)), '') AS inner_order_id,
NULLIF(LTRIM(RTRIM(undefined2)), '') AS order_id,
NULLIF(LTRIM(RTRIM(undefined10)), '') AS sku_id,
TRY_CONVERT(datetime, REPLACE(undefined5, '/', '-')) AS pay_time,
TRY_CONVERT(datetime, REPLACE(undefined6, '/', '-')) AS afs_send_time,
NULLIF(LTRIM(RTRIM(undefined8)), '') AS shop_name,
NULLIF(LTRIM(RTRIM(undefined62)), '') AS order_labels,
NULLIF(LTRIM(RTRIM(undefined72)), '') AS express_no,
NULLIF(LTRIM(RTRIM(undefined73)), '') AS afs_courier_name,
NULLIF(LTRIM(RTRIM(undefined74)), '') AS province,
NULLIF(LTRIM(RTRIM(undefined75)), '') AS city,
NULLIF(LTRIM(RTRIM(undefined66)), '') AS afs_warehouse_name,
NULLIF(LTRIM(RTRIM(undefined32)), '') AS order_type,
TRY_CONVERT(datetime, REPLACE(undefined33, '/', '-')) AS aftersale_apply_time
INTO #afs_raw
FROM dbo.jst_goods_profit_kpi_items_afs
WHERE undefined5 IS NOT NULL
AND undefined5 LIKE @payDatePrefix;
CREATE NONCLUSTERED INDEX IX_afs_raw_dedup ON #afs_raw(inner_order_id, sku_id);
WITH ranked AS (
SELECT
*,
ROW_NUMBER() OVER (
PARTITION BY inner_order_id, sku_id
ORDER BY
CASE WHEN express_no IS NULL OR express_no = '' THEN 1 ELSE 0 END,
CASE WHEN afs_send_time IS NULL THEN 1 ELSE 0 END,
afs_send_time DESC,
pay_time DESC
) AS rn
FROM #afs_raw
WHERE inner_order_id IS NOT NULL
AND inner_order_id <> ''
AND sku_id IS NOT NULL
AND sku_id <> ''
)
SELECT
inner_order_id,
order_id,
sku_id,
pay_time,
afs_send_time,
shop_name,
order_labels,
express_no,
afs_courier_name,
province,
city,
afs_warehouse_name,
order_type,
aftersale_apply_time
INTO #afs_dedup
FROM ranked
WHERE rn = 1;
CREATE NONCLUSTERED INDEX IX_afs_dedup_order_id ON #afs_dedup(order_id);
CREATE NONCLUSTERED INDEX IX_afs_dedup_sku_id ON #afs_dedup(sku_id);
SELECT DISTINCT express_no
INTO #express_keys
FROM #afs_dedup
WHERE express_no IS NOT NULL
AND express_no <> '';
SELECT DISTINCT order_id
INTO #order_keys
FROM #afs_dedup
WHERE order_id IS NOT NULL
AND order_id <> '';
CREATE UNIQUE CLUSTERED INDEX IX_order_keys ON #order_keys(order_id);
SELECT DISTINCT sku_id
INTO #sku_keys
FROM #afs_dedup
WHERE sku_id IS NOT NULL
AND sku_id <> '';
CREATE UNIQUE CLUSTERED INDEX IX_sku_keys ON #sku_keys(sku_id);
SELECT
NULLIF(LTRIM(RTRIM(e.logistic_id)), '') AS logistic_id,
MAX(NULLIF(LTRIM(RTRIM(e.so_id)), '')) AS so_id,
MAX(NULLIF(LTRIM(RTRIM(e.logistic_company_name)), '')) AS logistic_company_name,
MAX(NULLIF(LTRIM(RTRIM(e.order_status_name)), '')) AS order_status_name,
MAX(NULLIF(LTRIM(RTRIM(e.aftersale_status_name)), '')) AS aftersale_status_name,
MAX(NULLIF(LTRIM(RTRIM(e.remark)), '')) AS remark,
MAX(NULLIF(LTRIM(RTRIM(e.buyer_message)), '')) AS buyer_message,
MAX(NULLIF(LTRIM(RTRIM(e.remark_content)), '')) AS remark_content,
MAX(NULLIF(LTRIM(RTRIM(e.shop_name)), '')) AS shop_name,
MAX(NULLIF(LTRIM(RTRIM(e.warehouse_name)), '')) AS warehouse_name,
MIN(e.send_time) AS send_time,
MIN(e.fetch_time) AS fetch_time,
MAX(NULLIF(LTRIM(RTRIM(e.receiver_address)), '')) AS receiver_address
INTO #express_by_logistic
FROM dbo.jst_order_express_details e
INNER JOIN #express_keys k
ON k.express_no = NULLIF(LTRIM(RTRIM(e.logistic_id)), '')
GROUP BY NULLIF(LTRIM(RTRIM(e.logistic_id)), '');
CREATE UNIQUE CLUSTERED INDEX IX_express_by_logistic ON #express_by_logistic(logistic_id);
SELECT
NULLIF(LTRIM(RTRIM(e.so_id)), '') AS so_id,
MAX(NULLIF(LTRIM(RTRIM(e.logistic_id)), '')) AS logistic_id,
MAX(NULLIF(LTRIM(RTRIM(e.logistic_company_name)), '')) AS logistic_company_name,
MAX(NULLIF(LTRIM(RTRIM(e.order_status_name)), '')) AS order_status_name,
MAX(NULLIF(LTRIM(RTRIM(e.aftersale_status_name)), '')) AS aftersale_status_name,
MAX(NULLIF(LTRIM(RTRIM(e.remark)), '')) AS remark,
MAX(NULLIF(LTRIM(RTRIM(e.buyer_message)), '')) AS buyer_message,
MAX(NULLIF(LTRIM(RTRIM(e.remark_content)), '')) AS remark_content,
MAX(NULLIF(LTRIM(RTRIM(e.shop_name)), '')) AS shop_name,
MAX(NULLIF(LTRIM(RTRIM(e.warehouse_name)), '')) AS warehouse_name,
MIN(e.send_time) AS send_time,
MIN(e.fetch_time) AS fetch_time,
MAX(NULLIF(LTRIM(RTRIM(e.receiver_address)), '')) AS receiver_address
INTO #express_by_order
FROM dbo.jst_order_express_details e
INNER JOIN #order_keys k
ON k.order_id = NULLIF(LTRIM(RTRIM(e.so_id)), '')
GROUP BY NULLIF(LTRIM(RTRIM(e.so_id)), '');
CREATE UNIQUE CLUSTERED INDEX IX_express_by_order ON #express_by_order(so_id);
SELECT
NULLIF(LTRIM(RTRIM(s.sku_id)), '') AS sku_id,
MAX(NULLIF(LTRIM(RTRIM(s.supplier_name)), '')) AS supplier_name
INTO #sku_supplier
FROM dbo.jst_sku s
INNER JOIN #sku_keys k
ON k.sku_id = NULLIF(LTRIM(RTRIM(s.sku_id)), '')
GROUP BY NULLIF(LTRIM(RTRIM(s.sku_id)), '');
CREATE UNIQUE CLUSTERED INDEX IX_sku_supplier ON #sku_supplier(sku_id);
SELECT
d.inner_order_id,
d.order_id,
d.sku_id,
d.pay_time,
COALESCE(el.send_time, eo.send_time, d.afs_send_time) AS send_time,
COALESCE(el.fetch_time, eo.fetch_time) AS fetch_time,
COALESCE(el.shop_name, eo.shop_name, d.shop_name) AS shop_name,
COALESCE(el.warehouse_name, eo.warehouse_name, d.afs_warehouse_name) AS warehouse_name,
COALESCE(el.logistic_company_name, eo.logistic_company_name, d.afs_courier_name) AS courier_name,
COALESCE(d.express_no, el.logistic_id, eo.logistic_id, N'') AS express_no,
COALESCE(el.order_status_name, eo.order_status_name, N'') AS order_status_name,
COALESCE(el.aftersale_status_name, eo.aftersale_status_name, N'') AS aftersale_status_name,
COALESCE(el.remark, eo.remark, N'') AS remark,
COALESCE(el.buyer_message, eo.buyer_message, N'') AS buyer_message,
COALESCE(el.remark_content, eo.remark_content, N'') AS remark_content,
COALESCE(d.order_labels, N'') AS order_labels,
COALESCE(d.province, N'') AS province,
COALESCE(d.city, N'') AS city,
COALESCE(el.receiver_address, eo.receiver_address, N'') AS receiver_address,
COALESCE(d.order_type, N'') AS order_type,
d.aftersale_apply_time,
COALESCE(s.supplier_name, N'') AS supplier_name,
COALESCE(shop_meta.responsible_department, N'') AS shop_responsible_department
INTO #pickup_enriched
FROM #afs_dedup d
LEFT JOIN #express_by_logistic el ON d.express_no = el.logistic_id
LEFT JOIN #express_by_order eo ON d.order_id = eo.so_id
LEFT JOIN #sku_supplier s ON d.sku_id = s.sku_id
OUTER APPLY (
SELECT TOP 1 NULLIF(LTRIM(RTRIM(responsibleDepartment)), N'') AS responsible_department
FROM dbo.fys_shopInfo
WHERE NULLIF(LTRIM(RTRIM(erpShopName)), N'') = COALESCE(el.shop_name, eo.shop_name, d.shop_name)
) shop_meta;
SELECT
src.*,
CASE
WHEN (${excludedLabelSql}) THEN 1
WHEN ISNULL(src.shop_responsible_department, N'') LIKE N'%灞变笢椤圭洰缁?%' THEN 1
WHEN src.supplier_name IN (${q('\u865a\u62df')}, ${q('\u666f\u74f4\u5341\u65b9')}) THEN 1
WHEN src.order_status_name IN (${q('\u88ab\u62c6\u5206')}, ${q('\u88ab\u5408\u5e76')}) THEN 1
WHEN (${excludedCourierSql}) THEN 1
WHEN ISNULL(src.express_no, N'') LIKE ${q('%F800%')} THEN 1
WHEN (${excludedRemarkSql}) THEN 1
WHEN ISNULL(src.shop_name, N'') LIKE ${q('%\u4eac\u559c%')}
AND ISNULL(src.aftersale_status_name, ${q(NO_AFTERSALE)}) <> ${q(NO_AFTERSALE)} THEN 1
WHEN ISNULL(src.shop_name, N'') LIKE ${q('%\u6837\u54c1\u4e13\u7528\u5e97\u94fa%')} THEN 1
WHEN ISNULL(src.order_status_name, N'') LIKE ${q('%\u53d6\u6d88%')}
AND src.send_time IS NULL
AND ISNULL(src.aftersale_status_name, ${q(NO_AFTERSALE)}) IN (N'', ${q(NO_AFTERSALE)}, ${q(NO_REFUND_REQUEST)}) THEN 1
WHEN src.aftersale_apply_time IS NOT NULL
AND src.pay_time IS NOT NULL
AND src.aftersale_apply_time >= src.pay_time
AND DATEDIFF(SECOND, src.pay_time, src.aftersale_apply_time) <= 24 * 3600 THEN 1
WHEN ISNULL(src.order_type, N'') LIKE ${q('%\u9884\u552e%')}
OR ISNULL(src.order_status_name, N'') LIKE ${q('%\u9884\u552e%')} THEN 1
ELSE 0
END AS is_excluded,
CASE
WHEN src.pay_time IS NULL THEN 1
WHEN ISNULL(src.shop_name, N'') LIKE ${q('%\u4eac\u559c%')}
AND src.send_time IS NOT NULL
AND src.send_time >= src.pay_time
AND DATEDIFF(SECOND, src.pay_time, src.send_time) <= 24 * 3600 THEN 0
WHEN (${timelyNoFetchCourierSql})
AND src.fetch_time IS NULL
AND src.send_time IS NOT NULL
AND src.send_time >= src.pay_time
AND DATEDIFF(SECOND, src.pay_time, src.send_time) <= 24 * 3600 THEN 0
WHEN (${timelyLabelSql})
AND src.send_time IS NOT NULL
AND src.send_time >= src.pay_time
AND DATEDIFF(SECOND, src.pay_time, src.send_time) <= 24 * 3600 THEN 0
WHEN src.fetch_time IS NOT NULL
AND src.fetch_time >= src.pay_time
AND DATEDIFF(SECOND, src.pay_time, src.fetch_time) <= 24 * 3600 THEN 0
ELSE 1
END AS is_timeout,
CASE
WHEN src.fetch_time IS NOT NULL
AND src.pay_time IS NOT NULL
AND src.fetch_time >= src.pay_time
THEN CAST(DATEDIFF(SECOND, src.pay_time, src.fetch_time) AS DECIMAL(20, 2)) / 3600.0
ELSE NULL
END AS pickup_hours,
CASE
WHEN src.receiver_address LIKE ${q('%-%-%')}
THEN REVERSE(LEFT(REVERSE(src.receiver_address), CHARINDEX('-', REVERSE(src.receiver_address)) - 1))
ELSE N''
END AS district
INTO #pickup_scored
FROM #pickup_enriched src;
SELECT *
INTO #pickup_valid
FROM #pickup_scored
WHERE is_excluded = 0;
SELECT
COUNT_BIG(*) AS total_count,
SUM(CASE WHEN is_timeout = 1 THEN 1 ELSE 0 END) AS timeout_count,
CAST(ROUND(AVG(pickup_hours), 2) AS DECIMAL(10, 2)) AS avg_hours
FROM #pickup_valid;
SELECT
ISNULL(NULLIF(warehouse_name, N''), ${q(UNKNOWN_WAREHOUSE)}) AS warehouse_name,
COUNT_BIG(*) AS total_count,
SUM(CASE WHEN is_timeout = 1 THEN 1 ELSE 0 END) AS timeout_count,
CAST(ROUND(AVG(pickup_hours), 2) AS DECIMAL(10, 2)) AS avg_hours
FROM #pickup_valid
GROUP BY ISNULL(NULLIF(warehouse_name, N''), ${q(UNKNOWN_WAREHOUSE)});
SELECT
ISNULL(NULLIF(supplier_name, N''), ${q(UNKNOWN_SUPPLIER)}) AS supplier_name,
COUNT_BIG(*) AS total_count,
SUM(CASE WHEN is_timeout = 1 THEN 1 ELSE 0 END) AS timeout_count,
CAST(ROUND(AVG(pickup_hours), 2) AS DECIMAL(10, 2)) AS avg_hours
FROM #pickup_valid
GROUP BY ISNULL(NULLIF(supplier_name, N''), ${q(UNKNOWN_SUPPLIER)});
SELECT
ISNULL(NULLIF(province, N''), ${q(UNKNOWN_PROVINCE)}) AS province,
COUNT_BIG(*) AS total_count,
SUM(CASE WHEN is_timeout = 1 THEN 1 ELSE 0 END) AS timeout_count,
CAST(ROUND(AVG(pickup_hours), 2) AS DECIMAL(10, 2)) AS avg_hours
FROM #pickup_valid
GROUP BY ISNULL(NULLIF(province, N''), ${q(UNKNOWN_PROVINCE)});
SELECT
ISNULL(NULLIF(courier_name, N''), ${q(UNKNOWN_COURIER)}) AS courier_name,
COUNT_BIG(*) AS total_count,
SUM(CASE WHEN is_timeout = 1 THEN 1 ELSE 0 END) AS timeout_count,
CAST(ROUND(AVG(pickup_hours), 2) AS DECIMAL(10, 2)) AS avg_hours
FROM #pickup_valid
GROUP BY ISNULL(NULLIF(courier_name, N''), ${q(UNKNOWN_COURIER)});
SELECT
ISNULL(NULLIF(province, N''), ${q(UNKNOWN_PROVINCE)}) AS province,
ISNULL(NULLIF(city, N''), N'') AS city,
ISNULL(NULLIF(district, N''), N'') AS district,
ISNULL(NULLIF(courier_name, N''), ${q(UNKNOWN_COURIER)}) AS courier_name,
COUNT_BIG(*) AS total_count,
SUM(CASE WHEN is_timeout = 1 THEN 1 ELSE 0 END) AS timeout_count,
CAST(ROUND(AVG(pickup_hours), 2) AS DECIMAL(10, 2)) AS avg_hours
FROM #pickup_valid
GROUP BY
ISNULL(NULLIF(province, N''), ${q(UNKNOWN_PROVINCE)}),
ISNULL(NULLIF(city, N''), N''),
ISNULL(NULLIF(district, N''), N''),
ISNULL(NULLIF(courier_name, N''), ${q(UNKNOWN_COURIER)});
`;
const TABLE_DELETES = [
'summary_daily_pickup_region_courier',
'summary_daily_pickup_courier',
'summary_daily_pickup_region',
'summary_daily_pickup_supplier',
'summary_daily_pickup_warehouse',
'summary_daily_pickup',
];
const toInt = (value) => Number.parseInt(value, 10) || 0;
const toFloat = (value) => Number.parseFloat(value) || 0;
const calcPickupRate = (totalCount, timeoutCount) => (
totalCount > 0 ? Number((((totalCount - timeoutCount) / totalCount) * 100).toFixed(2)) : 0
);
const TRANSIENT_LOCAL_WRITE_ERROR_CODES = new Set([
'ETIMEDOUT',
'ECONNRESET',
'ECONNREFUSED',
'EPIPE',
'PROTOCOL_CONNECTION_LOST',
'PROTOCOL_ENQUEUE_AFTER_FATAL_ERROR',
'PROTOCOL_ENQUEUE_AFTER_QUIT',
'PROTOCOL_PACKETS_OUT_OF_ORDER',
]);
const isDateString = (value) => /^\d{4}-\d{2}-\d{2}$/.test(value) && dayjs(value).format('YYYY-MM-DD') === value;
const buildPayDatePrefix = (statDate) => `${dayjs(statDate).subtract(1, 'day').format('YYYY/M/D')}%`;
const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));
const isTransientLocalWriteError = (error) => {
if (!error) {
return false;
}
const message = String(error.message || '').toLowerCase();
const code = String(error.code || '');
return (
TRANSIENT_LOCAL_WRITE_ERROR_CODES.has(code) ||
message.includes('timed out') ||
message.includes('closed state') ||
message.includes('connection lost') ||
message.includes('socket hang up')
);
};
const chunkRows = (rows, size) => {
const chunks = [];
for (let index = 0; index < rows.length; index += size) {
chunks.push(rows.slice(index, index + size));
}
return chunks;
};
const buildInsertSql = (tableName, columns, rowCount) => {
const valuePlaceholder = `(${columns.map(() => '?').join(', ')})`;
const placeholders = new Array(rowCount).fill(valuePlaceholder).join(', ');
const quotedColumns = columns.map((column) => `\`${column}\``).join(', ');
return `INSERT INTO ${tableName} (${quotedColumns}) VALUES ${placeholders}`;
};
const executeInsertChunks = async (connection, tableName, columns, rows, chunkSize = 500) => {
if (rows.length === 0) {
return;
}
for (const chunk of chunkRows(rows, chunkSize)) {
const sql = buildInsertSql(tableName, columns, chunk.length);
const params = chunk.flat();
await connection.query(sql, params);
}
};
const normalizeOverallRow = (statDate, row = {}) => {
const totalCount = toInt(row.total_count);
const timeoutCount = toInt(row.timeout_count);
if (totalCount === 0 && timeoutCount === 0) {
return null;
}
return [
statDate,
totalCount,
timeoutCount,
calcPickupRate(totalCount, timeoutCount),
toFloat(row.avg_hours),
];
};
const normalizeNamedRows = (statDate, rows, nameField) => rows
.map((row) => {
const totalCount = toInt(row.total_count);
const timeoutCount = toInt(row.timeout_count);
return [
statDate,
row[nameField] || '',
totalCount,
timeoutCount,
calcPickupRate(totalCount, timeoutCount),
toFloat(row.avg_hours),
];
});
const normalizeRegionCourierRows = (statDate, rows) => rows
.map((row) => {
const totalCount = toInt(row.total_count);
const timeoutCount = toInt(row.timeout_count);
return [
statDate,
row.province || UNKNOWN_PROVINCE,
row.city || '',
row.district || '',
row.courier_name || UNKNOWN_COURIER,
totalCount,
timeoutCount,
calcPickupRate(totalCount, timeoutCount),
toFloat(row.avg_hours),
];
});
const fetchPickupRecordsets = async (statDate) => {
const pool = await cloudDb.getPool();
const request = pool.request();
request.input('payDatePrefix', buildPayDatePrefix(statDate));
const result = await request.query(BACKFILL_PICKUP_SQL);
return result.recordsets;
};
const replacePickupSummaryForDateOnce = async (statDate, recordsets) => {
const [
overallRows = [],
warehouseRows = [],
supplierRows = [],
regionRows = [],
courierRows = [],
regionCourierRows = [],
] = recordsets;
const overallRow = normalizeOverallRow(statDate, overallRows[0]);
const warehouseInsertRows = normalizeNamedRows(statDate, warehouseRows, 'warehouse_name');
const supplierInsertRows = normalizeNamedRows(statDate, supplierRows, 'supplier_name');
const regionInsertRows = normalizeNamedRows(statDate, regionRows, 'province');
const courierInsertRows = normalizeNamedRows(statDate, courierRows, 'courier_name');
const regionCourierInsertRows = normalizeRegionCourierRows(statDate, regionCourierRows);
const connection = await localDb.pool.getConnection();
try {
await connection.beginTransaction();
for (const tableName of TABLE_DELETES) {
await connection.query(`DELETE FROM ${tableName} WHERE stat_date = ?`, [statDate]);
}
if (overallRow) {
await executeInsertChunks(
connection,
'summary_daily_pickup',
['stat_date', 'total_order_count', 'timeout_count', 'pickup_rate', 'avg_pickup_hours'],
[overallRow],
1
);
}
await executeInsertChunks(
connection,
'summary_daily_pickup_warehouse',
['stat_date', 'warehouse_name', 'total_order_count', 'timeout_count', 'pickup_rate', 'avg_pickup_hours'],
warehouseInsertRows
);
await executeInsertChunks(
connection,
'summary_daily_pickup_supplier',
['stat_date', 'supplier_name', 'total_order_count', 'timeout_count', 'pickup_rate', 'avg_pickup_hours'],
supplierInsertRows
);
await executeInsertChunks(
connection,
'summary_daily_pickup_region',
['stat_date', 'province', 'total_order_count', 'timeout_count', 'pickup_rate', 'avg_pickup_hours'],
regionInsertRows
);
await executeInsertChunks(
connection,
'summary_daily_pickup_courier',
['stat_date', 'courier_name', 'total_order_count', 'timeout_count', 'pickup_rate', 'avg_pickup_hours'],
courierInsertRows
);
await executeInsertChunks(
connection,
'summary_daily_pickup_region_courier',
['stat_date', 'province', 'city', 'district', 'courier_name', 'total_order_count', 'timeout_count', 'pickup_rate', 'avg_pickup_hours'],
regionCourierInsertRows
);
await connection.commit();
console.log(
`[pickup-backfill] ${statDate} total=${overallRow ? overallRow[1] : 0} timeout=${overallRow ? overallRow[2] : 0} `
+ `warehouse=${warehouseInsertRows.length} supplier=${supplierInsertRows.length} `
+ `region=${regionInsertRows.length} courier=${courierInsertRows.length} regionCourier=${regionCourierInsertRows.length}`
);
} catch (error) {
try {
await connection.rollback();
} catch (rollbackError) {
console.warn(`[pickup-backfill] rollback failed for ${statDate}: ${rollbackError.message}`);
}
throw error;
} finally {
connection.release();
}
};
const replacePickupSummaryForDate = async (statDate, recordsets) => {
const maxAttempts = 3;
for (let attempt = 1; attempt <= maxAttempts; attempt += 1) {
try {
await replacePickupSummaryForDateOnce(statDate, recordsets);
return;
} catch (error) {
const shouldRetry = attempt < maxAttempts && isTransientLocalWriteError(error);
if (!shouldRetry) {
throw error;
}
console.warn(
`[pickup-backfill] local write retry for ${statDate} (${attempt}/${maxAttempts}) due to: ${error.message}`
);
await sleep(attempt * 1000);
}
}
};
const backfillRange = async (startDate, endDate) => {
const start = dayjs(startDate);
const end = dayjs(endDate);
const totalDays = end.diff(start, 'day') + 1;
if (totalDays <= 0) {
throw new Error(`Invalid date range: ${startDate} -> ${endDate}`);
}
for (let offset = 0; offset < totalDays; offset += 1) {
const statDate = start.add(offset, 'day').format('YYYY-MM-DD');
const dayStart = Date.now();
console.log(`[pickup-backfill] start ${statDate} (${offset + 1}/${totalDays}) payPrefix=${buildPayDatePrefix(statDate)}`);
const recordsets = await fetchPickupRecordsets(statDate);
await replacePickupSummaryForDate(statDate, recordsets);
console.log(`[pickup-backfill] done ${statDate} elapsed=${Date.now() - dayStart}ms`);
}
};
const main = async () => {
const [startDate, endDate = startDate] = process.argv.slice(2);
if (!isDateString(startDate) || !isDateString(endDate)) {
throw new Error('Usage: node backend/scripts/backfillPickupHistory.js <YYYY-MM-DD> [YYYY-MM-DD]');
}
console.log(`[pickup-backfill] range ${startDate} -> ${endDate}`);
try {
await backfillRange(startDate, endDate);
try {
await refreshAllCache();
console.log('[pickup-backfill] cache refresh done');
} catch (error) {
console.warn(`[pickup-backfill] cache refresh skipped: ${error.message}`);
}
} finally {
await Promise.allSettled([
cloudDb.close(),
database.close(),
redis.close(),
]);
}
};
main().catch((error) => {
console.error('[pickup-backfill] failed');
console.error(error);
process.exit(1);
});