93 lines
3.1 KiB
JavaScript
93 lines
3.1 KiB
JavaScript
import { BigQuery } from '@google-cloud/bigquery';
|
||
import { Storage } from '@google-cloud/storage';
|
||
|
||
// Cross-project setup:
|
||
// - BigQuery data source: fynd-jio-commerceml-prod
|
||
// - GCS destination: fynd-boltic-prod
|
||
const bigquery = new BigQuery({
|
||
projectId: process.env.BIGQUERY_PROJECT || 'fynd-jio-commerceml-prod'
|
||
});
|
||
const storage = new Storage({
|
||
projectId: process.env.STORAGE_PROJECT || 'fynd-jio-commerceml-prod',
|
||
credentials: {}
|
||
});
|
||
|
||
export const exportMergedJson = async (req, res) => {
|
||
try {
|
||
const datasetId = process.env.DATASET || "temp_zenith_data";
|
||
const sourceTable = process.env.SOURCE_TABLE || "pr_training_data";
|
||
const bucketName = 'pr_dataset_storage';
|
||
const objectPath = '261/643e93f5ffc8101656a5629b/GET_ALL_PRODUCTS/catalog-*.json';
|
||
const uri = `gs://${bucketName}/${objectPath}`;
|
||
|
||
// Get deletion criteria from request body or environment
|
||
const companyIdToDelete = req.body?.companyId;
|
||
const applicationIdToDelete = req.body?.applicationId;
|
||
|
||
// 1️⃣ Direct export from source table to GCS (cross-project)
|
||
const sourceTablePath = `${bigquery.projectId}.${datasetId}.${sourceTable}`;
|
||
|
||
const exportSql = `
|
||
EXPORT DATA
|
||
OPTIONS(
|
||
uri='${uri}',
|
||
format='JSON',
|
||
overwrite=true
|
||
) AS
|
||
SELECT
|
||
ARRAY_CONCAT_AGG(JSON_EXTRACT_ARRAY(data)) AS merged_array
|
||
FROM
|
||
\`${sourceTablePath}\`
|
||
`;
|
||
const [queryJob] = await bigquery.createQueryJob({
|
||
query: exportSql,
|
||
useLegacySql: false,
|
||
});
|
||
const jobResponse = await queryJob.promise();
|
||
console.log(jobResponse);
|
||
|
||
// 2️⃣ Delete rows from source table after successful export
|
||
// if (companyIdToDelete && applicationIdToDelete) {
|
||
// console.log('Starting deletion of processed rows...');
|
||
// const [deleteJob] = await bigquery.createQueryJob({
|
||
// query: `
|
||
// DELETE FROM \`${sourceTablePath}\`
|
||
// WHERE companyId = @companyId AND applicationId = @applicationId
|
||
// `,
|
||
// params: {
|
||
// companyId: companyIdToDelete,
|
||
// applicationId: applicationIdToDelete
|
||
// },
|
||
// useLegacySql: false,
|
||
// });
|
||
|
||
// await deleteJob.promise();
|
||
// console.log(`Successfully deleted rows with companyId: ${companyIdToDelete}, applicationId: ${applicationIdToDelete}`);
|
||
// } else {
|
||
// console.log('No deletion criteria provided, skipping row cleanup');
|
||
// }
|
||
|
||
const options = {
|
||
version: 'v4',
|
||
action: 'read',
|
||
expires: Date.now() + 15 * 60 * 1000,
|
||
extensionHeaders: {},
|
||
queryParams: {},
|
||
method: 'POST'
|
||
};
|
||
|
||
const signedUrl = await storage.bucket(bucketName).file(objectPath.replace('*', '000000000000')).getSignedUrl(options);
|
||
|
||
return {
|
||
signedUrl,
|
||
success: true,
|
||
message: `Success: exported to ${uri}`,
|
||
companyId: companyIdToDelete,
|
||
applicationId: applicationIdToDelete,
|
||
destination: uri
|
||
};
|
||
} catch (err) {
|
||
console.error('Export or cleanup failed:', err);
|
||
throw new Error(`Export failed: ${err.message}`);
|
||
}
|
||
}; |