batch-speech-utils/example/cx-analysis.js

const { request } = require('undici');
const { pipeline } = require('stream');
const fs = require('fs');
const path = require('path');
const os = require('os');
const fsP = require('fs').promises;
const createAudioService = require('..');
const audioService = createAudioService();

// Constants
const JAMBONZ_API_BASE_URL = 'https://jambonz.one/api/v1';

// Utility functions


async function readJSONFile(filePath) {
  try {
    const data = await fsP.readFile(filePath, 'utf8');
    return JSON.parse(data);
  } catch (err) {
    throw new Error(`Error reading JSON file: ${err.message}`);
  }
}

async function writeJSONFile(filePath, data) {
  try {
    await fs.writeFileSync(filePath, JSON.stringify(data, null, 2));
  } catch (err) {
    throw new Error(`Error writing JSON file: ${err.message}`);
  }
}

function extractJambonzTrace(data, audioService) {
  const participants = [];
  let duration = 0;
  const uniqueNumbers = new Set();
  const result = {
    conversation: {
      'as heard': {
        'transcripts': [],
        'transcription vendor': '',
      }
    }
  };

  data.resourceSpans.forEach((resourceSpan) => {
    resourceSpan.instrumentationLibrarySpans.forEach((librarySpan) => {
      librarySpan.spans.forEach((span) => {
        const startTime = span.startTimeUnixNano;
        const endTime = span.endTimeUnixNano;
        span.attributes.forEach((attribute) => {
          if (attribute.value && attribute.value.stringValue) {
            switch (attribute.key) {
              case 'from':
                if (!uniqueNumbers.has(attribute.value.stringValue)) {
                  uniqueNumbers.add(attribute.value.stringValue);
                  participants.push({
                    'type': 'human',
                    'initiatedConversation': true,
                    'id': {
                      'name': null,
                      'phone': attribute.value.stringValue
                    }
                  });
                }
                break;
              case 'duration':
                duration += parseInt(attribute.value.stringValue || '0', 10);
                break;
              case 'to':
                if (!uniqueNumbers.has(attribute.value.stringValue)) {
                  uniqueNumbers.add(attribute.value.stringValue);
                  participants.push({
                    'type': 'machine',
                    'initiatedConversation': false,
                    'id': {
                      'name': 'jambonz.one',
                      'phone': attribute.value.stringValue
                    }
                  });
                }
                break;
              case 'http.body':
                if (attribute.value) {
                  const httpBody = JSON.parse(attribute.value.stringValue);
                  if (httpBody.speech && httpBody.speech.alternatives && httpBody.speech.alternatives.length > 0) {
                    const alternative = httpBody.speech.alternatives[0];
                    result.conversation['as heard']['transcripts'].push({
                      transcript: alternative.transcript,
                      confidence: alternative.confidence,
                      vendor:httpBody.speech.vendor.name,
                      startTime: parseInt(startTime, 10),
                      endTime: parseInt(endTime, 10)
                    });
                    result.conversation['as heard']['transcription vendor'] = httpBody.speech.vendor.name;
                  }
                }
                break;
              case 'stt.result':
                if (attribute.value) {
                  const sttResult = JSON.parse(attribute.value.stringValue);
                  sttResult.alternatives.forEach((alternative) => {
                    result.conversation['as heard']['transcripts'].push({
                      transcript: alternative.transcript,
                      confidence: alternative.confidence,
                      startTime: parseInt(startTime, 10),
                      endTime: parseInt(endTime, 10)
                    });
                    if (!result.conversation['as heard']['transcription vendor']) {
                      result.conversation['as heard']['transcription vendor'] = sttResult.vendor.name;
                    }
                  });
                }
                break;
            }
          }
        });
      });
    });
  });

  // Sort the transcripts by start time
  result.conversation['as heard']['transcripts'].sort((a, b) => a.startTime - b.startTime);

  console.log(result.conversation['as heard'].transcripts);
  return { 'participants': participants, 'duration': duration, 'conversation': result.conversation };
}


async function fetchCallDetails(callId) {
  const headers = {
    'Authorization': `Bearer ${process.env.JAMBONZ_API_TOKEN}`,
    'Accept': '*/*',
  };
    // eslint-disable-next-line max-len
  const url = `${JAMBONZ_API_BASE_URL}/Accounts/${process.env.ACCOUNT_SID}/RecentCalls?page=1&count=25&filter=${callId}`;
  try {
    const { statusCode, body } = await request(url, { headers });
    if (statusCode !== 200) {
      throw new Error(`Request failed with status code: ${statusCode}`);
    }
    var data = await body.json();
    data = data.data[0];
    const response = {
      'call_start' : data.answered_at,
      'duration_ms':data.duration * 1000,
      'trace_id': data.trace_id,
      'recording_url' : data.recording_url
    };
    // await writeJSONFile('response_call_details.json', targetCall);
    console.log('Call details has been received');
    return response;
  } catch (error) {
    console.error('Error fetching data:', error);
    throw error;
  }
}

async function fetchJambonzTrace(trace_id) {
  const url = `${JAMBONZ_API_BASE_URL}/Accounts/${process.env.ACCOUNT_SID}/RecentCalls/trace/${trace_id}`;
  const headers = {
    'Authorization': `Bearer ${process.env.JAMBONZ_API_TOKEN}`,
    'Accept': '*/*',
  };

  try {
    const { statusCode, body } = await request(url, { headers });
    if (statusCode !== 200) {
      throw new Error(`Request failed with status code: ${statusCode}`);
    }
    const data = await body.json();
    await writeJSONFile('trace.json', data);
    console.log('Trace received');
    const jsonData = await readJSONFile('./trace.json');
    return extractJambonzTrace(jsonData, audioService);
  } catch (error) {
    console.error('Error fetching trace data:', error);
    throw error;
  }
}

async function downloadFile(url, tempFilePath, finalFilePath, token, callDetails) {
  try {
    const { statusCode, body } = await request(url, {
      method: 'GET',
      headers: {
        'Authorization': `Bearer ${token}`,
        'Accept': '*/*'
      }
    });

    if (statusCode !== 200) {
      throw new Error(`Failed to download file: ${statusCode}`);
    }

    const tempWriteStream = fs.createWriteStream(tempFilePath);

    pipeline(
      body,
      tempWriteStream,
      async(err) => {
        if (err) {
          console.error('Pipeline failed:', err);
        } else {
          try {
            await processAndRedactFile(tempFilePath, finalFilePath, callDetails);
          } catch (processingError) {
            console.error('Error processing file:', processingError);
          } finally {
            // Clean up the temporary file
            fs.unlink(tempFilePath, (err) => {
              if (err) {
                console.error('Error deleting temp file:', err);
              } else {
                console.log('Temporary file deleted.');
              }
            });
          }
        }
      }
    );
  } catch (error) {
    console.error('Error fetching file:', error);
  }
}


async function processAndRedactFile(audioFilePath, outputFilePath, callDetails) {
  const credentials = { 'vendor': 'deepgram', 'apiKey': process.env.DEEPGRAM_API_KEY };

  try {
    const transcriptionResults = await audioService.transcribe(credentials, audioFilePath);
    // const timestampsToRedact = transcriptionResults.redactionTimestamps;

    // await audioService.redact({
    //   credentials: credentials,
    //   transcriptionData: timestampsToRedact,
    //   audioPath: audioFilePath,
    //   audioOutputPath: outputFilePath
    // });

    const jambonzTrace = await fetchJambonzTrace(callDetails.trace_id);
    const {'redactionTimestamps':_, ...transcript} = transcriptionResults;
    jambonzTrace.duration = callDetails.duration_ms;
    jambonzTrace.call_start = callDetails.call_start;
    jambonzTrace.transcript = transcript;
    const callStartMillis = new Date(callDetails.call_start).getTime();
    jambonzTrace.transcript.speechEvents = jambonzTrace.transcript.speechEvents.map((event) => {
      const startTimestamp = new Date(callStartMillis + event.start * 1000).toISOString();
      // const endTimestamp = new Date(callStartMillis + event.end * 1000).toISOString();

      return {
        spokenAt: startTimestamp,
        ...event

        // start: startTimestamp,
        // end: endTimestamp,
      };
    });
    const inputData = jambonzTrace;
    const deepgramTranscripts = inputData.transcript.speechEvents;
    const asHeardTranscripts = inputData.conversation['as heard'].transcripts;

    // Determine the minimum length to avoid out-of-bounds errors
    const minLength = Math.min(deepgramTranscripts.length, asHeardTranscripts.length);

    // Transform the data
    const turns = [];

    for (let i = 0; i < minLength ; i++) {
      const offer = deepgramTranscripts[i];
      const response = deepgramTranscripts[i + 1];
      const follow_up = deepgramTranscripts[i + 2];
      const offerAsHeard = asHeardTranscripts[i];
      const responseAsHeard = asHeardTranscripts[i + 1];

      turns.push({
        offer: offer.sentence,
        response: {
          asTranscribed: {
            transcript: response.sentence,
            vendor: "deepgram",
            confidence: null
          },
          asHeard: {
            transcript: responseAsHeard ? responseAsHeard.transcript : '',
            vendor: responseAsHeard ? responseAsHeard.vendor : '',
            confidence: responseAsHeard ? responseAsHeard.confidence : 0

          }
        },
        follow_up: follow_up.sentence
      });
      i++;
    }

    // Write the output to a file
    const outputData = {
      startTime: inputData.call_start,
      endTime: new Date(new Date(inputData.call_start).getTime() + inputData.duration).toISOString(),
      recognizer: "deepgram",
      turns: turns
    };

    await writeJSONFile(`${process.env.OUTPUT_PATH}/transcription.json`, jambonzTrace);
    await writeJSONFile(`${process.env.OUTPUT_PATH}/cx_transcription.json`, outputData);

    console.log('File redacted and saved successfully.');
  } catch (error) {
    console.error('Failed to redact audio:', error);
    throw error;
  }
}

// Main execution
async function main() {
  try {
    const [,, callId] = process.argv;
    if (!callId) {
      throw new Error('Missing required command line arguments. Usage: node app.js <callId>');
    }
    const callDetails = await fetchCallDetails(callId);
    const dateFormat = callDetails.recording_url.split('record/')[1];
    const url = `${JAMBONZ_API_BASE_URL}/Accounts/${process.env.ACCOUNT_SID}/RecentCalls/${callId}/record/${dateFormat}`;
    //
    //
    // // console.log(callDetails);
    const finalFilePath = path.resolve(__dirname, path.join(process.env.OUTPUT_PATH, 'redacted_audio.wav'));
    const tempFolder = process.env.TEMP_FOLDER || os.tmpdir();
    const tempFilePath = path.join(tempFolder, 'tempDownloadedFile.mp3');
    // //
    // //
    await downloadFile(url, tempFilePath, finalFilePath, process.env.JAMBONZ_API_TOKEN, callDetails);
  } catch (error) {
    console.error('Error in main process:', error);
    process.exit(1);
  }
}

main();