GoogleSpeechTranscriptionService extends AbstractJobProducer implements TranscriptionService { public void activate(ComponentContext cc) { enabled = OsgiUtil.getOptCfgAsBoolean(cc.getProperties(), ENABLED_CONFIG).get(); if (!enabled) { logger.info("Service disabled. If you want to enable it, please update the service configuration."); return; } clientId = OsgiUtil.getComponentContextProperty(cc, GOOGLE_CLOUD_CLIENT_ID); clientSecret = OsgiUtil.getComponentContextProperty(cc, GOOGLE_CLOUD_CLIENT_SECRET); clientToken = OsgiUtil.getComponentContextProperty(cc, GOOGLE_CLOUD_REFRESH_TOKEN); storageBucket = OsgiUtil.getComponentContextProperty(cc, GOOGLE_CLOUD_BUCKET); Option<String> tokenOpt = OsgiUtil.getOptCfg(cc.getProperties(), GOOGLE_CLOUD_TOKEN_ENDPOINT_URL); if (tokenOpt.isSome()) { tokenEndpoint = tokenOpt.get(); logger.info("Access token endpoint is set to {}", tokenEndpoint); } else { logger.info("Default access token endpoint will be used"); } Option<String> profanityOpt = OsgiUtil.getOptCfg(cc.getProperties(), PROFANITY_FILTER); if (profanityOpt.isSome()) { profanityFilter = Boolean.parseBoolean(profanityOpt.get()); logger.info("Profanity filter is set to {}", profanityFilter); } else { logger.info("Default profanity filter will be used"); } Option<String> languageOpt = OsgiUtil.getOptCfg(cc.getProperties(), GOOGLE_SPEECH_LANGUAGE); if (languageOpt.isSome()) { defaultLanguage = languageOpt.get(); logger.info("Language used is {}", defaultLanguage); } else { logger.info("Default language will be used"); } Option<String> encodingOpt = OsgiUtil.getOptCfg(cc.getProperties(), ENCODING_EXTENSION); if (encodingOpt.isSome()) { defaultEncoding = encodingOpt.get(); logger.info("Encoding used is {}", defaultEncoding); } else { logger.info("Default encoding will be used"); } Option<String> wfOpt = OsgiUtil.getOptCfg(cc.getProperties(), WORKFLOW_CONFIG); if (wfOpt.isSome()) { workflowDefinitionId = wfOpt.get(); } logger.info("Workflow definition is {}", workflowDefinitionId); Option<String> intervalOpt = OsgiUtil.getOptCfg(cc.getProperties(), DISPATCH_WORKFLOW_INTERVAL_CONFIG); if (intervalOpt.isSome()) { try { workflowDispatchInterval = Long.parseLong(intervalOpt.get()); } catch (NumberFormatException e) { logger.warn("Invalid configuration for Workflow dispatch interval. Default used instead: {}", workflowDispatchInterval); } } logger.info("Workflow dispatch interval is {} seconds", workflowDispatchInterval); Option<String> bufferOpt = OsgiUtil.getOptCfg(cc.getProperties(), COMPLETION_CHECK_BUFFER_CONFIG); if (bufferOpt.isSome()) { try { completionCheckBuffer = Long.parseLong(bufferOpt.get()); } catch (NumberFormatException e) { logger.warn("Invalid configuration for {} : {}. Default used instead: {}", new Object[]{COMPLETION_CHECK_BUFFER_CONFIG, bufferOpt.get(), completionCheckBuffer}); } } logger.info("Completion check buffer is {} seconds", completionCheckBuffer); Option<String> maxProcessingOpt = OsgiUtil.getOptCfg(cc.getProperties(), MAX_PROCESSING_TIME_CONFIG); if (maxProcessingOpt.isSome()) { try { maxProcessingSeconds = Long.parseLong(maxProcessingOpt.get()); } catch (NumberFormatException e) { logger.warn("Invalid configuration for maximum processing time. Default used instead: {}", maxProcessingSeconds); } } logger.info("Maximum time a job is checked after it should have ended is {} seconds", maxProcessingSeconds); Option<String> cleaupOpt = OsgiUtil.getOptCfg(cc.getProperties(), CLEANUP_RESULTS_DAYS_CONFIG); if (cleaupOpt.isSome()) { try { cleanupResultDays = Integer.parseInt(cleaupOpt.get()); } catch (NumberFormatException e) { logger.warn("Invalid configuration for clean up days. Default used instead: {}", cleanupResultDays); } } logger.info("Cleanup result files after {} days", cleanupResultDays); systemAccount = OsgiUtil.getContextProperty(cc, OpencastConstants.DIGEST_USER_PROPERTY); scheduledExecutor.scheduleWithFixedDelay(new WorkflowDispatcher(), 120, workflowDispatchInterval, TimeUnit.SECONDS); scheduledExecutor.scheduleWithFixedDelay(new ResultsFileCleanup(), 1, 1, TimeUnit.DAYS); Option<String> optTo = OsgiUtil.getOptCfg(cc.getProperties(), NOTIFICATION_EMAIL_CONFIG); if (optTo.isSome()) { toEmailAddress = optTo.get(); } else { optTo = OsgiUtil.getOptContextProperty(cc, OpencastConstants.ADMIN_EMAIL_PROPERTY); if (optTo.isSome()) { toEmailAddress = optTo.get(); } } if (toEmailAddress != null) { logger.info("Notification email set to {}", toEmailAddress); } else { logger.warn("Email notification disabled"); } Option<String> optCluster = OsgiUtil.getOptContextProperty(cc, OpencastConstants.ENVIRONMENT_NAME_PROPERTY); if (optCluster.isSome()) { clusterName = optCluster.get(); } logger.info("Environment name is {}", clusterName); logger.info("Activated!"); } GoogleSpeechTranscriptionService(); void activate(ComponentContext cc); @Override Job startTranscription(String mpId, Track track, String language); @Override Job startTranscription(String mpId, Track track); @Override void transcriptionDone(String mpId, Object obj); @Override String getLanguage(); @Override void transcriptionError(String mpId, Object obj); String getTranscriptionResults(String jobId); @Override MediaPackageElement getGeneratedTranscription(String mpId, String jobId); String getTranscriptionStatus(String mpId); void setServiceRegistry(ServiceRegistry serviceRegistry); void setSecurityService(SecurityService securityService); void setUserDirectoryService(UserDirectoryService userDirectoryService); void setOrganizationDirectoryService(OrganizationDirectoryService organizationDirectoryService); void setSmtpService(SmtpService service); void setWorkspace(Workspace ws); void setWorkingFileRepository(WorkingFileRepository wfr); void setDatabase(TranscriptionDatabase service); void setAssetManager(AssetManager service); void setWorkflowService(WorkflowService service); static final String DEFAULT_WF_DEF; static final String ENABLED_CONFIG; static final String GOOGLE_SPEECH_LANGUAGE; static final String PROFANITY_FILTER; static final String WORKFLOW_CONFIG; static final String DISPATCH_WORKFLOW_INTERVAL_CONFIG; static final String COMPLETION_CHECK_BUFFER_CONFIG; static final String MAX_PROCESSING_TIME_CONFIG; static final String NOTIFICATION_EMAIL_CONFIG; static final String CLEANUP_RESULTS_DAYS_CONFIG; static final String GOOGLE_CLOUD_CLIENT_ID; static final String GOOGLE_CLOUD_CLIENT_SECRET; static final String GOOGLE_CLOUD_REFRESH_TOKEN; static final String GOOGLE_CLOUD_BUCKET; static final String GOOGLE_CLOUD_TOKEN_ENDPOINT_URL; static final String ENCODING_EXTENSION; }
@Test public void testWorkflowDispatcherRunTranscriptionCompletedState() throws Exception { service.activate(cc); database.storeJobControl(MP_ID, TRACK_ID, JOB_ID, TranscriptionJobControl.Status.InProgress.name(), TRACK_DURATION, DATE_EXPECTED, PROVIDER); database.storeJobControl("mpId2", "audioTrack2", "jobId2", TranscriptionJobControl.Status.InProgress.name(), TRACK_DURATION, DATE_EXPECTED, PROVIDER); database.updateJobControl(JOB_ID, TranscriptionJobControl.Status.TranscriptionComplete.name()); Capture<Set<String>> capturedMpIds = mockAssetManagerAndWorkflow(GoogleSpeechTranscriptionService.DEFAULT_WF_DEF, true); WorkflowDispatcher dispatcher = service.new WorkflowDispatcher(); dispatcher.run(); Assert.assertEquals(1, capturedMpIds.getValue().size()); Assert.assertEquals(MP_ID, capturedMpIds.getValue().iterator().next()); TranscriptionJobControl job = database.findByJob(JOB_ID); Assert.assertNotNull(job); Assert.assertEquals(TranscriptionJobControl.Status.Closed.name(), job.getStatus()); }
@Test public void testWorkflowDispatcherRunProgressState() throws Exception { service.activate(cc); InputStream stream = GoogleSpeechTranscriptionServiceTest.class.getResourceAsStream("/" + PULLED_TRANSCRIPTION_FILE); database.storeJobControl(MP_ID, TRACK_ID, JOB_ID, TranscriptionJobControl.Status.InProgress.name(), 0, null, PROVIDER); database.storeJobControl("mpId2", "audioTrack2", "jobId2", TranscriptionJobControl.Status.InProgress.name(), TRACK_DURATION, DATE_EXPECTED, PROVIDER); EasyMock.expect(workspace.putInCollection(EasyMock.anyObject(String.class), EasyMock.anyObject(String.class), EasyMock.anyObject(InputStream.class))).andReturn(new URI("http: EasyMock.replay(workspace); HttpEntity httpEntity = EasyMock.createNiceMock(HttpEntity.class); EasyMock.expect(httpEntity.getContent()).andReturn(stream); CloseableHttpResponse response = EasyMock.createNiceMock(CloseableHttpResponse.class); StatusLine status = EasyMock.createNiceMock(StatusLine.class); EasyMock.expect(response.getStatusLine()).andReturn(status).anyTimes(); EasyMock.expect(response.getEntity()).andReturn(httpEntity).anyTimes(); EasyMock.expect(status.getStatusCode()).andReturn(HttpStatus.SC_OK).anyTimes(); EasyMock.replay(httpEntity, response, status); Capture<HttpGet> capturedGet = Capture.newInstance(); EasyMock.expect(httpClient.execute(EasyMock.capture(capturedGet))).andReturn(response); EasyMock.replay(httpClient); Capture<Set<String>> capturedMpIds = mockAssetManagerAndWorkflow(GoogleSpeechTranscriptionService.DEFAULT_WF_DEF, true); WorkflowDispatcher dispatcher = service.new WorkflowDispatcher(); dispatcher.run(); Assert.assertEquals(RESULT_URL + JOB_ID, capturedGet.getValue().getURI().toString()); Assert.assertEquals(1, capturedMpIds.getValue().size()); Assert.assertEquals(MP_ID, capturedMpIds.getValue().iterator().next()); TranscriptionJobControl job = database.findByJob(JOB_ID); Assert.assertNotNull(job); Assert.assertEquals(TranscriptionJobControl.Status.Closed.name(), job.getStatus()); }