import * as fs from 'fs';
import csv from 'csv-parser';
import { parse } from 'fast-csv';
import * as readline from "readline";
import { parse as csvParse } from 'fast-csv';
import { writeToPath } from '@fast-csv/format';
import * as path from 'path';

interface Mapping {
    sentralColumn: string;
    proserveColumn: string;
}

interface ParityReportEntry {
    proserveId: string;
    sentralId: string;
    proserveColumn: string;
    sentralColumn: string;
    proserveValue: string;
    sentralValue: string;
    status: 'Matched' | 'Mismatched' | 'Skipped';
}

async function parseCSVFast(filePath: string): Promise<any[]> {
    console.log(`Starting to parse CSV file: ${filePath}`);
    return new Promise((resolve, reject) => {
        const results: any[] = [];
        const startTime = Date.now();
        let rowCount = 0;

        fs.createReadStream(filePath)
            .pipe(parse({ headers: true, ignoreEmpty: true }))
            .on('error', (error: any) => {
                console.error(`Error parsing CSV file ${filePath}:`, error);
                reject(error);
            })
            .on('data', (row: any) => {
                results.push(row);
                rowCount++;
                if (rowCount % 5000 === 0) {
                    console.log(`Parsed ${rowCount} records from ${filePath}...`);
                }
            })
            .on('end', () => {
                const duration = ((Date.now() - startTime) / 1000).toFixed(2);
                console.log(
                    `Finished parsing CSV file: ${filePath}. Total records: ${rowCount}. Time taken: ${duration}s`,
                );
                resolve(results);
            });
    });
}

/**
 * Generates a parity report by comparing Sentral and Proserve data.
 * @param sentralData - Array of records from Sentral.
 * @param proserveData - Array of records from Proserve.
 * @param mappings - Array of field mappings between Sentral and Proserve.
 * @param recordIdMapping - Array of [Proserve ID, Sentral ID] tuples.
 * @param sentralIdColumn - The column name for Sentral IDs.
 * @param proserveIdColumn - The column name for Proserve IDs.
 * @returns An object containing the report entries and summary statistics.
 */
function generateParityReport(
    sentralData: any[],
    proserveData: any[],
    mappings: Mapping[],
    recordIdMapping: [string, string][],
    sentralIdColumn: string,
    proserveIdColumn: string,
): { report: ParityReportEntry[]; summary: any } {
    console.log('Starting parity report generation...');
    const report: ParityReportEntry[] = [];
    let totalRecords = 0;
    let failedCount = 0;
    let passedCount = 0;
    let totalMismatches = 0;

    const sentralMap = new Map(sentralData.map((row) => [row[sentralIdColumn], row]));
    const proserveMap = new Map(proserveData.map((row) => [row[proserveIdColumn], row]));
    let skipped = 0;

    const totalMappings = recordIdMapping.length;
    console.log(`Processing ${totalMappings} record ID mappings...`);

    recordIdMapping.forEach(([proserveId, sentralId], index) => {
        if (index > 0 && index % 1000 === 0) {
            console.log(`Processed ${index} / ${totalMappings} record ID mappings...`);
        }

        const sentralRow = sentralMap.get(sentralId);
        const proserveRow = proserveMap.get(proserveId);

        if (!sentralRow || !proserveRow) {
            skipped++;
            // Record the missing rows with a specific status
            report.push({
                proserveId,
                sentralId,
                proserveColumn: '',
                sentralColumn: '',
                proserveValue: '',
                sentralValue: '',
                status: 'Skipped', // Changed to 'Skipped' for clarity
            });
            return; // Skip if there's no corresponding row in either dataset
        }

        totalRecords++;
        let recordHasMismatch = false;

        mappings.forEach((mapping) => {
            let sentralValue = sentralRow[mapping.sentralColumn];
            let proserveValue = proserveRow[mapping.proserveColumn];

            if (typeof sentralValue === 'string') {
                sentralValue = sentralValue.replace(/\r\n/g, '\n').replace(/\r/g, '\n').trim();
            }
            if (typeof proserveValue === 'string') {
                proserveValue = proserveValue.replace(/\r\n/g, '\n').replace(/\r/g, '\n').trim();
            }

            const isMatch =
                (sentralValue === '' || sentralValue == null) &&
                (proserveValue === '' || proserveValue == null)
                    ? true
                    : sentralValue === proserveValue;

            report.push({
                proserveId,
                sentralId,
                proserveColumn: mapping.proserveColumn,
                sentralColumn: mapping.sentralColumn,
                proserveValue: proserveValue || '',
                sentralValue: sentralValue || '',
                status: isMatch ? 'Matched' : 'Mismatched',
            });

            if (!isMatch) {
                recordHasMismatch = true;
                totalMismatches++;
            }
        });

        if (recordHasMismatch) {
            failedCount++;
        } else {
            passedCount++;
        }
    });

    console.log(`Parity report generation completed.`);
    console.log(`Total Records Processed: ${totalRecords}`);
    console.log(`Passed Records: ${passedCount}`);
    console.log(`Failed Records: ${failedCount}`);
    console.log(`Total Mismatches Found: ${totalMismatches}`);
    console.log(`Total Records Skipped (missing data): ${skipped}`);

    const summary = {
        totalRecords,
        passedCount,
        failedCount,
        totalMismatches,
        skipped,
    };

    return { report, summary };
}

/**
 * Saves the parity report to multiple CSV files in batches of 10,000 records each.
 * Each batch is saved with a unique file path based on the base file path provided.
 * @param report - Array of ParityReportEntry to be saved.
 * @param summary - Summary object containing report statistics.
 * @param baseFilePath - Base path to save the CSV report batches. The function appends batch numbers to this base path.
 */
async function saveReportToCSVInBatches(report: ParityReportEntry[], summary: any, baseFilePath: string): Promise<void> {
    console.log(`Starting to save CSV report to batches based on base path: ${baseFilePath}`);

    const BATCH_SIZE = 100000;
    let batch: ParityReportEntry[] = [];
    let batchNumber = 1;

    // Extract directory and base name from the baseFilePath
    const directory = path.dirname(baseFilePath);
    const ext = path.extname(baseFilePath);
    const baseName = path.basename(baseFilePath, ext);

    for (let i = 0; i < report.length; i++) {
        batch.push(report[i]);

        if (batch.length === BATCH_SIZE) {
            // Generate dynamic file path for the current batch
            const batchFileName = `${baseName}_batch${batchNumber}${ext}`;
            const batchFilePath = path.join(directory, batchFileName);

            // Write the current batch to its file
            await writeBatchToFile(batch, batchFilePath, batchNumber);

            batchNumber++;
            batch = [];
        }
    }

    // Write any remaining records in the final batch
    if (batch.length > 0) {
        const batchFileName = `${baseName}_batch${batchNumber}${ext}`;
        const batchFilePath = path.join(directory, batchFileName);
        await writeBatchToFile(batch, batchFilePath, batchNumber);
    }

    console.log(`All batches have been successfully saved to CSV files.`);
}

/**
 * Writes a batch of records to a specified CSV file.
 * @param batchToWrite - Array of ParityReportEntry representing the current batch.
 * @param filePath - Path to save the current batch CSV file.
 * @param batchNumber - The current batch number for logging purposes.
 */
async function writeBatchToFile(batchToWrite: ParityReportEntry[], filePath: string, batchNumber: number): Promise<void> {
    return new Promise((resolve, reject) => {
        const headers = [
            'Proserve ID',
            'Sentral ID',
            'Proserve Column',
            'Sentral Column',
            'Proserve Value',
            'Sentral Value',
            'Status',
            'Object',
            'Job Date',
        ];

        const fileStream = fs.createWriteStream(filePath);
        // Write headers
        fileStream.write(headers.join(',') + '\n');

        // Write each record
        batchToWrite.forEach(entry => {
            const values = [
                entry.proserveId,
                entry.sentralId,
                entry.proserveColumn,
                entry.sentralColumn,
                `"${entry.proserveValue.replace(/"/g, '""')}"`, // Handle quotes
                `"${entry.sentralValue.replace(/"/g, '""')}"`,
                entry.status,
                'Opportunity', // Static value for 'Object'
                '2024-11-16',    // Static value for 'Job Date'
            ];
            fileStream.write(values.join(',') + '\n');
        });

        fileStream.end();

        fileStream.on('finish', () => {
            console.log(`Batch ${batchNumber}: Successfully saved to ${filePath}`);
            resolve();
        });

        fileStream.on('error', (error) => {
            console.error(`Batch ${batchNumber}: Error writing to ${filePath}:`, error);
            reject(error);
        });
    });
}

async function readCsvIntoMap(filePath: string): Promise<Map<string, string[]>> {
    const fileStream = fs.createReadStream(filePath);
    const rl = readline.createInterface({
        input: fileStream,
        crlfDelay: Infinity,
    });

    const map = new Map<string, string[]>();

    for await (const line of rl) {
        const values = line.split(',');
        if (values.length > 1) {
            const key = values[0].trim(); // Assuming the first column is the key
            const rest = values.slice(1).map(val => val.trim()); // Rest are values
            map.set(key, rest);
        }
    }

    return map;
}

async function checkParity() {
    console.log('Starting parity check process...');
    try {
        // Define file paths
        const sentralFilePath = '/Users/svvemur/Documents/SentralData/SnetralMerged.csv';
        const proserveFilePath = '/Users/svvemur/Downloads/ProserveQueriesAndData/ProserveMerged.csv';
        const sentralIdColumn = 'Id';
        const proserveIdColumn = 'Id';
        const fieldMappingsPath = '/Users/svvemur/workspaces/AWSCRMTestAutomation/src/AWSCRMTestAutomation/TO-BE Data model - Opportunity.csv';
        const recordIdMappingPath = '/Users/svvemur/workspaces/AWSCRMTestAutomation/src/AWSCRMTestAutomation/output.csv';
        const baseReportFilePath = './Opportunity_20241116.csv'; // Base path for report batches

        console.log('Reading record ID mappings...');
        const recordIdMapping = await readFirstTwoColumns(recordIdMappingPath);
        console.log(`Loaded ${recordIdMapping.length} record ID mappings.`);

        console.log('Parsing Sentral CSV data...');
        const sentralData = await parseCSVFast(sentralFilePath);

        console.log('Parsing Proserve CSV data...');
        const proserveData = await parseCSVFast(proserveFilePath);

        console.log('Generating field mappings...');
        const mappings: Mapping[] = await generateFieldMappings(fieldMappingsPath);
        console.log(`Generated ${mappings.length} field mappings.`);

        console.log('Generating parity report...');
        const { report, summary } = generateParityReport(
            sentralData,
            proserveData,
            mappings,
            recordIdMapping,
            sentralIdColumn,
            proserveIdColumn,
        );

        console.log('Data parity check completed.');
        console.log(`Total mismatches: ${summary.totalMismatches}`);
        console.log(`Total records skipped: ${summary.skipped}`);

        if (report.length === 0) {
            console.log('Data is in parity, no discrepancies found.');
        } else {
            console.log('Saving parity report to CSV batches...');
            await saveReportToCSVInBatches(report, summary, baseReportFilePath);
            console.log('CSV batch report generation completed.');
            console.log(`Total mismatches: ${summary.totalMismatches}`);
        }

        console.log('Parity check process completed successfully.');
    } catch (error) {
        console.error('Error processing files:', error);
    }
}

/**
 * Generates field mappings by reading from a CSV file.
 * @param filePath - Path to the field mappings CSV file.
 * @returns A Promise that resolves to an array of Mapping objects.
 */
function generateFieldMappings(filePath: string): Promise<Mapping[]> {
    console.log(`Generating field mappings from file: ${filePath}`);
    return new Promise((resolve, reject) => {
        fs.readFile(filePath, 'utf8', (err, data) => {
            if (err) {
                console.error(`Error reading field mappings file ${filePath}:`, err);
                return reject(err);
            }

            const mappings: Mapping[] = [];
            const lines = data.split('\n');

            const headers = lines[0].split(',');
            const fieldNameIndex = headers.indexOf('Field Name');
            const fieldReplacedIndex = headers.indexOf('Field Replaced from ProServe instance');

            if (fieldNameIndex === -1 || fieldReplacedIndex === -1) {
                const errorMsg = 'Required columns not found in field mappings CSV file.';
                console.error(errorMsg);
                return reject(new Error(errorMsg));
            }

            for (let i = 1; i < lines.length; i++) {
                const columns = lines[i].split(',');

                if (columns[fieldReplacedIndex]) {
                    mappings.push({
                        sentralColumn: columns[fieldNameIndex].trim(),
                        proserveColumn: columns[fieldReplacedIndex].trim(),
                    });
                }

                if (i % 1000 === 0) {
                    console.log(`Processed ${i} lines for field mappings...`);
                }
            }

            console.log(`Completed generating field mappings. Total mappings: ${mappings.length}`);
            resolve(mappings);
        });
    });
}

/**
 * Reads the first two columns from a CSV file.
 * @param filePath - Path to the CSV file.
 * @returns A Promise that resolves to an array of [string, string] tuples representing the first two columns.
 */
async function readFirstTwoColumns(filePath: string): Promise<[string, string][]> {
    console.log(`Reading first two columns from file: ${filePath}`);
    return new Promise<[string, string][]>((resolve, reject) => {
        const results: [string, string][] = [];

        const readStream = fs.createReadStream(filePath, { encoding: 'utf8' });

        readStream.on('error', (err) => {
            console.error(`Error reading the file ${filePath}:`, err.message);
            reject(`Error reading the file: ${err.message}`);
        });

        const parser = csv()
            .on('headers', (headers: string[]) => {
                if (headers.length < 2) {
                    const errorMsg = 'CSV does not contain at least two columns.';
                    console.error(errorMsg);
                    parser.destroy(new Error(errorMsg));
                }
                console.log(`CSV headers: ${headers.slice(0, 2).join(', ')}`);
            })
            .on('data', (row: any) => {
                try {
                    const keys = Object.keys(row);
                    if (keys.length < 2) {
                        return;
                    }

                    const col1 = row[keys[0]].trim();
                    const col2 = row[keys[1]].trim();

                    results.push([col1, col2]);

                    if (results.length % 1000 === 0) {
                        console.log(`Read ${results.length} record ID mappings...`);
                    }
                } catch (err) {
                    console.error(`Error processing a row:`, err);
                }
            })
            .on('end', () => {
                console.log(`Finished reading record ID mappings. Total mappings: ${results.length}`);
                resolve(results);
            })
            .on('error', (err: Error) => {
                console.error(`Error processing the CSV ${filePath}:`, err.message);
                reject(`Error processing the CSV: ${err.message}`);
            });

        readStream.pipe(parser);
    });
}

export default checkParity;


Modify the code to use a proper csv writer as there are issues in current code