如何通过Node.js实现Cloudant DB多JSON文件上传及失败记录迁移

Hey there! Let's tackle these two Cloudant + Node.js tasks one by one. I'll break down each step with practical code examples that you can adapt to your setup.

First, we'll use the official IBM Cloudant SDK for Node.js. Here's how to get it done:

Step 1: Install dependencies #

First, install the required packages:

npm install @ibm-cloud/cloudant ibm-cloud-sdk-core

Step 2: Write the upload script #

This script will read all JSON files from a directory, parse them, and bulk upload to your Cloudant database:

const { CloudantV1 } = require('@ibm-cloud/cloudant');
const { IamAuthenticator } = require('ibm-cloud-sdk-core');
const fs = require('fs');
const path = require('path');

// Initialize Cloudant client (replace with your credentials)
const authenticator = new IamAuthenticator({ apikey: 'YOUR_CLOUDANT_API_KEY' });
const cloudant = CloudantV1.newInstance({
  authenticator: authenticator,
  serviceUrl: 'YOUR_CLOUDANT_SERVICE_URL'
});

// Function to read all JSON files from a directory
const loadJsonDocsFromDir = (dirPath) => {
  const docs = [];
  const files = fs.readdirSync(dirPath);

  files.forEach(file => {
    if (path.extname(file) === '.json') {
      const fullPath = path.join(dirPath, file);
      try {
        const fileContent = fs.readFileSync(fullPath, 'utf8');
        const doc = JSON.parse(fileContent);
        docs.push(doc);
      } catch (error) {
        console.error(`Skipping invalid JSON file ${file}:`, error.message);
      }
    }
  });

  return docs;
};

// Function to bulk upload documents to Cloudant
const bulkUploadDocs = async (dbName, docs) => {
  try {
    const response = await cloudant.postBulkDocs({
      db: dbName,
      bulkDocs: { docs: docs }
    });
    console.log('Bulk upload completed successfully!');
    console.log('Upload results:', response.result);
  } catch (error) {
    console.error('Bulk upload failed:', error);
  }
};

// Execute the workflow
const main = () => {
  const jsonFilesDir = './your-json-files-directory'; // Replace with your directory path
  const targetDb = 'your-target-cloudant-db'; // Replace with your DB name

  const docsToUpload = loadJsonDocsFromDir(jsonFilesDir);
  if (docsToUpload.length === 0) {
    console.log('No valid JSON files found to upload.');
    return;
  }

  bulkUploadDocs(targetDb, docsToUpload);
};

main();

Key Notes: #

• Replace YOUR_CLOUDANT_API_KEY and YOUR_CLOUDANT_SERVICE_URL with your actual Cloudant credentials.
• Ensure your JSON files are valid (no syntax errors) — the script will skip invalid files and log errors.
• The postBulkDocs method is efficient for uploading multiple documents in one request.

This workflow involves three main steps: fetching targeted records from the Fail DB, bulk inserting them into the Reprocess DB, and saving each record as a separate JSON file.

Step 1: Use the same dependencies #

Ensure you've already installed the Cloudant SDK as shown in Problem 1.

Step 2: Write the reprocessing script #

const { CloudantV1 } = require('@ibm-cloud/cloudant');
const { IamAuthenticator } = require('ibm-cloud-sdk-core');
const fs = require('fs');
const path = require('path');

// Initialize Cloudant client
const authenticator = new IamAuthenticator({ apikey: 'YOUR_CLOUDANT_API_KEY' });
const cloudant = CloudantV1.newInstance({
  authenticator: authenticator,
  serviceUrl: 'YOUR_CLOUDANT_SERVICE_URL'
});

// Configuration
const config = {
  failDb: 'fail-db', // Your source failure DB
  reprocessDb: 'reprocess-db', // Your target reprocessing DB
  targetDate: '2018-01-06', // Date to fetch records from
  maxRecords: 20, // Number of records to fetch
  outputDir: './reprocess-records' // Directory to save JSON files
};

// Create output directory if it doesn't exist
if (!fs.existsSync(config.outputDir)) {
  fs.mkdirSync(config.outputDir, { recursive: true });
}

// Fetch records from Fail DB for the target date
const fetchFailedRecords = async () => {
  // Define timestamp range for the entire target day
  const startOfDay = `${config.targetDate}T00:00:00.000Z`;
  const endOfDay = `${config.targetDate}T23:59:59.999Z`;

  try {
    const response = await cloudant.postFind({
      db: config.failDb,
      selector: {
        timestamp: {
          $gte: startOfDay,
          $lte: endOfDay
        },
        DocType: 'CustFail',
        Response: 'Fail'
      },
      limit: config.maxRecords,
      sort: [{ timestamp: 'asc' }] // Optional: sort by timestamp
    });

    return response.result.docs;
  } catch (error) {
    console.error('Failed to fetch records:', error);
    return [];
  }
};

// Bulk insert records into Reprocess DB
const bulkInsertToReprocessDb = async (docs) => {
  // Remove _rev field since it's specific to the source DB
  const sanitizedDocs = docs.map(doc => {
    const { _rev, ...cleanDoc } = doc;
    return cleanDoc;
  });

  try {
    const response = await cloudant.postBulkDocs({
      db: config.reprocessDb,
      bulkDocs: { docs: sanitizedDocs }
    });
    console.log('Successfully inserted records into Reprocess DB!');
    console.log('Insert results:', response.result);
  } catch (error) {
    console.error('Bulk insert failed:', error);
  }
};

// Save each record as a separate JSON file
const saveRecordsToFiles = (docs) => {
  docs.forEach(doc => {
    const fileName = `${doc._id}.json`;
    const filePath = path.join(config.outputDir, fileName);

    try {
      fs.writeFileSync(filePath, JSON.stringify(doc, null, 2), 'utf8');
      console.log(`Saved record to ${filePath}`);
    } catch (error) {
      console.error(`Failed to save ${fileName}:`, error);
    }
  });
};

// Run the entire workflow
const runReprocessWorkflow = async () => {
  const records = await fetchFailedRecords();

  if (records.length === 0) {
    console.log('No records found for the specified date.');
    return;
  }

  await bulkInsertToReprocessDb(records);
  saveRecordsToFiles(records);
};

runReprocessWorkflow();

Key Notes: #

• Again, replace your Cloudant credentials in the initialization section.
• The postFind method uses a selector to filter records by timestamp range, document type, and failure status. Adjust the selector if you need different filters.
• We remove the _rev field before inserting into the Reprocess DB because this revision ID is only valid in the source Fail DB.
• Records are saved with their _id as the filename to ensure uniqueness.

内容的提问来源于stack exchange，提问作者Diya