diff --git a/src/rev_ai/__init__.py b/src/rev_ai/__init__.py index 30e6470..681aac8 100644 --- a/src/rev_ai/__init__.py +++ b/src/rev_ai/__init__.py @@ -7,4 +7,5 @@ CaptionType, GroupChannelsType, CustomVocabulary, TopicExtractionJob, TopicExtractionResult, \ Topic, Informant, SpeakerName, LanguageIdentificationJob, LanguageIdentificationResult, \ LanguageConfidence, SentimentAnalysisResult, SentimentValue, SentimentMessage, \ - SentimentAnalysisJob, CustomerUrlData, RevAiApiDeploymentConfigMap, RevAiApiDeployment + SentimentAnalysisJob, CustomerUrlData, RevAiApiDeploymentConfigMap, RevAiApiDeployment, \ + ForcedAlignmentJob, ForcedAlignmentResult, Monologue, ElementAlignment diff --git a/src/rev_ai/forced_alignment_client.py b/src/rev_ai/forced_alignment_client.py new file mode 100644 index 0000000..d58b922 --- /dev/null +++ b/src/rev_ai/forced_alignment_client.py @@ -0,0 +1,85 @@ +# -*- coding: utf-8 -*- +"""Client used for interacting with our forced alignment api""" + +import json +from .generic_api_client import GenericApiClient +from .models.forced_alignment import ForcedAlignmentJob, ForcedAlignmentResult + + +class ForcedAlignmentClient(GenericApiClient): + """Client for interacting with the Rev AI forced alignment api""" + + # Default version of Rev AI forced alignment api + api_version = 'v1' + + # Default api name of Rev AI forced alignment api + api_name = 'alignment' + + def __init__(self, access_token): + """Constructor + + :param access_token: access token which authorizes all requests and links them to your + account. Generated on the settings page of your account dashboard + on Rev AI. + """ + GenericApiClient.__init__(self, access_token, self.api_name, self.api_version, + ForcedAlignmentJob.from_json, ForcedAlignmentResult.from_json) + + def submit_job_url( + self, + source_config=None, + source_transcript_config=None, + transcript_text=None, + metadata=None, + delete_after_seconds=None, + notification_config=None, + language=None): + """Submit a job to the Rev AI forced alignment api. + + :param source_config: CustomerUrlData object containing url of the source media and + optional authentication headers to use when accessing the source url + :param source_transcript_config: CustomerUrlData object containing url of the transcript file and + optional authentication headers to use when accessing the transcript url + :param transcript_text: The text of the transcript to be aligned (no punctuation, just words) + :param metadata: info to associate with the alignment job + :param delete_after_seconds: number of seconds after job completion when job is auto-deleted + :param notification_config: CustomerUrlData object containing the callback url to + invoke on job completion as a webhook and optional authentication headers to use when + calling the callback url + :param language: Language code for the audio and transcript. One of: "en", "es", "fr" + :returns: ForcedAlignmentJob object + :raises: HTTPError + """ + if not source_config: + raise ValueError('source_config must be provided') + if not (source_transcript_config or transcript_text): + raise ValueError('Either source_transcript_config or transcript_text must be provided') + if source_transcript_config and transcript_text: + raise ValueError('Only one of source_transcript_config or transcript_text may be provided') + + payload = self._enhance_payload({ + 'source_config': source_config.to_dict() if source_config else None, + 'source_transcript_config': source_transcript_config.to_dict() if source_transcript_config else None, + 'transcript_text': transcript_text, + 'language': language + }, metadata, None, delete_after_seconds, notification_config) + + return self._submit_job(payload) + + def get_result_json(self, id_): + """Get result of a forced alignment job as json. + + :param id_: id of job to be requested + :returns: job result data as raw json + :raises: HTTPError + """ + return self._get_result_json(id_, {}, route='transcript') + + def get_result_object(self, id_): + """Get result of a forced alignment job as ForcedAlignmentResult object. + + :param id_: id of job to be requested + :returns: job result data as ForcedAlignmentResult object + :raises: HTTPError + """ + return self._get_result_object(id_, {}, route='transcript') \ No newline at end of file diff --git a/src/rev_ai/generic_api_client.py b/src/rev_ai/generic_api_client.py index 9ae0bfa..981b055 100644 --- a/src/rev_ai/generic_api_client.py +++ b/src/rev_ai/generic_api_client.py @@ -105,7 +105,7 @@ def get_list_of_jobs(self, limit=None, starting_after=None): return [self.parse_job_info(job) for job in response.json()] - def _get_result_json(self, id_, params): + def _get_result_json(self, id_, params, route='result'): """Get the result of a job. This method is special in that it is intended to be hidden by the implementation this is done because python standard is to pass options individually instead of as an object and our true clients should match this standard @@ -124,12 +124,12 @@ def _get_result_json(self, id_, params): response = self._make_http_request( "GET", - urljoin(self.base_url, 'jobs/{0}/result?{1}'.format(id_, '&'.join(query_params))) + urljoin(self.base_url, 'jobs/{0}/{1}?{2}'.format(id_, route, '&'.join(query_params))) ) return response.json() - def _get_result_object(self, id_, params): + def _get_result_object(self, id_, params, route='result'): """Get the result of a job. This method is special in that it is intended to be hidden by the implementation this is done because python standard is to pass options individually instead of as an object and our true clients should match this standard @@ -138,7 +138,7 @@ def _get_result_object(self, id_, params): :returns: job result data as object :raises: HTTPError """ - return self.parse_job_result(self._get_result_json(id_, params)) + return self.parse_job_result(self._get_result_json(id_, params, route)) def delete_job(self, id_): """Delete a specific job diff --git a/src/rev_ai/models/__init__.py b/src/rev_ai/models/__init__.py index 77e8097..6dde062 100644 --- a/src/rev_ai/models/__init__.py +++ b/src/rev_ai/models/__init__.py @@ -10,3 +10,4 @@ from .language_id import LanguageIdentificationJob, LanguageIdentificationResult, LanguageConfidence from .customer_url_data import CustomerUrlData from .revaiapi_deployment_config_constants import RevAiApiDeployment, RevAiApiDeploymentConfigMap +from .forced_alignment import ForcedAlignmentJob, ForcedAlignmentResult, Monologue, ElementAlignment diff --git a/src/rev_ai/models/forced_alignment/__init__.py b/src/rev_ai/models/forced_alignment/__init__.py new file mode 100644 index 0000000..174bedd --- /dev/null +++ b/src/rev_ai/models/forced_alignment/__init__.py @@ -0,0 +1,6 @@ +"""Module containing models for Rev AI forced alignment""" + +from .forced_alignment_job import ForcedAlignmentJob +from .forced_alignment_result import ForcedAlignmentResult, Monologue, ElementAlignment + +__all__ = ['ForcedAlignmentJob', 'ForcedAlignmentResult', 'Monologue', 'ElementAlignment'] \ No newline at end of file diff --git a/src/rev_ai/models/forced_alignment/forced_alignment_job.py b/src/rev_ai/models/forced_alignment/forced_alignment_job.py new file mode 100644 index 0000000..cf2e410 --- /dev/null +++ b/src/rev_ai/models/forced_alignment/forced_alignment_job.py @@ -0,0 +1,67 @@ +# -*- coding: utf-8 -*- +"""Contains ForcedAlignmentJob dataclass""" + +from typing import Dict, Any +from ..asynchronous.job_status import JobStatus + + +class ForcedAlignmentJob: + def __init__( + self, id_, created_on, status, + completed_on=None, + callback_url=None, + metadata=None, + media_url=None, + failure=None, + failure_detail=None, + processed_duration_seconds=None, + delete_after_seconds=None): + """Dataclass containing information about a Rev AI forced alignment job + + :param id: unique identifier for this job + :param status: current job status + :param created_on: date and time at which this job was created + :param completed_on: date and time at which this job was completed + :param metadata: customer-provided metadata + :param type: type of job (always "alignment") + :param media_url: URL of the media to be aligned + :param failure: details about job failure if status is "failed" + """ + self.id = id_ + self.created_on = created_on + self.status = status + self.completed_on = completed_on + self.callback_url = callback_url + self.metadata = metadata + self.media_url = media_url + self.failure = failure + self.failure_detail = failure_detail + self.processed_duration_seconds = processed_duration_seconds + self.delete_after_seconds = delete_after_seconds + + def __eq__(self, other): + """Override default equality operator""" + if isinstance(other, self.__class__): + return self.__dict__ == other.__dict__ + return False + + @classmethod + def from_json(cls, json: Dict[str, Any]) -> 'ForcedAlignmentJob': + """Alternate constructor used for parsing json + + :param json: json dictionary to convert + :returns: ForcedAlignmentJob + """ + return cls( + id_=json['id'], + created_on=json['created_on'], + status=JobStatus.from_string(json['status']), + completed_on=json.get('completed_on'), + callback_url=json.get('callback_url'), + metadata=json.get('metadata'), + media_url=json.get('media_url'), + failure=json.get('failure'), + failure_detail=json.get('failure_detail'), + processed_duration_seconds=json.get('processed_duration_seconds'), + delete_after_seconds=json.get('delete_after_seconds') + ) diff --git a/src/rev_ai/models/forced_alignment/forced_alignment_result.py b/src/rev_ai/models/forced_alignment/forced_alignment_result.py new file mode 100644 index 0000000..1d5daf3 --- /dev/null +++ b/src/rev_ai/models/forced_alignment/forced_alignment_result.py @@ -0,0 +1,78 @@ +# -*- coding: utf-8 -*- +"""Contains ForcedAlignmentResult dataclass""" + +from dataclasses import dataclass +from typing import List, Dict, Any + + +@dataclass +class ElementAlignment: + """Dataclass containing information about an aligned word + + :param value: the word that was aligned + :param ts: start time of the word in seconds + :param end_ts: end time of the word in seconds + :param type: type of element (always "text") + """ + value: str + ts: float + end_ts: float + type: str = "text" + + @staticmethod + def from_json(json: Dict[str, Any]) -> 'ElementAlignment': + """Creates an ElementAlignment from the given json dictionary + + :param json: json dictionary to convert + :returns: ElementAlignment + """ + return ElementAlignment( + value=json.get('value'), + ts=json.get('ts'), + end_ts=json.get('end_ts'), + type=json.get('type', 'text') + ) + + +@dataclass +class Monologue: + """Dataclass containing information about a monologue section + + :param speaker: speaker identifier + :param elements: list of words in this monologue with timing information + """ + speaker: int + elements: List[ElementAlignment] + + @staticmethod + def from_json(json: Dict[str, Any]) -> 'Monologue': + """Creates a Monologue from the given json dictionary + + :param json: json dictionary to convert + :returns: Monologue + """ + return Monologue( + speaker=json.get('speaker', 0), + elements=[ElementAlignment.from_json(element) for element in json.get('elements', [])] + ) + + +@dataclass +class ForcedAlignmentResult: + """Dataclass containing the result of a forced alignment job + + :param monologues: A Monologue object per speaker containing the words + they spoke with timing information + """ + monologues: List[Monologue] + + @staticmethod + def from_json(json: Dict[str, Any]) -> 'ForcedAlignmentResult': + """Creates a ForcedAlignmentResult from the given json dictionary + + :param json: json dictionary to convert + :returns: ForcedAlignmentResult + """ + return ForcedAlignmentResult( + monologues=[Monologue.from_json(monologue) for monologue in json.get('monologues', [])] + ) \ No newline at end of file diff --git a/test.py b/test.py index 80399df..267a35a 100644 --- a/test.py +++ b/test.py @@ -1 +1,2 @@ import src.rev_ai.apiclient as client +from src.rev_ai.forced_alignment_client import ForcedAlignmentClient diff --git a/tests/test_forced_alignment_client.py b/tests/test_forced_alignment_client.py new file mode 100644 index 0000000..825e255 --- /dev/null +++ b/tests/test_forced_alignment_client.py @@ -0,0 +1,257 @@ +# -*- coding: utf-8 -*- +"""Unit tests for Forced Alignment Client""" + +import json +import pytest +from src.rev_ai.forced_alignment_client import ForcedAlignmentClient +from src.rev_ai import __version__ +from src.rev_ai.models.forced_alignment import ForcedAlignmentJob, ForcedAlignmentResult, Monologue, ElementAlignment +from src.rev_ai.models import JobStatus, CustomerUrlData + +try: + from urllib.parse import urljoin +except ImportError: + from urlparse import urljoin + +TOKEN = 'token' +JOB_ID = '1' +METADATA = 'test' +MEDIA_URL = 'https://www.rev.ai/FTC_Sample_1.mp3' +TRANSCRIPT_URL = 'https://www.rev.ai/FTC_Sample_1.txt' +TRANSCRIPT_TEXT = """Hi, my name is Jack Groetzinger and I'm going to be talking about how to recruit a cofounder and a team for your startup. I'm going to first give a quick overview of what we're going to talk about. We're going to start out by talking about the sorts of folks you might be looking for for your startup. Then we'll dive right into tactics and approaches you can use to get inbound interest in your company. We'll talk a little bit about how to get people excited for your startup, and then we'll talk about the screening process, how to figure out if someone is a good match and someone you should make an offer to. We'll then cover how you make an offer and we'll wrap up by covering a few specific issues, one is technical hiring, which can be a little different from more general recruiting and finally we'll talk about specifically what you should do when you're looking for a cofounder. Many of the things we're going to be talking about are applicable to both finding a cofounder and finding employees, but there are a few specific issues that are relevant to when you're looking for a co-founder to be part of your startup, so we'll talk about that I want to give a quick background in myself, I am the cofounder of SeatGeek. We are a search engine for sports and concert tickets based in New York City, this is my third startup and for all three we built a team so I have some experience with this. At SeetGeek we have 17 folks and we have an incredible team. And I just want to talk about some things that we've done that I think have helped us get there as sort of the overall thesis for this, one thing that's worth keeping in mind is that recruiting is a lot of work. Some people think that you can raise money and spend a few weeks building your team and then move on to more interesting things, that's totally not the case, recruiting is something that you're always doing, I as a co-founder probably spend regularly 30% of my time trying to find new people to add to our team. But if you, you it's a lot of work because it's such an important thing to assess the success of a startup, and if you invest that time, then the dividends pay off.""" +CREATED_ON = '2025-01-11T22:42:23.45Z' +LANGUAGE = 'en' + + +class TestForcedAlignmentClient: + def test_constructor_with_success(self): + client = ForcedAlignmentClient(TOKEN) + + headers = client.default_headers + + assert headers.get('User-Agent') == 'RevAi-PythonSDK/{}'.format(__version__) + assert headers.get('Authorization') == 'Bearer {}'.format(TOKEN) + assert client.base_url == 'https://api.rev.ai/alignment/v1/' + + @pytest.mark.parametrize('token', [None, '']) + def test_constructor_with_no_token(self, token): + with pytest.raises(ValueError, match='access_token must be provided'): + ForcedAlignmentClient(token) + + def test_submit_job_url_with_source_config_success(self, mock_session, make_mock_response): + client = ForcedAlignmentClient(TOKEN) + url = urljoin(client.base_url, 'jobs') + source_config = CustomerUrlData(MEDIA_URL) + transcript_text = TRANSCRIPT_TEXT + data = { + 'id': JOB_ID, + 'created_on': CREATED_ON, + 'status': 'in_progress', + 'metadata': METADATA, + 'type': 'alignment', + 'language': LANGUAGE + } + response = make_mock_response(url=url, json_data=data) + mock_session.request.return_value = response + + res = client.submit_job_url( + source_config=source_config, + transcript_text=transcript_text, + metadata=METADATA, + language=LANGUAGE) + + assert res == ForcedAlignmentJob( + id=JOB_ID, + created_on=CREATED_ON, + status=JobStatus.IN_PROGRESS, + metadata=METADATA, + type='alignment') + + mock_session.request.assert_called_once_with( + "POST", + url, + json={ + 'source_config': source_config.to_dict(), + 'transcript_text': transcript_text, + 'metadata': METADATA, + 'language': LANGUAGE + }, + headers=client.default_headers) + + def test_submit_job_url_with_transcript_config_success(self, mock_session, make_mock_response): + client = ForcedAlignmentClient(TOKEN) + url = urljoin(client.base_url, 'jobs') + source_config = CustomerUrlData(MEDIA_URL) + source_transcript_config = CustomerUrlData(TRANSCRIPT_URL) + data = { + 'id': JOB_ID, + 'created_on': CREATED_ON, + 'status': 'in_progress', + 'metadata': METADATA, + 'type': 'alignment' + } + response = make_mock_response(url=url, json_data=data) + mock_session.request.return_value = response + + res = client.submit_job_url( + source_config=source_config, + source_transcript_config=source_transcript_config, + metadata=METADATA) + + assert res == ForcedAlignmentJob( + id=JOB_ID, + created_on=CREATED_ON, + status=JobStatus.IN_PROGRESS, + metadata=METADATA, + type='alignment') + + mock_session.request.assert_called_once_with( + "POST", + url, + json={ + 'source_config': source_config.to_dict(), + 'source_transcript_config': source_transcript_config.to_dict(), + 'metadata': METADATA + }, + headers=client.default_headers) + + def test_submit_job_url_missing_source_config(self): + client = ForcedAlignmentClient(TOKEN) + with pytest.raises(ValueError, match='source_config must be provided'): + client.submit_job_url(transcript_text=TRANSCRIPT_TEXT) + + def test_submit_job_url_missing_transcript(self): + client = ForcedAlignmentClient(TOKEN) + source_config = CustomerUrlData(MEDIA_URL) + with pytest.raises(ValueError, match='Either source_transcript_config or transcript_text must be provided'): + client.submit_job_url(source_config=source_config) + + def test_submit_job_url_both_transcript_options(self): + client = ForcedAlignmentClient(TOKEN) + source_config = CustomerUrlData(MEDIA_URL) + source_transcript_config = CustomerUrlData(TRANSCRIPT_URL) + with pytest.raises(ValueError, match='Only one of source_transcript_config or transcript_text may be provided'): + client.submit_job_url( + source_config=source_config, + source_transcript_config=source_transcript_config, + transcript_text=TRANSCRIPT_TEXT) + + def test_get_result_json_with_success(self, mock_session, make_mock_response): + client = ForcedAlignmentClient(TOKEN) + url = urljoin(client.base_url, 'jobs/{}/transcript'.format(JOB_ID)) + data = { + 'monologues': [ + { + 'speaker': 0, + 'elements': [ + { + 'type': 'text', + 'value': 'Hi', + 'ts': 0.25, + 'end_ts': 0.5 + }, + { + 'type': 'text', + 'value': 'my', + 'ts': 0.6, + 'end_ts': 0.75 + }, + { + 'type': 'text', + 'value': 'name', + 'ts': 0.8, + 'end_ts': 1.1 + }, + { + 'type': 'text', + 'value': 'is', + 'ts': 1.2, + 'end_ts': 1.3 + }, + { + 'type': 'text', + 'value': 'Jack', + 'ts': 1.4, + 'end_ts': 1.7 + } + ] + } + ] + } + response = make_mock_response(url=url, json_data=data) + mock_session.request.return_value = response + + res = client.get_result_json(JOB_ID) + + assert res == data + mock_session.request.assert_called_once_with( + "GET", + url, + headers=client.default_headers) + + def test_get_result_object_with_success(self, mock_session, make_mock_response): + client = ForcedAlignmentClient(TOKEN) + url = urljoin(client.base_url, 'jobs/{}/transcript'.format(JOB_ID)) + data = { + 'monologues': [ + { + 'speaker': 0, + 'elements': [ + { + 'type': 'text', + 'value': 'Hi', + 'ts': 0.25, + 'end_ts': 0.5 + }, + { + 'type': 'text', + 'value': 'my', + 'ts': 0.6, + 'end_ts': 0.75 + }, + { + 'type': 'text', + 'value': 'name', + 'ts': 0.8, + 'end_ts': 1.1 + }, + { + 'type': 'text', + 'value': 'is', + 'ts': 1.2, + 'end_ts': 1.3 + }, + { + 'type': 'text', + 'value': 'Jack', + 'ts': 1.4, + 'end_ts': 1.7 + } + ] + } + ] + } + response = make_mock_response(url=url, json_data=data) + mock_session.request.return_value = response + + res = client.get_result_object(JOB_ID) + + assert res == ForcedAlignmentResult( + monologues=[ + Monologue( + speaker=0, + elements=[ + ElementAlignment(value='Hi', ts=0.25, end_ts=0.5, type='text'), + ElementAlignment(value='my', ts=0.6, end_ts=0.75, type='text'), + ElementAlignment(value='name', ts=0.8, end_ts=1.1, type='text'), + ElementAlignment(value='is', ts=1.2, end_ts=1.3, type='text'), + ElementAlignment(value='Jack', ts=1.4, end_ts=1.7, type='text') + ] + ) + ] + ) + + mock_session.request.assert_called_once_with( + "GET", + url, + headers=client.default_headers) \ No newline at end of file