AmberscriptTranscriptionService.java
/*
* Licensed to The Apereo Foundation under one or more contributor license
* agreements. See the NOTICE file distributed with this work for additional
* information regarding copyright ownership.
*
*
* The Apereo Foundation licenses this file to you under the Educational
* Community License, Version 2.0 (the "License"); you may not use this file
* except in compliance with the License. You may obtain a copy of the License
* at:
*
* http://opensource.org/licenses/ecl2.txt
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations under
* the License.
*
*/
package org.opencastproject.transcription.amberscript;
import org.opencastproject.assetmanager.api.AssetManager;
import org.opencastproject.assetmanager.api.fn.Enrichments;
import org.opencastproject.assetmanager.api.query.AQueryBuilder;
import org.opencastproject.assetmanager.api.query.AResult;
import org.opencastproject.assetmanager.util.Workflows;
import org.opencastproject.job.api.AbstractJobProducer;
import org.opencastproject.job.api.Job;
import org.opencastproject.mediapackage.Catalog;
import org.opencastproject.mediapackage.MediaPackageElement;
import org.opencastproject.mediapackage.MediaPackageElementBuilder;
import org.opencastproject.mediapackage.MediaPackageElementBuilderFactory;
import org.opencastproject.mediapackage.MediaPackageElementFlavor;
import org.opencastproject.mediapackage.MediaPackageElementParser;
import org.opencastproject.mediapackage.MediaPackageElements;
import org.opencastproject.mediapackage.MediaPackageException;
import org.opencastproject.mediapackage.Track;
import org.opencastproject.metadata.dublincore.DublinCore;
import org.opencastproject.metadata.dublincore.DublinCoreCatalog;
import org.opencastproject.metadata.dublincore.DublinCoreValue;
import org.opencastproject.metadata.dublincore.DublinCores;
import org.opencastproject.security.api.DefaultOrganization;
import org.opencastproject.security.api.Organization;
import org.opencastproject.security.api.OrganizationDirectoryService;
import org.opencastproject.security.api.SecurityService;
import org.opencastproject.security.api.UserDirectoryService;
import org.opencastproject.security.util.SecurityUtil;
import org.opencastproject.serviceregistry.api.ServiceRegistry;
import org.opencastproject.serviceregistry.api.ServiceRegistryException;
import org.opencastproject.systems.OpencastConstants;
import org.opencastproject.transcription.api.TranscriptionService;
import org.opencastproject.transcription.api.TranscriptionServiceException;
import org.opencastproject.transcription.persistence.TranscriptionDatabase;
import org.opencastproject.transcription.persistence.TranscriptionDatabaseException;
import org.opencastproject.transcription.persistence.TranscriptionJobControl;
import org.opencastproject.transcription.persistence.TranscriptionProviderControl;
import org.opencastproject.util.NotFoundException;
import org.opencastproject.util.OsgiUtil;
import org.opencastproject.util.data.Option;
import org.opencastproject.workflow.api.ConfiguredWorkflow;
import org.opencastproject.workflow.api.WorkflowDefinition;
import org.opencastproject.workflow.api.WorkflowInstance;
import org.opencastproject.workflow.api.WorkflowService;
import org.opencastproject.workingfilerepository.api.WorkingFileRepository;
import org.opencastproject.workspace.api.Workspace;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpEntity;
import org.apache.http.HttpStatus;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.ContentType;
import org.apache.http.entity.mime.HttpMultipartMode;
import org.apache.http.entity.mime.MultipartEntityBuilder;
import org.apache.http.entity.mime.content.FileBody;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.util.EntityUtils;
import org.json.simple.JSONObject;
import org.json.simple.parser.JSONParser;
import org.json.simple.parser.ParseException;
import org.osgi.service.component.ComponentContext;
import org.osgi.service.component.annotations.Activate;
import org.osgi.service.component.annotations.Component;
import org.osgi.service.component.annotations.Deactivate;
import org.osgi.service.component.annotations.Reference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
@Component(
immediate = true,
service = { TranscriptionService.class,AmberscriptTranscriptionService.class },
property = {
"service.description=AmberScript Transcription Service",
"provider=amberscript"
}
)
public class AmberscriptTranscriptionService extends AbstractJobProducer implements TranscriptionService {
private static final Logger logger = LoggerFactory.getLogger(AmberscriptTranscriptionService.class);
private static final String JOB_TYPE = "org.opencastproject.transcription.amberscript";
public static final String SUBMISSION_COLLECTION = "amberscript-submission";
private static final String TRANSCRIPT_COLLECTION = "amberscript-transcripts";
private static final int CONNECTION_TIMEOUT = 60000; // ms, 1 minute
private static final int SOCKET_TIMEOUT = 60000; // ms, 1 minute
private static final String BASE_URL = "https://qs.amberscript.com";
private static final String STATUS_OPEN = "OPEN";
private static final String STATUS_DONE = "DONE";
private static final String STATUS_ERROR = "ERROR";
private static final String ERROR_NO_SPEECH = "No speech found";
private static final String PROVIDER = "amberscript";
private AssetManager assetManager;
private OrganizationDirectoryService organizationDirectoryService;
private ScheduledExecutorService scheduledExecutor;
private SecurityService securityService;
private ServiceRegistry serviceRegistry;
private TranscriptionDatabase database;
private UserDirectoryService userDirectoryService;
private WorkflowService workflowService;
private WorkingFileRepository wfr;
private Workspace workspace;
// Only used by unit tests
private Workflows wfUtil;
private enum Operation {
StartTranscription
}
// Enum for configuring which metadata field shall be used
// for determining the number of speakers of an event
private enum SpeakerMetadataField {
creator, contributor, both
}
// service configuration keys
private static final String ENABLED_CONFIG = "enabled";
private static final String CLIENT_KEY = "client.key";
private static final String LANGUAGE = "language";
private static final String LANGUAGE_FROM_DUBLINCORE = "language.from.dublincore";
private static final String LANGUAGE_CODE_MAP = "language.code.map";
private static final String AMBERSCRIPTJOBTYPE = "jobtype";
private static final String WORKFLOW_CONFIG = "workflow";
private static final String DISPATCH_WORKFLOW_INTERVAL_CONFIG = "workflow.dispatch.interval";
private static final String MAX_PROCESSING_TIME_CONFIG = "max.overdue.time";
private static final String CLEANUP_RESULTS_DAYS_CONFIG = "cleanup.results.days";
private static final String SPEAKER = "speaker";
private static final String SPEAKER_FROM_DUBLINCORE = "speaker.from.dublincore";
private static final String SPEAKER_METADATA_FIELD = "speaker.metadata.field";
private static final String TRANSCRIPTIONTYPE = "transcriptiontype";
private static final String GLOSSARY = "glossary";
private static final String TRANSCRIPTIONSTYLE = "cleanread";
private static final String TARGETLANGUAGE = "targetlanguage";
// service configuration default values
private boolean enabled = false;
private String clientKey;
private String language = "en";
/** determines if the transcription language should be taken from the dublincore */
private boolean languageFromDublinCore;
private String amberscriptJobType = "direct";
private String workflowDefinitionId = "amberscript-attach-transcripts";
private long workflowDispatchIntervalSeconds = 60;
private long maxProcessingSeconds = 8 * 24 * 60 * 60; // maximum runtime for jobType perfect is 8 days
private int cleanupResultDays = 7;
private int numberOfSpeakers = 1;
private boolean speakerFromDublinCore = true;
private SpeakerMetadataField speakerMetadataField = SpeakerMetadataField.creator;
private String transcriptionType = "transcription";
private String glossary = "";
private String transcriptionStyle = "cleanread";
private String targetLanguage = "";
/**
* Contains mappings from several possible ways of writing a language name/code to the
* corresponding amberscript language code
*/
private AmberscriptLangUtil amberscriptLangUtil;
private String systemAccount;
public AmberscriptTranscriptionService() {
super(JOB_TYPE);
}
@Activate
public void activate(ComponentContext cc) {
Option<Boolean> enabledOpt = OsgiUtil.getOptCfgAsBoolean(cc.getProperties(), ENABLED_CONFIG);
if (enabledOpt.isSome()) {
enabled = enabledOpt.get();
}
if (!enabled) {
logger.info("Amberscript Transcription Service disabled."
+ " If you want to enable it, please update the service configuration.");
return;
}
Option<String> clientKeyOpt = OsgiUtil.getOptCfg(cc.getProperties(), CLIENT_KEY);
if (clientKeyOpt.isSome()) {
clientKey = clientKeyOpt.get();
} else {
logger.warn("API key was not set.");
return;
}
Option<String> languageOpt = OsgiUtil.getOptCfg(cc.getProperties(), LANGUAGE);
if (languageOpt.isSome()) {
language = languageOpt.get();
logger.info("Default language is set to '{}'.", language);
} else {
logger.info("Default language '{}' will be used.", language);
}
Option<String> languageFromDublinCoreOpt = OsgiUtil.getOptCfg(cc.getProperties(), LANGUAGE_FROM_DUBLINCORE);
if (languageFromDublinCoreOpt.isSome()) {
try {
languageFromDublinCore = Boolean.parseBoolean(languageFromDublinCoreOpt.get());
} catch (Exception e) {
logger.warn("Configuration value for '{}' is invalid, defaulting to false.", LANGUAGE_FROM_DUBLINCORE);
}
}
logger.info("Configuration value for '{}' is set to '{}'.", LANGUAGE_FROM_DUBLINCORE, languageFromDublinCore);
amberscriptLangUtil = AmberscriptLangUtil.getInstance();
int customMapEntriesCount = 0;
Option<String> langCodeMapOpt = OsgiUtil.getOptCfg(cc.getProperties(), LANGUAGE_CODE_MAP);
if (langCodeMapOpt.isSome()) {
try {
String langCodeMapStr = langCodeMapOpt.get();
if (langCodeMapStr != null) {
for (String mapping : langCodeMapStr.split(",")) {
String[] mapEntries = mapping.split(":");
amberscriptLangUtil.addCustomMapping(mapEntries[0], mapEntries[1]);
customMapEntriesCount += 1;
}
}
} catch (Exception e) {
logger.warn("Configuration '{}' is invalid. Using just default mapping.", LANGUAGE_CODE_MAP);
}
}
logger.info("Language code map was set. Added '{}' additional entries.", customMapEntriesCount);
Option<String> amberscriptJobTypeOpt = OsgiUtil.getOptCfg(cc.getProperties(), AMBERSCRIPTJOBTYPE);
if (amberscriptJobTypeOpt.isSome()) {
amberscriptJobType = amberscriptJobTypeOpt.get();
logger.info("Default Amberscript job type is set to '{}'.", amberscriptJobType);
} else {
logger.info("Default Amberscript job type '{}' will be used.", amberscriptJobType);
}
Option<String> wfOpt = OsgiUtil.getOptCfg(cc.getProperties(), WORKFLOW_CONFIG);
if (wfOpt.isSome()) {
workflowDefinitionId = wfOpt.get();
logger.info("Workflow is set to '{}'.", workflowDefinitionId);
} else {
logger.info("Default workflow '{}' will be used.", workflowDefinitionId);
}
Option<String> intervalOpt = OsgiUtil.getOptCfg(cc.getProperties(), DISPATCH_WORKFLOW_INTERVAL_CONFIG);
if (intervalOpt.isSome()) {
try {
workflowDispatchIntervalSeconds = Long.parseLong(intervalOpt.get());
} catch (NumberFormatException e) {
logger.warn("Configured '{}' is invalid. Using default.", DISPATCH_WORKFLOW_INTERVAL_CONFIG);
}
}
logger.info("Workflow dispatch interval is {} seconds.", workflowDispatchIntervalSeconds);
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("Configured '{}' is invalid. Using default.", MAX_PROCESSING_TIME_CONFIG);
}
}
logger.info("Maximum processing time for transcription job is {} seconds.", maxProcessingSeconds);
Option<String> cleanupOpt = OsgiUtil.getOptCfg(cc.getProperties(), CLEANUP_RESULTS_DAYS_CONFIG);
if (cleanupOpt.isSome()) {
try {
cleanupResultDays = Integer.parseInt(cleanupOpt.get());
} catch (NumberFormatException e) {
logger.warn("Configured '{}' is invalid. Using default.", CLEANUP_RESULTS_DAYS_CONFIG);
}
}
logger.info("Cleanup result files after {} days.", cleanupResultDays);
Option<String> speakerOpt = OsgiUtil.getOptCfg(cc.getProperties(), SPEAKER);
if (speakerOpt.isSome()) {
try {
numberOfSpeakers = Integer.parseInt(speakerOpt.get());
} catch (NumberFormatException e) {
logger.warn("Configured '{}' is invalid. Using default.", SPEAKER);
}
}
logger.info("Default number of speakers is set to '{}'.", numberOfSpeakers);
Option<String> speakerFromDublinCoreOpt = OsgiUtil.getOptCfg(cc.getProperties(), SPEAKER_FROM_DUBLINCORE);
if (speakerFromDublinCoreOpt.isSome()) {
try {
speakerFromDublinCore = Boolean.parseBoolean(speakerFromDublinCoreOpt.get());
} catch (Exception e) {
logger.warn("Configuration value for '{}' is invalid, defaulting to true.", SPEAKER_FROM_DUBLINCORE);
}
}
logger.info("Configuration value for '{}' is set to '{}'.", SPEAKER_FROM_DUBLINCORE, speakerFromDublinCore);
Option<String> speakerMetadataFieldOpt = OsgiUtil.getOptCfg(cc.getProperties(), SPEAKER_METADATA_FIELD);
if (speakerMetadataFieldOpt.isSome()) {
try {
speakerMetadataField = SpeakerMetadataField.valueOf(speakerMetadataFieldOpt.get());
} catch (IllegalArgumentException e) {
logger.warn("Value '{}' is invalid for configuration '{}'. Using default: '{}'.",
speakerMetadataFieldOpt.get(), SPEAKER_METADATA_FIELD, speakerMetadataField);
}
}
logger.info("Default metadata field for calculating the amount of speakers is set to '{}'.", speakerMetadataField);
Option<String> transcriptionTypeOpt = OsgiUtil.getOptCfg(cc.getProperties(), TRANSCRIPTIONTYPE);
if (transcriptionTypeOpt.isSome()) {
if (List.of("transcription", "captions", "translatedSubtitles").contains(transcriptionType)) {
transcriptionType = transcriptionTypeOpt.get();
logger.info("Default transcription type is set to '{}'.", transcriptionType);
} else {
logger.warn("Value '{}' is invalid for configuration '{}'. Using default: '{}'.",
transcriptionTypeOpt.get(), TRANSCRIPTIONTYPE, transcriptionType);
}
} else {
logger.info("Default transcription type '{}' will be used.", transcriptionType);
}
Option<String> glossaryOpt = OsgiUtil.getOptCfg(cc.getProperties(), GLOSSARY);
if (glossaryOpt.isSome()) {
glossary = glossaryOpt.get();
logger.info("Default glossary is set to '{}'.", glossary);
} else {
logger.info("No glossary will be used by default");
}
Option<String> transcriptionStyleOpt = OsgiUtil.getOptCfg(cc.getProperties(), TRANSCRIPTIONSTYLE);
if (transcriptionStyleOpt.isSome()) {
if (List.of("cleanread", "verbatim").contains(transcriptionStyle)) {
transcriptionStyle = transcriptionStyleOpt.get();
logger.info("Default transcription style is set to '{}'.", transcriptionStyle);
} else {
logger.warn("Value '{}' is invalid for configuration '{}'. Using default: '{}'.",
transcriptionStyleOpt.get(), TRANSCRIPTIONSTYLE, transcriptionStyle);
}
} else {
logger.info("Default transcription style '{}' will be used.", transcriptionStyle);
}
Option<String> targetLanguageOpt = OsgiUtil.getOptCfg(cc.getProperties(), TARGETLANGUAGE);
if (targetLanguageOpt.isSome()) {
targetLanguage = targetLanguageOpt.get();
logger.info("Default target language is set to '{}'.", targetLanguage);
} else {
logger.info("Transcriptions won't be translated");
}
systemAccount = OsgiUtil.getContextProperty(cc, OpencastConstants.DIGEST_USER_PROPERTY);
scheduledExecutor = Executors.newScheduledThreadPool(2);
scheduledExecutor.scheduleWithFixedDelay(new WorkflowDispatcher(), 120, workflowDispatchIntervalSeconds,
TimeUnit.SECONDS);
scheduledExecutor.scheduleWithFixedDelay(new ResultsFileCleanup(), 1, 1, TimeUnit.DAYS);
logger.info("Activated.");
}
@Deactivate
public void deactivate() {
if (scheduledExecutor != null) {
scheduledExecutor.shutdown();
}
}
@Override
public Job startTranscription(String mpId, Track track) throws TranscriptionServiceException {
throw new UnsupportedOperationException("Not supported.");
}
@Override
public Job startTranscription(String mpId, Track track, String... args) throws TranscriptionServiceException {
if (!enabled) {
throw new TranscriptionServiceException("AmberScript Transcription Service disabled."
+ " If you want to enable it, please update the service configuration.");
}
String language = null;
if (languageFromDublinCore) {
for (Catalog catalog : track.getMediaPackage().getCatalogs(MediaPackageElements.EPISODE)) {
try (InputStream in = workspace.read(catalog.getURI())) {
DublinCoreCatalog dublinCatalog = DublinCores.read(in);
String dublinCoreLang = dublinCatalog.getFirst(DublinCore.PROPERTY_LANGUAGE);
if (dublinCoreLang != null) {
language = amberscriptLangUtil.getLanguageCodeOrNull(dublinCoreLang);
}
if (language != null) {
break;
}
} catch (IOException | NotFoundException e) {
logger.error(String.format("Unable to load dublin core catalog for event '%s'",
track.getMediaPackage().getIdentifier()), e);
}
}
}
if (language == null) {
if (args.length > 0 && StringUtils.isNotBlank(args[0])) {
language = args[0];
} else {
language = getLanguage();
}
}
String jobType;
if (args.length > 1 && StringUtils.isNotBlank(args[1])) {
jobType = args[1];
} else {
jobType = getAmberscriptJobType();
}
int numberOfSpeakers = 0;
if (speakerFromDublinCore) {
Set<String> speakers = new HashSet<>();
for (Catalog catalog : track.getMediaPackage().getCatalogs(MediaPackageElements.EPISODE)) {
try (InputStream in = workspace.read(catalog.getURI())) {
DublinCoreCatalog dublinCatalog = DublinCores.read(in);
if (speakerMetadataField.equals(SpeakerMetadataField.creator)
|| speakerMetadataField.equals(SpeakerMetadataField.both)) {
dublinCatalog.get(DublinCore.PROPERTY_CREATOR).stream()
.map(DublinCoreValue::getValue).forEach(speakers::add);
}
if (speakerMetadataField.equals(SpeakerMetadataField.contributor)
|| speakerMetadataField.equals(SpeakerMetadataField.both)) {
dublinCatalog.get(DublinCore.PROPERTY_CONTRIBUTOR).stream()
.map(DublinCoreValue::getValue).forEach(speakers::add);
}
} catch (IOException | NotFoundException e) {
logger.error("Unable to load dublin core catalog for event '{}'",
track.getMediaPackage().getIdentifier(), e);
}
}
if (speakers.size() >= 1) {
numberOfSpeakers = speakers.size();
}
}
if (numberOfSpeakers == 0) {
if (args.length > 2 && StringUtils.isNotBlank(args[2])) {
numberOfSpeakers = Integer.parseInt(args[2]);
} else {
numberOfSpeakers = getNumberOfSpeakers();
}
}
String transcriptionType;
if (args.length > 3 && StringUtils.isNotBlank(args[3])) {
transcriptionType = args[3];
} else {
transcriptionType = getTranscriptionType();
}
String glossary;
if (args.length > 4 && args[4] != null) {
glossary = args[4];
} else {
glossary = getGlossary();
}
String transcriptionStyle;
if (args.length > 5 && StringUtils.isNotBlank(args[5])) {
transcriptionStyle = args[5];
} else {
transcriptionStyle = getTranscriptionStyle();
}
String targetLanguage;
if (args.length > 6 && args[6] != null) {
targetLanguage = args[6];
} else {
targetLanguage = getTargetLanguage();
}
logger.info("New transcription job for mpId '{}' language '{}' job type '{}' speakers '{}' transcription type '{}'"
+ "glossary '{}'.", mpId, language, jobType, numberOfSpeakers, transcriptionType, glossary);
try {
return serviceRegistry.createJob(JOB_TYPE, Operation.StartTranscription.name(), Arrays.asList(
mpId, MediaPackageElementParser.getAsXml(track), language, jobType, Integer.toString(numberOfSpeakers),
transcriptionType, glossary, transcriptionStyle, targetLanguage));
} catch (ServiceRegistryException e) {
throw new TranscriptionServiceException("Unable to create a job", e);
} catch (MediaPackageException e) {
throw new TranscriptionServiceException("Invalid track '" + track.toString() + "'", e);
}
}
@Override
public void transcriptionDone(String mpId, Object results) { }
private void transcriptionDone(String mpId, String jobId) {
try {
logger.info("Transcription done for mpId '{}'.", mpId);
if (getAndSaveJobResult(jobId)) {
database.updateJobControl(jobId, TranscriptionJobControl.Status.TranscriptionComplete.name());
} else {
logger.debug("Unable to get and save the transcription result for mpId '{}'.", mpId);
}
} catch (IOException | TranscriptionServiceException e) {
logger.warn("Could not save transcription results file for mpId '{}': {}", mpId, e.toString());
} catch (TranscriptionDatabaseException e) {
logger.warn("Transcription results file were saved but state in db not updated for mpId '{}': ", mpId, e);
}
}
@Override
public void transcriptionError(String mpId, Object obj) throws TranscriptionServiceException {
JSONObject jsonObj = null;
String jobId = null;
try {
jsonObj = (JSONObject) obj;
jobId = (String) jsonObj.get("name");
// Update state in database
database.updateJobControl(jobId, TranscriptionJobControl.Status.Error.name());
TranscriptionJobControl jobControl = database.findByJob(jobId);
logger.warn(String.format("Error received for media package %s, job id %s",
jobControl.getMediaPackageId(), jobId));
// Send notification email
} catch (TranscriptionDatabaseException e) {
logger.warn("Transcription error. State in db could not be updated to error for mpId {}, jobId {}", mpId, jobId);
throw new TranscriptionServiceException("Could not update transcription job control db", e);
}
}
@Override
public Map<String, Object> getReturnValues(String mpId, String jobId) throws TranscriptionServiceException {
throw new TranscriptionServiceException("Method not implemented");
}
@Override
public String getLanguage() {
return language;
}
public String getAmberscriptJobType() {
return amberscriptJobType;
}
public int getNumberOfSpeakers() {
return numberOfSpeakers;
}
public String getTranscriptionType() {
return transcriptionType;
}
public String getGlossary() {
return glossary;
}
public String getTranscriptionStyle() {
return transcriptionStyle;
}
public String getTargetLanguage() {
return targetLanguage;
}
// Called by workflow
@Override
protected String process(Job job) throws Exception {
Operation op = null;
String operation = job.getOperation();
List<String> arguments = job.getArguments();
String result = "";
op = Operation.valueOf(operation);
switch (op) {
case StartTranscription:
String mpId = arguments.get(0);
Track track = (Track) MediaPackageElementParser.getFromXml(arguments.get(1));
String languageCode = arguments.get(2);
String jobType = arguments.get(3);
String numberOfSpeakers = arguments.get(4);
String transcriptionType = arguments.get(5);
String glossary = arguments.get(6);
String transcriptionStyle = arguments.get(7);
String targetLanguage = arguments.get(8);
createRecognitionsJob(mpId, track, languageCode, jobType, numberOfSpeakers, transcriptionType, glossary,
transcriptionStyle, targetLanguage);
break;
default:
throw new IllegalStateException("Don't know how to handle operation '" + operation + "'");
}
return result;
}
void createRecognitionsJob(String mpId, Track track, String languageCode, String jobType, String numberOfSpeakers,
String transcriptionType, String glossary, String transcriptionStyle, String targetLanguage)
throws TranscriptionServiceException {
// Timeout 3 hours (needs to include the time for the remote service to
// fetch the media URL before sending final response)
CloseableHttpClient httpClient = makeHttpClient(3 * 3600 * 1000);
CloseableHttpResponse response = null;
String submitUrl = BASE_URL + "/jobs/upload-media"
+ "?apiKey=" + clientKey
+ "&language=" + languageCode
+ "&jobType=" + jobType
+ "&numberOfSpeakers=" + numberOfSpeakers
+ "&transcriptionType=" + transcriptionType
+ "&transcriptionStyle=" + transcriptionStyle;
if (StringUtils.isNotBlank(glossary)) {
submitUrl += "&glossary=" + glossary;
}
if (StringUtils.isNotBlank(targetLanguage)) {
submitUrl += "&targetLanguage=" + targetLanguage;
}
try {
FileBody fileBody = new FileBody(workspace.get(track.getURI()), ContentType.DEFAULT_BINARY);
MultipartEntityBuilder builder = MultipartEntityBuilder.create();
builder.setMode(HttpMultipartMode.BROWSER_COMPATIBLE);
builder.addPart("file", fileBody);
HttpEntity multipartEntity = builder.build();
HttpPost httpPost = new HttpPost(submitUrl);
httpPost.setEntity(multipartEntity);
response = httpClient.execute(httpPost);
int code = response.getStatusLine().getStatusCode();
HttpEntity entity = response.getEntity();
String jsonString = EntityUtils.toString(response.getEntity());
JSONParser jsonParser = new JSONParser();
JSONObject jsonObject = (JSONObject) jsonParser.parse(jsonString);
logger.debug("Submitting new transcription job: {}" + System.lineSeparator()
+ "Response: {}", removePrivateInfo(submitUrl), jsonString);
JSONObject result = (JSONObject) jsonObject.get("jobStatus");
String jobId = (String) result.get("jobId");
switch (code) {
case HttpStatus.SC_OK: // 200
logger.info("mp {} has been submitted to AmberScript service with jobId {}.", mpId, jobId);
database.storeJobControl(mpId, track.getIdentifier(), jobId, TranscriptionJobControl.Status.InProgress.name(),
track.getDuration() == null ? 0 : track.getDuration().longValue(), new Date(), PROVIDER);
EntityUtils.consume(entity);
return;
default:
String error = (String) result.get("error");
String message = (String) result.get("message");
String msg = String.format("Unable to submit job: API returned %s - %s: %s", code, error, message);
logger.warn(msg);
throw new TranscriptionServiceException(msg);
}
} catch (Exception e) {
logger.warn("Exception when calling the captions endpoint", e);
throw new TranscriptionServiceException("Exception when calling the captions endpoint", e);
} finally {
try {
httpClient.close();
if (response != null) {
response.close();
}
} catch (IOException e) {
}
}
}
boolean checkJobResults(String jobId) throws TranscriptionServiceException {
String mpId = "unknown";
CloseableHttpClient httpClient = makeHttpClient();
CloseableHttpResponse response = null;
String checkUrl = BASE_URL + "/jobs/status?jobId=" + jobId + "&apiKey=" + clientKey;
try {
HttpGet httpGet = new HttpGet(checkUrl);
response = httpClient.execute(httpGet);
int code = response.getStatusLine().getStatusCode();
HttpEntity entity = response.getEntity();
String jsonString = EntityUtils.toString(entity);
EntityUtils.consume(entity);
logger.debug("AmberScript API call was '{}'." + System.lineSeparator() + "Response: {}",
removePrivateInfo(checkUrl), jsonString);
JSONParser jsonParser = new JSONParser();
JSONObject jsonObject = (JSONObject) jsonParser.parse(jsonString);
switch (code) {
case HttpStatus.SC_OK:
JSONObject result = (JSONObject) jsonObject.get("jobStatus");
String status = (String) result.get("status");
switch (status) {
case STATUS_OPEN:
logger.debug("Captions job '{}' has not finished yet.", jobId);
return false;
case STATUS_ERROR:
var errorMsg = (String) result.get("errorMsg");
throw new TranscriptionServiceException(
String.format("Captions job '%s' failed: %s", jobId, errorMsg),
code,
ERROR_NO_SPEECH.equals(errorMsg));
case STATUS_DONE:
logger.info("Captions job '{}' has finished.", jobId);
TranscriptionJobControl jc = database.findByJob(jobId);
if (jc != null) {
mpId = jc.getMediaPackageId();
}
transcriptionDone(mpId, jobId);
return true;
default:
return false; // only here to obey checkstyle
}
default:
String error = (String) jsonObject.get("error");
String errorMessage = (String) jsonObject.get("errorMessage");
logger.warn("Error while checking status: {}."
+ System.lineSeparator() + "{}: {}", code, error, errorMessage);
throw new TranscriptionServiceException(
String.format("Captions job '%s' failed: Return Code %d", jobId, code), code);
}
} catch (TranscriptionDatabaseException | IOException | ParseException e) {
logger.warn("Error while checking status: {}", e.toString());
} finally {
try {
httpClient.close();
if (response != null) {
response.close();
}
} catch (IOException e) {
}
}
return false;
}
private boolean getAndSaveJobResult(String jobId) throws TranscriptionServiceException, IOException {
CloseableHttpClient httpClient = makeHttpClient();
CloseableHttpResponse response = null;
String transcriptUrl = BASE_URL + "/jobs/export?format=srt&jobId=" + jobId + "&apiKey=" + clientKey;
boolean done = false;
try {
HttpGet httpGet = new HttpGet(transcriptUrl);
response = httpClient.execute(httpGet);
int code = response.getStatusLine().getStatusCode();
logger.debug("AmberScript API {} http response {}", removePrivateInfo(transcriptUrl), code);
switch (code) {
case HttpStatus.SC_OK: // 200
HttpEntity entity = response.getEntity();
logger.info("Retrieved details for transcription with jobid: '{}'", jobId);
// Save the result subrip (srt) file into a collection
workspace.putInCollection(TRANSCRIPT_COLLECTION, jobId + ".srt", entity.getContent());
done = true;
break;
default:
logger.warn("Error retrieving details for transcription with jobid: '{}', return status: {}.", jobId, code);
break;
}
} catch (Exception e) {
throw new TranscriptionServiceException(String.format(
"Exception when calling the transcription service for jobid: %s", jobId), e);
} finally {
try {
httpClient.close();
if (response != null) {
response.close();
}
} catch (IOException e) {
}
}
return done;
}
@Override
// Called by the attach workflow operation
public MediaPackageElement getGeneratedTranscription(String mpId, String jobId, MediaPackageElement.Type type)
throws TranscriptionServiceException {
try {
// If jobId is unknown, look for all jobs associated to that mpId
if (jobId == null || "null".equals(jobId)) {
jobId = null;
for (TranscriptionJobControl jc : database.findByMediaPackage(mpId)) {
if (TranscriptionJobControl.Status.Closed.name().equals(jc.getStatus())
|| TranscriptionJobControl.Status.TranscriptionComplete.name().equals(jc.getStatus())) {
jobId = jc.getTranscriptionJobId();
}
}
}
if (jobId == null) {
throw new TranscriptionServiceException(
"No completed or closed transcription job found in database for media package " + mpId);
}
// Results already saved?
URI uri = workspace.getCollectionURI(TRANSCRIPT_COLLECTION, jobId + ".srt");
logger.info("Looking for transcript at URI: {}", uri);
try {
workspace.get(uri);
logger.info("Found captions at URI: {}", uri);
} catch (Exception e) {
logger.info("Results not saved: getting from service for jobId {}", jobId);
// Not saved yet so call the transcription service to get the results
checkJobResults(jobId);
}
MediaPackageElementBuilder builder = MediaPackageElementBuilderFactory.newInstance().newElementBuilder();
logger.debug("Returning MPE with results file URI: {}", uri);
return builder.elementFromURI(uri, type, new MediaPackageElementFlavor("captions", "srt"));
} catch (TranscriptionDatabaseException e) {
throw new TranscriptionServiceException("Job id not informed and could not find transcription", e);
}
}
/**
* Get mediapackage transcription status
*
* @param mpId, mediapackage id
* @return transcription status
* @throws TranscriptionServiceException
*/
public String getTranscriptionStatus(String mpId) throws TranscriptionServiceException {
try {
for (TranscriptionJobControl jc : database.findByMediaPackage(mpId)) {
return jc.getStatus();
}
} catch (TranscriptionDatabaseException e) {
throw new TranscriptionServiceException("Mediapackage id transcription status unknown", e);
}
return "Unknown";
}
/**
* Creates a closable http client with default configuration.
*
* @return closable http client
*/
protected CloseableHttpClient makeHttpClient() {
return makeHttpClient(SOCKET_TIMEOUT);
}
/**
* Creates a closable http client.
*
* @param socketTimeout http socket timeout value in milliseconds
*/
protected CloseableHttpClient makeHttpClient(int socketTimeout) {
RequestConfig reqConfig = RequestConfig.custom()
.setConnectTimeout(AmberscriptTranscriptionService.CONNECTION_TIMEOUT)
.setSocketTimeout(socketTimeout)
.setConnectionRequestTimeout(AmberscriptTranscriptionService.CONNECTION_TIMEOUT)
.build();
CloseableHttpClient httpClient = HttpClientBuilder.create()
.useSystemProperties()
.setDefaultRequestConfig(reqConfig)
.build();
return httpClient;
}
// Called when a transcription job has been submitted
protected void deleteStorageFile(String filename) {
try {
logger.debug("Removing {} from collection {}.", filename, SUBMISSION_COLLECTION);
wfr.deleteFromCollection(SUBMISSION_COLLECTION, filename, false);
} catch (IOException e) {
logger.warn("Unable to remove submission file {} from collection {}", filename, SUBMISSION_COLLECTION);
}
}
@Reference
public void setServiceRegistry(ServiceRegistry serviceRegistry) {
this.serviceRegistry = serviceRegistry;
}
@Reference
public void setSecurityService(SecurityService securityService) {
this.securityService = securityService;
}
@Reference
public void setUserDirectoryService(UserDirectoryService userDirectoryService) {
this.userDirectoryService = userDirectoryService;
}
@Reference
public void setOrganizationDirectoryService(OrganizationDirectoryService organizationDirectoryService) {
this.organizationDirectoryService = organizationDirectoryService;
}
@Reference
public void setWorkspace(Workspace ws) {
this.workspace = ws;
}
@Reference
public void setWorkingFileRepository(WorkingFileRepository wfr) {
this.wfr = wfr;
}
@Reference
public void setDatabase(TranscriptionDatabase service) {
this.database = service;
}
@Reference
public void setAssetManager(AssetManager service) {
this.assetManager = service;
}
@Reference
public void setWorkflowService(WorkflowService service) {
this.workflowService = service;
}
@Override
protected ServiceRegistry getServiceRegistry() {
return serviceRegistry;
}
@Override
protected SecurityService getSecurityService() {
return securityService;
}
@Override
protected UserDirectoryService getUserDirectoryService() {
return userDirectoryService;
}
@Override
protected OrganizationDirectoryService getOrganizationDirectoryService() {
return organizationDirectoryService;
}
// Only used by unit tests!
void setWfUtil(Workflows wfUtil) {
this.wfUtil = wfUtil;
}
class WorkflowDispatcher implements Runnable {
/**
* {@inheritDoc}
*
* @see java.lang.Thread#run()
*/
@Override
public void run() {
logger.debug("WorkflowDispatcher waking up...");
try {
// Find jobs that are in progress and jobs that had transcription complete
long providerId;
TranscriptionProviderControl providerInfo = database.findIdByProvider(PROVIDER);
if (providerInfo != null) {
providerId = providerInfo.getId();
} else {
logger.debug("No jobs yet for provider {}.", PROVIDER);
return;
}
List<TranscriptionJobControl> jobs = database.findByStatus(TranscriptionJobControl.Status.InProgress.name(),
TranscriptionJobControl.Status.TranscriptionComplete.name());
for (TranscriptionJobControl j : jobs) {
// Don't process jobs for other services
if (j.getProviderId() != providerId) {
continue;
}
String mpId = j.getMediaPackageId();
String jobId = j.getTranscriptionJobId();
// If the job in progress, check if it should already have finished.
if (TranscriptionJobControl.Status.InProgress.name().equals(j.getStatus())) {
// If job should already have been completed, try to get the results.
if (j.getDateExpected().getTime() < System.currentTimeMillis()) {
try {
if (!checkJobResults(jobId)) {
// Job still running, not finished, so check if it should have finished more than N seconds ago
if (j.getDateExpected().getTime() + maxProcessingSeconds * 1000 < System.currentTimeMillis()) {
// Processing for too long, mark job as canceled and don't check anymore
database.updateJobControl(jobId, TranscriptionJobControl.Status.Canceled.name());
}
// else Job still running, not finished
continue;
}
} catch (TranscriptionServiceException e) {
try {
database.updateJobControl(jobId, TranscriptionJobControl.Status.Canceled.name());
continue;
} catch (TranscriptionDatabaseException ex) {
logger.warn("Could not cancel job '{}'.", jobId);
}
}
} else {
continue; // Not time to check yet
}
}
// Jobs that get here have state TranscriptionCompleted
try {
DefaultOrganization defaultOrg = new DefaultOrganization();
securityService.setOrganization(defaultOrg);
securityService.setUser(SecurityUtil.createSystemUser(systemAccount, defaultOrg));
// Find the episode
final AQueryBuilder q = assetManager.createQuery();
final AResult r = q.select(q.snapshot()).where(q.mediaPackageId(mpId).and(q.version().isLatest())).run();
if (r.getSize() == 0) {
logger.warn("Media package {} no longer exists in the asset manager. It was likely deleted. "
+ "Dropping the generated transcription.", mpId);
database.updateJobControl(jobId, TranscriptionJobControl.Status.Error.name());
continue;
}
String org = Enrichments.enrich(r).getSnapshots().stream().findFirst().get().getOrganizationId();
Organization organization = organizationDirectoryService.getOrganization(org);
if (organization == null) {
logger.warn("Media package {} has an unknown organization {}. Skipped.", mpId, org);
continue;
}
securityService.setOrganization(organization);
// Build workflow
Map<String, String> params = new HashMap<String, String>();
params.put("transcriptionJobId", jobId);
WorkflowDefinition wfDef = workflowService.getWorkflowDefinitionById(workflowDefinitionId);
// Apply workflow
// wfUtil is only used by unit tests
Workflows workflows = wfUtil != null ? wfUtil : new Workflows(assetManager, workflowService);
Set<String> mpIds = new HashSet<String>();
mpIds.add(mpId);
List<WorkflowInstance> wfList = workflows
.applyWorkflowToLatestVersion(mpIds, ConfiguredWorkflow.workflow(wfDef, params));
String wfId = wfList.size() > 0 ? Long.toString(wfList.get(0).getId()) : "Unknown";
// Update state in the database
database.updateJobControl(jobId, TranscriptionJobControl.Status.Closed.name());
logger.info("Attach transcription workflow {} scheduled for mp {}, transcription service job {}",
wfId, mpId, jobId);
} catch (Exception e) {
logger.warn("Attach transcription workflow could NOT be scheduled for mp {}, amberscript job {}, {}: {}",
mpId, jobId, e.getClass().getName(), e.getMessage());
}
}
} catch (TranscriptionDatabaseException e) {
logger.warn("Could not read transcription job control database: {}", e.getMessage());
}
}
}
class ResultsFileCleanup implements Runnable {
@Override
public void run() {
logger.info("ResultsFileCleanup waking up...");
try {
// Cleans up results files older than CLEANUP_RESULT_FILES_DAYS days
wfr.cleanupOldFilesFromCollection(TRANSCRIPT_COLLECTION, cleanupResultDays);
wfr.cleanupOldFilesFromCollection(SUBMISSION_COLLECTION, cleanupResultDays);
} catch (IOException e) {
logger.warn("Could not cleanup old submission and transcript results files", e);
}
}
}
private String removePrivateInfo(String unsafeString) {
String safeString = unsafeString.replace(clientKey, "__api-key-was-hidden__");
return safeString;
}
}