From 4f8d71b7cccf5c92d1154e2c95847ad5be2be8eb Mon Sep 17 00:00:00 2001 From: Hongbo Wu Date: Wed, 24 Aug 2022 15:25:47 +0800 Subject: [PATCH] Add a pubsub subscription listening on the new article to automatically synthesize article --- packages/api/src/routers/svc/speech.ts | 97 ------------ .../api/src/routers/svc/text_to_speech.ts | 140 ++++++++++++++++++ packages/api/src/server.ts | 2 +- packages/api/src/services/speech.ts | 35 ++++- 4 files changed, 174 insertions(+), 100 deletions(-) delete mode 100644 packages/api/src/routers/svc/speech.ts create mode 100644 packages/api/src/routers/svc/text_to_speech.ts diff --git a/packages/api/src/routers/svc/speech.ts b/packages/api/src/routers/svc/speech.ts deleted file mode 100644 index 87ddfb981..000000000 --- a/packages/api/src/routers/svc/speech.ts +++ /dev/null @@ -1,97 +0,0 @@ -import express from 'express' -import cors from 'cors' -import { corsConfig } from '../../utils/corsConfig' -import { getRepository } from '../../entity/utils' -import { getPageById } from '../../elastic/pages' -import { synthesizeTextToSpeech } from '../../utils/textToSpeech' -import { Speech, SpeechState } from '../../entity/speech' -import { buildLogger } from '../../utils/logger' -import { getClaimsByToken } from '../../utils/auth' -import { setSpeechFailure } from '../../services/speech' - -const logger = buildLogger('app.dispatch') - -export function speechServiceRouter() { - const router = express.Router() - - router.options('/', cors({ ...corsConfig, maxAge: 600 })) - // eslint-disable-next-line @typescript-eslint/no-misused-promises - router.post('/', async (req, res) => { - logger.info('Speech svc request', { - // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment - body: req.body, - }) - const token = req.query.token as string - try { - if (!(await getClaimsByToken(token))) { - logger.info('Unauthorized request', { token }) - return res.status(200).send('UNAUTHORIZED') - } - } catch (error) { - logger.error('Unauthorized request', { token, error }) - return res.status(200).send('UNAUTHORIZED') - } - - const { userId, speechId } = req.body as { - userId: string - speechId: string - } - if (!userId || !speechId) { - return res.status(200).send('Invalid data') - } - - logger.info(`Create article speech`, { - body: { - userId, - speechId, - }, - labels: { - source: 'CreateArticleSpeech', - }, - }) - const speech = await getRepository(Speech).findOneBy({ - id: speechId, - user: { id: userId }, - }) - if (!speech) { - return res.status(200).send('Speech not found') - } - - const page = await getPageById(speech.elasticPageId) - if (!page) { - await setSpeechFailure(speech.id) - return res.status(200).send('Page not found') - } - - try { - const startTime = Date.now() - const speechOutput = await synthesizeTextToSpeech({ - id: speech.id, - text: page.content, - languageCode: page.language, - voice: speech.voice, - textType: 'ssml', - }) - logger.info('Created speech', { - audioFileName: speechOutput.audioFileName, - speechMarksFileName: speechOutput.speechMarksFileName, - duration: Date.now() - startTime, - }) - - // set state to completed - await getRepository(Speech).update(speech.id, { - audioFileName: speechOutput.audioFileName, - speechMarksFileName: speechOutput.speechMarksFileName, - state: SpeechState.COMPLETED, - }) - - res.status(200).send('OK') - } catch (error) { - logger.error(`Error creating article speech`, { error }) - await setSpeechFailure(speech.id) - res.status(500).send('Error creating article speech') - } - }) - - return router -} diff --git a/packages/api/src/routers/svc/text_to_speech.ts b/packages/api/src/routers/svc/text_to_speech.ts new file mode 100644 index 000000000..05f0f60b4 --- /dev/null +++ b/packages/api/src/routers/svc/text_to_speech.ts @@ -0,0 +1,140 @@ +/* eslint-disable @typescript-eslint/no-misused-promises */ +/* eslint-disable @typescript-eslint/no-unsafe-assignment */ +/* eslint-disable @typescript-eslint/explicit-module-boundary-types */ +import express from 'express' +import cors from 'cors' +import { corsConfig } from '../../utils/corsConfig' +import { getRepository } from '../../entity/utils' +import { getPageById } from '../../elastic/pages' +import { Speech, SpeechState } from '../../entity/speech' +import { buildLogger } from '../../utils/logger' +import { getClaimsByToken } from '../../utils/auth' +import { + setSpeechFailure, + shouldSynthesize, + synthesize, +} from '../../services/speech' +import { readPushSubscription } from '../../datalayer/pubsub' + +const logger = buildLogger('app.dispatch') + +export function speechServiceRouter() { + const router = express.Router() + + router.options( + '/auto-synthesis', + cors({ ...corsConfig, maxAge: 600 }) + ) + // eslint-disable-next-line @typescript-eslint/no-misused-promises + router.post('/auto-synthesize', async (req, res) => { + logger.info('auto-synthesize') + const { message: msgStr, expired } = readPushSubscription(req) + + if (!msgStr) { + return res.status(400).send('Bad Request') + } + + if (expired) { + logger.info('discarding expired message') + return res.status(200).send('Expired') + } + + try { + const data: { userId: string; type: string; id: string } = + JSON.parse(msgStr) + const { userId, type, id } = data + if (!userId || !type) { + logger.info('No userId or type found in message') + return res.status(400).send('Bad Request') + } + + if (type.toUpperCase() !== 'PAGE') { + logger.info('Not a page') + return res.status(200).send('Not a page') + } + + // checks if this page needs to be synthesized automatically + const page = await getPageById(id) + if (!page) { + logger.info('No page found', { id }) + return res.status(200).send('No page found') + } + + if (await shouldSynthesize(userId, page)) { + logger.info('page needs to be synthesized') + // initialize state + const speech = await getRepository(Speech).save({ + user: { id: userId }, + elasticPageId: id, + state: SpeechState.INITIALIZED, + }) + await synthesize(page, speech) + logger.info('page synthesized') + } + + res.status(200).send('Page should not synthesize') + } catch (err) { + logger.error('Auto synthesize failed', err) + res.status(500).send(err) + } + }) + + router.options('/', cors({ ...corsConfig, maxAge: 600 })) + // eslint-disable-next-line @typescript-eslint/no-misused-promises + router.post('/', async (req, res) => { + logger.info('Synthesize svc request', { + // eslint-disable-next-line @typescript-eslint/no-unsafe-assignment + body: req.body, + }) + const token = req.query.token as string + try { + if (!(await getClaimsByToken(token))) { + logger.info('Unauthorized request', { token }) + return res.status(200).send('UNAUTHORIZED') + } + } catch (error) { + logger.error('Unauthorized request', { token, error }) + return res.status(200).send('UNAUTHORIZED') + } + + const { userId, speechId } = req.body as { + userId: string + speechId: string + } + if (!userId || !speechId) { + return res.status(200).send('Invalid data') + } + + logger.info(`Create article speech`, { + body: { + userId, + speechId, + }, + labels: { + source: 'CreateArticleSpeech', + }, + }) + const speech = await getRepository(Speech).findOneBy({ + id: speechId, + user: { id: userId }, + }) + if (!speech) { + return res.status(200).send('Speech not found') + } + + const page = await getPageById(speech.elasticPageId) + if (!page) { + await setSpeechFailure(speech.id) + return res.status(200).send('Page not found') + } + + try { + await synthesize(page, speech) + } catch (error) { + logger.error(`Error synthesizing article`, { error }) + res.status(500).send('Error synthesizing article') + } + }) + + return router +} diff --git a/packages/api/src/server.ts b/packages/api/src/server.ts index 7c9273f92..bc82e067f 100755 --- a/packages/api/src/server.ts +++ b/packages/api/src/server.ts @@ -45,7 +45,7 @@ import { uploadServiceRouter } from './routers/svc/upload' import rateLimit from 'express-rate-limit' import { webhooksServiceRouter } from './routers/svc/webhooks' import { integrationsServiceRouter } from './routers/svc/integrations' -import { speechServiceRouter } from './routers/svc/speech' +import { speechServiceRouter } from './routers/svc/text_to_speech' const PORT = process.env.PORT || 4000 diff --git a/packages/api/src/services/speech.ts b/packages/api/src/services/speech.ts index 997773624..ef2c4ee2d 100644 --- a/packages/api/src/services/speech.ts +++ b/packages/api/src/services/speech.ts @@ -3,6 +3,7 @@ import { Speech, SpeechState } from '../entity/speech' import { searchPages } from '../elastic/pages' import { Page } from '../elastic/types' import { SortBy, SortOrder } from '../utils/search' +import { synthesizeTextToSpeech } from '../utils/textToSpeech' export const setSpeechFailure = async (id: string) => { // update state @@ -12,11 +13,11 @@ export const setSpeechFailure = async (id: string) => { } /* - * We should not transcribe the page when: + * We should not synthesize the page when: ** 1. User has no recent listens the last 30 days ** 2. User has a recent listen but the page was saved after the listen */ -export const shouldTranscribe = async ( +export const shouldSynthesize = async ( userId: string, page: Page ): Promise => { @@ -44,3 +45,33 @@ export const shouldTranscribe = async ( page.savedAt < recentListenedPage[0].listenedAt ) } + +export const synthesize = async (page: Page, speech: Speech): Promise => { + try { + console.log('synthesizing', speech.id) + const startTime = Date.now() + const speechOutput = await synthesizeTextToSpeech({ + id: speech.id, + text: page.content, + languageCode: page.language, + voice: speech.voice, + textType: 'ssml', + }) + console.log('Synthesized article', { + audioFileName: speechOutput.audioFileName, + speechMarksFileName: speechOutput.speechMarksFileName, + duration: Date.now() - startTime, + }) + + // set state to completed + await getRepository(Speech).update(speech.id, { + audioFileName: speechOutput.audioFileName, + speechMarksFileName: speechOutput.speechMarksFileName, + state: SpeechState.COMPLETED, + }) + } catch (error) { + console.log('Error synthesize article', error) + await setSpeechFailure(speech.id) + throw error + } +}